Use PySequoia to parse signatures & not break on PQC signatures

CHANGES/2237.bugfix

@@ -0,0 +1 @@
+Don't blow up on encountering PQC signatures.

pulp_container/app/registry_api.py

@@ -1545,8 +1545,10 @@ def put(self, request, path, pk):
         except binascii.Error:
             raise ManifestSignatureInvalid(digest=pk)
 
-        signature_json = extract_data_from_signature(signature_raw, manifest.digest)
-        if signature_json is None:
+        try:
+            signature_json = extract_data_from_signature(signature_raw, manifest.digest)
+        except ValueError as exc:
+            log.warning("Error processing signature on upload: {}".format(exc))
             raise ManifestSignatureInvalid(digest=pk)
 
         sig_digest = hashlib.sha256(signature_raw).hexdigest()

pulp_container/app/tasks/sign.py

@@ -118,7 +118,10 @@ async def create_signature(manifest, reference, signing_service):
             data = sig_fp.read()
             encoded_sig = base64.b64encode(data).decode()
             sig_digest = hashlib.sha256(data).hexdigest()
-            sig_json = extract_data_from_signature(data, manifest.digest)
+            try:
+                sig_json = extract_data_from_signature(data, manifest.digest)
+            except ValueError:
+                raise
             manifest_digest = sig_json["critical"]["image"]["docker-manifest-digest"]
 
             signature = ManifestSignature(

pulp_container/app/tasks/sync_stages.py

@@ -391,8 +391,10 @@ def create_manifest(self, manifest_data, raw_text_data, media_type, digest=None)
     def _create_signature_declarative_content(
         self, signature_raw, man_dc, name=None, signature_b64=None
     ):
-        signature_json = extract_data_from_signature(signature_raw, man_dc.content.digest)
-        if signature_json is None:
+        try:
+            signature_json = extract_data_from_signature(signature_raw, man_dc.content.digest)
+        except ValueError as exc:
+            log.warning("Error processing signature on sync: {}".format(str(exc)))
             return
 
         sig_digest = hashlib.sha256(signature_raw).hexdigest()

pulp_container/app/utils.py

@@ -1,9 +1,6 @@
 import base64
 import hashlib
 import fnmatch
-import re
-import subprocess
-import gnupg
 import json
 import logging
 import time
@@ -15,6 +12,8 @@
 from functools import partial
 from rest_framework.exceptions import Throttled
 
+from pysequoia.packet import PacketPile, Tag
+
 from pulpcore.plugin.models import Artifact, Task
 from pulpcore.plugin.util import get_domain
 
@@ -32,9 +31,6 @@
     SIGNATURE_SCHEMA,
 )
 
-KEY_ID_REGEX_COMPILED = re.compile(r"keyid ([0-9A-F]+)")
-TIMESTAMP_REGEX_COMPILED = re.compile(r"created ([0-9]+)")
-
 signature_validator = Draft7Validator(SIGNATURE_SCHEMA)
 
 log = logging.getLogger(__name__)
@@ -86,6 +82,20 @@ def urlpath_sanitize(*args):
     return "/".join(segments)
 
 
+def keyid_from_fingerprint(fingerprint):
+    """Derive a key ID from an OpenPGP fingerprint.
+
+    For v4 fingerprints (40 hex chars / 20 bytes), the key ID is the last 8 bytes.
+    For v6 fingerprints (64 hex chars / 32 bytes), the key ID is the first 8 bytes.
+    """
+    if len(fingerprint) == 40:
+        return fingerprint[-16:]
+    elif len(fingerprint) == 64:
+        return fingerprint[:16]
+    else:
+        raise ValueError(f"Unexpected fingerprint length: {len(fingerprint)}")
+
+
 def extract_data_from_signature(signature_raw, man_digest):
     """
     Extract data from an "integrated" signature, aka a signed non-encrypted document.
@@ -98,37 +108,56 @@ def extract_data_from_signature(signature_raw, man_digest):
         dict: JSON representation of the document and available data about signature
 
     """
