Improve OpenPGP keyring test coverage

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",
+                    )
+                ],
+            },
+        ),
+    ]

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"),
         ]

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,16 +28,20 @@ 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": [],
             }
             signed_content = public_subkey
             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)

pulpcore/app/serializers/openpgp.py

@@ -21,6 +21,7 @@ class NestedOpenPGPSignatureSerializer(NoArtifactContentSerializer):
     class Meta:
         model = models.OpenPGPSignature
         fields = (
+            "signature_type",
             "issuer",
             "created",
             "expiration_time",

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
 

pulpcore/app/viewsets/__init__.py

@@ -88,6 +88,7 @@
 from .vulnerability_report import VulnerabilityReportViewSet
 from .openpgp import (
     OpenPGPDistributionViewSet,
+    OpenPGPKeyringVersionViewSet,
     OpenPGPKeyringViewSet,
     OpenPGPPublicKeyViewSet,
     OpenPGPPublicSubkeyViewSet,

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,13 +40,16 @@ class Meta:
 
 
 class OpenPGPPublicSubkeyFilter(ContentFilter):
+    fingerprint = django_filters.CharFilter(lookup_expr="iexact")
+
     class Meta:
         model = models.OpenPGPPublicSubkey
         fields = ["fingerprint"]
 
 
 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"],
+    }