Update signing key fields to include a version prefix

Author: daviddavis

pulp_rpm/app/migrations/0070_add_rpm_signing_keys.py

@@ -5,15 +5,26 @@
 
 
 class Migration(migrations.Migration):
-
     dependencies = [
-        ('rpm', '0069_DATA_fix_signing_fingerprint'),
+        ("rpm", "0069_DATA_fix_signing_fingerprint"),
     ]
 
     operations = [
         migrations.AddField(
-            model_name='package',
-            name='signing_keys',
-            field=django.contrib.postgres.fields.ArrayField(base_field=models.TextField(), default=None, null=True, size=None),
+            model_name="package",
+            name="signing_keys",
+            field=django.contrib.postgres.fields.ArrayField(
+                base_field=models.TextField(), default=None, null=True, size=None
+            ),
+        ),
+        migrations.AlterField(
+            model_name="rpmpackagesigningresult",
+            name="package_signing_fingerprint",
+            field=models.TextField(),
+        ),
+        migrations.AlterField(
+            model_name="rpmrepository",
+            name="package_signing_fingerprint",
+            field=models.TextField(null=True),
         ),
     ]

pulp_rpm/app/models/content.py

@@ -37,7 +37,7 @@ def sign(
         Args:
             filename: The absolute path to the package to be signed.
             env_vars: (optional) Dict of env_vars to be passed to the signing script.
-            pubkey_fingerprint: The V4 fingerprint that correlates with the private key to use.
+            pubkey_fingerprint: The raw fingerprint that correlates with the private key to use.
         """
         if not pubkey_fingerprint:
             raise ValueError("A pubkey_fingerprint must be provided.")
@@ -57,7 +57,7 @@ async def asign(
         Args:
             filename: The absolute path to the package to be signed.
             env_vars: (optional) Dict of env_vars to be passed to the signing script.
-            pubkey_fingerprint: The V4 fingerprint that correlates with the private key to use.
+            pubkey_fingerprint: The raw fingerprint that correlates with the private key to use.
         """
         if not pubkey_fingerprint:
             raise ValueError("A pubkey_fingerprint must be provided.")
@@ -107,16 +107,16 @@ class RpmPackageSigningResult(BaseModel):
         original_package_sha256 (String):
             The sha256 digest of the original package artifact.
         package_signing_fingerprint (String):
-            The fingerprint used to sign the package. This value is copied from the repo's
-            package_signing_fingerprint field.
+            The prefixed fingerprint used to sign the package (e.g. 'v4:<hex>').
+            This value is copied from the repo's package_signing_fingerprint field.
 
     Relations:
         result_package (ForeignKey):
             The resulting package that was signed by @package_signing_fingerprint.
     """
 
     original_package_sha256 = models.TextField(max_length=64)
-    package_signing_fingerprint = models.TextField(max_length=40)
+    package_signing_fingerprint = models.TextField()
     result_package = models.ForeignKey(Content, on_delete=models.CASCADE)
 
     class Meta:

pulp_rpm/app/models/repository.py

@@ -208,7 +208,8 @@ class RpmRepository(Repository, AutoAddObjPermsMixin):
         package_signing_service (RpmPackageSigningService):
             Signing service to be used on package signing operations related to this repository.
         package_signing_fingerprint (String):
-            The V4 fingerprint (160 bits) to be used by @package_signing_service.
+            The fingerprint to be used by @package_signing_service.
+            Format: 'v<N>:<hex-fingerprint>' or 'keyid:<16-hex>'.
         repo_config (JSON): repo configuration that will be served by distribution
         compression_type(pulp_rpm.app.constants.COMPRESSION_TYPES):
             Compression type to use for metadata files.
@@ -238,7 +239,7 @@ class RpmRepository(Repository, AutoAddObjPermsMixin):
     package_signing_service = models.ForeignKey(
         RpmPackageSigningService, on_delete=models.SET_NULL, null=True
     )
-    package_signing_fingerprint = models.TextField(null=True, max_length=40)
+    package_signing_fingerprint = models.TextField(null=True)
     last_sync_details = models.JSONField(default=dict)
     retain_package_versions = models.PositiveIntegerField(default=0)
 

pulp_rpm/app/serializers/repository.py

@@ -42,6 +42,7 @@
     UlnRemote,
 )
 from pulp_rpm.app.schema import COPY_CONFIG_SCHEMA
+from pulp_rpm.app.shared_utils import SIGNING_FINGERPRINT_RE, normalize_signing_fingerprint
 from urllib.parse import urlparse
 from textwrap import dedent
 
@@ -89,10 +90,11 @@ class RpmRepositorySerializer(RepositorySerializer):
     )
     package_signing_fingerprint = serializers.CharField(
         help_text=_(
-            "The pubkey V4 fingerprint (160 bits) to be passed to the package signing service."
-            "The signing service will use that on signing operations related to this repository."
+            "The pubkey fingerprint to be passed to the package signing service. "
+            "Format: 'v<N>:<hex-fingerprint>' or 'keyid:<16-hex-char>'. "
+            "Example: 'v4:ABCDEF1234567890ABCDEF1234567890ABCDEF12'."
         ),
-        max_length=40,
+        max_length=68,
         required=False,
         allow_null=True,
         default=None,
@@ -194,6 +196,16 @@ def to_representation(self, instance):
                 data[field] = None
         return data
 
+    def validate_package_signing_fingerprint(self, value):
+        if value is None:
+            return value
+        value = normalize_signing_fingerprint(value)
+        if not SIGNING_FINGERPRINT_RE.match(value):
+            raise serializers.ValidationError(
+                _("Invalid format. Expected 'v<N>:<hex-fingerprint>' or 'keyid:<16-hex>'.")
+            )
+        return value
+
     def validate(self, data):
         """Validate data."""
         if checksum_type := data.get("checksum_type"):

pulp_rpm/app/shared_utils.py

@@ -1,3 +1,4 @@
+import re
 import shutil
 import subprocess
 import tempfile
@@ -13,6 +14,20 @@
 from pulpcore.plugin.exceptions import InvalidSignatureError
 from pulp_rpm.app.rpm_version import RpmVersion
 
+# Matches versioned fingerprints (v3/v4/v5/v6) and legacy 16-char key IDs.
+SIGNING_FINGERPRINT_RE = re.compile(r"^(v\d:[0-9A-F]{32,64}|keyid:[0-9A-F]{16})$")
+
+
+def normalize_signing_fingerprint(value):
+    """Uppercase the hex portion of a prefixed signing fingerprint."""
+    prefix, sep, hex_part = value.partition(":")
+    return f"{prefix}:{hex_part.upper()}" if sep else value
+
+
+def parse_signing_fingerprint(value):
+    """Extract the raw hex value from a prefixed signing fingerprint."""
+    return value.split(":", 1)[1]
+
 
 def annotate_with_age(qs):
     """Provide an "age" score for each Package object in the queryset.

pulp_rpm/app/tasks/signing.py

@@ -22,6 +22,7 @@
 from pulp_rpm.app.models.content import RpmPackageSigningResult, RpmPackageSigningService
 from pulp_rpm.app.models.package import Package
 from pulp_rpm.app.models.repository import RpmRepository
+from pulp_rpm.app.shared_utils import parse_signing_fingerprint
 
 
 log = logging.getLogger(__name__)
@@ -55,14 +56,16 @@ def _verify_package_fingerprint(path, signing_fingerprint):
             f"{completed_process.stderr}."
         )
 
