comments

cloudinit/gpg.py

@@ -23,6 +23,10 @@
 HOME = "GNUPGHOME"
 
 
+class GpgVerificationError(Exception):
+    """GpgVerificationError is raised when a signature verification fails."""
+
+
 class GPG:
     def __init__(self):
         self.gpg_started = False
@@ -87,7 +91,7 @@ def import_key(self, key: pathlib.Path) -> None:
         except subp.ProcessExecutionError as error:
             LOG.warning("Failed to import key %s: %s", key, error)
 
-    def decrypt(self, data: str) -> str:
+    def decrypt(self, data: str, *, require_signature=False) -> str:
         """Process data using gpg.
 
         This can be used to decrypt encrypted data, verify signed data,
@@ -97,14 +101,19 @@ def decrypt(self, data: str) -> str:
         :return: decrypted data
         :raises: ProcessExecutionError if gpg fails to decrypt data
         """
-        return subp.subp(
+        result = subp.subp(
             [
                 "gpg",
                 "--decrypt",
             ],
             data=data,
             update_env=self.env,
-        ).stdout
+        )
+        if require_signature and "gpg: Good signature" not in result.stderr:
+            raise GpgVerificationError(
+                "Signature verification required, but no signature found"
+            )
+        return result.stdout
 
     def dearmor(self, key: str) -> str:
         """Dearmor gpg key, dearmored key gets returned

cloudinit/sources/__init__.py

@@ -637,34 +637,34 @@ def get_url_params(self):
         return URLParams(max_wait, timeout, retries, sec_between_retries)
 
     def get_userdata(self, apply_filter=False):
-        require_pgp = self.sys_cfg.get("user_data", {}).get(
-            "require_pgp", False
+        require_signature = self.sys_cfg.get("user_data", {}).get(
+            "require_signature", False
         )
         if self.userdata is None:
             self.userdata = self.ud_proc.process(
-                self.get_userdata_raw(), require_pgp
+                self.get_userdata_raw(), require_signature
             )
         if apply_filter:
             return self._filter_xdata(self.userdata)
         return self.userdata
 
     def get_vendordata(self):
-        require_pgp = self.sys_cfg.get("vendor_data", {}).get(
-            "require_pgp", False
+        require_signature = self.sys_cfg.get("vendor_data", {}).get(
+            "require_signature", False
         )
         if self.vendordata is None:
             self.vendordata = self.ud_proc.process(
-                self.get_vendordata_raw(), require_pgp
+                self.get_vendordata_raw(), require_signature
             )
         return self.vendordata
 
     def get_vendordata2(self):
-        require_pgp = self.sys_cfg.get("vendor_data2", {}).get(
-            "require_pgp", False
+        require_signature = self.sys_cfg.get("vendor_data2", {}).get(
+            "require_signature", False
         )
         if self.vendordata2 is None:
             self.vendordata2 = self.ud_proc.process(
-                self.get_vendordata2_raw(), require_pgp
+                self.get_vendordata2_raw(), require_signature
             )
         return self.vendordata2
 

cloudinit/stages.py

@@ -793,7 +793,7 @@ def _consume_userdata_if_enabled(self, frequency: str) -> None:
         Base config can have a definition like:
           user_data:
             enabled: false
-            require_pgp: true
+            require_signature: true
         or a deprecated `allow_userdata` key.
 
         Parse them and maybe consume userdata accordingly.

cloudinit/user_data.py

@@ -85,20 +85,22 @@ def __init__(self, paths):
         self.paths = paths
         self.ssl_details = util.fetch_ssl_details(paths)
 
-    def process(self, blob, require_pgp=False):
+    def process(self, blob, require_signature=False):
         accumulating_msg = MIMEMultipart()
         if isinstance(blob, list):
             for b in blob:
                 self._process_msg(
-                    convert_string(b), accumulating_msg, require_pgp
+                    convert_string(b), accumulating_msg, require_signature
                 )
         else:
             self._process_msg(
-                convert_string(blob), accumulating_msg, require_pgp
+                convert_string(blob), accumulating_msg, require_signature
             )
         return accumulating_msg
 
-    def _process_msg(self, base_msg: Message, append_msg, require_pgp=False):
+    def _process_msg(
+        self, base_msg: Message, append_msg, require_signature=False
+    ):
         def find_ctype(payload):
             return handlers.type_from_starts_with(payload)
 
@@ -118,9 +120,9 @@ def find_ctype(payload):
             if ctype in TYPE_NEEDED + ["text/x-shellscript"]:
                 ctype = find_ctype(payload) or ctype
 
-            if require_pgp and ctype != ENCRYPT_TYPE:
+            if require_signature and ctype != ENCRYPT_TYPE:
                 error_message = (
-                    "'require_pgp' was set true in cloud-init's base "
+                    "'require_signature' was set true in cloud-init's base "
                     f"configuration, but content type is {ctype}."
                 )
                 raise RuntimeError(error_message)
@@ -152,7 +154,9 @@ def find_ctype(payload):
                             for key_path in keys_dir.iterdir():
                                 gpg_context.import_key(key_path)
                         try:
-                            payload = gpg_context.decrypt(payload)
+                            payload = gpg_context.decrypt(
+                                payload, require_signature=require_signature
+                            )
                         except subp.ProcessExecutionError as e:
                             raise RuntimeError(
                                 "Failed decrypting user data payload of type "

doc/rtd/explanation/format.rst

@@ -190,7 +190,7 @@ message only. To do so, the following
 .. code-block:: yaml
 
     user_data:
-      require_pgp: true
+      require_signature: true
 
 
 For a detailed guide to creating the PGP message and image necessary for this

doc/rtd/howto/pgp.rst

@@ -93,8 +93,8 @@ cloud-config file:
 
 Save this file to your working directory as `cloud-config.yaml`.
 
-Encrypt the user data using the public key of the encrypting user and the
-private key of the signing user:
+Encrypt the user data using the public key of the encrypting user and
+sign it using the private key of the signing user:
 
 .. code-block:: bash
 
@@ -134,8 +134,7 @@ While it is more steps to export the keys in this way as opposed to
 using the existing key ring in the snapshot, we do this for a few reasons:
 
 * Users may not want these keys in any key ring by default on a new instance
-* Keys may be exported from any system without needing to be concerned with
-  the implementation details of the gpg implementation
+* Exporting keys is easier than copying key rings
 
 Note that on launch, cloud-init will import there keys into a temporary
 key ring that is removed after the user data is processed. The default
@@ -151,7 +150,7 @@ require that cloud-init only process PGP messages. To do so, create a file
 .. code-block:: text
 
     user_data:
-      require_pgp: true
+      require_signature: true
 
 Retrieve our encrypted and signed user data
 ===========================================

doc/rtd/reference/base_config_reference.rst

@@ -267,8 +267,9 @@ Format is a dict with the following keys:
 * ``enabled``: A boolean value to enable or disable the use of user data.
   One use case is to prevent users from accidentally breaking closed
   appliances. Default: ``true``.
-* ``require_pgp``: A boolean indicating whether to require PGP verification
-  of the user data. Default: ``false``. This should be true if
+* ``require_signature``: A boolean indicating whether to require a PGP
+  signed message for user data. Default: ``false``.
+  This should be true if
   using :ref:`signed user data<user_data_formats-pgp>`.
 
 ``vendor_data``/``vendor_data2``
@@ -281,8 +282,8 @@ Format is a dict with the following keys:
 
 * ``enabled``: A boolean indicating whether to enable or disable the vendor
   data. Default: ``true``.
-* ``require_pgp``: A boolean indicating whether to require PGP verification
-  of the vendor data. Default: ``false``.
+* ``require_signature``: A boolean indicating whether to require a PGP
+  signed message for vendor data. Default: ``false``.
 * ``prefix``: A path to a binary to prefix to any executed scripts.
   An example of usage would be to prefix a script with ``strace`` to
   debug a script.

tests/integration_tests/userdata/test_pgp.py

@@ -316,16 +316,43 @@ def test_unparseable_userdata(
 
 
 @pytest.mark.user_data(USER_DATA.format("no"))
-def test_pgp_required(client: IntegrationInstance):
+def test_signature_required(client: IntegrationInstance):
     client.write_to_file(
-        "/etc/cloud/cloud.cfg.d/99_pgp.cfg", "user_data:\n  require_pgp: true"
+        "/etc/cloud/cloud.cfg.d/99_pgp.cfg",
+        "user_data:\n  require_signature: true",
     )
     client.execute("cloud-init clean --logs")
     client.restart()
 
     result = client.execute("cloud-init status --format=json")
     assert result.failed
     assert (
-        "'require_pgp' was set true in cloud-init's base configuration, but "
-        "content type is text/cloud-config"
+        "'require_signature' was set true in cloud-init's base configuration, "
+        "but content type is text/cloud-config"
     ) in result.stdout
+
+
+@pytest.mark.parametrize(
+    "pgp_client", [("encrypted_userdata", "valid_keys_image")], indirect=True
+)
+def test_encrypted_message_but_required_signature(
+    pgp_client: IntegrationInstance,
+):
+    """Ensure fail if we require signature but only have encrypted message."""
+    client = pgp_client
+    assert client.execute("test -f /var/tmp/encrypted")
+    verify_clean_boot(client)
+
+    client.write_to_file(
+        "/etc/cloud/cloud.cfg.d/99_pgp.cfg",
+        "user_data:\n  require_signature: true",
+    )
+    client.execute("cloud-init clean --logs")
+    client.restart()
+
+    result = client.execute("cloud-init status --format=json")
+    assert result.failed
+    assert (
+        "Signature verification required, but no signature found"
+        in result.stdout
+    )

tests/unittests/cloudinit/test_user_data.py

@@ -114,7 +114,7 @@ def test_pgp_decryption_failure(self, mocker):
     def test_pgp_required(self, mocker):
         mocker.patch("cloudinit.subp.subp", side_effect=my_subp)
         ud_proc = user_data.UserDataProcessor({})
-        message = ud_proc.process(GOOD_MESSAGE, require_pgp=True)
+        message = ud_proc.process(GOOD_MESSAGE, require_signature=True)
         parts = [p for p in message.walk() if not user_data.is_skippable(p)]
         assert len(parts) == 1
         part = parts[0]
@@ -126,4 +126,4 @@ def test_pgp_required_with_no_pgp_message(self, mocker):
         with pytest.raises(
             RuntimeError, match="content type is text/cloud-config"
         ):
-            ud_proc.process(CLOUD_CONFIG, require_pgp=True)
+            ud_proc.process(CLOUD_CONFIG, require_signature=True)