New features: inline secrets, custom Jinja filters! (#10)

* fixed *.auto.tfvars variable autodeclaration

* updated copyright year

* added custom jinja2 filters support

* simplified directory_remove

* added inline encryption support

Author: ribejara-te

src/filters.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def deepformat(value, params):
+    if isinstance(value, dict):
+        return {
+            deepformat(key, params): deepformat(value, params)
+            for key, value in value.items()
+        }
+    if isinstance(value, list):
+        return [
+            deepformat(item, params)
+            for item in value
+        ]
+    if isinstance(value, str):
+        return value.format(**params)
+    return value
+
+
+__all__ = [
+    deepformat,
+]

src/helpers.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# Copyright 2023 Cisco Systems, Inc.
+# Copyright 2024 Cisco Systems, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,6 +17,7 @@
 import fnmatch
 import glob
 import json
+import os
 import pathlib
 import shutil
 import tempfile
@@ -26,6 +27,9 @@
 import jinja2
 import yaml
 
+import filters
+from tools import encryption_decrypt
+
 
 def directory_copy(srcpath, dstpath, ignore=[]):
     """Copy the contents of the dir in 'srcpath' to the dir in 'dstpath'.
@@ -62,19 +66,12 @@ def directory_remove(path, keep=[]):
     if not path.is_dir():
         return
 
-    temp = pathlib.Path(tempfile.TemporaryDirectory(dir=pathlib.Path().cwd()).name)
-    for item in keep:
-        itempath = path.joinpath(item)
-        if itempath.exists():
-            shutil.move(itempath, temp.joinpath(item))
-
-    shutil.rmtree(path)
-
-    for item in keep:
-        itempath = temp.joinpath(item)
-        if itempath.exists():
-            shutil.move(itempath, path.joinpath(item))
-    directory_remove(temp)
+    for item in path.iterdir():
+        if item.name not in keep:
+            if item.is_dir():
+                shutil.rmtree(item)
+            else:
+                item.unlink()
 
 
 def json_read(patterns):
@@ -151,6 +148,31 @@ def hcl2_read(patterns):
                 continue
             with open(path, "r") as f:
                 data = deepmerge.always_merger.merge(data, hcl2.load(f))
+    return hcl2_decrypt(data)
+
+
+def hcl2_decrypt(data):
+    """Decrypts all strings in 'data'.
+
+    Keyword arguments:
+      data[any]: any HCL2-sourced data structure
+    """
+    if isinstance(data, str) and data.startswith("ENC[") and data.endswith("]"):
+        key_path = os.getenv("STACKS_PRIVATE_KEY_PATH")
+        if not key_path:
+            raise Exception("could not decrypt data: STACKS_PRIVATE_KEY_PATH is not set")
+        if not pathlib.Path(key_path).exists():
+            raise Exception(f"could not decrypt data: STACKS_PRIVATE_KEY_PATH ({key_path}) does not exist")
+        return encryption_decrypt.main(data, key_path)
+
+    elif isinstance(data, list):
+        for i in range(len(data)):
+            data[i] = hcl2_decrypt(data[i])
+
+    elif isinstance(data, dict):
+        for k, v in data.items():
+            data[k] = hcl2_decrypt(v)
+
     return data
 
 
@@ -167,8 +189,20 @@ def jinja2_render(patterns, data):
             path = pathlib.Path(path)
             if not path.is_file():
                 continue
-            with open(path, "r") as fin:
-                template = jinja2.Template(fin.read())
-            rendered = template.render(data)
-            with open(path, "w") as fout:
-                fout.write(rendered)
+            try:
+                with open(path, "r") as fin:
+                    template = jinja2.Template(fin.read())
+
+                rendered = template.render(data | {
+                    func.__name__: func
+                    for func in filters.__all__
+                })
+
+                with open(path, "w") as fout:
+                    fout.write(rendered)
+            except jinja2.exceptions.UndefinedError as e:
+                print(f"Failure to render {path}: {e}", file=sys.stderr)
+                sys.exit(1)
+            except jinja2.exceptions.TemplateSyntaxError as e:
+                print(f"Failure to render {path} at line {e.lineno}, in statement {e.source}: {e}", file=sys.stderr)
+                sys.exit(1)

src/postinit.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# Copyright 2023 Cisco Systems, Inc.
+# Copyright 2024 Cisco Systems, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -17,10 +17,10 @@
 import glob
 import pathlib
 
-import helpers
-
 import git
 
+import helpers
+
 
 # define context
 cwd         = pathlib.Path().cwd()                              # current working directory

src/preinit.py

@@ -1,6 +1,6 @@
 #!/usr/bin/env python3
 
-# Copyright 2023 Cisco Systems, Inc.
+# Copyright 2024 Cisco Systems, Inc.
 #
 # Licensed under the Apache License, Version 2.0 (the "License");
 # you may not use this file except in compliance with the License.
@@ -87,7 +87,7 @@
     list(variable.keys())[0]
     for variable in helpers.hcl2_read([workdir.joinpath("*.tf")]).get("variable", [])
 ]
-for variable in {**variables, **helpers.hcl2_read([workdir.joinpath("*.tfvars")])}.keys():  # also autodeclare variables in *.auto.tfvars files
+for variable in {**variables, **helpers.hcl2_read([workdir.joinpath("*.auto.tfvars")])}.keys():  # also autodeclare variables in *.auto.tfvars files
     if variable not in variables_declared:
         universe["variable"][variable] = {}
 

src/requirements.txt

@@ -1,5 +1,6 @@
-deepmerge==1.0.1
-GitPython==3.1.30
-Jinja2==3.1.2
-python-hcl2==3.0.5
-PyYAML==6.0
+cryptography==43.0.1
+deepmerge==2.0
+GitPython==3.1.43
+Jinja2==3.1.4
+python-hcl2==4.3.5
+PyYAML==6.0.2

src/tools/cli_wrapper.py

@@ -0,0 +1,29 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def main(func):
+    import argparse, inspect, json
+    parser = argparse.ArgumentParser()
+    for key, value in inspect.signature(func).parameters.items():
+        parser.add_argument(
+            f"--{key.replace('_','-')}",
+            action = argparse.BooleanOptionalAction if isinstance(value.default, bool) else None,
+            default = value.default,
+            required = value.default == value.empty,
+        )
+    output = func(**vars(parser.parse_args()))
+    if output is not None:
+        print(json.dumps(output, indent=2))

src/tools/encryption_decrypt.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def main(string, private_key_path):
+    import base64
+    import cryptography.hazmat.backends
+    import cryptography.hazmat.primitives.asymmetric.padding
+    import cryptography.hazmat.primitives.hashes
+    import cryptography.hazmat.primitives.padding
+    import cryptography.hazmat.primitives.serialization
+
+    symmetric_key_encrypted_base64, encryptor_tag_base64, init_vector_base64, string_encrypted_base64 = string.removeprefix("ENC[").removesuffix("]").split(";")
+
+    with open(private_key_path, "rb") as key_file:
+        symmetric_key = cryptography.hazmat.primitives.serialization.load_pem_private_key(
+            key_file.read(),
+            password = None,
+            backend = cryptography.hazmat.backends.default_backend(),
+        ).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,
+            )
+        )
+
+    decryptor = cryptography.hazmat.primitives.ciphers.Cipher(
+        cryptography.hazmat.primitives.ciphers.algorithms.AES(symmetric_key),
+        cryptography.hazmat.primitives.ciphers.modes.GCM(
+            base64.b64decode(init_vector_base64.encode()),
+            base64.b64decode(encryptor_tag_base64.encode()),
+        ),
+        backend = cryptography.hazmat.backends.default_backend(),
+    ).decryptor()
+    padded = decryptor.update(base64.b64decode(string_encrypted_base64.encode())) + decryptor.finalize()
+
+    unpadder = cryptography.hazmat.primitives.padding.PKCS7(128).unpadder()
+    unpadded = unpadder.update(padded) + unpadder.finalize()
+
+    return unpadded.decode("utf-8")
+
+
+if __name__ == "__main__":
+    import cli_wrapper
+    cli_wrapper.main(main)

src/tools/encryption_encrypt.py

@@ -0,0 +1,62 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def main(string, public_key_path):
+    import base64
+    import cryptography.hazmat.backends
+    import cryptography.hazmat.primitives.asymmetric.padding
+    import cryptography.hazmat.primitives.ciphers
+    import cryptography.hazmat.primitives.hashes
+    import cryptography.hazmat.primitives.padding
+    import cryptography.hazmat.primitives.serialization
+    import os
+
+    padder = cryptography.hazmat.primitives.padding.PKCS7(128).padder()
+    padded = padder.update(string.encode()) + padder.finalize()
+
+    symmetric_key = os.urandom(32)
+
+    init_vector = os.urandom(12)
+    init_vector_base64 = base64.b64encode(init_vector).decode("utf-8")
+
+    encryptor = cryptography.hazmat.primitives.ciphers.Cipher(cryptography.hazmat.primitives.ciphers.algorithms.AES(symmetric_key), cryptography.hazmat.primitives.ciphers.modes.GCM(init_vector), backend=cryptography.hazmat.backends.default_backend()).encryptor()
+
+    string_encrypted = encryptor.update(padded) + encryptor.finalize()
+    string_encrypted_base64 = base64.b64encode(string_encrypted).decode("utf-8")
+
+    encryptor_tag_base64 = base64.b64encode(encryptor.tag).decode("utf-8")
+
+    with open(public_key_path, "rb") as f:
+        public_key = cryptography.hazmat.primitives.serialization.load_pem_public_key(
+            f.read(),
+            backend = cryptography.hazmat.backends.default_backend(),
+        )
+
+    symmetric_key_encrypted_base64 = base64.b64encode(public_key.encrypt(
+        symmetric_key,
+        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,
+        )
+    )).decode("utf-8")
+
+    return f"ENC[{symmetric_key_encrypted_base64};{encryptor_tag_base64};{init_vector_base64};{string_encrypted_base64}]"
+
+
+if __name__ == "__main__":
+    import cli_wrapper
+    cli_wrapper.main(main)

src/tools/encryption_generate_key.py

@@ -0,0 +1,42 @@
+#!/usr/bin/env python3
+
+# Copyright 2024 Cisco Systems, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+def main(public_key_path, private_key_path):
+    import cryptography.hazmat.backends
+    import cryptography.hazmat.primitives.serialization
+    import cryptography.hazmat.primitives.asymmetric.rsa
+
+    key = cryptography.hazmat.primitives.asymmetric.rsa.generate_private_key(
+        backend = cryptography.hazmat.backends.default_backend(),
+        key_size = 2**11,
+        public_exponent = 2**16+1,
+    )
+    with open(private_key_path, "wb") as f:
+        f.write(key.private_bytes(
+            encoding = cryptography.hazmat.primitives.serialization.Encoding.PEM,
+            format = cryptography.hazmat.primitives.serialization.PrivateFormat.PKCS8,
+            encryption_algorithm = cryptography.hazmat.primitives.serialization.NoEncryption(),
+        ))
+    with open(public_key_path, "wb") as f:
+        f.write(key.public_key().public_bytes(
+            encoding = cryptography.hazmat.primitives.serialization.Encoding.PEM,
+            format = cryptography.hazmat.primitives.serialization.PublicFormat.SubjectPublicKeyInfo,
+        ))
+
+
+if __name__ == "__main__":
+    import cli_wrapper
+    cli_wrapper.main(main)