[RBAC PR 3] Add configurable default-access role to RBAC

datajunction-server/datajunction_server/config.py

@@ -213,6 +213,12 @@ class Settings(BaseSettings):  # pragma: no cover
     # - "restrictive": Deny by default
     default_access_policy: str = "permissive"  # or "restrictive"
 
+    # Optional role name whose scopes are evaluated as a fallback when no
+    # explicit grant matches. Lets a deployment express graceful defaults such
+    # as "everyone gets read on *" without flipping the whole policy to
+    # permissive. Applied before the default_access_policy fallback.
+    default_access_role: Optional[str] = None
+
     # Require configured break-glass admins before serving requests.
     # Restrictive default access also enables this check automatically.
     # RBAC_ADMIN_USERS uses JSON list syntax, for example ["admin-user"].

datajunction-server/datajunction_server/internal/access/authorization/context.py

@@ -3,7 +3,7 @@
 """
 
 from fastapi import Depends
-from dataclasses import dataclass
+from dataclasses import dataclass, field
 from typing import List, Optional
 
 from sqlalchemy import select
@@ -14,7 +14,7 @@
 from datajunction_server.internal.access.group_membership import (
     get_group_membership_service,
 )
-from datajunction_server.database.rbac import RoleAssignment, Role
+from datajunction_server.database.rbac import RoleAssignment, Role, RoleScope
 from datajunction_server.database.user import User
 
 from datajunction_server.utils import (
@@ -47,6 +47,9 @@ class AuthContext:
     oauth_provider: Optional[str]
     role_assignments: List[RoleAssignment]  # Direct + groups, flattened
     is_admin: bool = False
+    # Scopes from the configured default-access role, evaluated as a fallback
+    # alongside the user's own grants.
+    default_scopes: List[RoleScope] = field(default_factory=list)
 
     @classmethod
     async def from_user(
@@ -71,15 +74,35 @@ async def from_user(
             session=session,
             user=user,
         )
+        default_scopes = await cls.get_default_scopes(session=session)
 
         return cls(
             user_id=user.id,
             username=user.username,
             oauth_provider=user.oauth_provider,
             role_assignments=assignments,
             is_admin=bool(user.is_admin),
+            default_scopes=default_scopes,
         )
 
+    @classmethod
+    async def get_default_scopes(
+        cls,
+        session: AsyncSession,
+    ) -> List[RoleScope]:
+        """
+        Load the scopes of the configured default-access role, if any.
+
+        Returns an empty list when no default role is configured or the named
+        role does not exist, so authorization simply falls through to the
+        default_access_policy.
+        """
+        role_name = settings.default_access_role
+        if not role_name:
+            return []
+        default_role = await Role.get_by_name(session, role_name)
+        return list(default_role.scopes) if default_role else []
+
     @classmethod
     async def get_effective_assignments(
         cls,

datajunction-server/datajunction_server/internal/access/authorization/service.py

@@ -6,7 +6,10 @@
 from abc import ABC, abstractmethod
 from datetime import datetime, timezone
 from functools import lru_cache
-from typing import List
+from typing import List, TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from datajunction_server.database.rbac import RoleScope
 
 
 from datajunction_server.models.access import (
@@ -160,7 +163,28 @@ def authorize(
                 )
                 for request in requests
             ]
-        return [self._make_decision(auth_context, request) for request in requests]
+        candidate_scopes = self.candidate_scopes(auth_context)
+        return [self._make_decision(request, candidate_scopes) for request in requests]
+
+    @classmethod
+    def candidate_scopes(cls, auth_context: AuthContext) -> List["RoleScope"]:
+        """
+        Collect every scope that could grant a request for this context.
+
+        This is the union of the principal's own (non-expired) role scopes and
+        the configured default-access role's scopes. Collecting all candidates
+        up front (rather than short-circuiting source by source) keeps the
+        decision a single resolve step, leaving room for future deny/precedence
+        rules without restructuring.
+        """
+        scopes: List["RoleScope"] = []
+        now = datetime.now(timezone.utc)
+        for assignment in auth_context.role_assignments:
+            if assignment.expires_at and assignment.expires_at < now:
+                continue
+            scopes.extend(assignment.role.scopes)
+        scopes.extend(auth_context.default_scopes)
+        return scopes
 
     def authorize_explicit_grants(
         self,
@@ -177,21 +201,30 @@ def authorize_explicit_grants(
 
     def _make_decision(
         self,
-        auth_context: AuthContext,
         request: ResourceRequest,
+        candidate_scopes: List["RoleScope"],
     ) -> AccessDecision:
         """
         Convert ResourceRequest to AccessDecision.
+
+        Evaluates the candidate scopes (explicit grants + default-access role)
+        collected once per authorize() call and approves if any grants the
+        request. Otherwise falls back to the configured default_access_policy.
         """
-        has_grant = self.has_permission(
-            assignments=auth_context.role_assignments,
-            action=request.verb,
-            resource_type=request.access_object.resource_type,
-            resource_name=request.access_object.name,
+        granted = any(
+            self._scope_grants_permission(
+                scope,
+                request.verb,
+                request.access_object.resource_type,
+                request.access_object.name,
+            )
+            for scope in candidate_scopes
         )
+        if granted:
+            return AccessDecision(request=request, approved=True)
         return AccessDecision(
             request=request,
-            approved=(has_grant or settings.default_access_policy == "permissive"),
+            approved=(settings.default_access_policy == "permissive"),
         )
 
     def _make_explicit_grant_decision(