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 0000000000..261b532e41 --- /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 3c96e03e8f..8b459cffa2 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 @@ -140,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): @@ -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/app/openpgp.py b/pulpcore/app/openpgp.py index 7335970bd9..1448f43caf 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": [], @@ -28,9 +28,11 @@ 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, + "fingerprint": packet.fingerprint.upper(), "created": packet.key_created, "signatures": [], } @@ -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], @@ -69,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/serializers/openpgp.py b/pulpcore/app/serializers/openpgp.py index 7858149f97..25b6191301 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", diff --git a/pulpcore/app/util.py b/pulpcore/app/util.py index 76f2a5b47f..e7cb92dbec 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 diff --git a/pulpcore/app/viewsets/__init__.py b/pulpcore/app/viewsets/__init__.py index 379ee75c7f..430585e1cb 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 4da6e9887d..ff8a164a1c 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 @@ -175,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"], + } diff --git a/pulpcore/tests/functional/api/test_openpgp.py b/pulpcore/tests/functional/api/test_openpgp.py index 3adbc93076..b014d13e63 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