fix(auth): admins can access any task; expose super_admin_mode; introduce authorize() PDP

openrag/api/dependencies/auth.py

@@ -192,14 +192,20 @@ async def require_task_owner(
     task_id=Depends(request_task_id),
     user=Depends(current_user),
     job_service=Depends(get_job_service),
+    auth_service=Depends(get_auth_service),
 ):
     task_details = await job_service.get_task_details(task_id)
     if not task_details:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail=f"Task '{task_id}' not found",
         )
-    if task_details.get("user_id") != user.get("id"):
+    # Delegate the decision to the central PDP (AuthService.authorize, reached via DI
+    # — the api layer must not import services directly) rather than inlining policy
+    # here — admins may access any task (parity with the admin jobs list), owners
+    # their own. Without this, an admin opening another user's job in the admin UI
+    # hits a 403 dead-end.
+    if not auth_service.authorize(user=user, action="task:access", resource=task_details):
         raise HTTPException(
             status_code=status.HTTP_403_FORBIDDEN,
             detail="You do not have permission to access this task",

openrag/api/main.py

@@ -319,8 +319,15 @@ def root_redirect():
 
 @app.get("/config", summary="Get current configuration", tags=["Configuration"], dependencies=[Depends(require_admin)])
 def get_config():
-    """Return the loaded application settings for admins."""
-    return settings
+    """Return the loaded application settings for admins, plus runtime auth flags
+    the admin UI needs. ``super_admin_mode`` governs whether an admin bypasses
+    partition-membership checks; the UI permission layer mirrors the backend rule
+    instead of assuming every admin has partition access.
+    """
+    from api.dependencies.auth import SUPER_ADMIN_MODE
+    from fastapi.encoders import jsonable_encoder
+
+    return {**jsonable_encoder(settings), "super_admin_mode": SUPER_ADMIN_MODE}
 
 
 # Router mounts. Phase 10F finished moving these into

openrag/services/orchestrators/auth_service.py

@@ -528,6 +528,37 @@ def check_partition_access(
             )
         return True
 
+    @classmethod
+    def authorize(cls, *, user: Any, action: str, resource: dict[str, Any] | None = None) -> bool:
+        """Single authorization decision point (PDP) — deny-by-default.
+
+        The seam every capability check should route through so policy lives in one
+        place instead of being inlined at call sites (which is how the rules drift).
+        Each ``action`` owns its rule here; an unknown action is a programming error
+        and is denied loudly rather than silently allowed.
+
+        Intentionally minimal today (the only migrated action is task access). Other
+        checks (partition roles via :meth:`check_partition_access`) will migrate here.
+
+        TODO(org): add an org-admin tier — allow when ``user`` is an admin of the
+        organization that owns ``resource`` (``resource['org_id'] == user's org``).
+        Until orgs exist, ``is_admin`` is the system-wide admin signal.
+        """
+        match action:
+            case "task:access":
+                # Admins may access any task — consistent with the admin jobs list,
+                # which lists every user's tasks. Otherwise only the task's owner.
+                if cls._uget(user, "is_admin", False):
+                    return True
+                # Deny-by-default: both identities must be present before comparing,
+                # so a missing/partially-hydrated principal or a malformed task
+                # (user_id=None) can never match by two Nones.
+                owner_id = (resource or {}).get("user_id")
+                user_id = cls._uget(user, "id", None)
+                return owner_id is not None and user_id is not None and owner_id == user_id
+            case _:
+                raise ValueError(f"Unknown authorization action: {action!r}")
+
     @classmethod
     def validate_file_quota(
         cls,

tests/unit/api/dependencies/test_auth.py

@@ -9,6 +9,7 @@
 )
 from core.utils.exceptions import AuthError
 from fastapi import HTTPException
+from services.orchestrators.auth_service import AuthService
 
 
 class FakeAuthService:
@@ -194,12 +195,42 @@ async def test_require_task_owner_reads_task_details_through_job_service():
         task_id="task-1",
         user={"id": 7},
         job_service=job_service,
+        auth_service=AuthService,
     )
 
     assert details == {"user_id": 7, "filename": "a.pdf"}
     assert job_service.detail_checks == ["task-1"]
 
 
+@pytest.mark.asyncio
+async def test_require_task_owner_allows_admin_for_another_users_task():
+    # Admins may open any task (parity with the admin jobs list) — no 403 dead-end.
+    job_service = FakeJobService(details={"user_id": 2, "filename": "b.pdf"})
+
+    details = await require_task_owner(
+        task_id="task-2",
+        user={"id": 1, "is_admin": True},
+        job_service=job_service,
+        auth_service=AuthService,
+    )
+
+    assert details == {"user_id": 2, "filename": "b.pdf"}
+
+
+@pytest.mark.asyncio
+async def test_require_task_owner_rejects_non_owner_non_admin():
+    job_service = FakeJobService(details={"user_id": 2, "filename": "b.pdf"})
+
+    with pytest.raises(HTTPException) as exc:
+        await require_task_owner(
+            task_id="task-2",
+            user={"id": 1, "is_admin": False},
+            job_service=job_service,
+            auth_service=AuthService,
+        )
+    assert exc.value.status_code == 403
+
+
 @pytest.mark.asyncio
 async def test_check_user_file_quota_reads_pending_count_through_job_service():
     job_service = FakeJobService(pending_count=2)

tests/unit/services/orchestrators/test_auth_service.py

@@ -585,6 +585,31 @@ def test_check_partition_access_super_admin_bypass():
     )
 
 
+def test_authorize_task_access_owner_admin_and_deny():
+    # Owner is allowed.
+    assert AuthService.authorize(user={"id": 7, "is_admin": False}, action="task:access", resource={"user_id": 7})
+    # Admin may access any task.
+    assert AuthService.authorize(user={"id": 1, "is_admin": True}, action="task:access", resource={"user_id": 2})
+    # Non-owner non-admin is denied (deny-by-default).
+    assert not AuthService.authorize(user={"id": 1, "is_admin": False}, action="task:access", resource={"user_id": 2})
+    # Missing resource denies rather than crashes.
+    assert not AuthService.authorize(user={"id": 1, "is_admin": False}, action="task:access")
+
+
+def test_authorize_task_access_denies_missing_identities():
+    # Deny-by-default: two missing ids must NOT match (no None == None bypass).
+    assert not AuthService.authorize(user=None, action="task:access", resource={"user_id": None})
+    assert not AuthService.authorize(user={"id": None}, action="task:access", resource={"user_id": None})
+    # A real user vs a malformed task with no owner is denied.
+    assert not AuthService.authorize(user={"id": 5}, action="task:access", resource={"user_id": None})
+
+
+def test_authorize_unknown_action_raises():
+    # An unknown action is a programming error — denied loudly, never silently allowed.
+    with pytest.raises(ValueError, match="Unknown authorization action"):
+        AuthService.authorize(user={"id": 1, "is_admin": True}, action="bogus:action")
+
+
 def test_validate_file_quota():
     # Admin bypass.
     AuthService.validate_file_quota({"is_admin": True}, pending_task_count=99, default_quota=1)