Add a fingerprint to ManifestSignature

closes #2261

Author: dralley

CHANGES/2261.feature

@@ -0,0 +1 @@
+Add a signature fingerprint to the ManifestSignature model for improved forwards compatibility with OpenPGP v6

pulp_container/app/migrations/0049_manifestsignature_fingerprint.py

@@ -0,0 +1,78 @@
+# Generated by Django 5.2.11 on 2026-04-15 02:37
+
+import base64
+import logging
+
+from django.db import migrations, models
+
+from pysequoia.packet import PacketPile, Tag
+
+
+logger = logging.getLogger(__name__)
+
+
+def keyid_from_fingerprint(fingerprint):
+    if len(fingerprint) == 40:
+        return fingerprint[-16:]
+    elif len(fingerprint) == 64:
+        return fingerprint[:16]
+    else:
+        raise ValueError(f"Unexpected fingerprint length: {len(fingerprint)}")
+
+
+def populate_fingerprint(apps, schema_editor):
+    ManifestSignature = apps.get_model("container", "ManifestSignature")
+
+    batch = []
+    for sig in ManifestSignature.objects.filter(fingerprint__isnull=True).iterator():
+        try:
+            signature_raw = base64.b64decode(sig.data)
+            pile = PacketPile.from_bytes(signature_raw)
+        except Exception as exc:
+            logger.warning("Could not parse signature %s, skipping fingerprint extraction", sig.pk)
+            logger.warning(str(exc))
+            continue
+
+        fingerprint = None
+        for packet in pile:
+            if packet.tag == Tag.Signature:
+                if packet.issuer_fingerprint is not None:
+                    fingerprint = packet.issuer_fingerprint.upper()
+                    break
+                elif packet.issuer_key_id is not None:
+                    # No fingerprint available, only key_id — nothing new to store
+                    break
+
+        if fingerprint:
+            assert sig.key_id == keyid_from_fingerprint(fingerprint), (
+                f"Signature {sig.pk}: key_id {sig.key_id} does not match "
+                f"fingerprint {fingerprint}"
+            )
+            sig.fingerprint = fingerprint
+            batch.append(sig)
+
+        if len(batch) > 500:
+            ManifestSignature.objects.bulk_update(batch, ["fingerprint"])
+            batch.clear()
+
+    if batch:
+        ManifestSignature.objects.bulk_update(batch, ["fingerprint"])
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('container', '0048_containerremote_includes_excludes'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='manifestsignature',
+            name='fingerprint',
+            field=models.TextField(db_index=True, null=True),
+        ),
+        migrations.RunPython(
+            populate_fingerprint,
+            reverse_code=migrations.RunPython.noop,
+            elidable=True,
+        ),
+    ]

pulp_container/app/models.py

@@ -399,7 +399,8 @@ class ManifestSignature(Content):
         digest (models.TextField): A signature sha256 digest prepended with its algorithm `sha256:`.
         type (models.TextField): A signature type as specified in signature metadata. Currently
                                  it's only "atomic container signature".
-        key_id (models.TextField): A key id identified by gpg (last 8 bytes of the fingerprint).
+        key_id (models.TextField): A PGP key id (last 8 bytes of the fingerprint).
+        fingerprint (models.TextField): A PGP key fingerprint
         timestamp (models.PositiveIntegerField): A signature timestamp identified by gpg.
         creator (models.TextField): A signature creator.
         data (models.TextField): A signature, base64 encoded.
@@ -417,6 +418,7 @@ class ManifestSignature(Content):
     digest = models.TextField()
     type = models.TextField(choices=SIGNATURE_CHOICES)
     key_id = models.TextField(db_index=True)
+    fingerprint = models.TextField(null=True, db_index=True)
     timestamp = models.PositiveIntegerField()
     creator = models.TextField(blank=True)
     data = models.TextField()

pulp_container/app/registry_api.py

@@ -1745,6 +1745,7 @@ def put(self, request, path, pk):
             digest=f"sha256:{sig_digest}",
             type=SIGNATURE_TYPE.ATOMIC_SHORT,
             key_id=signature_json["signing_key_id"],
+            fingerprint=signature_json["signing_key_fingerprint"],
             timestamp=signature_json["signature_timestamp"],
             creator=signature_json["optional"].get("creator"),
             data=signature_dict["content"],

