Skip to content

Commit d3421ec

Browse files
hosted-pulp-botAlcovedecko
authored
Fix #1101: Filter domain list by user's DomainOrg permissions (#1166)
* Fix domain list filtering by DomainOrg permissions - Add scope_queryset method to DomainBasedPermission class - Filter Domain querysets based on DomainOrg associations (user, group, org_id) - Superusers see all domains, unauthenticated users see none - Update existing test_user_list_domain_permissions to verify filtering - Add 10 comprehensive test cases covering user isolation, org-based visibility, group-based access, superuser permissions, default domain handling, deduplication, basic auth, and model guard functionality Fixes #1101 * fix: address review findings in domain list filtering - Replace `if group_pks:` with `user.groups.exists()` in both `_has_domain_access` and `scope_queryset` to avoid evaluating a lazy QuerySet in boolean context. - Add `domain_org_factory` fixture for creating DomainOrg entries with cleanup, following the `domain_factory` pattern. - Rewrite `test_group_based_domain_visibility` and `test_domain_deduplication` to use API fixtures (`gen_group`, `domain_factory`, `GroupsUsersApi`) instead of direct ORM writes that are invisible to the API server. - Remove unused fixture parameters (`gen_user`, `django_user_model`) and replace Mock with SimpleNamespace in `test_scope_queryset_model_guard`. * style: fix ruff format and import sorting in review fixes * style: fix all ruff violations in test_domain_based_permissions - Sort imports (I001) - Remove unused variable assignments (F841) - Rename test_user_permissions_without_orgId to snake_case (N802) - Add noqa for unused monitor_task fixture param (ARG001) - Reformat to ruff line-length 120 * style: fix import grouping in test_scope_queryset_model_guard Separate third-party (django) from first-party (pulp_service) imports per ruff I001 with project config. * fix: add transactional_db to domain_org_factory fixture The fixture performs ORM writes that must be visible to the API server running in a separate process. Using transactional_db ensures the changes are committed rather than wrapped in a rolled-back transaction. * fix: use API-only approach for group and dedup tests Remove domain_org_factory and all direct ORM access from tests. Instead, rely on the post_save signal that auto-creates DomainOrg entries (with group and org_id) when a domain is created via the API. This eliminates cross-process DB visibility issues entirely. * perf: avoid double query on user.groups in permission checks Use group_pks.exists() instead of user.groups.exists() to reuse the same queryset rather than issuing two separate queries against user.groups. * perf: materialize group_pks with list() to avoid extra query Use list() to fetch group PKs in a single query. The if check is then a Python truthiness test on a list, avoiding the extra EXISTS query that group_pks.exists() would issue. * fix: use basic auth for superuser domain list test The superuser test was using anonymous_user + x-rh-identity header, which authenticates via RHTermsBasedRegistryAuthentication and does not set is_superuser=True on the user object. Use basic auth instead so the request authenticates as the actual Django superuser and scope_queryset correctly bypasses filtering. * fix: restore Domain import needed by scope_queryset The upstream ruff cleanup removed the Domain import since the original code didn't reference it directly. But scope_queryset uses qs.model is not Domain, so the import is required. Co-authored-by: Alcove <alcove@localhost> Co-authored-by: André "decko" de Brito <decko@redhat.com>
1 parent ddb0a0b commit d3421ec

2 files changed

Lines changed: 509 additions & 18 deletions

File tree

pulp_service/pulp_service/app/authorization.py

Lines changed: 32 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
from django.db.models import Q
99
from rest_framework.permissions import SAFE_METHODS, BasePermission
1010

11+
from pulpcore.plugin.models import Domain
1112
from pulpcore.plugin.util import extract_pk, get_domain_pk
1213

1314
from pulp_service.app.models import DomainOrg
@@ -31,8 +32,7 @@ def _has_domain_access(self, domain_pk, org_id, user):
3132
"""
3233
query = Q(domains__pk=domain_pk, user=user)
3334

34-
# Fetch group ids once to avoid multiple DB hits
35-
group_pks = user.groups.values_list("pk", flat=True)
35+
group_pks = list(user.groups.values_list("pk", flat=True))
3636
if group_pks:
3737
query |= Q(domains__pk=domain_pk, group_id__in=group_pks)
3838

@@ -123,6 +123,36 @@ def get_org_id(self, decoded_header_content):
123123
return None
124124
return None
125125

126+
def scope_queryset(self, view, qs):
127+
"""
128+
Filter Domain querysets to only show domains the user has DomainOrg access to.
129+
"""
130+
if qs.model is not Domain:
131+
return qs
132+
133+
request = view.request
134+
user = request.user
135+
136+
if user.is_superuser:
137+
return qs
138+
139+
if not user.is_authenticated:
140+
return qs.none()
141+
142+
decoded_header = self.get_decoded_identity_header(request)
143+
org_id = self.get_org_id(decoded_header)
144+
145+
query = Q(domain_orgs__user=user)
146+
147+
group_pks = list(user.groups.values_list("pk", flat=True))
148+
if group_pks:
149+
query |= Q(domain_orgs__group_id__in=group_pks)
150+
151+
if org_id is not None:
152+
query |= Q(domain_orgs__org_id=org_id)
153+
154+
return qs.filter(query).distinct()
155+
126156

127157
class AllowUnauthPull(BasePermission):
128158
"""

0 commit comments

Comments
 (0)