feat: add decryption capability in userdata with Jinja template

cloudinit/templater.py

@@ -12,9 +12,12 @@
 
 # noqa: E402
 
+import base64
 import collections
 import logging
 import re
+import shutil
+import subprocess
 import sys
 from typing import Any
 
@@ -143,6 +146,45 @@ def replacer(match):
     )
 
 
+def gpgdecrypt(cipher, default=None):
+    LOG.debug("Decrypting: %s", cipher)
+
+    if not shutil.which("gpg"):
+        if default is not None:
+            return default
+        return "cloudinit: GPG executable cannot be found"
+
+    p = subprocess.Popen(
+        ["gpg", "--decrypt"],
+        stdin=subprocess.PIPE,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+    )
+    p.stdin.write(base64.b64decode(cipher))
+    p.stdin.close()
+    try:
+        return_code = p.wait(10)
+        if return_code == 0:
+            return p.stdout.read().decode()
+        LOG.warning(
+            "Decryption failed: gpg returned status code %d. Stderr: %s",
+            return_code,
+            p.stderr.read().decode(),
+        )
+
+    except subprocess.TimeoutExpired:
+        LOG.warning(
+            "Decryption failed: gpg didn't terminate in time. Aborting",
+            p.stderr.read().decode(),
+        )
+        p.terminate()
+
+    if default is not None:
+        return default
+
+    return "cloudinit: unable to decrypt GPG message"
+
+
 def detect_template(text):
     def jinja_render(content, params):
         # keep_trailing_newline is in jinja2 2.7+, not 2.6
@@ -155,7 +197,7 @@ def jinja_render(content, params):
                         undefined=UndefinedJinjaVariable,
                         trim_blocks=True,
                         extensions=["jinja2.ext.do"],
-                    ).render(**params)
+                    ).render(gpgdecrypt=gpgdecrypt, **params)
                     + add
                 )
         except TemplateSyntaxError as template_syntax_error:

doc/userdata.txt

@@ -62,6 +62,46 @@ finds.  However, certain types of user-data are handled specially.
    mechanism provided for running only once.  The boothook must take
    care of this itself.
 
+=== Sentive values ===
+Sharing secret in plain text in userdata is generally considered unsafe 
+(example in AWS documentation[1]). This is because user-data can usally 
+be read back, often with a simple "read" permission, allowing a user with 
+limited access to discover potentially sensitive secrets on existing 
+deploying VM.
+
+A common pattern is to leverage managed secret services from cloud 
+providers, along with an instance profile. However, this often implies 
+multiple elements, such as cloud specific logic to provision and manage 
+the secrets, as well as fetching them inside the VM instance. In may also 
+induce costs and create a potential attack vector as the VM will be able 
+to operate on the cloud provider API, and for example, perform denial of 
+service attach by consuming operqtional quota.
+
+When using the jinja templating, it is possible to pass encrypted 
+sensitive value with GPG and decrypt them at runtime, using the 
+`gpgdecrypt`. This requires to install to install the GPG key with 
+"encrypt" usage prior, for example during the baking processing of the 
+VM image.
+
+Note that this requires to have the GPG binary available in the PATH that 
+cloudinit will use, as a ``gpg`` command.
+
+"gpgdecrypt" function takes one required argument and one optional. The 
+first argument is the base64-encoded encrypted data to decrypt, and the 
+second argument is the default value, to return in case the decryption 
+process fail.
+
+ [1] https://docs.aws.amazon.com/codeguru/detector-library/cloudformation/exposed-ec2-user-data-secret-cloudformation/
+
+Here is a complete example:
+
+```
+## template: jinja
+runcmd:
+- mkdir -p /var/run/secrets/example.com
+- echo "{{ gpgdecrypt("bG9yZW0gaXBzdW0K") }}" > /var/run/secrets/example.com/mykey.pem
+```
+
 === Examples ===
 There are examples in the examples subdirectory.
 Additionally, the 'tools' directory contains 'write-mime-multipart',

test-requirements.txt

@@ -19,3 +19,6 @@ passlib
 # This one is currently used only by the CloudSigma and SmartOS datasources.
 # If these datasources are removed, this is no longer needed.
 pyserial
+
+# For encrypting sensitive values in userdata
+python-gnupg

tests/unittests/test_templating.py

@@ -4,10 +4,14 @@
 #
 # This file is part of cloud-init. See LICENSE file for license information.
 
+import base64
 import logging
+import os
+import tempfile
 import textwrap
 from unittest import mock
 
+import gnupg
 import pytest
 
 from cloudinit import templater
