feat: Add domain-admin content list endpoint

CHANGES/+domain-admin-content.feature

@@ -0,0 +1 @@
+Added a domain-scoped content list endpoint (`/pulp/api/v3/content/domains/`) that lets domain administrators list all content in a domain, including content not in any repository.

pulpcore/app/viewsets/__init__.py

@@ -20,6 +20,7 @@
     ArtifactFilter,
     ArtifactViewSet,
     ContentFilter,
+    ContentDomainViewSet,
     ContentViewSet,
     ListContentViewSet,
     ReadOnlyContentViewSet,

pulpcore/app/viewsets/content.py

@@ -206,6 +206,48 @@ def routable(cls):
         return True
 
 
+class ContentDomainViewSet(BaseContentViewSet, mixins.ListModelMixin):
+    """Endpoint for domain administrators to list all content in a domain.
+
+    Unlike ListContentViewSet, which scopes content to the repositories a user can
+    see, this endpoint returns every content unit in the current domain, including
+    content that is not in any repository. It is restricted to domain administrators
+    (users holding domain-level core.view_content). Filtering, including
+    pulp_label_select, works the same as the standard content list.
+    """
+
+    endpoint_name = "content/domains"
+
+    DEFAULT_ACCESS_POLICY = {
+        "statements": [
+            {
+                "action": ["list"],
+                "principal": "authenticated",
+                "effect": "allow",
+                "condition": "has_model_or_domain_perms:core.view_content",
+            },
+        ],
+        "queryset_scoping": {"function": "scope_queryset"},
+    }
+    LOCKED_ROLES = {
+        "core.content_domain_viewer": ["core.view_content"],
+    }
+
+    @classmethod
+    def routable(cls):
+        """Do not hide from the routers."""
+        return True
+
+    def scope_queryset(self, qs):
+        """Return all content in the current domain.
+
+        Repository-based scoping (BaseContentViewSet.scope_queryset) is intentionally
+        bypassed: the access policy already restricts this endpoint to domain
+        administrators, and get_queryset has filtered to the request's domain.
+        """
+        return qs
+
+
 class ContentViewSet(
     BaseContentViewSet,
     mixins.CreateModelMixin,

pulpcore/tests/functional/api/test_domain_admin_content.py

@@ -0,0 +1,90 @@
+import os
+import uuid
+
+import pytest
+from django.conf import settings
+
+pytestmark = pytest.mark.skipif(not settings.DOMAIN_ENABLED, reason="Domains not enabled.")
+
+
+@pytest.mark.parallel
+def test_domain_admin_lists_all_domain_content(
+    pulpcore_bindings,
+    file_bindings,
+    domain_factory,
+    gen_user,
+    monitor_task,
+    tmp_path,
+):
+    domain = domain_factory()
+
+    # Orphan content (not in any repository) uploaded into the domain, with a label.
+    temp_file = tmp_path / str(uuid.uuid4())
+    temp_file.write_bytes(os.urandom(128))
+    created = monitor_task(
+        file_bindings.ContentFilesApi.create(
+            relative_path="a.txt",
+            file=str(temp_file),
+            pulp_domain=domain.name,
+            pulp_labels={"key_a": "value_a"},
+        ).task
+    ).created_resources
+    href = next(h for h in created if "/content/" in h)
+
+    # A domain-scoped content admin (NOT a repository owner, NOT a superuser).
+    user = gen_user(domain_roles=[("core.content_domain_viewer", domain.pulp_href)])
+
+    with user:
+        # Sees the orphan content via the domain-admin endpoint.
+        result = pulpcore_bindings.ContentDomainsApi.list(pulp_domain=domain.name)
+        assert result.count == 1
+        assert result.results[0].pulp_href == href
+
+        # Label filter works.
+        filtered = pulpcore_bindings.ContentDomainsApi.list(
+            pulp_domain=domain.name, pulp_label_select="key_a=value_a"
+        )
+        assert filtered.count == 1
+        no_match = pulpcore_bindings.ContentDomainsApi.list(
+            pulp_domain=domain.name, pulp_label_select="key_a=nope"
+        )
+        assert no_match.count == 0
+
+        # Contrast: the repository-scoped endpoint hides the orphan content.
+        repo_scoped = pulpcore_bindings.ContentApi.list(pulp_domain=domain.name)
+        assert repo_scoped.count == 0
+
+
+@pytest.mark.parallel
+def test_non_admin_denied(
+    pulpcore_bindings,
+    domain_factory,
+    gen_user,
+):
+    domain = domain_factory()
+    user = gen_user()  # no roles
+
+    with user:
+        with pytest.raises(pulpcore_bindings.ApiException) as ctx:
+            pulpcore_bindings.ContentDomainsApi.list(pulp_domain=domain.name)
+        assert ctx.value.status == 403
+
+
+@pytest.mark.parallel
+def test_admin_of_one_domain_denied_in_another(
+    pulpcore_bindings,
+    domain_factory,
+    gen_user,
+):
+    domain_a = domain_factory()
+    domain_b = domain_factory()
+    # Domain admin for domain A only.
+    user = gen_user(domain_roles=[("core.content_domain_viewer", domain_a.pulp_href)])
+
+    with user:
+        # Allowed in the domain they administer.
+        pulpcore_bindings.ContentDomainsApi.list(pulp_domain=domain_a.name)
+        # Denied in a domain they do not administer.
+        with pytest.raises(pulpcore_bindings.ApiException) as ctx:
+            pulpcore_bindings.ContentDomainsApi.list(pulp_domain=domain_b.name)
+        assert ctx.value.status == 403

pulpcore/tests/unit/test_content_domain_viewset.py

@@ -0,0 +1,30 @@
+from pulpcore.app.viewsets.content import ContentDomainViewSet
+
+
+def test_endpoint_urlpattern():
+    assert ContentDomainViewSet.urlpattern() == "content/domains"
+    assert ContentDomainViewSet.routable() is True
+
+
+def test_access_policy_gates_on_domain_perms():
+    statements = ContentDomainViewSet.DEFAULT_ACCESS_POLICY["statements"]
+    assert statements == [
+        {
+            "action": ["list"],
+            "principal": "authenticated",
+            "effect": "allow",
+            "condition": "has_model_or_domain_perms:core.view_content",
+        }
+    ]
+
+
+def test_locked_role_grants_view_content():
+    assert ContentDomainViewSet.LOCKED_ROLES == {
+        "core.content_domain_viewer": ["core.view_content"],
+    }
+
+
+def test_scope_queryset_is_identity():
+    sentinel = object()
+    # scope_queryset must not touch the queryset (repository scoping is bypassed)
+    assert ContentDomainViewSet.scope_queryset(ContentDomainViewSet(), sentinel) is sentinel