diff --git a/CHANGES/+domain-admin-content.feature b/CHANGES/+domain-admin-content.feature new file mode 100644 index 00000000000..83c87d88542 --- /dev/null +++ b/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. diff --git a/pulpcore/app/viewsets/__init__.py b/pulpcore/app/viewsets/__init__.py index 379ee75c7fc..9ed92f48f97 100644 --- a/pulpcore/app/viewsets/__init__.py +++ b/pulpcore/app/viewsets/__init__.py @@ -20,6 +20,7 @@ ArtifactFilter, ArtifactViewSet, ContentFilter, + ContentDomainViewSet, ContentViewSet, ListContentViewSet, ReadOnlyContentViewSet, diff --git a/pulpcore/app/viewsets/content.py b/pulpcore/app/viewsets/content.py index a23ad0e13dd..4d08445140b 100644 --- a/pulpcore/app/viewsets/content.py +++ b/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, diff --git a/pulpcore/tests/functional/api/test_domain_admin_content.py b/pulpcore/tests/functional/api/test_domain_admin_content.py new file mode 100644 index 00000000000..11b4763e680 --- /dev/null +++ b/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 diff --git a/pulpcore/tests/unit/test_content_domain_viewset.py b/pulpcore/tests/unit/test_content_domain_viewset.py new file mode 100644 index 00000000000..5f116aea625 --- /dev/null +++ b/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