diff --git a/pulp_service/pulp_service/app/authorization.py b/pulp_service/pulp_service/app/authorization.py index 6c837f9e..8fa8f71f 100644 --- a/pulp_service/pulp_service/app/authorization.py +++ b/pulp_service/pulp_service/app/authorization.py @@ -6,12 +6,13 @@ import jq from django.db.models import Q +from django.http import Http404 from rest_framework.permissions import SAFE_METHODS, BasePermission from pulpcore.plugin.models import Domain from pulpcore.plugin.util import extract_pk, get_domain_pk -from pulp_service.app.models import DomainOrg, FeatureContentGuard +from pulp_service.app.models import DomainOrg _logger = logging.getLogger(__name__) org_id_var = ContextVar("org_id") @@ -20,19 +21,15 @@ user_id_var = ContextVar("user_id") group_var = ContextVar("group") -# The domain whose PyPI views require the lightwell-network feature entitlement (see -# has_permission()). Other domains' PyPI views keep the pre-existing "any SAFE_METHOD -# request is allowed" behavior. +# The hardcoded domain name used by the lightwell read-only group check (see +# _has_lightwell_readonly_group_access and _check_safe_method_access). LIGHTWELL_DOMAIN_NAME = "lightwell" -# Feature entitlement required to read the lightwell domain's PyPI views (simple API, etc.) -# for orgs that don't have a DomainOrg association with the domain. See has_permission(). -LIGHTWELL_NETWORK_FEATURE = "lightwell-network" # Hardcoded group whose members get read-only (SAFE_METHODS) access to the lightwell # domain's non-PyPI endpoints (repository/content/artifact listing, Pulp REST API, Maven # content API), independent of any DomainOrg association. This is a plain Group -- there is # no DomainOrg row backing this access, unlike the normal group-based access model. It does -# NOT grant write access, and it does NOT apply to PyPI views, which are still gated -# exclusively by the lightwell-network feature check (see _check_pypi_safe_method_access()). +# NOT grant write access, and it does NOT apply to PyPI views, which are gated +# by the distribution's content guard (see _check_pypi_safe_method_access()). LIGHTWELL_READONLY_GROUP_NAME = "Lightwell-ReadOnly" @@ -56,52 +53,6 @@ def _has_domain_access(self, domain_pk, org_id, user): return DomainOrg.objects.filter(query).exists() - def _has_pypi_read_access(self, request, domain): - """ - Checks SAFE_METHOD access to the lightwell domain's PyPI views. - - Users with a DomainOrg association bypass the feature check entirely (existing - permission model). Everyone else -- including unauthenticated users -- must belong - to an org that has the lightwell-network feature entitlement. - """ - user = request.user - domain_pk = domain.pk if domain is not None else get_domain_pk() - - decoded_header_content = self.get_decoded_identity_header(request) - org_id = self.get_org_id(decoded_header_content) - - if user.is_authenticated and self._has_domain_access(domain_pk, org_id, user): - _logger.info("lightwell PyPI access GRANTED via DomainOrg: user=%s org_id=%s", user, org_id) - return True - - if org_id is None: - _logger.info( - "lightwell PyPI access DENIED: no org_id in identity header and no DomainOrg match (user=%s)", - user, - ) - return False - - allowed = self._has_lightwell_network_feature(org_id) - _logger.info( - "lightwell PyPI access %s via %s feature check: org_id=%s user=%s", - "GRANTED" if allowed else "DENIED", - LIGHTWELL_NETWORK_FEATURE, - org_id, - user, - ) - return allowed - - def _has_lightwell_network_feature(self, org_id): - """ - Checks (with caching) whether `org_id` has the lightwell-network feature, via the - same cache/lookup mechanism as FeatureContentGuard. - """ - guard = FeatureContentGuard(features=[LIGHTWELL_NETWORK_FEATURE], pulp_domain=None) - try: - return guard.check_feature(org_id) - except PermissionError: - return False - def _has_lightwell_readonly_group_access(self, user): """ Members of the hardcoded LIGHTWELL_READONLY_GROUP_NAME group get read-only @@ -113,25 +64,71 @@ def _has_lightwell_readonly_group_access(self, user): """ return user.is_authenticated and user.groups.filter(name=LIGHTWELL_READONLY_GROUP_NAME).exists() - def _check_pypi_safe_method_access(self, request, view, domain): + def _check_pypi_safe_method_access(self, request, view, domain): # noqa: PLR0911 """ Returns True/False for a SAFE_METHOD request to a PyPI `view`, or None if `view` isn't a PyPI view (the caller should then fall through to the standard DomainOrg-based checks below). - Only the lightwell domain's PyPI views require a DomainOrg association or the - lightwell-network feature entitlement. Other domains' PyPI views keep the - pre-existing "any SAFE_METHOD request is allowed" behavior; non-PyPI endpoints - (Maven, repository listing, Pulp REST API, ...) never hit this method. + If the distribution has a content guard, access is gated by guard.cast().permit(): + users with a DomainOrg association bypass the guard (domain owner privilege); + everyone else must satisfy the guard. Distributions without a content guard allow + all SAFE_METHOD access (the pre-existing default behavior). """ from pulp_python.app.pypi.views import PyPIMixin if not isinstance(view, PyPIMixin): return None - if domain and domain.name == LIGHTWELL_DOMAIN_NAME: - return self._has_pypi_read_access(request, domain) - return True + try: + distribution = view.distribution + except Http404: + return True + except Exception: + _logger.exception("Unexpected error resolving distribution for PyPI permission check") + return False + + guard = distribution.content_guard + if not guard: + return True + + user = request.user + domain_pk = domain.pk if domain is not None else get_domain_pk() + decoded_header = self.get_decoded_identity_header(request) + org_id = self.get_org_id(decoded_header) + + if user.is_authenticated and self._has_domain_access(domain_pk, org_id, user): + _logger.info( + "Content-guarded PyPI access GRANTED via DomainOrg: user=%s org_id=%s", + user, + org_id, + ) + return True + + try: + casted_guard = guard.cast() + except Exception: + _logger.exception("Failed to resolve content guard type for distribution") + return False + + try: + casted_guard.permit(request) + _logger.info( + "Content-guarded PyPI access GRANTED via content guard: org_id=%s user=%s", + org_id, + user, + ) + return True + except PermissionError: + _logger.info( + "Content-guarded PyPI access DENIED via content guard: org_id=%s user=%s", + org_id, + user, + ) + return False + except Exception: + _logger.exception("Unexpected error evaluating content guard permit") + return False def _check_safe_method_access(self, request, view, domain, user): """ diff --git a/pulp_service/pulp_service/app/urls.py b/pulp_service/pulp_service/app/urls.py index 61fc48f4..4e66f29e 100644 --- a/pulp_service/pulp_service/app/urls.py +++ b/pulp_service/pulp_service/app/urls.py @@ -3,11 +3,11 @@ from .admin import admin_site from .viewsets import ( CreateDomainView, - MigrateDomainView, DatabaseTriggersView, DebugAuthenticationHeadersView, InternalServerErrorCheck, InternalServerErrorCheckWithException, + MigrateDomainView, OOMKillTriggerView, RDSConnectionTestDispatcherView, RedirectCheck, diff --git a/pulp_service/pulp_service/tests/functional/test_authentication.py b/pulp_service/pulp_service/tests/functional/test_authentication.py index e3e3bee2..88309ec7 100644 --- a/pulp_service/pulp_service/tests/functional/test_authentication.py +++ b/pulp_service/pulp_service/tests/functional/test_authentication.py @@ -122,7 +122,7 @@ def test_get_requests_without_auth_to_simple_api( ): """Test that all domains (other than "lightwell") allow GET requests without authentication but block other methods. The "lightwell" domain is excluded from this - behavior -- see test_lightwell_feature_permission.py. + behavior -- see test_content_guard_permission.py. """ # Create a user with credentials to set up the domain setup_user = { diff --git a/pulp_service/pulp_service/tests/functional/test_lightwell_feature_permission.py b/pulp_service/pulp_service/tests/functional/test_content_guard_permission.py similarity index 54% rename from pulp_service/pulp_service/tests/functional/test_lightwell_feature_permission.py rename to pulp_service/pulp_service/tests/functional/test_content_guard_permission.py index b445f4bd..289b0c9b 100644 --- a/pulp_service/pulp_service/tests/functional/test_lightwell_feature_permission.py +++ b/pulp_service/pulp_service/tests/functional/test_content_guard_permission.py @@ -1,17 +1,15 @@ """ -Functional tests for the lightwell-network feature check enforced by DomainBasedPermission, -scoped specifically to the "lightwell" domain's PyPI views (simple API, package metadata, -etc.). Other domains' PyPI views, and all non-PyPI endpoints (even within the lightwell -domain), keep the pre-existing permission model unaffected by this feature check. +Functional tests for the content-guard-driven access check enforced by DomainBasedPermission +on PyPI views. When a PyPI distribution has a content guard, SAFE_METHOD access is gated +by guard.cast().permit(request). Distributions without content guards keep the pre-existing +open-access behavior. These follow the pattern used in test_feature_service.py: they exercise the real Features Service (no mocking) using known staging accounts. Org LIGHTWELL_ENTITLED_ORG_ID has the lightwell-network feature; org LIGHTWELL_NOT_ENTITLED_ORG_ID does not. -NOTE: the feature check is keyed off the literal domain name "lightwell" (see -pulp_service.app.authorization.LIGHTWELL_DOMAIN_NAME), so the domain created here can't use -a random per-test suffix like most other functional tests in this suite. These tests assume -they run against an ephemeral Pulp instance where no "lightwell" domain already exists. +The check is domain-name-agnostic: any domain's distributions can be protected by configuring +a content guard, with no code changes. """ import json @@ -24,11 +22,10 @@ from pulp_service.tests.functional.constants import ( LIGHTWELL_ENTITLED_ORG_ID, + LIGHTWELL_NETWORK_FEATURE, LIGHTWELL_NOT_ENTITLED_ORG_ID, ) -LIGHTWELL_DOMAIN_NAME = "lightwell" - # An org with no DomainOrg association with the test domains and no lightwell-network # feature entitlement; only used to own the test domain/repo/distribution. DOMAIN_OWNER_ORG_ID = "555555555" @@ -49,19 +46,22 @@ def _identity_header(org_id, username): def configure_pypi_distribution( anonymous_user, gen_object_with_cleanup, + add_to_cleanup, pulpcore_bindings, python_bindings, + service_content_guards_api_client, bindings_cfg, ): """ Creates a domain owned by DOMAIN_OWNER_ORG_ID, with a Python repository and a PyPI distribution. + Optionally assigns a FeatureContentGuard with the given features. Returns a (domain_name, pypi_simple_url, repos_url, owner_header) tuple. """ owner_header = _identity_header(DOMAIN_OWNER_ORG_ID, "lightwell-test-owner") - def _configure(domain_name): + def _configure(domain_name, features=None): with anonymous_user: pulpcore_bindings.DomainsApi.api_client.default_headers["x-rh-identity"] = owner_header @@ -79,14 +79,31 @@ def _configure(domain_name): python_bindings.RepositoriesPythonApi, {"name": str(uuid4())}, pulp_domain=domain_name ) + distro_params = {"name": str(uuid4()), "base_path": str(uuid4()), "repository": repo.pulp_href} + if features is not None: + from pulpcore.client.pulp_service import ServiceFeatureContentGuard + + service_content_guards_api_client.api_client.default_headers["x-rh-identity"] = owner_header + guard = service_content_guards_api_client.create( + service_feature_content_guard=ServiceFeatureContentGuard( + name=f"guard-{uuid4()}", + header_name="x-rh-identity", + features=features, + jq_filter=".identity.org_id", + ), + pulp_domain=domain_name, + ) + add_to_cleanup(service_content_guards_api_client, guard.pulp_href) + distro_params["content_guard"] = guard.pulp_href + python_bindings.DistributionsPypiApi.api_client.default_headers["x-rh-identity"] = owner_header - base_path = str(uuid4()) gen_object_with_cleanup( python_bindings.DistributionsPypiApi, - {"name": str(uuid4()), "base_path": base_path, "repository": repo.pulp_href}, + distro_params, pulp_domain=domain_name, ) + base_path = distro_params["base_path"] pypi_url = urljoin(bindings_cfg.host, f"/api/pypi/{domain_name}/{base_path}/simple/") repos_url = urljoin(bindings_cfg.host, f"/api/pulp/{domain_name}/api/v3/repositories/python/python/") return domain_name, pypi_url, repos_url, owner_header @@ -96,23 +113,25 @@ def _configure(domain_name): pulpcore_bindings.DomainsApi.api_client.default_headers.pop("x-rh-identity", None) python_bindings.RepositoriesPythonApi.api_client.default_headers.pop("x-rh-identity", None) python_bindings.DistributionsPypiApi.api_client.default_headers.pop("x-rh-identity", None) + service_content_guards_api_client.api_client.default_headers.pop("x-rh-identity", None) @pytest.fixture -def configure_lightwell_pypi_distribution(configure_pypi_distribution): - """Same as configure_pypi_distribution, but uses the literal "lightwell" domain name that - DomainBasedPermission gates behind the lightwell-network feature.""" +def configure_guarded_pypi_distribution(configure_pypi_distribution): + """Creates a domain with a FeatureContentGuard requiring the lightwell-network feature + on its PyPI distribution. Uses a unique domain name.""" def _configure(): - return configure_pypi_distribution(LIGHTWELL_DOMAIN_NAME) + domain_name = f"guarded-{uuid4()}" + return configure_pypi_distribution(domain_name, features=[LIGHTWELL_NETWORK_FEATURE]) return _configure -def test_org_without_feature_denied_on_lightwell_pypi_simple_api(configure_lightwell_pypi_distribution): - """A user whose org doesn't have the lightwell-network feature and has no DomainOrg - association gets 403 on the lightwell domain's PyPI simple API.""" - _, pypi_url, _, _ = configure_lightwell_pypi_distribution() +def test_org_without_feature_denied_on_guarded_pypi_simple_api(configure_guarded_pypi_distribution): + """A user whose org doesn't have the required feature and has no DomainOrg + association gets 403 on a content-guarded PyPI simple API.""" + _, pypi_url, _, _ = configure_guarded_pypi_distribution() headers = {"x-rh-identity": _identity_header(LIGHTWELL_NOT_ENTITLED_ORG_ID, "not-entitled-user")} response = requests.get(pypi_url, headers=headers, timeout=30) @@ -120,10 +139,10 @@ def test_org_without_feature_denied_on_lightwell_pypi_simple_api(configure_light assert response.status_code == 403 -def test_org_with_feature_allowed_on_lightwell_pypi_simple_api(configure_lightwell_pypi_distribution): - """A user whose org has the lightwell-network feature can read the lightwell domain's - PyPI simple API, even without a DomainOrg association.""" - _, pypi_url, _, _ = configure_lightwell_pypi_distribution() +def test_org_with_feature_allowed_on_guarded_pypi_simple_api(configure_guarded_pypi_distribution): + """A user whose org has the required feature can read a content-guarded PyPI + simple API, even without a DomainOrg association.""" + _, pypi_url, _, _ = configure_guarded_pypi_distribution() headers = {"x-rh-identity": _identity_header(LIGHTWELL_ENTITLED_ORG_ID, "entitled-user")} response = requests.get(pypi_url, headers=headers, timeout=30) @@ -131,10 +150,10 @@ def test_org_with_feature_allowed_on_lightwell_pypi_simple_api(configure_lightwe assert response.status_code == 200 -def test_domain_org_association_bypasses_feature_check(configure_lightwell_pypi_distribution): - """The domain owner (has a DomainOrg association) can read the lightwell domain's PyPI - simple API regardless of the lightwell-network feature.""" - _, pypi_url, _, owner_header = configure_lightwell_pypi_distribution() +def test_domain_org_association_bypasses_content_guard(configure_guarded_pypi_distribution): + """The domain owner (has a DomainOrg association) can read the guarded PyPI + simple API regardless of the content guard.""" + _, pypi_url, _, owner_header = configure_guarded_pypi_distribution() headers = {"x-rh-identity": owner_header} response = requests.get(pypi_url, headers=headers, timeout=30) @@ -142,21 +161,19 @@ def test_domain_org_association_bypasses_feature_check(configure_lightwell_pypi_ assert response.status_code == 200 -def test_unauthenticated_denied_on_lightwell_pypi_simple_api(configure_lightwell_pypi_distribution): - """Without any identity at all (no org_id to check a feature for), the lightwell domain's - PyPI simple API must not be readable.""" - _, pypi_url, _, _ = configure_lightwell_pypi_distribution() +def test_unauthenticated_denied_on_guarded_pypi_simple_api(configure_guarded_pypi_distribution): + """Without any identity at all, a content-guarded PyPI simple API must not be readable.""" + _, pypi_url, _, _ = configure_guarded_pypi_distribution() response = requests.get(pypi_url, timeout=30) assert response.status_code in (401, 403) -def test_write_operations_unaffected_by_feature_check(configure_lightwell_pypi_distribution): - """The lightwell-network feature only grants read access -- an entitled org with no - DomainOrg association must still be denied write access to the lightwell domain's PyPI - views.""" - _, pypi_url, _, _ = configure_lightwell_pypi_distribution() +def test_write_operations_unaffected_by_content_guard(configure_guarded_pypi_distribution): + """The content guard only gates SAFE_METHOD access -- an entitled org with no + DomainOrg association must still be denied write access.""" + _, pypi_url, _, _ = configure_guarded_pypi_distribution() headers = {"x-rh-identity": _identity_header(LIGHTWELL_ENTITLED_ORG_ID, "entitled-write-user")} response = requests.post(pypi_url, headers=headers, data={}, timeout=30) @@ -164,11 +181,11 @@ def test_write_operations_unaffected_by_feature_check(configure_lightwell_pypi_d assert response.status_code in (401, 403) -def test_non_pypi_endpoints_unaffected_by_feature_check(configure_lightwell_pypi_distribution): +def test_non_pypi_endpoints_unaffected_by_content_guard(configure_guarded_pypi_distribution): """Non-PyPI endpoints (here, the Pulp REST API's repository listing) must keep using the - existing DomainOrg-based permission model, even within the lightwell domain: the - lightwell-network feature grants no access to them.""" - _, _, repos_url, _ = configure_lightwell_pypi_distribution() + existing DomainOrg-based permission model: the content guard on a PyPI distribution + grants no access to non-PyPI endpoints.""" + _, _, repos_url, _ = configure_guarded_pypi_distribution() headers = {"x-rh-identity": _identity_header(LIGHTWELL_ENTITLED_ORG_ID, "entitled-rest-user")} response = requests.get(repos_url, headers=headers, timeout=30) @@ -176,11 +193,10 @@ def test_non_pypi_endpoints_unaffected_by_feature_check(configure_lightwell_pypi assert response.status_code == 403 -def test_other_domains_pypi_simple_api_unaffected_by_feature_check(configure_pypi_distribution): - """Domains other than "lightwell" must keep the pre-existing behavior: any SAFE_METHOD - request -- even unauthenticated -- can read the PyPI simple API, regardless of the - lightwell-network feature.""" - domain_name = f"not-lightwell-{uuid4()}" +def test_unguarded_distribution_allows_unauthenticated_access(configure_pypi_distribution): + """Distributions without content guards allow any SAFE_METHOD request, included + unauthenticated -- the pre-existing behavior.""" + domain_name = f"unguarded-{uuid4()}" _, pypi_url, _, _ = configure_pypi_distribution(domain_name) response = requests.get(pypi_url, timeout=30) @@ -190,8 +206,7 @@ def test_other_domains_pypi_simple_api_unaffected_by_feature_check(configure_pyp def test_public_domain_allows_unauthenticated_pypi_access(configure_pypi_distribution): """A public- domain's PyPI simple API stays open to unauthenticated SAFE_METHOD - requests -- unaffected by the lightwell-network feature check (which never applies to - public- domains, or to non-"lightwell" domains in general).""" + requests -- unaffected by content guards.""" domain_name = f"public-{uuid4()}" _, pypi_url, _, _ = configure_pypi_distribution(domain_name) diff --git a/pulp_service/pulp_service/tests/functional/test_domain_based_permissions.py b/pulp_service/pulp_service/tests/functional/test_domain_based_permissions.py index 728c59f8..4e48db82 100644 --- a/pulp_service/pulp_service/tests/functional/test_domain_based_permissions.py +++ b/pulp_service/pulp_service/tests/functional/test_domain_based_permissions.py @@ -112,7 +112,7 @@ def test_user_list_domain_permissions(pulpcore_bindings, anonymous_user, gen_obj assert response.results[0].name == domain_name -def test_only_owners_can_delete_domain(pulpcore_bindings, anonymous_user, gen_object_with_cleanup, monitor_task): # noqa: ARG001 +def test_only_owners_can_delete_domain(pulpcore_bindings, anonymous_user, gen_object_with_cleanup): user1_orgid1 = { "identity": { "org_id": 1, diff --git a/pulp_service/pulp_service/tests/functional/test_lightwell_readonly_group_permission.py b/pulp_service/pulp_service/tests/functional/test_lightwell_readonly_group_permission.py index e1ef983b..6d84d2a2 100644 --- a/pulp_service/pulp_service/tests/functional/test_lightwell_readonly_group_permission.py +++ b/pulp_service/pulp_service/tests/functional/test_lightwell_readonly_group_permission.py @@ -8,14 +8,14 @@ hardcoded group grants read-only (SAFE_METHODS) access to the lightwell domain, independent of any DomainOrg association. It does not grant write access, and it does not apply to the lightwell domain's PyPI views, which remain gated exclusively by the lightwell-network -feature check (see test_lightwell_feature_permission.py) -- group membership must not bypass +feature check (see test_content_guard_permission.py) -- group membership must not bypass that check. These follow the pattern used in test_group_based_permissions.py (group setup via gen_group -/ UsersApi / GroupsUsersApi) and test_lightwell_feature_permission.py (the "lightwell" +/ UsersApi / GroupsUsersApi) and test_content_guard_permission.py (the "lightwell" domain/PyPI fixtures). -NOTE: like test_lightwell_feature_permission.py, this is keyed off the literal domain name +NOTE: like test_content_guard_permission.py, this is keyed off the literal domain name "lightwell" (see pulp_service.app.authorization.LIGHTWELL_DOMAIN_NAME), so the domain created here can't use a random per-test suffix. These tests assume they run against an ephemeral Pulp instance where no "lightwell" domain already exists, and are not run concurrently with @@ -69,9 +69,11 @@ def lightwell_readonly_group(gen_group): def configure_lightwell_domain( anonymous_user, gen_object_with_cleanup, + add_to_cleanup, pulpcore_bindings, file_bindings, python_bindings, + service_content_guards_api_client, bindings_cfg, ): """ @@ -104,11 +106,30 @@ def configure_lightwell_domain( python_bindings.RepositoriesPythonApi, {"name": str(uuid4())}, pulp_domain=LIGHTWELL_DOMAIN_NAME ) + from pulpcore.client.pulp_service import ServiceFeatureContentGuard + + service_content_guards_api_client.api_client.default_headers["x-rh-identity"] = owner_header + guard = service_content_guards_api_client.create( + service_feature_content_guard=ServiceFeatureContentGuard( + name=f"lightwell-guard-{uuid4()}", + header_name="x-rh-identity", + features=["lightwell-network"], + jq_filter=".identity.org_id", + ), + pulp_domain=LIGHTWELL_DOMAIN_NAME, + ) + add_to_cleanup(service_content_guards_api_client, guard.pulp_href) + python_bindings.DistributionsPypiApi.api_client.default_headers["x-rh-identity"] = owner_header pypi_base_path = str(uuid4()) gen_object_with_cleanup( python_bindings.DistributionsPypiApi, - {"name": str(uuid4()), "base_path": pypi_base_path, "repository": repo.pulp_href}, + { + "name": str(uuid4()), + "base_path": pypi_base_path, + "repository": repo.pulp_href, + "content_guard": guard.pulp_href, + }, pulp_domain=LIGHTWELL_DOMAIN_NAME, ) @@ -121,6 +142,7 @@ def configure_lightwell_domain( file_bindings.RepositoriesFileApi.api_client.default_headers.pop("x-rh-identity", None) python_bindings.RepositoriesPythonApi.api_client.default_headers.pop("x-rh-identity", None) python_bindings.DistributionsPypiApi.api_client.default_headers.pop("x-rh-identity", None) + service_content_guards_api_client.api_client.default_headers.pop("x-rh-identity", None) @pytest.fixture diff --git a/pulp_service/pulp_service/tests/unit/test_domain_based_permission.py b/pulp_service/pulp_service/tests/unit/test_domain_based_permission.py index 0ff9c313..8c2ebe57 100644 --- a/pulp_service/pulp_service/tests/unit/test_domain_based_permission.py +++ b/pulp_service/pulp_service/tests/unit/test_domain_based_permission.py @@ -56,7 +56,7 @@ def _make_domain(name): return domain -def _make_pypi_view(): +def _make_pypi_view(content_guard=None): """Create a mock view that is an instance of PyPIMixin.""" from pulp_python.app.pypi.views import PyPIMixin @@ -64,9 +64,32 @@ class FakePyPIView(PyPIMixin): pass view = FakePyPIView() + view._distro = SimpleNamespace(content_guard=content_guard) return view +def _make_content_guard(permits=True): + """Create a mock content guard. If permits=False, cast().permit() raises PermissionError""" + guard = MagicMock() + casted = MagicMock() + guard.cast.return_value = casted + if not permits: + casted.permit.side_effect = PermissionError("Access denied.") + return guard + + +def _make_pypi_view_with_error(error): + """Create a mock PyPI view whose distribution property raises the given error.""" + from pulp_python.app.pypi.views import PyPIMixin + + class FakePyPIView(PyPIMixin): + @property + def distribution(self): + raise error + + return FakePyPIView() + + def _make_regular_view(): """Create a mock view that is NOT a PyPIMixin.""" return MagicMock() @@ -124,18 +147,15 @@ def test_authenticated_get_public_domain_allowed(self): assert permission.has_permission(request, view) is True - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - def test_public_domain_pypi_view_never_checks_feature(self, mock_feature_check): + def test_public_domain_pypi_view_never_checks_feature(self): permission = DomainBasedPermission() domain = _make_domain("public-something") request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain) - view = _make_pypi_view() + view = _make_pypi_view(content_guard=_make_content_guard()) assert permission.has_permission(request, view) is True - mock_feature_check.assert_not_called() - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - def test_anonymous_get_pypi_view_non_lightwell_domain_allowed(self, mock_feature_check): + def test_anonymous_get_pypi_view_non_lightwell_domain_allowed(self): """Domains other than "lightwell" keep the pre-existing behavior: any SAFE_METHOD request to a PyPI view is allowed, without any feature check.""" permission = DomainBasedPermission() @@ -144,20 +164,16 @@ def test_anonymous_get_pypi_view_non_lightwell_domain_allowed(self, mock_feature view = _make_pypi_view() assert permission.has_permission(request, view) is True - mock_feature_check.assert_not_called() - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - def test_authenticated_get_pypi_view_non_lightwell_domain_allowed(self, mock_feature_check): + def test_authenticated_get_pypi_view_non_lightwell_domain_allowed(self): permission = DomainBasedPermission() domain = _make_domain("some-other-domain") request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain) view = _make_pypi_view() assert permission.has_permission(request, view) is True - mock_feature_check.assert_not_called() - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - def test_anonymous_get_pypi_view_no_domain_allowed(self, mock_feature_check): + def test_anonymous_get_pypi_view_no_domain_allowed(self): """No domain resolved on the request (shouldn't happen in practice for PyPI views, but matches the pre-existing default-allow behavior).""" permission = DomainBasedPermission() @@ -165,157 +181,139 @@ def test_anonymous_get_pypi_view_no_domain_allowed(self, mock_feature_check): view = _make_pypi_view() assert permission.has_permission(request, view) is True - mock_feature_check.assert_not_called() -class TestLightwellDomainPyPIFeatureCheck: - """Verify the lightwell-network feature check, scoped specifically to the "lightwell" - domain's PyPI views.""" +class TestContentGuardPyPICheck: + """Verify the content-guard driven access checks on PyPI views. The check is + domain-name agnostic: any distribution with content guard enforces it.""" - def test_anonymous_no_org_id_denied(self): - """Anonymous requests carry no org_id, so there's nothing to check the feature for.""" + def test_no_guard_allows_anonymous(self): + """Distributions without content guards allow anonymous SAFE_METHOD access.""" permission = DomainBasedPermission() - domain = _make_domain("lightwell") + domain = _make_domain("any-domain") request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain) view = _make_pypi_view() - assert permission.has_permission(request, view) is False + assert permission.has_permission(request, view) is True - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_authenticated_with_feature_allowed(self, mock_domain_org, mock_feature_check): - mock_domain_org.filter.return_value.exists.return_value = False - mock_feature_check.return_value = True + def test_guard_permits_access(self): + """Content guard permits -> access granted.""" permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="20368420") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) assert permission.has_permission(request, view) is True - mock_feature_check.assert_called_once_with("20368420") + guard.cast.return_value.permit.assert_called_once_with(request) - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_authenticated_without_feature_denied(self, mock_domain_org, mock_feature_check): - mock_domain_org.filter.return_value.exists.return_value = False - mock_feature_check.return_value = False + def test_guard_denies_access(self): + """Content guard denies -> access denied.""" permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="1979710") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=False) + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) assert permission.has_permission(request, view) is False - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - def test_unauthenticated_with_org_id_and_feature_allowed(self, mock_feature_check): - """An org_id can be present even without a fully authenticated user; the feature - check still applies (and still gates access) in that case.""" - mock_feature_check.return_value = True + def test_anonymous_no_header_with_guard_denied(self): + """Anonymous request with no identity header and no DomainOrg falls through + to the content guard, which denies access.""" permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="20368420") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=False) + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain) + view = _make_pypi_view(content_guard=guard) - assert permission.has_permission(request, view) is True + assert permission.has_permission(request, view) is False - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_domain_org_association_bypasses_feature_check(self, mock_domain_org, mock_feature_check): - """Users with a DomainOrg association must not be denied by (or even trigger) the - feature check.""" + def test_domain_org_bypasses_guard(self, mock_domain_org): + """Users with a DomainOrg association bypass the content guard entirely.""" mock_domain_org.filter.return_value.exists.return_value = True permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="1979710") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=False) + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) assert permission.has_permission(request, view) is True - mock_feature_check.assert_not_called() + guard.cast.return_value.permit.assert_not_called() - @patch("pulp_service.app.models.FeatureContentGuard._get_cached_result", return_value=None) - @patch("pulp_service.app.models.FeatureContentGuard._check_for_feature", side_effect=PermissionError) @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_feature_service_failure_fails_closed(self, mock_domain_org, mock_check_feature, mock_cache_result): - """If the Features Service call fails, access must be denied, not silently allowed.""" + def test_no_domain_org_falls_through_to_guard(self, mock_domain_org): + """Without a DomainOrg association, the content guard is checked.""" mock_domain_org.filter.return_value.exists.return_value = False permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="1979710") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) - assert permission.has_permission(request, view) is False + assert permission.has_permission(request, view) is True + guard.cast.return_value.permit.assert_called_once() + + def test_guard_works_on_any_domain_name(self): + """The content guard check is domain-name-agnostic, no hardcoded names.""" + for domain_name in ["lightwell", "my-domain", "domain-xyz", "domain-abc"]: + permission = DomainBasedPermission() + guard = _make_content_guard(permits=True) + domain = _make_domain(domain_name) + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + assert permission.has_permission(request, view) is True -class TestLightwellPyPIAccessLogging: - """ - Verify _has_pypi_read_access logs its allow/deny decision and the reason for it (DomainOrg - match, missing org_id, or feature check outcome), so incidents can be diagnosed from logs - alone -- see the discussion that prompted this in the "brand new user can read lightwell - pypi" investigation. - """ + +class TestContentGuardAccessLogging: + """Verify that the content-guard PyPI access check logs its allow/deny decision + and the reason for it (DomainOrg match or content guard outcome).""" @patch("pulp_service.app.authorization.DomainOrg.objects") def test_domain_org_grant_is_logged(self, mock_domain_org, caplog): mock_domain_org.filter.return_value.exists.return_value = True permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="1979710") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=False) + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) with caplog.at_level("INFO", logger="pulp_service.app.authorization"): assert permission.has_permission(request, view) is True - assert any( - "GRANTED via DomainOrg" in record.message and "1979710" in record.message for record in caplog.records - ) + assert any("GRANTED via DomainOrg" in record.message and "12345" in record.message for record in caplog.records) @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_missing_org_id_denial_is_logged(self, mock_domain_org, caplog): + def test_guard_grant_is_logged(self, mock_domain_org, caplog): mock_domain_org.filter.return_value.exists.return_value = False permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain) - view = _make_pypi_view() - - with caplog.at_level("INFO", logger="pulp_service.app.authorization"): - assert permission.has_permission(request, view) is False - - assert any("DENIED: no org_id" in record.message for record in caplog.records) - - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") - @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_feature_check_outcome_is_logged(self, mock_domain_org, mock_feature_check, caplog): - mock_domain_org.filter.return_value.exists.return_value = False - mock_feature_check.return_value = True - permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="20368420") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) with caplog.at_level("INFO", logger="pulp_service.app.authorization"): assert permission.has_permission(request, view) is True assert any( - "GRANTED via lightwell-network feature check" in record.message and "20368420" in record.message - for record in caplog.records + "GRANTED via content guard" in record.message and "12345" in record.message for record in caplog.records ) - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature") @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_feature_check_denial_is_logged(self, mock_domain_org, mock_feature_check, caplog): + def test_guard_denial_is_logged(self, mock_domain_org, caplog): mock_domain_org.filter.return_value.exists.return_value = False - mock_feature_check.return_value = False permission = DomainBasedPermission() - domain = _make_domain("lightwell") - request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="1979710") - view = _make_pypi_view() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=False) + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) with caplog.at_level("INFO", logger="pulp_service.app.authorization"): assert permission.has_permission(request, view) is False assert any( - "DENIED via lightwell-network feature check" in record.message and "1979710" in record.message - for record in caplog.records + "DENIED via content guard" in record.message and "12345" in record.message for record in caplog.records ) @@ -360,22 +358,21 @@ def test_readonly_group_member_write_non_pypi_lightwell_denied(self, mock_domain assert permission.has_permission(request, MagicMock()) is False - @patch("pulp_service.app.authorization.DomainBasedPermission._has_lightwell_network_feature", return_value=False) + @patch("pulp_service.app.authorization.get_domain_pk", return_value=42) @patch("pulp_service.app.authorization.DomainOrg.objects") - def test_readonly_group_member_pypi_view_still_requires_feature(self, mock_domain_org, mock_feature_check): - """Group membership must not bypass the lightwell-network feature check on PyPI - views -- a member with no feature entitlement and no DomainOrg association is - still denied.""" + def test_readonly_group_member_pypi_view_still_requires_guard(self, mock_domain_org, mock_get_domain_pk): + """Group membership must not bypass the content guard on PyPI views -- a member + with no DomainOrg association is still denied if the guard denies.""" mock_domain_org.filter.return_value.exists.return_value = False permission = DomainBasedPermission() domain = _make_domain("lightwell") + guard = _make_content_guard(permits=False) request = _make_request( method="GET", user=_make_authenticated_user(in_readonly_group=True), domain=domain, org_id="1979710" ) - view = _make_pypi_view() + view = _make_pypi_view(content_guard=guard) assert permission.has_permission(request, view) is False - mock_feature_check.assert_called_once_with("1979710") @patch("pulp_service.app.authorization.get_domain_pk", return_value=42) @patch("pulp_service.app.authorization.DomainOrg.objects") @@ -456,3 +453,114 @@ def test_superuser_always_allowed(self): view = _make_regular_view() assert permission.has_permission(request, view) is True + + +class TestDistributionResolutionErrors: + """Verify exception handling when view.distribution raises.""" + + def test_http404_returns_true(self): + """Http404 lets view handle 404 naturally -- doesn't deny access at permission layer.""" + from django.http import Http404 + + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain) + view = _make_pypi_view_with_error(Http404("not found")) + + assert permission.has_permission(request, view) is True + + @patch("pulp_service.app.authorization.get_domain_pk", return_value=42) + @patch("pulp_service.app.authorization.DomainOrg.objects") + def test_unexpected_error_fails_closed(self, mock_domain_org, mock_get_domain_pk): + """Unexpected errors fail closed even if user would otherwise have domain access.""" + mock_domain_org.filter.return_value.exists.return_value = True + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view_with_error(RuntimeError("db timeout")) + + assert permission.has_permission(request, view) is False + + def test_unexpected_error_is_logged(self, caplog): + """Unexpected errors are logged with exception details.""" + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain) + view = _make_pypi_view_with_error(RuntimeError("db timeout")) + + with caplog.at_level("ERROR", logger="pulp_service.app.authorization"): + permission.has_permission(request, view) + + assert any("Unexpected error resolving distribution" in r.message for r in caplog.records) + + +class TestContentGuardCastFailure: + """Verify that guard.cast() failures fail closed instead of bubbling as 500s.""" + + def test_cast_failure_returns_false(self): + """If guard.cast() raises (orphaned FK, race condition), fail closed.""" + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + guard.cast.side_effect = Exception("content guard subclass row deleted") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + + assert permission.has_permission(request, view) is False + + def test_cast_failure_is_logged(self, caplog): + """cast() failures are logged with exception details.""" + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + guard.cast.side_effect = Exception("content guard subclass row deleted") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + + with caplog.at_level("ERROR", logger="pulp_service.app.authorization"): + permission.has_permission(request, view) + + assert any("Failed to resolve content guard" in r.message for r in caplog.records) + + @patch("pulp_service.app.authorization.DomainOrg.objects") + def test_cast_failure_not_reached_when_domain_org_matches(self, mock_domain_org): + """DomainOrg bypass short-circuits before cast() is ever called.""" + mock_domain_org.filter.return_value.exists.return_value = True + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + guard.cast.side_effect = Exception("should not be called") + request = _make_request(method="GET", user=_make_authenticated_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + + assert permission.has_permission(request, view) is True + guard.cast.assert_not_called() + + +class TestPermitUnexpectedError: + """Verify that unexpected errors from casted_guard.permit() fail closed instead of raising 500s.""" + + def test_permit_unexpected_error_fails_closed(self): + """If permit() raises something other than PermissionError, deny access.""" + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + guard.cast.return_value.permit.side_effect = RuntimeError("unexpected guard failure") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + + assert permission.has_permission(request, view) is False + + def test_permit_unexpected_error_is_logged(self, caplog): + """Unexpected permit() errors are logged with exception details.""" + permission = DomainBasedPermission() + domain = _make_domain("any-domain") + guard = _make_content_guard(permits=True) + guard.cast.return_value.permit.side_effect = RuntimeError("unexpected guard failure") + request = _make_request(method="GET", user=_make_anonymous_user(), domain=domain, org_id="12345") + view = _make_pypi_view(content_guard=guard) + + with caplog.at_level("ERROR", logger="pulp_service.app.authorization"): + permission.has_permission(request, view) + + assert any("Unexpected error" in r.message and "guard permit" in r.message for r in caplog.records)