-    gpg = gnupg.GPG()
-    crypt_obj = gpg.decrypt(signature_raw, extra_args=["--skip-verify"])
-    if not crypt_obj.data:
-        log.info(
-            "It is not possible to read the signed document, GPG error: {}".format(crypt_obj.stderr)
+    try:
+        pile = PacketPile.from_bytes(signature_raw)
+    except Exception as exc:
+        raise ValueError(
+            "Signed document for manifest {} is un-parseable: {}".format(man_digest, str(exc))
         )
-        return
+
+    literal_data = None
+    signing_key_id = None
+    signing_key_fingerprint = None
+    signature_timestamp = None
+
+    for packet in pile:
+        if packet.tag == Tag.Literal:
+            literal_data = bytes(packet.literal_data)
+        elif packet.tag == Tag.Signature:
+            if packet.issuer_key_id is not None:
+                signing_key_id = packet.issuer_key_id.upper()
+            elif packet.issuer_fingerprint is not None:
+                signing_key_fingerprint = packet.issuer_fingerprint.upper()
+                signing_key_id = keyid_from_fingerprint(signing_key_fingerprint)
+            else:
+                raise ValueError(
+                    "Signature for manifest {} has no fingerprint or key_id".format(man_digest)
+                )
+            if packet.signature_created is not None:
+                signature_timestamp = int(packet.signature_created.timestamp())
+
+    if not literal_data:
+        raise ValueError("Signature for manifest {} has no literal data".format(man_digest))
 
     try:
-        sig_json = json.loads(crypt_obj.data)
+        sig_json = json.loads(literal_data)
     except Exception as exc:
-        log.info(
-            "Signed document cannot be parsed to create a signature for {}."
-            " Error: {}".format(man_digest, str(exc))
+        raise ValueError(
+            "Signed document cannot be parsed to create a signature for {}. Error: {}".format(
+                man_digest, str(exc)
+            )
         )
-        return
 
     errors = []
     for error in signature_validator.iter_errors(sig_json):
         errors.append(f'{".".join(error.path)}: {error.message}')
 
     if errors:
-        log.info("The signature for {} is not synced due to: {}".format(man_digest, errors))
-        return
-
-    # decrypted and unverified signatures do not have prepopulated the key_id and timestamp
-    # fields; thus, it is necessary to use the debugging utilities of gpg to extract these
-    # fields since they are not encrypted and still readable without decrypting the signature first
-    packets = subprocess.check_output(["gpg", "--list-packets"], input=signature_raw).decode()
-    sig_json["signing_key_id"] = KEY_ID_REGEX_COMPILED.search(packets).group(1)
-    sig_json["signature_timestamp"] = TIMESTAMP_REGEX_COMPILED.search(packets).group(1)
+        raise ValueError("The signature for {} is not synced due to: {}".format(man_digest, errors))
+
+    sig_json["signing_key_id"] = signing_key_id
+    sig_json["signing_key_fingerprint"] = signing_key_fingerprint
+    sig_json["signature_timestamp"] = signature_timestamp
 
     return sig_json
 

pulp_container/tests/functional/api/test_sync_signatures.py

@@ -159,17 +159,17 @@ def test_sync_image_with_pqc_signatures(
     ).results
     assert len(signatures) > 0
 
-    # Assert that signature using the "old" Red Hat signing release keys exist
+    # Assert that a signature using one of the "old" Red Hat signing release keys exist
     expected_key_ids = ["199E2F91FD431D51", "E60D446E63405576"]
     assert any(s.key_id in expected_key_ids for s in signatures), (
         f"No signature found with key_ids {expected_key_ids}; "
         f"found key_ids: {sorted({s.key_id for s in signatures})}"
     )
 
-    # First 8 bytes (16 hex chars) of the fingerprint the Red Hat PQC (ML-DSA-87) signing key
-    # which is FCD355B305707A62DA143AB6E422397E50FE8467A2A95343D246D6276AFEDF8F
+    # Assert that a signature using the Red Hat PQC (ML-DSA-87) signing key exists
+    # Fingerprint: FCD355B305707A62DA143AB6E422397E50FE8467A2A95343D246D6276AFEDF8F
+    # Key ID => first 8 bytes (16 hex chars)
     expected_key_id = "FCD355B305707A62"
-    pytest.xfail("pulp_container does not yet support PQC (ML-DSA-87) signatures")
     assert any(s.key_id == expected_key_id for s in signatures), (
         f"No signature found with key_id {expected_key_id!r}; "
         f"found key_ids: {sorted({s.key_id for s in signatures})}"

pyproject.toml

@@ -26,6 +26,7 @@ dependencies = [
   "jsonschema>=4.4,<4.27",
   "pulpcore>=3.73.2,<3.115",
   "pyjwt[crypto]>=2.4,<2.13",
+  "pysequoia==0.1.32"
 ]
 
 [project.urls]