From 8ecb6cb4c501740ca8a0b643026a1aa864d86afa Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 10:02:19 -0400 Subject: [PATCH 1/7] Improve OpenPGP keyring test coverage - v4/v6 key uploads (RSA, Ed25519) - PQC keys (ML-DSA-65, ML-DSA-87) - content listing/filtering - serialization round-trips via distribution - signature verification using keys served from the keyring - key revocation - idempotent uploads - private key rejection Also fix three bugs: - Typo: distribution permission said "gem" instead of "openpgp" - content_handler_list_directory crashed when no repository version exists - represent() querysets were unordered, producing nondeterministic output Assisted-By: Claude Opus 4.6 --- ...pgp_distribution_permission_description.py | 25 + pulpcore/app/models/openpgp.py | 23 +- pulpcore/tests/functional/api/test_openpgp.py | 460 +++++++++++++++++- 3 files changed, 479 insertions(+), 29 deletions(-) create mode 100644 pulpcore/app/migrations/0155_fix_openpgp_distribution_permission_description.py diff --git a/pulpcore/app/migrations/0155_fix_openpgp_distribution_permission_description.py b/pulpcore/app/migrations/0155_fix_openpgp_distribution_permission_description.py new file mode 100644 index 00000000000..261b532e410 --- /dev/null +++ b/pulpcore/app/migrations/0155_fix_openpgp_distribution_permission_description.py @@ -0,0 +1,25 @@ +# Generated by Django 5.2.14 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ("core", "0154_task_api_version"), + ] + + operations = [ + migrations.AlterModelOptions( + name="openpgpdistribution", + options={ + "default_related_name": "%(app_label)s_%(model_name)s", + "permissions": [ + ( + "manage_roles_openpgpdistribution", + "Can manage roles on openpgp distributions", + ) + ], + }, + ), + ] diff --git a/pulpcore/app/models/openpgp.py b/pulpcore/app/models/openpgp.py index 3c96e03e8f0..9b7f3ec5442 100644 --- a/pulpcore/app/models/openpgp.py +++ b/pulpcore/app/models/openpgp.py @@ -48,20 +48,23 @@ def represent(self, repository_version=None): else: content_filter = {} data = self.packet() - # note: because these queries aren't ordered, the result may be nondetermininistic - for signature in self.openpgp_signatures.filter(**content_filter): + for signature in self.openpgp_signatures.filter(**content_filter).order_by("pk"): data += signature.packet() - for user_id in self.user_ids.filter(**content_filter): + for user_id in self.user_ids.filter(**content_filter).order_by("pk"): data += user_id.packet() - for signature in user_id.openpgp_signatures.filter(**content_filter): + for signature in user_id.openpgp_signatures.filter(**content_filter).order_by("pk"): data += signature.packet() - for user_attribute in self.user_attributes.filter(**content_filter): + for user_attribute in self.user_attributes.filter(**content_filter).order_by("pk"): data += user_attribute.packet() - for signature in user_attribute.openpgp_signatures.filter(**content_filter): + for signature in user_attribute.openpgp_signatures.filter(**content_filter).order_by( + "pk" + ): data += signature.packet() - for public_subkey in self.public_subkeys.filter(**content_filter): + for public_subkey in self.public_subkeys.filter(**content_filter).order_by("pk"): data += public_subkey.packet() - for signature in public_subkey.openpgp_signatures.filter(**content_filter): + for signature in public_subkey.openpgp_signatures.filter(**content_filter).order_by( + "pk" + ): data += signature.packet() return armor(data, ArmorKind.PublicKey).strip() # avoid trailing newline @@ -222,6 +225,8 @@ def content_handler(self, path): def content_handler_list_directory(self, rel_path): if rel_path == "": repository_version = self.repository_version or self.repository.latest_version() + if repository_version is None: + return set() fingerprints = OpenPGPPublicKey.objects.filter( pk__in=repository_version.content ).values_list("fingerprint", flat=True) @@ -231,5 +236,5 @@ def content_handler_list_directory(self, rel_path): class Meta: default_related_name = "%(app_label)s_%(model_name)s" permissions = [ - ("manage_roles_openpgpdistribution", "Can manage roles on gem distributions"), + ("manage_roles_openpgpdistribution", "Can manage roles on openpgp distributions"), ] diff --git a/pulpcore/tests/functional/api/test_openpgp.py b/pulpcore/tests/functional/api/test_openpgp.py index 3adbc930768..b014d13e637 100644 --- a/pulpcore/tests/functional/api/test_openpgp.py +++ b/pulpcore/tests/functional/api/test_openpgp.py @@ -1,4 +1,49 @@ +import re +import uuid + import pytest +import requests + +from pulpcore.pytest_plugin import ( + KEY_V4_ED25519_PRIVATE, + KEY_V4_ED25519_PUBLIC, + KEY_V4_RSA2K_PUBLIC, + KEY_V4_RSA4K_PUBLIC, + KEY_V6_ED25519_PRIVATE, + KEY_V6_ED25519_PUBLIC, + KEY_V6_MLDSA65_ED25519_PUBLIC, + KEY_V6_MLDSA87_ED448_PUBLIC, + KEY_V6_RSA4K_PUBLIC, +) +from pulpcore.tests.functional.utils import PulpTaskError + + +def _download_key(url): + response = requests.get(url) + response.raise_for_status() + return response.content + + +def _upload_key(tmpdir, pulpcore_bindings, monitor_task, key_data, keyring): + key_path = tmpdir / f"{uuid.uuid4()}.asc" + if isinstance(key_data, str): + key_path.write_text(key_data, "UTF-8") + else: + key_path.write_binary(key_data) + result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( + file=str(key_path), repository=keyring.pulp_href + ) + monitor_task(result.task) + keyring = pulpcore_bindings.RepositoriesOpenpgpKeyringApi.read(keyring.pulp_href) + return keyring + + +def _upload_key_from_url(tmpdir, pulpcore_bindings, monitor_task, url, keyring): + key_data = _download_key(url) + return _upload_key(tmpdir, pulpcore_bindings, monitor_task, key_data, keyring) + + +# ── Sample keys from IETF OpenPGP samples draft ────────────────────────── ALICE_PUB = """ -----BEGIN PGP PUBLIC KEY BLOCK----- @@ -18,7 +63,6 @@ -----END PGP PUBLIC KEY BLOCK----- """ - ALICE_REVOCATION = """ -----BEGIN PGP PUBLIC KEY BLOCK----- Comment: Alice's revocation certificate @@ -31,7 +75,6 @@ -----END PGP PUBLIC KEY BLOCK----- """ - ALICE_REVOKED = """ -----BEGIN PGP PUBLIC KEY BLOCK----- @@ -51,7 +94,6 @@ -----END PGP PUBLIC KEY BLOCK----- """ - BOB_PUB = """ -----BEGIN PGP PUBLIC KEY BLOCK----- Comment: Bob's OpenPGP certificate @@ -117,29 +159,407 @@ -----END PGP PUBLIC KEY BLOCK----- """ +# Known fingerprints for the IETF sample keys above +ALICE_FINGERPRINT = "EB85BB5FA33A75E15E944E63F231550C4F47E38E" +BOB_FINGERPRINT = "D1A66E1A23B182C9980F788CFBFCC82A015E7330" + @pytest.mark.parallel -def test_key_upload(tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task): - keyring = openpgp_keyring_factory() +class TestOpenPGPKeyUpload: + """Test uploading various OpenPGP key types.""" - alice_pub = tmpdir / "alice.pub" - alice_pub.write_text(ALICE_PUB, "UTF-8") + @pytest.mark.parametrize( + "key_url,fingerprint_len", + [ + (KEY_V4_RSA2K_PUBLIC, 40), + (KEY_V4_RSA4K_PUBLIC, 40), + (KEY_V4_ED25519_PUBLIC, 40), + (KEY_V6_ED25519_PUBLIC, 64), + (KEY_V6_RSA4K_PUBLIC, 64), + (KEY_V6_MLDSA65_ED25519_PUBLIC, 64), + (KEY_V6_MLDSA87_ED448_PUBLIC, 64), + ], + ids=[ + "v4-rsa2k", + "v4-rsa4k", + "v4-ed25519", + "v6-ed25519", + "v6-rsa4k", + "pqc-mldsa65", + "pqc-mldsa87", + ], + ) + def test_upload_key( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + key_url, + fingerprint_len, + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key_from_url(tmpdir, pulpcore_bindings, monitor_task, key_url, keyring) - alice_revoked = tmpdir / "alice.revoked" - alice_revoked.write_text(ALICE_REVOKED, "UTF-8") + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + assert keys.count == 1 + key = keys.results[0] + assert len(key.fingerprint) == fingerprint_len + assert re.fullmatch(rf"[0-9A-F]{{{fingerprint_len}}}", key.fingerprint) - bob_pub = tmpdir / "bob.pub" - bob_pub.write_text(BOB_PUB, "UTF-8") + user_ids = pulpcore_bindings.ContentOpenpgpUseridApi.list( + repository_version=keyring.latest_version_href + ) + assert user_ids.count >= 1 - result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( - file=str(alice_pub), repository=keyring.pulp_href + signatures = pulpcore_bindings.ContentOpenpgpSignatureApi.list( + repository_version=keyring.latest_version_href + ) + assert signatures.count >= 1 + + def test_reject_private_key_upload( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + key_data = _download_key(KEY_V4_ED25519_PRIVATE) + key_path = tmpdir / "private.key" + key_path.write_binary(key_data) + + result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( + file=str(key_path), repository=keyring.pulp_href + ) + with pytest.raises(PulpTaskError) as exc_info: + monitor_task(result.task) + assert exc_info.value.task.state == "failed" + + +@pytest.mark.parallel +class TestOpenPGPKeyContent: + """Test content listing, filtering, and structure.""" + + def test_multiple_keys_in_keyring( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + + for url in [KEY_V4_RSA2K_PUBLIC, KEY_V4_ED25519_PUBLIC, KEY_V6_ED25519_PUBLIC]: + keyring = _upload_key_from_url(tmpdir, pulpcore_bindings, monitor_task, url, keyring) + + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + assert keys.count == 3 + + v4_keys = [k for k in keys.results if len(k.fingerprint) == 40] + v6_keys = [k for k in keys.results if len(k.fingerprint) == 64] + assert len(v4_keys) == 2 + assert len(v6_keys) == 1 + + def test_filter_public_key_by_fingerprint( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, BOB_PUB, keyring) + + alice_results = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + fingerprint=ALICE_FINGERPRINT + ) + assert alice_results.count == 1 + assert alice_results.results[0].fingerprint == ALICE_FINGERPRINT + + bob_results = pulpcore_bindings.ContentOpenpgpPublickeyApi.list(fingerprint=BOB_FINGERPRINT) + assert bob_results.count == 1 + assert bob_results.results[0].fingerprint == BOB_FINGERPRINT + + def test_filter_signature_by_issuer( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + + alice_keyid = ALICE_FINGERPRINT[-16:] + sigs = pulpcore_bindings.ContentOpenpgpSignatureApi.list( + repository_version=keyring.latest_version_href, issuer=alice_keyid + ) + assert sigs.count >= 1 + for sig in sigs.results: + assert sig.issuer == alice_keyid + + def test_user_id_content( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + + user_ids = pulpcore_bindings.ContentOpenpgpUseridApi.list( + repository_version=keyring.latest_version_href + ) + assert user_ids.count == 1 + assert "Alice Lovelace" in user_ids.results[0].user_id + + def test_subkey_content(self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + + subkeys = pulpcore_bindings.ContentOpenpgpPublicsubkeyApi.list( + repository_version=keyring.latest_version_href + ) + assert subkeys.count >= 1 + for subkey in subkeys.results: + assert len(subkey.fingerprint) == 40 + assert re.fullmatch(r"[0-9A-F]{40}", subkey.fingerprint) + + def test_idempotent_upload( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + assert keys.count == 1 + assert keys.results[0].fingerprint == ALICE_FINGERPRINT + + def test_key_revocation_merged( + self, tmpdir, openpgp_keyring_factory, pulpcore_bindings, monitor_task + ): + """Upload a key, then upload the same key with a revocation signature merged in.""" + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_REVOKED, keyring) + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + assert keys.count == 1 + assert keys.results[0].fingerprint == ALICE_FINGERPRINT + + sigs = pulpcore_bindings.ContentOpenpgpSignatureApi.list( + repository_version=keyring.latest_version_href + ) + sig_types = {sig.signature_type for sig in sigs.results} + assert 0x20 in sig_types, "Expected a key revocation signature (type 0x20)" + + @pytest.mark.parametrize( + "pub_data,revocation_data", + [ + (ALICE_PUB, ALICE_REVOCATION), + (BOB_PUB, BOB_REVOCATION), + ], + ids=["alice-ed25519", "bob-rsa"], ) - monitor_task(result.task) - result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( - file=str(bob_pub), repository=keyring.pulp_href + def test_standalone_revocation_certificate( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + pub_data, + revocation_data, + ): + """Upload a standalone revocation certificate (no public key, just the signature).""" + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, pub_data, keyring) + + key_path = tmpdir / f"{uuid.uuid4()}.asc" + key_path.write_text(revocation_data, "UTF-8") + result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( + file=str(key_path), repository=keyring.pulp_href + ) + with pytest.raises(PulpTaskError): + monitor_task(result.task) + + +@pytest.mark.parallel +class TestOpenPGPKeySerialization: + """Test key serialization via distribution and round-trip integrity.""" + + def _create_distribution( + self, pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ): + body = { + "base_path": str(uuid.uuid4()), + "name": str(uuid.uuid4()), + "repository": keyring.pulp_href, + } + return gen_object_with_cleanup(pulpcore_bindings.DistributionsOpenpgpApi, body) + + @pytest.mark.parametrize( + "key_url", + [KEY_V4_RSA2K_PUBLIC, KEY_V6_ED25519_PUBLIC, KEY_V6_MLDSA65_ED25519_PUBLIC], + ids=["v4-rsa2k", "v6-ed25519", "pqc-mldsa65"], ) - monitor_task(result.task) - result = pulpcore_bindings.ContentOpenpgpPublickeyApi.create( - file=str(alice_revoked), repository=keyring.pulp_href + def test_key_serialization_roundtrip( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + gen_object_with_cleanup, + http_get, + distribution_base_url, + key_url, + ): + from pysequoia import Cert + + keyring = openpgp_keyring_factory() + keyring = _upload_key_from_url(tmpdir, pulpcore_bindings, monitor_task, key_url, keyring) + + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + uploaded_fingerprint = keys.results[0].fingerprint + + distro = self._create_distribution( + pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ) + + keyid = uploaded_fingerprint[-16:] + download_url = distribution_base_url(distro.base_url) + f"/{keyid}.pub" + downloaded_key = http_get(download_url) + + cert = Cert.from_bytes(downloaded_key) + assert cert.fingerprint.upper() == uploaded_fingerprint + + def test_distribution_lists_keys( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + gen_object_with_cleanup, + http_get, + distribution_base_url, + ): + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, ALICE_PUB, keyring) + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, BOB_PUB, keyring) + + distro = self._create_distribution( + pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ) + + listing_url = distribution_base_url(distro.base_url) + "/" + listing = http_get(listing_url) + listing_text = listing.decode("utf-8") + + alice_keyid = ALICE_FINGERPRINT[-16:] + bob_keyid = BOB_FINGERPRINT[-16:] + listing_lower = listing_text.lower() + assert f"{alice_keyid.lower()}.pub" in listing_lower + assert f"{bob_keyid.lower()}.pub" in listing_lower + + +@pytest.mark.parallel +class TestOpenPGPSignatureVerification: + """Test that keys from the keyring can verify signatures.""" + + def _create_distribution( + self, pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ): + body = { + "base_path": str(uuid.uuid4()), + "name": str(uuid.uuid4()), + "repository": keyring.pulp_href, + } + return gen_object_with_cleanup(pulpcore_bindings.DistributionsOpenpgpApi, body) + + @pytest.mark.parametrize( + "private_url,public_url", + [ + (KEY_V4_ED25519_PRIVATE, KEY_V4_ED25519_PUBLIC), + (KEY_V6_ED25519_PRIVATE, KEY_V6_ED25519_PUBLIC), + ], + ids=["v4-ed25519", "v6-ed25519"], ) - monitor_task(result.task) + def test_verify_signature( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + gen_object_with_cleanup, + http_get, + distribution_base_url, + private_url, + public_url, + ): + from pysequoia import Cert, Sig, SignatureMode, sign, verify + + secret_data = _download_key(private_url) + cert = Cert.from_bytes(secret_data) + fingerprint = cert.fingerprint.upper() + signer = cert.secrets.signer() + + test_data = b"Hello, OpenPGP!\n" + sig_bytes = sign(signer, test_data, mode=SignatureMode.DETACHED) + + pub_data = _download_key(public_url) + keyring = openpgp_keyring_factory() + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, pub_data, keyring) + + distro = self._create_distribution( + pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ) + + keyid = fingerprint[-16:] + download_url = distribution_base_url(distro.base_url) + f"/{keyid}.pub" + served_key = http_get(download_url) + + served_certs = Cert.split_bytes(served_key) + sig = Sig.from_bytes(sig_bytes.encode() if isinstance(sig_bytes, str) else sig_bytes) + result = verify(bytes=test_data, store=lambda key_ids: served_certs, signature=sig) + assert result.valid_sigs + + def test_verify_signature_multiple_keys_in_keyring( + self, + tmpdir, + openpgp_keyring_factory, + pulpcore_bindings, + monitor_task, + gen_object_with_cleanup, + http_get, + distribution_base_url, + ): + """Sign with one key, verify against a keyring containing multiple keys.""" + from pysequoia import Cert, Sig, SignatureMode, sign, verify + + secret_data = _download_key(KEY_V4_ED25519_PRIVATE) + cert = Cert.from_bytes(secret_data) + fingerprint = cert.fingerprint.upper() + signer = cert.secrets.signer() + + test_data = b"Hello, OpenPGP!\n" + sig_bytes = sign(signer, test_data, mode=SignatureMode.DETACHED) + + keyring = openpgp_keyring_factory() + + pub_data = _download_key(KEY_V4_ED25519_PUBLIC) + keyring = _upload_key(tmpdir, pulpcore_bindings, monitor_task, pub_data, keyring) + keyring = _upload_key_from_url( + tmpdir, pulpcore_bindings, monitor_task, KEY_V4_RSA2K_PUBLIC, keyring + ) + keyring = _upload_key_from_url( + tmpdir, pulpcore_bindings, monitor_task, KEY_V6_ED25519_PUBLIC, keyring + ) + + keys = pulpcore_bindings.ContentOpenpgpPublickeyApi.list( + repository_version=keyring.latest_version_href + ) + assert keys.count == 3 + + distro = self._create_distribution( + pulpcore_bindings, monitor_task, gen_object_with_cleanup, keyring + ) + + keyid = fingerprint[-16:] + download_url = distribution_base_url(distro.base_url) + f"/{keyid}.pub" + served_key = http_get(download_url) + + served_certs = Cert.split_bytes(served_key) + sig = Sig.from_bytes(sig_bytes.encode() if isinstance(sig_bytes, str) else sig_bytes) + result = verify(bytes=test_data, store=lambda key_ids: served_certs, signature=sig) + assert result.valid_sigs From 59a6beb506f4017e62de83edea33864d32281db7 Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 10:54:58 -0400 Subject: [PATCH 2/7] Be defensive about key structure A public key packet needs to be before any other packets --- pulpcore/app/openpgp.py | 8 ++++++++ 1 file changed, 8 insertions(+) diff --git a/pulpcore/app/openpgp.py b/pulpcore/app/openpgp.py index 7335970bd99..528650b25aa 100644 --- a/pulpcore/app/openpgp.py +++ b/pulpcore/app/openpgp.py @@ -28,6 +28,8 @@ def read_public_key(data): signed_content = public_key elif tag == Tag.PublicSubkey: + if public_key is None: + raise ValueError("Not a public key.") public_subkey = { "raw_data": body, "fingerprint": packet.fingerprint, @@ -38,6 +40,8 @@ def read_public_key(data): public_key["public_subkeys"].append(public_subkey) elif tag == Tag.UserID: + if public_key is None: + raise ValueError("Not a public key.") user_id = { "raw_data": body, "user_id": packet.user_id, @@ -47,6 +51,8 @@ def read_public_key(data): public_key["user_ids"].append(user_id) elif tag == Tag.UserAttribute: + if public_key is None: + raise ValueError("Not a public key.") user_attribute = { "raw_data": body, "sha256": hashlib.sha256(body).hexdigest(), @@ -56,6 +62,8 @@ def read_public_key(data): public_key["user_attributes"].append(user_attribute) elif tag == Tag.Signature: + if signed_content is None: + raise ValueError("Not a public key.") sig_attrs = { "sha256": hashlib.sha256(body).hexdigest(), "signature_type": body[1], From f600fe774af4de87d6f638f9e9a9a1735b93e29a Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 13:34:28 -0400 Subject: [PATCH 3/7] Make filters case-insensitive by default And store fingerprints uppercase consistently in the DB. --- pulpcore/app/openpgp.py | 6 +++--- pulpcore/app/viewsets/openpgp.py | 8 +++++++- 2 files changed, 10 insertions(+), 4 deletions(-) diff --git a/pulpcore/app/openpgp.py b/pulpcore/app/openpgp.py index 528650b25aa..1448f43caf6 100644 --- a/pulpcore/app/openpgp.py +++ b/pulpcore/app/openpgp.py @@ -18,7 +18,7 @@ def read_public_key(data): raise ValueError("Multiple public keys found.") public_key = { "raw_data": body, - "fingerprint": packet.fingerprint, + "fingerprint": packet.fingerprint.upper(), "created": packet.key_created, "user_ids": [], "user_attributes": [], @@ -32,7 +32,7 @@ def read_public_key(data): raise ValueError("Not a public key.") public_subkey = { "raw_data": body, - "fingerprint": packet.fingerprint, + "fingerprint": packet.fingerprint.upper(), "created": packet.key_created, "signatures": [], } @@ -77,7 +77,7 @@ def read_public_key(data): if packet.key_validity_period is not None: sig_attrs["key_expiration_time"] = packet.key_validity_period if packet.issuer_key_id is not None: - sig_attrs["issuer"] = packet.issuer_key_id + sig_attrs["issuer"] = packet.issuer_key_id.upper() if packet.signers_user_id is not None: sig_attrs["signers_user_id"] = packet.signers_user_id signed_content["signatures"].append(sig_attrs) diff --git a/pulpcore/app/viewsets/openpgp.py b/pulpcore/app/viewsets/openpgp.py index 4da6e9887d7..378ea716cf3 100644 --- a/pulpcore/app/viewsets/openpgp.py +++ b/pulpcore/app/viewsets/openpgp.py @@ -1,3 +1,5 @@ +import django_filters + from pulpcore.app import models from pulpcore.app.serializers.openpgp import ( OpenPGPDistributionSerializer, @@ -11,13 +13,14 @@ from pulpcore.app.viewsets.base import NAME_FILTER_OPTIONS, RolesMixin from pulpcore.app.viewsets.content import ContentFilter, ReadOnlyContentViewSet from pulpcore.app.viewsets.publication import DistributionFilter, DistributionViewSet -from pulpcore.app.viewsets.repository import RepositoryViewSet +from pulpcore.app.viewsets.repository import RepositoryVersionViewSet, RepositoryViewSet from pulpcore.plugin.actions import ModifyRepositoryActionMixin from pulpcore.plugin.viewsets import NoArtifactContentUploadViewSet class OpenPGPSignatureFilter(ContentFilter): # Wishlist: filter by expired + issuer = django_filters.CharFilter(lookup_expr="iexact") class Meta: model = models.OpenPGPSignature @@ -37,6 +40,8 @@ class Meta: class OpenPGPPublicSubkeyFilter(ContentFilter): + fingerprint = django_filters.CharFilter(lookup_expr="iexact") + class Meta: model = models.OpenPGPPublicSubkey fields = ["fingerprint"] @@ -44,6 +49,7 @@ class Meta: class OpenPGPPublicKeyFilter(ContentFilter): # Wishlist: filter by user id + fingerprint = django_filters.CharFilter(lookup_expr="iexact") class Meta: model = models.OpenPGPPublicKey From f6bd99f9421a5ac6a962d230de2ac94f1ab4e1ac Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 10:52:27 -0400 Subject: [PATCH 4/7] Add an access policy for OpenPGP keyring distributions Add a viewset for repository versions. Assisted-By: Claude Opus 4.6 --- pulpcore/app/viewsets/__init__.py | 1 + pulpcore/app/viewsets/openpgp.py | 107 +++++++++++++++++++++++++++++- 2 files changed, 106 insertions(+), 2 deletions(-) diff --git a/pulpcore/app/viewsets/__init__.py b/pulpcore/app/viewsets/__init__.py index 379ee75c7fc..430585e1cb3 100644 --- a/pulpcore/app/viewsets/__init__.py +++ b/pulpcore/app/viewsets/__init__.py @@ -88,6 +88,7 @@ from .vulnerability_report import VulnerabilityReportViewSet from .openpgp import ( OpenPGPDistributionViewSet, + OpenPGPKeyringVersionViewSet, OpenPGPKeyringViewSet, OpenPGPPublicKeyViewSet, OpenPGPPublicSubkeyViewSet, diff --git a/pulpcore/app/viewsets/openpgp.py b/pulpcore/app/viewsets/openpgp.py index 378ea716cf3..ff8a164a1cd 100644 --- a/pulpcore/app/viewsets/openpgp.py +++ b/pulpcore/app/viewsets/openpgp.py @@ -181,10 +181,113 @@ class OpenPGPKeyringViewSet(RepositoryViewSet, ModifyRepositoryActionMixin, Role } -class OpenPGPDistributionViewSet(DistributionViewSet): +class OpenPGPKeyringVersionViewSet(RepositoryVersionViewSet): + parent_viewset = OpenPGPKeyringViewSet + + DEFAULT_ACCESS_POLICY = { + "statements": [ + { + "action": ["list", "retrieve"], + "principal": "authenticated", + "effect": "allow", + "condition": "has_repository_model_or_domain_or_obj_perms:core.view_openpgpkeyring", + }, + { + "action": ["destroy"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_repository_model_or_domain_or_obj_perms:core.delete_openpgpkeyring", + "has_repository_model_or_domain_or_obj_perms:core.view_openpgpkeyring", + ], + }, + { + "action": ["repair"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_repository_model_or_domain_or_obj_perms:core.repair_openpgpkeyring", + "has_repository_model_or_domain_or_obj_perms:core.view_openpgpkeyring", + ], + }, + ], + } + + +class OpenPGPDistributionViewSet(DistributionViewSet, RolesMixin): endpoint_name = "openpgp" queryset = models.OpenPGPDistribution.objects.all() serializer_class = OpenPGPDistributionSerializer filterset_class = OpenPGPDistributionFilter + queryset_filtering_required_permission = "core.view_openpgpdistribution" - # DEFAULT_ACCESS_POLICY + DEFAULT_ACCESS_POLICY = { + "statements": [ + { + "action": ["list", "my_permissions"], + "principal": "authenticated", + "effect": "allow", + }, + { + "action": ["retrieve"], + "principal": "authenticated", + "effect": "allow", + "condition": "has_model_or_domain_or_obj_perms:core.view_openpgpdistribution", + }, + { + "action": ["create"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_model_or_domain_perms:core.add_openpgpdistribution", + "has_repo_or_repo_ver_param_model_or_domain_or_obj_perms:" + "core.view_openpgpkeyring", + ], + }, + { + "action": ["update", "partial_update", "set_label", "unset_label"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_model_or_domain_or_obj_perms:core.change_openpgpdistribution", + "has_model_or_domain_or_obj_perms:core.view_openpgpdistribution", + "has_repo_or_repo_ver_param_model_or_domain_or_obj_perms:" + "core.view_openpgpkeyring", + ], + }, + { + "action": ["destroy"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_model_or_domain_or_obj_perms:core.delete_openpgpdistribution", + "has_model_or_domain_or_obj_perms:core.view_openpgpdistribution", + ], + }, + { + "action": ["list_roles", "add_role", "remove_role"], + "principal": "authenticated", + "effect": "allow", + "condition": [ + "has_model_or_domain_or_obj_perms:core.manage_roles_openpgpdistribution", + ], + }, + ], + "creation_hooks": [ + { + "function": "add_roles_for_object_creator", + "parameters": {"roles": "core.openpgpdistribution_owner"}, + }, + ], + "queryset_scoping": {"function": "scope_queryset"}, + } + LOCKED_ROLES = { + "core.openpgpdistribution_creator": ["core.add_openpgpdistribution"], + "core.openpgpdistribution_owner": [ + "core.view_openpgpdistribution", + "core.change_openpgpdistribution", + "core.delete_openpgpdistribution", + "core.manage_roles_openpgpdistribution", + ], + "core.openpgpdistribution_viewer": ["core.view_openpgpdistribution"], + } From 2c768019b2e0f07b5c76f157438db5edeebf5aef Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 17:30:43 -0400 Subject: [PATCH 5/7] Fix OpenPGPSignature.expired returning None instead of bool The `expired` property used short-circuit evaluation (`self.expiration_time and ...`) which returns None when expiration_time is None. The serializer declares `expired` as a non-nullable BooleanField, so None causes a pydantic ValidationError in the generated client bindings. Wrap the expression in bool() so it always returns False when expiration_time is unset. Assisted-By: Claude Opus 4.6 --- pulpcore/app/models/openpgp.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulpcore/app/models/openpgp.py b/pulpcore/app/models/openpgp.py index 9b7f3ec5442..8b459cffa26 100644 --- a/pulpcore/app/models/openpgp.py +++ b/pulpcore/app/models/openpgp.py @@ -143,7 +143,7 @@ class OpenPGPSignature(_OpenPGPContent): @property def expired(self): - return self.expiration_time and timezone.now() > self.created + self.expiration_time + return bool(self.expiration_time and timezone.now() > self.created + self.expiration_time) @property def key_expired(self): From 76a77314258822b4a156f6c96964b0d331025143 Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 17:30:54 -0400 Subject: [PATCH 6/7] Fix get_viewset_for_model when multiple viewsets share a model MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When an app registers multiple NamedModelViewSet subclasses for the same model (e.g. RepositoryVersionViewSet and OpenPGPKeyringVersionViewSet both use RepositoryVersion), the lookup previously skipped the model entirely, causing LookupError in the import/export system. Resolve the ambiguity by preferring the base class in the inheritance hierarchy. Subclass viewsets still participate in URL routing and provide their own access policies — they just aren't the canonical answer for model-to-viewset reverse lookups. Assisted-By: Claude Opus 4.6 --- pulpcore/app/util.py | 19 +++++++++++++------ 1 file changed, 13 insertions(+), 6 deletions(-) diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 76f2a5b47fb..e7cb92dbec4 100644 --- a/pulpcore/app/util.py +++ b/pulpcore/app/util.py @@ -260,14 +260,21 @@ def get_viewset_for_model(model_obj, ignore_error=False): # go through the viewset registry to find the viewset for the passed-in model for app in pulp_plugin_configs(): for model, viewsets in app.named_viewsets.items(): - # There may be multiple viewsets for a model. In this - # case, we can't reverse the mapping. if len(viewsets) == 1: viewset = viewsets[0] - _model_viewset_cache.setdefault(model, viewset) - if model is model_class: - model_viewset = viewset - break + else: + # Multiple viewsets for the same model (e.g. RepositoryVersionViewSet + # and OpenPGPKeyringVersionViewSet both use RepositoryVersion). + # Prefer the base class — subclass viewsets exist for URL routing + # and custom access policies, not for model-to-viewset reverse lookups. + bases = [vs for vs in viewsets if all(issubclass(other, vs) for other in viewsets)] + if len(bases) != 1: + continue + viewset = bases[0] + _model_viewset_cache.setdefault(model, viewset) + if model is model_class: + model_viewset = viewset + break if model_viewset is not None: break From 60c2214cfd79dc9dd34d6bec0d3678b0bd23a027 Mon Sep 17 00:00:00 2001 From: Daniel Alley Date: Fri, 7 Aug 2026 17:50:16 -0400 Subject: [PATCH 7/7] Expose signature_type in OpenPGP signature serializer The field exists on the model but was missing from the serializer's fields tuple, causing an AttributeError in the client bindings when tests accessed it. Assisted-By: Claude Opus 4.6 --- pulpcore/app/serializers/openpgp.py | 1 + 1 file changed, 1 insertion(+) diff --git a/pulpcore/app/serializers/openpgp.py b/pulpcore/app/serializers/openpgp.py index 7858149f97c..25b61913013 100644 --- a/pulpcore/app/serializers/openpgp.py +++ b/pulpcore/app/serializers/openpgp.py @@ -21,6 +21,7 @@ class NestedOpenPGPSignatureSerializer(NoArtifactContentSerializer): class Meta: model = models.OpenPGPSignature fields = ( + "signature_type", "issuer", "created", "expiration_time",