@@ -16,6 +20,15 @@
 from tests.unittests import helpers as test_helpers
 
 
+@pytest.fixture()
+def ready_gpg_keyring():
+    with tempfile.TemporaryDirectory() as temp_dir:
+        os.environ["GNUPGHOME"] = temp_dir
+        gpg = gnupg.GPG(gnupghome=temp_dir)
+        yield gpg
+        del os.environ["GNUPGHOME"]
+
+
 class TestTemplates:
 
     jinja_utf8 = b"It\xe2\x80\x99s not ascii, {{name}}\n"
@@ -171,6 +184,126 @@ def test_jinja_do_extension_render_to_string(self):
             == expected_result
         )
 
+    def test_gpgdecrypt_with_valid_key(self, ready_gpg_keyring):
+        # Generate a new key
+        input_data = ready_gpg_keyring.gen_key_input(
+            name_email="cloudinit.test@local",
+            key_type="RSA",
+            key_length=2048,
+            key_usage="encrypt",
+            no_protection=True,
+        )
+        key = ready_gpg_keyring.gen_key(input_data)
+
+        encrypted_data = ready_gpg_keyring.encrypt(
+            b"Some sensitive data",
+            recipients=[key.fingerprint],
+            always_trust=True,
+        )
+
+        expected_result = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "Some sensitive data" > /var/secrets/value'
+        )
+        jinja_template = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "{{ gpgdecrypt("%s") }}" > /var/secrets/value\n'
+        ) % base64.b64encode(encrypted_data.data).decode()
+        assert (
+            templater.render_string(
+                self.add_header("jinja", jinja_template), {}
+            ).strip()
+            == expected_result
+        )
+
+    def test_gpgdecrypt_with_missing_key(self, ready_gpg_keyring):
+        # Generate a new key
+        input_data = ready_gpg_keyring.gen_key_input(
+            name_email="cloudinit.test@local",
+            key_type="RSA",
+            key_length=2048,
+            key_usage="encrypt",
+            no_protection=True,
+        )
+        key = ready_gpg_keyring.gen_key(input_data)
+
+        encrypted_data = ready_gpg_keyring.encrypt(
+            b"Some sensitive data",
+            recipients=[key.fingerprint],
+            always_trust=True,
+        )
+        assert (
+            str(
+                ready_gpg_keyring.delete_keys(
+                    key.fingerprint, secret=True, passphrase="unused"
+                )
+            )
+            == "ok"
+        )
+
+        expected_result = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "cloudinit: unable to decrypt GPG message"'
+            " > /var/secrets/value"
+        )
+        jinja_template = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "{{ gpgdecrypt("%s") }}" > /var/secrets/value\n'
+        ) % base64.b64encode(encrypted_data.data).decode()
+        assert (
+            templater.render_string(
+                self.add_header("jinja", jinja_template), {}
+            ).strip()
+            == expected_result
+        )
+
+    def test_gpgdecrypt_with_missing_key_and_default(self, ready_gpg_keyring):
+        # Generate a new key
+        input_data = ready_gpg_keyring.gen_key_input(
+            name_email="cloudinit.test@local",
+            key_type="RSA",
+            key_length=2048,
+            key_usage="encrypt",
+            no_protection=True,
+        )
+        key = ready_gpg_keyring.gen_key(input_data)
+
+        encrypted_data = ready_gpg_keyring.encrypt(
+            b"Some sensitive data",
+            recipients=[key.fingerprint],
+            always_trust=True,
+        )
+        assert (
+            str(
+                ready_gpg_keyring.delete_keys(
+                    key.fingerprint, secret=True, passphrase="unused"
+                )
+            )
+            == "ok"
+        )
+
+        expected_result = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "adefaultvalue" > /var/secrets/value'
+        )
+        jinja_template = (
+            "## template: jinja\n"
+            "runcmd:\n"
+            ' - echo "{{ gpgdecrypt("%s", "adefaultvalue") }}" > '
+            "/var/secrets/value\n"
+        ) % base64.b64encode(encrypted_data.data).decode()
+        assert (
+            templater.render_string(
+                self.add_header("jinja", jinja_template), {}
+            ).strip()
+            == expected_result
+        )
+
 
 class TestJinjaSyntaxParsingException:
     def test_jinja_syntax_parsing_exception_message(self):

tox.ini

@@ -187,6 +187,7 @@ deps =
     pytest-mock==3.6.1
     responses==0.18.0
     passlib==1.7.4
+    python-gnupg==0.5.6
 commands = {envpython} -m pytest -m "not hypothesis_slow" --cov=cloud-init --cov-branch {posargs:tests/unittests}
 
 [testenv:doc]