added should_decrypt to allow partitioned environments

Author: ribejara-te

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "stacks"
-version = "2.0.11"
+version = "2.0.12"
 description = "Stacks, the Terraform code pre-processor"
 readme = "README.md"
 requires-python = ">=3.10"

src/stacks/cmd/preinit.py

@@ -15,7 +15,7 @@ def preinit(ctx):
         "stacks_subenvironment": ctx.subenv or "",
         "stacks_instance": ctx.instance or "",
         "stacks_environments": {
-            item.name: helpers.hcl2_read([item.joinpath("env.tfvars")])  # TODO: replace 'env.tfvars' with '*.tfvars' after all stacks have been upgraded to v2
+            item.name: helpers.hcl2_read([item.joinpath("env.tfvars")], must_decrypt=False)  # TODO: replace 'env.tfvars' with '*.tfvars' after all stacks have been upgraded to v2
             for item in ctx.envs_dir.iterdir()
             if item.is_dir() and item.joinpath("env.tfvars").exists()
         },

src/stacks/helpers/config.py

@@ -9,12 +9,13 @@
 from .merge import merge
 
 
-def config_read(patterns, should_decrypt, decoderfunc, **decoderargs):
+def config_read(patterns, should_decrypt, must_decrypt, decoderfunc, **decoderargs):
     """Read configuration files in 'patterns' using 'decoderfunc' and return their merged contents.
 
     Keyword arguments:
       patterns[list]: patterns to configuration files, in ascending order of priority
-      should_decrypt[bool]: whether to decrypt data or not
+      should_decrypt[bool]: whether it should try to decrypt data or not
+      must_decrypt[bool]: whether decryption should succeed (if False and fails, returns encrypted value)
       decoderfunc[function]: function that parses a given configuration file into a data structure
       decoderargs[dict]: keyword arguments to pass to decoderfunc
     """
@@ -26,19 +27,19 @@ def config_read(patterns, should_decrypt, decoderfunc, **decoderargs):
             if path.is_file():
                 with open(path, "r") as f:
                     data = merge(data, decoderfunc(f, **decoderargs))
-    return decrypt(data) if should_decrypt else data
+    return decrypt(data, must_decrypt=must_decrypt) if should_decrypt else data
 
 
-def json_read(patterns, should_decrypt=True):
-    return config_read(patterns, should_decrypt, json.load)
+def json_read(patterns, should_decrypt=True, must_decrypt=True):
+    return config_read(patterns, should_decrypt, must_decrypt, json.load)
 
 
-def yaml_read(patterns, should_decrypt=True):
-    return config_read(patterns, should_decrypt, yaml.safe_load)
+def yaml_read(patterns, should_decrypt=True, must_decrypt=True):
+    return config_read(patterns, should_decrypt, must_decrypt, yaml.safe_load)
 
 
-def hcl2_read(patterns, should_decrypt=True):
-    return config_read(patterns, should_decrypt, hcl2.load)
+def hcl2_read(patterns, should_decrypt=True, must_decrypt=True):
+    return config_read(patterns, should_decrypt, must_decrypt, hcl2.load)
 
 
 def config_write(data, path, encoderfunc, **encoderargs):

src/stacks/helpers/crypto.py

@@ -84,12 +84,13 @@ def encrypt(public_key_path, string):
     return f"ENC[{symmetric_key_encrypted_base64};{encryptor_tag_base64};{init_vector_base64};{string_encrypted_base64}]"
 
 
-def decrypt(data, private_key_path=os.getenv("STACKS_PRIVATE_KEY_PATH")):
+def decrypt(data, private_key_path=os.getenv("STACKS_PRIVATE_KEY_PATH"), must_decrypt=True):
     """Decrypt 'data' using 'private_key_path'.
 
     Keyword arguments:
-      private_key_path[pathlib.Path]: path to private key
       data[any]: any data structure
+      private_key_path[pathlib.Path]: path to private key
+      must_decrypt[bool]: whether decryption should succeed (if False and fails, returns encrypted value)
     """
     if isinstance(data, str) and data.startswith("ENC[") and data.endswith("]"):
         (
@@ -105,14 +106,21 @@ def decrypt(data, private_key_path=os.getenv("STACKS_PRIVATE_KEY_PATH")):
                 password=None,
                 backend=cryptography.hazmat.backends.default_backend(),
             )
-        symmetric_key = private_key.decrypt(
-            base64.b64decode(symmetric_key_encrypted_base64.encode()),
-            cryptography.hazmat.primitives.asymmetric.padding.OAEP(
-                mgf=cryptography.hazmat.primitives.asymmetric.padding.MGF1(algorithm=cryptography.hazmat.primitives.hashes.SHA256()),
-                algorithm=cryptography.hazmat.primitives.hashes.SHA256(),
-                label=None,
-            ),
-        )
+
+        try:
+            symmetric_key = private_key.decrypt(
+                base64.b64decode(symmetric_key_encrypted_base64.encode()),
+                cryptography.hazmat.primitives.asymmetric.padding.OAEP(
+                    mgf=cryptography.hazmat.primitives.asymmetric.padding.MGF1(algorithm=cryptography.hazmat.primitives.hashes.SHA256()),
+                    algorithm=cryptography.hazmat.primitives.hashes.SHA256(),
+                    label=None,
+                ),
+            )
+        except ValueError as e:
+            if must_decrypt:
+                raise e
+            else:
+                return data
 
         init_vector = base64.b64decode(init_vector_base64.encode())
 
@@ -134,9 +142,9 @@ def decrypt(data, private_key_path=os.getenv("STACKS_PRIVATE_KEY_PATH")):
         return string_decrypted.decode("utf-8")
 
     elif isinstance(data, list):
-        return [decrypt(private_key_path=private_key_path, data=item) for item in data]
+        return [decrypt(private_key_path=private_key_path, data=item, must_decrypt=must_decrypt) for item in data]
 
     elif isinstance(data, dict):
-        return {key: decrypt(private_key_path=private_key_path, data=value) for key, value in data.items()}
+        return {key: decrypt(private_key_path=private_key_path, data=value, must_decrypt=must_decrypt) for key, value in data.items()}
 
     return data