+    raw_fingerprint = parse_signing_fingerprint(signing_fingerprint)
+
     # check for `key ID` followed by a string of hex digits
     key_ids = re.findall(r"key ID ([0-9A-Fa-f]+)", completed_process.stdout, re.IGNORECASE)
     # check for `key fingerprint:` followed by a string of hex digits
     fingerprints = re.findall(
         r"key fingerprint:\s*([0-9A-Fa-f]+)", completed_process.stdout, re.IGNORECASE
     )
     for candidate in key_ids + fingerprints:
-        if signing_fingerprint.lower().endswith(candidate.lower()):
+        if raw_fingerprint.lower().endswith(candidate.lower()):
             return True
 
     return False
@@ -79,7 +82,8 @@ def _update_signing_keys(package_file, keys):
 
 def _sign_file(package_file, signing_service, signing_fingerprint):
     """Sign a package and return the local path of the signed file."""
-    result = signing_service.sign(package_file.name, pubkey_fingerprint=signing_fingerprint)
+    raw_fingerprint = parse_signing_fingerprint(signing_fingerprint)
+    result = signing_service.sign(package_file.name, pubkey_fingerprint=raw_fingerprint)
     signed_package_path = Path(result["rpm_package"])
     if not signed_package_path.exists():
         raise Exception(f"Signing script did not create the signed package: {result}")
@@ -88,7 +92,8 @@ def _sign_file(package_file, signing_service, signing_fingerprint):
 
 async def _asign_file(package_file, signing_service, signing_fingerprint):
     """Sign a package asynchronously and return the local path of the signed file."""
