From a24b834191f5f848830ed2f19ad13ac0050633c8 Mon Sep 17 00:00:00 2001 From: Philip Weiss Date: Fri, 17 Jul 2026 18:42:42 -0700 Subject: [PATCH] Fail closed for RBAC grant delegation Require explicit MANAGE grants containing every delegated scope so permissive fallback cannot authorize role or group administration. Align RBAC enum serialization and history labels with migrated schemas. --- ...2359-rbacenumfix1_sync_rbac_enum_values.py | 17 ++ .../datajunction_server/api/groups.py | 5 +- .../datajunction_server/api/rbac.py | 17 +- .../datajunction_server/database/rbac.py | 13 +- .../internal/access/authorization/service.py | 144 ++++++++++++ .../access/authorization/validator.py | 9 +- .../datajunction_server/models/access.py | 12 +- .../tests/api/graphql/find_nodes_test.py | 26 ++- datajunction-server/tests/api/groups_test.py | 35 ++- datajunction-server/tests/api/rbac_test.py | 49 +++- .../tests/internal/authorization_test.py | 212 ++++++++++++++++++ 11 files changed, 520 insertions(+), 19 deletions(-) create mode 100644 datajunction-server/datajunction_server/alembic/versions/2026_07_17_2359-rbacenumfix1_sync_rbac_enum_values.py diff --git a/datajunction-server/datajunction_server/alembic/versions/2026_07_17_2359-rbacenumfix1_sync_rbac_enum_values.py b/datajunction-server/datajunction_server/alembic/versions/2026_07_17_2359-rbacenumfix1_sync_rbac_enum_values.py new file mode 100644 index 0000000000..04d7e2e529 --- /dev/null +++ b/datajunction-server/datajunction_server/alembic/versions/2026_07_17_2359-rbacenumfix1_sync_rbac_enum_values.py @@ -0,0 +1,17 @@ +"""Sync RBAC enum values with ORM serialization.""" + +from alembic import op + +revision = "rbacenumfix1" +down_revision = "cm0001jsonbgin" +branch_labels = None +depends_on = None + + +def upgrade(): + for value in ("HIERARCHY", "ROLE", "ROLE_ASSIGNMENT", "ROLE_SCOPE"): + op.execute(f"ALTER TYPE entitytype ADD VALUE IF NOT EXISTS '{value}'") + + +def downgrade(): + pass diff --git a/datajunction-server/datajunction_server/api/groups.py b/datajunction-server/datajunction_server/api/groups.py index 3a242edcf4..5d09151a13 100644 --- a/datajunction-server/datajunction_server/api/groups.py +++ b/datajunction-server/datajunction_server/api/groups.py @@ -35,7 +35,10 @@ async def enforce_group_administration(access_checker: AccessChecker) -> None: privileged, cross-cutting operation rather than a per-namespace one. """ access_checker.add_scope(ResourceType.NAMESPACE, "*", ResourceAction.MANAGE) - await access_checker.check(on_denied=AccessDenialMode.RAISE) + await access_checker.check( + on_denied=AccessDenialMode.RAISE, + require_explicit_grant=True, + ) @router.post("/groups/", response_model=GroupOutput, status_code=201) diff --git a/datajunction-server/datajunction_server/api/rbac.py b/datajunction-server/datajunction_server/api/rbac.py index ec26a42b78..efa196487e 100644 --- a/datajunction-server/datajunction_server/api/rbac.py +++ b/datajunction-server/datajunction_server/api/rbac.py @@ -64,7 +64,10 @@ async def enforce_scope_management( scope.scope_value, ResourceAction.MANAGE, ) - await access_checker.check(on_denied=AccessDenialMode.RAISE) + await access_checker.check( + on_denied=AccessDenialMode.RAISE, + require_explicit_grant=True, + ) async def log_activity( @@ -463,8 +466,16 @@ async def delete_scope_from_role( include_deleted=False, ) - access_checker.add_scope(scope_type, scope_value, ResourceAction.MANAGE) - await access_checker.check(on_denied=AccessDenialMode.RAISE) + await enforce_scope_management( + access_checker, + [ + RoleScopeInput( + action=action, + scope_type=scope_type, + scope_value=scope_value, + ), + ], + ) # Find the scope by composite key delete_stmt = ( diff --git a/datajunction-server/datajunction_server/database/rbac.py b/datajunction-server/datajunction_server/database/rbac.py index c00465a27d..5d4f2bd2b1 100644 --- a/datajunction-server/datajunction_server/database/rbac.py +++ b/datajunction-server/datajunction_server/database/rbac.py @@ -220,9 +220,18 @@ class RoleScope(Base): nullable=False, ) - action: Mapped[ResourceAction] = mapped_column(Enum(ResourceAction), nullable=False) + action: Mapped[ResourceAction] = mapped_column( + Enum( + ResourceAction, + values_callable=lambda enum: [item.value for item in enum], + ), + nullable=False, + ) scope_type: Mapped[ResourceType] = mapped_column( - Enum(ResourceType), + Enum( + ResourceType, + values_callable=lambda enum: [item.value for item in enum], + ), nullable=False, ) scope_value: Mapped[str] = mapped_column(String(500), nullable=False) diff --git a/datajunction-server/datajunction_server/internal/access/authorization/service.py b/datajunction-server/datajunction_server/internal/access/authorization/service.py index 73aabf4082..e9bf01b5d1 100644 --- a/datajunction-server/datajunction_server/internal/access/authorization/service.py +++ b/datajunction-server/datajunction_server/internal/access/authorization/service.py @@ -64,6 +64,21 @@ def authorize( The same list of requests with approved=True/False set on each """ + def authorize_explicit_grants( + self, + auth_context: AuthContext, + requests: list[ResourceRequest], + ) -> list[AccessDecision]: + """Deny control-plane access until a provider explicitly supports it.""" + return [ + AccessDecision( + request=request, + approved=False, + reason="explicit_grant_provider_required", + ) + for request in requests + ] + class RBACAuthorizationService(AuthorizationService): """ @@ -142,6 +157,19 @@ def authorize( ] return [self._make_decision(auth_context, request) for request in requests] + def authorize_explicit_grants( + self, + auth_context: AuthContext, + requests: list[ResourceRequest], + ) -> list[AccessDecision]: + """Authorize from assigned roles without policy fallback.""" + if auth_context.is_admin: + return self.authorize(auth_context, requests) + return [ + self._make_explicit_grant_decision(auth_context, request) + for request in requests + ] + def _make_decision( self, auth_context: AuthContext, @@ -161,6 +189,33 @@ def _make_decision( approved=(has_grant or settings.default_access_policy == "permissive"), ) + def _make_explicit_grant_decision( + self, + auth_context: AuthContext, + request: ResourceRequest, + ) -> AccessDecision: + """Resolve a request from assigned roles only.""" + has_grant = ( + self.has_scope_permission( + assignments=auth_context.role_assignments, + action=request.verb, + scope_type=request.access_object.resource_type, + scope_value=request.access_object.name, + ) + if request.scope_target + else self.has_permission( + assignments=auth_context.role_assignments, + action=request.verb, + resource_type=request.access_object.resource_type, + resource_name=request.access_object.name, + ) + ) + return AccessDecision( + request=request, + approved=has_grant, + reason="explicit_grant" if has_grant else "explicit_grant_required", + ) + @classmethod def resource_matches_pattern(cls, resource_name: str, pattern: str) -> bool: """ @@ -235,6 +290,87 @@ def has_permission( return False + @classmethod + def has_scope_permission( + cls, + assignments: List, + action: ResourceAction, + scope_type: ResourceType, + scope_value: str, + ) -> bool: + """Return whether an assigned scope contains the requested scope.""" + for assignment in assignments: + if assignment.expires_at and assignment.expires_at < datetime.now( + timezone.utc, + ): + continue + for granted_scope in assignment.role.scopes: + granted_actions = cls.PERMISSION_HIERARCHY.get( + granted_scope.action, + {granted_scope.action}, + ) + if action in granted_actions and cls.scope_contains_scope( + granted_scope.scope_type, + granted_scope.scope_value, + scope_type, + scope_value, + ): + return True + return False + + @classmethod + def scope_contains_scope( + cls, + granted_type: ResourceType, + granted_value: str, + delegated_type: ResourceType, + delegated_value: str, + ) -> bool: + """Return whether one supported scope pattern contains another.""" + granted_pattern = cls._parse_scope_pattern(granted_value) + delegated_pattern = cls._parse_scope_pattern(delegated_value) + if granted_pattern is None or delegated_pattern is None: + return False + + if granted_type != delegated_type and not ( + granted_type == ResourceType.NAMESPACE + and delegated_type == ResourceType.NODE + and granted_pattern[0] != "global" + ): + return False + + granted_kind, granted_prefix = granted_pattern + delegated_kind, delegated_prefix = delegated_pattern + if granted_kind == "global": + return True + if delegated_kind == "global": + return False + if granted_kind == "exact": + return delegated_kind == "exact" and granted_prefix == delegated_prefix + if delegated_kind == "exact": + return delegated_prefix.startswith(granted_prefix + SEPARATOR) + return delegated_prefix == granted_prefix or delegated_prefix.startswith( + granted_prefix + SEPARATOR, + ) + + @staticmethod + def _parse_scope_pattern(value: str) -> tuple[str, str] | None: + """Parse exact, trailing-wildcard, and global scope patterns.""" + if not value or value == "*": + return ("global", "") + if ( + value.startswith(SEPARATOR) + or value.endswith(SEPARATOR) + or SEPARATOR * 2 in value + ): + return None + if "*" not in value: + return ("exact", value) + if value.count("*") != 1 or not value.endswith(f"{SEPARATOR}*"): + return None + prefix = value[: -len(f"{SEPARATOR}*")] + return ("subtree", prefix) if prefix else None + @classmethod def _scope_grants_permission( cls, @@ -293,6 +429,14 @@ def authorize( """Approve all requests without checks (sync).""" return [AccessDecision(request=request, approved=True) for request in requests] + def authorize_explicit_grants( + self, + auth_context: AuthContext, + requests: list[ResourceRequest], + ) -> list[AccessDecision]: + """Approve control-plane checks when authorization is disabled.""" + return self.authorize(auth_context, requests) + @lru_cache(maxsize=None) def get_authorization_service() -> AuthorizationService: diff --git a/datajunction-server/datajunction_server/internal/access/authorization/validator.py b/datajunction-server/datajunction_server/internal/access/authorization/validator.py index 4111c489da..77c153aaf7 100644 --- a/datajunction-server/datajunction_server/internal/access/authorization/validator.py +++ b/datajunction-server/datajunction_server/internal/access/authorization/validator.py @@ -128,12 +128,14 @@ def add_scope( name=scope_value or "*", resource_type=scope_type, ), + scope_target=True, ), ) async def check( self, on_denied: AccessDenialMode = AccessDenialMode.FILTER, + require_explicit_grant: bool = False, ) -> list[AccessDecision]: """ Validate all requests using AuthorizationService. @@ -143,9 +145,14 @@ async def check( - FILTER: Return only approved (default) - RAISE: Raise exception if any denied - RETURN_ALL: Return all with approved field set + require_explicit_grant: Ignore provider fallback authorization """ auth_service = get_authorization_service() - access_decisions = auth_service.authorize(self.auth_context, self.requests) + access_decisions = ( + auth_service.authorize_explicit_grants(self.auth_context, self.requests) + if require_explicit_grant + else auth_service.authorize(self.auth_context, self.requests) + ) if on_denied == AccessDenialMode.RETURN: return access_decisions diff --git a/datajunction-server/datajunction_server/models/access.py b/datajunction-server/datajunction_server/models/access.py index a74a99d9b2..58657927e4 100644 --- a/datajunction-server/datajunction_server/models/access.py +++ b/datajunction-server/datajunction_server/models/access.py @@ -62,16 +62,24 @@ class ResourceRequest: """ Resource Requests provide the information that is available to grant access to a resource + + ``scope_target`` marks a delegated scope pattern so authorization can + prove containment instead of treating it as one concrete resource. """ verb: ResourceAction access_object: Resource + scope_target: bool = False def __hash__(self) -> int: - return hash((self.verb, self.access_object)) + return hash((self.verb, self.access_object, self.scope_target)) def __eq__(self, other) -> bool: - return self.verb == other.verb and self.access_object == other.access_object + return ( + self.verb == other.verb + and self.access_object == other.access_object + and self.scope_target == other.scope_target + ) def __str__(self) -> str: return ( diff --git a/datajunction-server/tests/api/graphql/find_nodes_test.py b/datajunction-server/tests/api/graphql/find_nodes_test.py index 9327a75520..44f7294dc1 100644 --- a/datajunction-server/tests/api/graphql/find_nodes_test.py +++ b/datajunction-server/tests/api/graphql/find_nodes_test.py @@ -1976,15 +1976,25 @@ async def _setup_team_ownership(client: AsyncClient) -> None: - default.repair_order_details.owners == ['team-analytics'] (group only) - default.repair_orders_fact still has 'dj' as an owner """ - # Register the group and add 'dj' as a member. - resp = await client.post("/groups/", params={"username": "team-analytics"}) - assert resp.status_code == 201, resp.text - - resp = await client.post( - "/groups/team-analytics/members/", - params={"member_username": "dj"}, + # Register the group and add 'dj' as a member. Group administration now + # requires an explicit MANAGE grant, so authorize this setup as passthrough. + from datajunction_server.internal.access.authorization import ( + PassthroughAuthorizationService, ) - assert resp.status_code == 201, resp.text + + with mock.patch( + "datajunction_server.internal.access.authorization." + "validator.get_authorization_service", + return_value=PassthroughAuthorizationService(), + ): + resp = await client.post("/groups/", params={"username": "team-analytics"}) + assert resp.status_code == 201, resp.text + + resp = await client.post( + "/groups/team-analytics/members/", + params={"member_username": "dj"}, + ) + assert resp.status_code == 201, resp.text # Re-assign a node so the group is the sole owner (dj no longer owns it). resp = await client.patch( diff --git a/datajunction-server/tests/api/groups_test.py b/datajunction-server/tests/api/groups_test.py index f0095b8fc3..c30866e0d6 100644 --- a/datajunction-server/tests/api/groups_test.py +++ b/datajunction-server/tests/api/groups_test.py @@ -5,7 +5,11 @@ import pytest from httpx import AsyncClient -from datajunction_server.internal.access.authorization import AuthorizationService +from datajunction_server.internal.access.authorization import ( + AuthorizationService, + PassthroughAuthorizationService, + RBACAuthorizationService, +) from datajunction_server.models import access VALIDATOR_AUTH_SERVICE = ( @@ -28,6 +32,18 @@ def authorize(self, auth_context, requests): for request in requests ] + def authorize_explicit_grants(self, auth_context, requests): + return self.authorize(auth_context, requests) + + +@pytest.fixture(autouse=True) +def allow_control_plane_by_default(mocker): + """Keep existing endpoint tests focused on group behavior.""" + mocker.patch( + VALIDATOR_AUTH_SERVICE, + lambda: PassthroughAuthorizationService(), + ) + # Group Registration Tests @@ -368,6 +384,23 @@ async def test_group_lifecycle( # MANAGE Enforcement Tests +@pytest.mark.asyncio +async def test_permissive_policy_cannot_authorize_group_administration( + module__client: AsyncClient, + mocker, +) -> None: + """Group administration requires an explicit global MANAGE grant.""" + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: RBACAuthorizationService()) + + response = await module__client.post( + "/groups/", + params={"username": "permissive-blocked-team"}, + ) + + assert response.status_code == 403 + assert "Access denied" in response.json()["message"] + + @pytest.mark.asyncio async def test_register_group_requires_manage( module__client: AsyncClient, diff --git a/datajunction-server/tests/api/rbac_test.py b/datajunction-server/tests/api/rbac_test.py index f799d2f6be..8bc289434b 100644 --- a/datajunction-server/tests/api/rbac_test.py +++ b/datajunction-server/tests/api/rbac_test.py @@ -11,7 +11,11 @@ from datajunction_server.database.history import History from datajunction_server.database.rbac import RoleScope, RoleAssignment from datajunction_server.database.user import User -from datajunction_server.internal.access.authorization import AuthorizationService +from datajunction_server.internal.access.authorization import ( + AuthorizationService, + PassthroughAuthorizationService, + RBACAuthorizationService, +) from datajunction_server.internal.history import ActivityType, EntityType from datajunction_server.models import access @@ -35,6 +39,9 @@ def authorize(self, auth_context, requests): for request in requests ] + def authorize_explicit_grants(self, auth_context, requests): + return self.authorize(auth_context, requests) + class ScopedManageAuthorizationService(AuthorizationService): """Grants MANAGE only on a fixed set of namespace patterns.""" @@ -56,6 +63,18 @@ def authorize(self, auth_context, requests): for request in requests ] + def authorize_explicit_grants(self, auth_context, requests): + return self.authorize(auth_context, requests) + + +@pytest.fixture(autouse=True) +def allow_control_plane_by_default(mocker): + """Keep existing endpoint tests focused on role behavior.""" + mocker.patch( + VALIDATOR_AUTH_SERVICE, + lambda: PassthroughAuthorizationService(), + ) + @pytest.mark.asyncio async def test_create_role_basic(client_with_basic: AsyncClient): @@ -964,6 +983,34 @@ async def test_audit_logging_for_role_assignments( # ============================================================================ +@pytest.mark.asyncio +@pytest.mark.parametrize("scope_type", ["node", "namespace"]) +async def test_global_scope_requires_explicit_manage_under_permissive_policy( + client_with_basic: AsyncClient, + mocker, + scope_type: str, +): + """Permissive policy cannot authorize global grant delegation.""" + mocker.patch(VALIDATOR_AUTH_SERVICE, lambda: RBACAuthorizationService()) + + response = await client_with_basic.post( + "/roles/", + json={ + "name": f"blocked-global-{scope_type}", + "scopes": [ + { + "action": "write", + "scope_type": scope_type, + "scope_value": "*", + }, + ], + }, + ) + + assert response.status_code == 403 + assert "Access denied" in response.json()["message"] + + @pytest.mark.asyncio async def test_create_role_with_scopes_requires_manage( client_with_basic: AsyncClient, diff --git a/datajunction-server/tests/internal/authorization_test.py b/datajunction-server/tests/internal/authorization_test.py index f4495d922a..49548a2425 100644 --- a/datajunction-server/tests/internal/authorization_test.py +++ b/datajunction-server/tests/internal/authorization_test.py @@ -1,6 +1,8 @@ """Tests for RBAC authorization logic.""" from datetime import datetime, timedelta, timezone +from types import SimpleNamespace +from typing import cast import pytest from sqlalchemy.ext.asyncio import AsyncSession @@ -12,6 +14,7 @@ AccessChecker, AccessDenialMode, AuthContext, + AuthorizationService, PassthroughAuthorizationService, RBACAuthorizationService, get_authorization_service, @@ -19,6 +22,7 @@ from datajunction_server.errors import DJAuthorizationException from datajunction_server.internal.access.authentication.basic import get_user from datajunction_server.models.access import ( + AccessDecision, Resource, ResourceAction, ResourceRequest, @@ -29,6 +33,15 @@ ) +class PermissiveFallbackAuthorizationService(AuthorizationService): + """Test provider with a permissive normal authorization path.""" + + name = "test_permissive_fallback" + + def authorize(self, auth_context, requests): + return [AccessDecision(request=request, approved=True) for request in requests] + + class TestResourceMatching: """Tests for wildcard pattern matching.""" @@ -148,6 +161,202 @@ def test_wildcard_in_middle_not_supported(self): # which is unlikely in practice +class TestScopeContainment: + """Tests for delegated scope containment.""" + + @pytest.mark.parametrize( + "granted_type,granted_value,delegated_type,delegated_value,expected", + [ + (ResourceType.NAMESPACE, "*", ResourceType.NAMESPACE, "*", True), + (ResourceType.NAMESPACE, "finance.*", ResourceType.NAMESPACE, "*", False), + (ResourceType.NODE, "*", ResourceType.NODE, "*", True), + (ResourceType.NODE, "finance.*", ResourceType.NODE, "*", False), + ( + ResourceType.NAMESPACE, + "finance.*", + ResourceType.NAMESPACE, + "finance.revenue.*", + True, + ), + ( + ResourceType.NAMESPACE, + "finance.revenue.*", + ResourceType.NAMESPACE, + "finance.*", + False, + ), + ( + ResourceType.NAMESPACE, + "finance.*", + ResourceType.NODE, + "finance.revenue", + True, + ), + (ResourceType.NAMESPACE, "*", ResourceType.NODE, "*", False), + ], + ) + def test_scope_contains_scope( + self, + granted_type, + granted_value, + delegated_type, + delegated_value, + expected, + ) -> None: + assert ( + RBACAuthorizationService.scope_contains_scope( + granted_type, + granted_value, + delegated_type, + delegated_value, + ) + is expected + ) + + @pytest.mark.parametrize( + "pattern", + ["finance*", ".*", "**", "finance.*.revenue", "finance..*"], + ) + def test_global_manage_rejects_malformed_scope(self, pattern: str) -> None: + assert not RBACAuthorizationService.scope_contains_scope( + ResourceType.NAMESPACE, + "*", + ResourceType.NAMESPACE, + pattern, + ) + + +@pytest.mark.parametrize("scope_type", [ResourceType.NODE, ResourceType.NAMESPACE]) +def test_explicit_authorization_ignores_permissive_fallback( + scope_type: ResourceType, + mocker, +) -> None: + mock_settings = mocker.patch( + "datajunction_server.internal.access.authorization.service.settings", + ) + mock_settings.default_access_policy = "permissive" + context = AuthContext(1, "no-grants", "basic", []) + request = ResourceRequest( + ResourceAction.MANAGE, + Resource(name="*", resource_type=scope_type), + scope_target=True, + ) + service = RBACAuthorizationService() + + assert service.authorize(context, [request])[0].approved is True + assert service.authorize_explicit_grants(context, [request])[0].approved is False + + +def test_explicit_authorization_uses_scope_containment() -> None: + assignment = SimpleNamespace( + expires_at=None, + role=SimpleNamespace( + scopes=[ + SimpleNamespace( + action=ResourceAction.MANAGE, + scope_type=ResourceType.NAMESPACE, + scope_value="finance.*", + ), + ], + ), + ) + context = AuthContext( + 1, + "finance-owner", + "basic", + [cast(RoleAssignment, assignment)], + ) + request = ResourceRequest( + ResourceAction.MANAGE, + Resource("finance.revenue.*", ResourceType.NAMESPACE), + scope_target=True, + ) + + decision = RBACAuthorizationService().authorize_explicit_grants( + context, + [request], + )[0] + assert decision.approved is True + assert decision.reason == "explicit_grant" + + +def test_custom_provider_must_opt_in_to_control_plane_authorization() -> None: + context = AuthContext(1, "custom-provider-user", "basic", []) + request = ResourceRequest( + ResourceAction.MANAGE, + Resource("*", ResourceType.NAMESPACE), + scope_target=True, + ) + service = PermissiveFallbackAuthorizationService() + + assert service.authorize(context, [request])[0].approved is True + assert service.authorize_explicit_grants(context, [request])[0].approved is False + + +def test_admin_authorizes_explicit_grants_via_bypass() -> None: + context = AuthContext(1, "root", "basic", [], is_admin=True) + request = ResourceRequest( + ResourceAction.MANAGE, + Resource("*", ResourceType.NAMESPACE), + scope_target=True, + ) + + decision = RBACAuthorizationService().authorize_explicit_grants( + context, + [request], + )[0] + + assert decision.approved is True + assert decision.reason == "admin" + + +def _scope(action, scope_type, scope_value): + return SimpleNamespace( + action=action, + scope_type=scope_type, + scope_value=scope_value, + ) + + +def _assignment(scopes, expires_at=None): + return cast( + RoleAssignment, + SimpleNamespace(expires_at=expires_at, role=SimpleNamespace(scopes=scopes)), + ) + + +def test_has_scope_permission_allows_exact_grant() -> None: + assignments = [ + _assignment( + [_scope(ResourceAction.MANAGE, ResourceType.NAMESPACE, "finance")], + ), + ] + + assert RBACAuthorizationService.has_scope_permission( + assignments=assignments, + action=ResourceAction.MANAGE, + scope_type=ResourceType.NAMESPACE, + scope_value="finance", + ) + + +def test_has_scope_permission_skips_expired_and_mismatched_scopes() -> None: + expired = _assignment( + [_scope(ResourceAction.MANAGE, ResourceType.NAMESPACE, "finance.*")], + expires_at=datetime.now(timezone.utc) - timedelta(hours=1), + ) + wrong_action = _assignment( + [_scope(ResourceAction.READ, ResourceType.NAMESPACE, "finance.*")], + ) + + assert not RBACAuthorizationService.has_scope_permission( + assignments=[expired, wrong_action], + action=ResourceAction.MANAGE, + scope_type=ResourceType.NAMESPACE, + scope_value="finance.revenue", + ) + + @pytest.mark.asyncio class TestRBACPermissionChecks: """Tests for RBAC permission checking.""" @@ -490,6 +699,9 @@ async def test_passthrough_service_approves_all( assert len(result) == 2 assert all(req.approved for req in result) + assert all( + req.approved for req in service.authorize_explicit_grants(user, requests) + ) async def test_rbac_service_with_permissions( self,