pulp_container/app/serializers.py

@@ -186,6 +186,7 @@ class ManifestSignatureSerializer(NoArtifactContentSerializer):
     digest = serializers.CharField(help_text="sha256 digest of the signature blob")
     type = serializers.CharField(help_text="Container signature type, e.g. 'atomic'")
     key_id = serializers.CharField(help_text="Signing key ID")
+    fingerprint = serializers.CharField(help_text="Signing key fingerprint", allow_null=True)
     timestamp = serializers.IntegerField(help_text="Timestamp of a signature")
     creator = serializers.CharField(help_text="Signature creator")
     signed_manifest = DetailRelatedField(
@@ -201,6 +202,7 @@ class Meta:
             "digest",
             "type",
             "key_id",
+            "fingerprint",
             "timestamp",
             "creator",
             "signed_manifest",

pulp_container/app/tasks/sign.py

@@ -128,6 +128,7 @@ async def create_signature(manifest, reference, signing_service):
                 digest=f"sha256:{sig_digest}",
                 type=SIGNATURE_TYPE.ATOMIC_SHORT,
                 key_id=sig_json["signing_key_id"],
+                fingerprint=sig_json["signing_key_fingerprint"],
                 timestamp=sig_json["signature_timestamp"],
                 creator=sig_json["optional"].get("creator"),
                 data=encoded_sig,

pulp_container/app/tasks/sync_stages.py

@@ -484,6 +484,7 @@ def _create_signature_declarative_content(
             digest=f"sha256:{sig_digest}",
             type=SIGNATURE_TYPE.ATOMIC_SHORT,
             key_id=signature_json["signing_key_id"],
+            fingerprint=signature_json["signing_key_fingerprint"],
             timestamp=signature_json["signature_timestamp"],
             creator=signature_json["optional"].get("creator"),
             data=signature_b64 or base64.b64encode(signature_raw).decode(),

pulp_container/app/utils.py

@@ -125,10 +125,13 @@ def extract_data_from_signature(signature_raw, man_digest):
         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:
+
+            if packet.issuer_fingerprint is not None:
                 signing_key_fingerprint = packet.issuer_fingerprint.upper()
-                signing_key_id = keyid_from_fingerprint(signing_key_fingerprint)
-            else:
+                if signing_key_id is None:
+                    signing_key_id = keyid_from_fingerprint(signing_key_fingerprint)
+
+            if signing_key_id is None and signing_key_fingerprint is None:
                 raise ValueError(
                     "Signature for manifest {} has no fingerprint or key_id".format(man_digest)
                 )

pulp_container/tests/functional/api/test_push_signatures.py

@@ -63,6 +63,7 @@ def test_assert_signed_image(
     assert manifest.digest in signature.name
     assert signature.signed_manifest == manifest.pulp_href
     assert signature.key_id == keyid
+    assert signature.fingerprint == fingerprint
 
     path = f"/extensions/v2/{full_path(distribution)}/signatures/{manifest.digest}"
     response, _ = local_registry.get_response("GET", path)
@@ -77,6 +78,7 @@ def test_assert_signed_image(
         decrypted = gpg.decrypt(raw_s)
 
         assert decrypted.key_id == keyid
+        assert decrypted.fingerprint == fingerprint
         assert decrypted.status == "signature valid"
 
         json_s = json.loads(decrypted.data)

pulp_container/tests/functional/api/test_sign_manifests.py

@@ -32,7 +32,7 @@ def test_sign_manifest(
     monitor_task,
 ):
     """Test whether a user can sign a manifest by leveraging a signing service."""
-    _, _, keyid = signing_gpg_metadata
+    _, fingerprint, keyid = signing_gpg_metadata
     sign_data = {"manifest_signing_service": container_signing_service.pulp_href}
 
     response = container_push_repository_api.sign(distribution.repository, sign_data)
@@ -49,6 +49,7 @@ def test_sign_manifest(
 
     signature = signatures.results[0]
     assert signature.key_id == keyid
+    assert signature.fingerprint == fingerprint
     assert signature.type == SIGNATURE_TYPE.ATOMIC_SHORT
 
     manifest = container_manifest_api.read(tag.tagged_manifest)