-    result = await signing_service.asign(package_file.name, pubkey_fingerprint=signing_fingerprint)
+    raw_fingerprint = parse_signing_fingerprint(signing_fingerprint)
+    result = await signing_service.asign(package_file.name, pubkey_fingerprint=raw_fingerprint)
     signed_package_path = Path(result["rpm_package"])
     if not signed_package_path.exists():
         raise Exception(f"Signing script did not create the signed package: {result}")

pulp_rpm/tests/functional/api/test_package_signing.py

@@ -94,17 +94,18 @@ def test_sign_package_on_upload(
     # Upload Package to Repository
     # The same file is uploaded, but signed with different keys each time
     for fingerprint in fingerprint_set:
+        prefixed = f"v4:{fingerprint}"
         repository = rpm_repository_factory(
             package_signing_service=rpm_package_signing_service.pulp_href,
-            package_signing_fingerprint=fingerprint,
+            package_signing_fingerprint=prefixed,
         )
         upload_response = rpm_package_api.create(
             file=str(file_to_upload.absolute()),
             repository=repository.pulp_href,
         )
         package_href = monitor_task(upload_response.task).created_resources[2]
         package = rpm_package_api.read(package_href)
-        assert package.signing_keys == [fingerprint]
+        assert package.signing_keys == [prefixed]
 
         # Verify that the final served package is signed
         publication = rpm_publication_factory(repository=repository.pulp_href)
@@ -121,7 +122,7 @@ def test_sign_package_on_upload(
         repository = rpm_repository_api.read(repository.pulp_href)
         assert (
             rpm_package_api.list(
-                repository_version=repository.latest_version_href, signing_key=fingerprint
+                repository_version=repository.latest_version_href, signing_key=prefixed
             ).count
             == 1
         )
@@ -224,9 +225,10 @@ def test_sign_chunked_package_on_upload(
     # Upload Package to Repository
     # The same file is uploaded, but signed with different keys each time
     for fingerprint in fingerprint_set:
+        prefixed = f"v4:{fingerprint}"
         repository = rpm_repository_factory(
             package_signing_service=rpm_package_signing_service.pulp_href,
-            package_signing_fingerprint=fingerprint,
+            package_signing_fingerprint=prefixed,
         )
         file_chunks_data = pulpcore_chunked_file_factory(file_to_upload)
         size = file_chunks_data["size"]
@@ -240,7 +242,7 @@ def test_sign_chunked_package_on_upload(
         )
         package_href = monitor_task(upload_response.task).created_resources[2]
         package = rpm_package_api.read(package_href)
-        assert package.signing_keys == [fingerprint]
+        assert package.signing_keys == [prefixed]
 
         # Verify that the final served package is signed
         publication = rpm_publication_factory(repository=repository.pulp_href)
@@ -271,6 +273,7 @@ def test_signed_repo_modify(
     """Ensure packages added via modify are signed before distribution."""
 
     gpg, fingerprint, _ = signing_gpg_metadata
+    prefixed = f"v4:{fingerprint}"
 
     rpm_tool = RpmTool(tmp_path)
     rpm_tool.import_pubkey_string(gpg.export_keys(fingerprint))
@@ -283,7 +286,7 @@ def test_signed_repo_modify(
 
     repository = rpm_repository_factory(
         package_signing_service=rpm_package_signing_service.pulp_href,
-        package_signing_fingerprint=fingerprint,
+        package_signing_fingerprint=prefixed,
     )
 
     created_package = rpm_package_factory(url=RPM_UNSIGNED_URL)
@@ -299,7 +302,7 @@ def test_signed_repo_modify(
         repository_version=repository.latest_version_href
     ).results[0]
     assert signed_package.pulp_href != created_package.pulp_href
-    assert signed_package.signing_keys == [fingerprint]
+    assert signed_package.signing_keys == [prefixed]
     assert sorted(task_result.created_resources) == sorted(
         [repository.latest_version_href, signed_package.pulp_href, signed_package.artifact]
     )
@@ -341,14 +344,15 @@ def test_already_signed_package(
     """Don't sign a package if it's already signed with our key."""
 
     _, fingerprint, _ = signing_gpg_metadata
+    prefixed = f"v4:{fingerprint}"
 
     repo_one = rpm_repository_factory(
         package_signing_service=rpm_package_signing_service.pulp_href,
-        package_signing_fingerprint=fingerprint,
+        package_signing_fingerprint=prefixed,
     )
     repo_two = rpm_repository_factory(
         package_signing_service=rpm_package_signing_service.pulp_href,
-        package_signing_fingerprint=fingerprint,
+        package_signing_fingerprint=prefixed,
     )
 
     created_package = rpm_package_factory(url=RPM_UNSIGNED_URL)
@@ -365,7 +369,7 @@ def test_already_signed_package(
         repository_version=repo_one.latest_version_href
     ).results
     signed_package_href = repo_one_packages[0].pulp_href
-    assert repo_one_packages[0].signing_keys == [fingerprint]
+    assert repo_one_packages[0].signing_keys == [prefixed]
     assert len(repo_one_packages) == 1
     assert sorted(task_result.created_resources) == sorted(
         [signed_package_href, repo_one_packages[0].artifact, repo_one.latest_version_href]
@@ -401,7 +405,7 @@ def test_signed_repo_rejects_on_demand_content(
     _, fingerprint, _ = signing_gpg_metadata
     destination_repo = rpm_repository_factory(
         package_signing_service=rpm_package_signing_service.pulp_href,
-        package_signing_fingerprint=fingerprint,
+        package_signing_fingerprint=f"v4:{fingerprint}",
     )
 
     packages = rpm_package_api.list(repository_version=source_repo.latest_version_href).results

pulp_rpm/tests/unit/test_signing.py

@@ -1,10 +1,58 @@
 from types import SimpleNamespace
 
+import pytest
 import subprocess
 
+from pulp_rpm.app.shared_utils import (
+    SIGNING_FINGERPRINT_RE,
+    normalize_signing_fingerprint,
+    parse_signing_fingerprint,
+)
 from pulp_rpm.app.tasks.signing import _verify_package_fingerprint
 
 
+@pytest.mark.parametrize(
+    "value",
+    [
+        "v3:0123456789ABCDEF0123456789ABCDEF",
+        "v4:0123456789ABCDEF0123456789ABCDEF01234567",
+        "v5:" + "A" * 64,
+        "v6:" + "B" * 64,
+        "keyid:0123456789ABCDEF",
+    ],
+)
+def test_signing_fingerprint_re_valid(value):
+    assert SIGNING_FINGERPRINT_RE.match(value)
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
+        "0123456789ABCDEF0123456789ABCDEF01234567",  # no prefix
+        "v4:ghijkl",  # non-hex
+        "v4:0123",  # too short
+        "keyid:0123",  # too short
+        "keyid:0123456789ABCDEF0",  # too long
+        "v4:abcdef1234567890abcdef1234567890abcdef12",  # lowercase
+    ],
+)
+def test_signing_fingerprint_re_invalid(value):
+    assert not SIGNING_FINGERPRINT_RE.match(value)
+
+
+def test_normalize_signing_fingerprint():
+    assert (
+        normalize_signing_fingerprint("v4:abcdef1234567890abcdef1234567890abcdef12")
+        == "v4:ABCDEF1234567890ABCDEF1234567890ABCDEF12"
+    )
+    assert normalize_signing_fingerprint("keyid:abcdef1234567890") == "keyid:ABCDEF1234567890"
+
+
+def test_parse_signing_fingerprint():
+    assert parse_signing_fingerprint("v4:ABCDEF12") == "ABCDEF12"
+    assert parse_signing_fingerprint("keyid:0123456789ABCDEF") == "0123456789ABCDEF"
+
+
 def test_verify_package_fingerprint_accepts_rpm_fingerprint(monkeypatch, tmp_path):
     rpm_output = """\
     Header OpenPGP V4 signature, key fingerprint: c6e7f081cf80e13146676e88829b606631645531: OK
@@ -18,7 +66,7 @@ def fake_run(*args, **kwargs):
 
     monkeypatch.setattr(subprocess, "run", fake_run)
     package = SimpleNamespace(name=str(tmp_path / "example.rpm"))
-    assert _verify_package_fingerprint(package, "c6e7f081cf80e13146676e88829b606631645531")
+    assert _verify_package_fingerprint(package, "v4:C6E7F081CF80E13146676E88829B606631645531")
 
 
 def test_verify_package_fingerprint_accepts_long_key_id(monkeypatch, tmp_path):
@@ -35,4 +83,4 @@ def fake_run(*args, **kwargs):
 
     monkeypatch.setattr(subprocess, "run", fake_run)
     package = SimpleNamespace(name=str(tmp_path / "example.rpm"))
-    assert _verify_package_fingerprint(package, "999f7cbf38ab71f4")
+    assert _verify_package_fingerprint(package, "keyid:999F7CBF38AB71F4")