unwrap before verify

cloudinit/gpg.py

@@ -108,12 +108,30 @@ def decrypt(self, data: str, *, require_signature=False) -> str:
                     data=data,
                     update_env=self.env,
                 )
-            except subp.ProcessExecutionError as e:
-                if e.exit_code == 2:
+            except subp.ProcessExecutionError:
+                # If the message is signed then encrypted (the default),
+                # the message can't be verified until it's decrypted
+                try:
+                    stdout, _ = subp.subp(
+                        ["gpg", "--unwrap"],
+                        data=data,
+                        update_env=self.env,
+                        decode=False,
+                    )
+                except subp.ProcessExecutionError as e:
                     raise GpgVerificationError(
-                        "Signature verification failed"
+                        "Signature verification failed. Could not unwrap."
+                    ) from e
+                try:
+                    subp.subp(
+                        ["gpg", "--verify"],
+                        data=stdout,
+                        update_env=self.env,
+                    )
+                except subp.ProcessExecutionError as e:
+                    raise GpgVerificationError(
+                        "Signature verification failed. Could not verify."
                     ) from e
-                raise
         result = subp.subp(
             [
                 "gpg",

cloudinit/subp.py

@@ -7,7 +7,7 @@
 import subprocess
 from errno import ENOEXEC
 from io import TextIOWrapper
-from typing import List, Optional, Union
+from typing import List, Literal, Optional, Union
 
 from cloudinit import performance
 
@@ -170,7 +170,7 @@ def subp(
     capture=True,
     shell=False,
     logstring=False,
-    decode="replace",
+    decode: Literal[False, "strict", "ignore", "replace"] = "replace",
     update_env=None,
     cwd=None,
     timeout=None,

tests/integration_tests/userdata/test_pgp.py

@@ -1,5 +1,7 @@
 """Test PGP signed and encrypted userdata."""
 
+from pathlib import Path
+
 import pytest
 
 from cloudinit import subp
@@ -16,20 +18,19 @@
 
 
 @pytest.fixture(scope="module")
-def gpg_dir(tmp_path_factory):
+def gpg_dir(tmp_path_factory: pytest.TempPathFactory):
     yield tmp_path_factory.mktemp("gpg_dir")
 
 
 @pytest.fixture(scope="module")
-def public_key(gpg_dir):
+def public_key(gpg_dir: Path):
     subp.subp(
         [
             "gpg",
             "--homedir",
             str(gpg_dir),
             "--quick-generate-key",
             "--batch",
-            # "loopback",
             "--passphrase",
             "",
             "signing_user",
@@ -197,6 +198,14 @@ def _invalidate_key(client, key_path):
 )
 def test_signed_and_encrypted(pgp_client: IntegrationInstance):
     client = pgp_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()
+
     assert client.execute("test -f /var/tmp/signed_and_encrypted")
     verify_clean_boot(client)
 
@@ -209,7 +218,7 @@ def test_signed_and_encrypted(pgp_client: IntegrationInstance):
     assert not client.execute("test -f /var/tmp/signed_and_encrypted")
     result = client.execute("cloud-init status --format=json")
     assert result.failed
-    assert "Failed decrypting user data" in result.stdout
+    assert "Signature verification failed. Could not verify." in result.stdout
 
     # Restore the public key, invalidate the private key, and ensure we fail
     client.execute("cp /var/tmp/pub_key /etc/cloud/keys/pub_key")
@@ -221,7 +230,7 @@ def test_signed_and_encrypted(pgp_client: IntegrationInstance):
     assert not client.execute("test -f /var/tmp/signed_and_encrypted")
     result = client.execute("cloud-init status --format=json")
     assert result.failed
-    assert "Failed decrypting user data" in result.stdout
+    assert "Signature verification failed. Could not unwrap." in result.stdout
 
 
 @pytest.mark.parametrize(