feat(authz): surface effective route permissions on GET /auth/me (Phase 4, #4063) (#5228)

* feat(authz): surface effective route permissions on GET /auth/me (Phase 4, #4063)

GET /api/v1/auth/me now returns the effective route permissions alongside
the user identity, so the frontend can hide actions the caller's role
cannot perform (RFC #4063 Phase 4).

The value reuses the AuthContext that AuthMiddleware already resolves per
request (including PAT-scope intersection and internal-caller semantics),
so /me adds zero extra provider evaluations; a middleware-less composition
falls back to the same resolution _authenticate uses.

Credential-creation responses (register/initialize) leave the field None:
they are public paths where the middleware does not run, and resolving
there would introduce fresh on-loop config loads on those routes.

* test(e2e): expect /auth/me permissions in auth-disabled contract

PR #5228 adds the effective route permissions to GET /auth/me, so the
strict toEqual against the bare AUTH_DISABLED_USER object no longer
holds: the received payload carries six extra keys (the permissions
array). Extend the expected payload with the full registered permission
set in _ALL_PERMISSIONS order — with authorization disabled the gateway
grants exactly that static list, so the pin stays deterministic.

The runtime frontend is unaffected (auth-disabled SSR never calls /me,
and userSchema strips unknown keys); only this contract pin needed the
new field.

* refactor(authz): public resolve_route_permissions_for_request wrapper

Address review nits on the middleware-less fallback: the router reached
into the private authz._is_internal_caller, so expose a thin public
wrapper pairing resolve_route_permissions with the internal-caller
heuristics, and use it from both _authenticate and the /me fallback so
the two cannot drift apart. Also drop an unused tmp_path parameter from
test_auth_disabled_me_includes_default_admin_permissions (_setup_auth
provisions its own tmp directory).

No behavior change: the wrapper delegates to the exact pair of calls the
fallback made before.

backend/app/gateway/auth/models.py

@@ -40,3 +40,7 @@ class UserResponse(BaseModel):
     system_role: Literal["admin", "user"]
     needs_setup: bool = False
     oauth_provider: str | None = Field(None, description="OAuth/SSO provider ID if the user logged in via SSO (e.g. 'keycloak')")
+    permissions: list[str] | None = Field(
+        None,
+        description=("Effective route permissions granted to this credential (RFC #4063 Phase 4). Only GET /api/v1/auth/me resolves them; credential-creation responses leave it None."),
+    )

backend/app/gateway/authz.py

@@ -288,6 +288,17 @@ async def resolve_route_permissions(user: User, *, is_internal: bool) -> list[st
     return [p for p in results if p is not None]
 
 
+async def resolve_route_permissions_for_request(request: Request, user: Any) -> list[str]:
+    """Resolve the effective route permissions for a request's authenticated user.
+
+    Public wrapper pairing ``resolve_route_permissions`` with the internal-caller
+    heuristics of ``_is_internal_caller`` (auth source, synthetic internal role,
+    internal auth header), so middleware-less consumers resolve exactly what
+    ``_authenticate`` resolves and the two cannot drift apart.
+    """
+    return await resolve_route_permissions(user, is_internal=_is_internal_caller(request, user))
+
+
 class _AuthorizationUnavailable(Exception):
     """Raised internally when the provider cannot be resolved for a route check.
 
@@ -497,8 +508,7 @@ async def _authenticate(request: Request) -> AuthContext:
     if user is None:
         return AuthContext(user=None, permissions=[])
 
-    is_internal = _is_internal_caller(request, user)
+    permissions = await resolve_route_permissions_for_request(request, user)
-    permissions = await resolve_route_permissions(user, is_internal=is_internal)
     return AuthContext(user=user, permissions=permissions)
 
 

backend/app/gateway/routers/auth.py

@@ -560,14 +560,26 @@ async def change_password(request: Request, response: Response, body: ChangePass
 
 @router.get("/me", response_model=UserResponse)
 async def get_me(request: Request):
-    """Get current authenticated user info."""
+    """Get current authenticated user info, including effective permissions."""
     user = await get_current_user_from_request(request)
+    auth = getattr(getattr(request, "state", None), "auth", None)
+    if auth is not None:
+        # AuthMiddleware already resolved the per-request permission set (with
+        # PAT-scope intersection and internal-caller semantics applied) — reuse
+        # it instead of evaluating the provider a second time.
+        permissions: list[str] = list(auth.permissions)
+    else:
+        # Middleware-less composition: resolve exactly as _authenticate does.
+        from app.gateway.authz import resolve_route_permissions_for_request
+
+        permissions = await resolve_route_permissions_for_request(request, user)
     return UserResponse(
         id=str(user.id),
         email=user.email,
         system_role=user.system_role,
         needs_setup=user.needs_setup,
         oauth_provider=user.oauth_provider,
+        permissions=permissions,
     )
 
 

backend/tests/test_auth_me_permissions.py

@@ -0,0 +1,215 @@
+"""GET /api/v1/auth/me effective-permissions tests (RFC #4063 Phase 4).
+
+Pins the /me wiring only: the field is populated from the AuthContext that
+AuthMiddleware already resolves per request, and a middleware-less call
+falls back to a fresh resolution with the same semantics ``_authenticate``
+uses. The permission-derivation semantics themselves (disabled / fail-closed
+/ fail-open / per-action requests) are pinned by
+test_authorization_route_permissions.py and are not re-tested here.
+"""
+
+import asyncio
+import os
+from types import SimpleNamespace
+
+import pytest
+from fastapi.testclient import TestClient
+
+os.environ.setdefault("AUTH_JWT_SECRET", "test-secret-key-auth-me-permissions-min-32")
+
+from app.gateway.authz import Permissions  # noqa: E402
+from deerflow.authz.provider import AuthzDecision, AuthzReason  # noqa: E402
+from deerflow.config.authorization_config import AuthorizationConfig  # noqa: E402
+
+_TEST_SECRET = "test-secret-key-auth-me-permissions-min-32"
+
+_ALL_PERMISSIONS = [
+    Permissions.THREADS_READ,
+    Permissions.THREADS_WRITE,
+    Permissions.THREADS_DELETE,
+    Permissions.RUNS_CREATE,
+    Permissions.RUNS_READ,
+    Permissions.RUNS_CANCEL,
+]
+
+
+class _RecordingProvider:
+    """Async-only provider recording every decision request it serves."""
+
+    name = "recording"
+
+    def __init__(self, *, denied: set[str] | None = None) -> None:
+        self.denied = denied or set()
+        self.requests = []
+
+    def authorize(self, request):
+        raise AssertionError("route authorization must use the async provider API")
+
+    async def aauthorize(self, request):
+        self.requests.append(request)
+        allowed = request.target not in self.denied
+        return AuthzDecision(
+            allow=allowed,
+            reasons=[AuthzReason(code="authz.allowed" if allowed else "authz.denied")],
+        )
+
+    def filter_resources(self, principal, resource_type, candidates):
+        raise AssertionError("route authorization must preserve per-action requests")
+
+
+@pytest.fixture(autouse=True)
+def _setup_auth(tmp_path):
+    """Fresh SQLite engine + auth config per test (test_initialize_admin pattern)."""
+    from app.gateway import deps
+    from app.gateway.auth.config import AuthConfig, set_auth_config
+    from app.gateway.routers.auth import _SETUP_STATUS_CACHE, _SETUP_STATUS_INFLIGHT
+    from deerflow.persistence.engine import close_engine, init_engine
+
+    set_auth_config(AuthConfig(jwt_secret=_TEST_SECRET))
+    url = f"sqlite+aiosqlite:///{tmp_path}/auth_me.db"
+    asyncio.run(init_engine("sqlite", url=url, sqlite_dir=str(tmp_path)))
+    deps._cached_local_provider = None
+    deps._cached_repo = None
+    _SETUP_STATUS_CACHE.clear()
+    _SETUP_STATUS_INFLIGHT.clear()
+    try:
+        yield
+    finally:
+        deps._cached_local_provider = None
+        deps._cached_repo = None
+        _SETUP_STATUS_CACHE.clear()
+        _SETUP_STATUS_INFLIGHT.clear()
+        asyncio.run(close_engine())
+
+
+@pytest.fixture(autouse=True)
+def _default_route_authorization_config(monkeypatch):
+    """Keep /me independent of a repository config.yaml (disabled by default)."""
+    monkeypatch.setattr(
+        "app.gateway.authz._get_route_authorization_config",
+        lambda: AuthorizationConfig(),
+    )
+
+
+@pytest.fixture()
+def client(_setup_auth):
+    from app.gateway.app import create_app
+    from app.gateway.auth.config import AuthConfig, set_auth_config
+
+    set_auth_config(AuthConfig(jwt_secret=_TEST_SECRET))
+    app = create_app()
+    # No context manager: the full lifespan requires config.yaml, the auth
+    # routes work without it (the persistence engine is set up by _setup_auth).
+    yield TestClient(app)
+
+
+def _initialize_admin(client: TestClient):
+    resp = client.post(
+        "/api/v1/auth/initialize",
+        json={"email": "admin@example.com", "password": "Str0ng!Pass99"},
+    )
+    assert resp.status_code == 201
+    return resp
+
+
+def _enable_authorization(monkeypatch, provider) -> None:
+    config = AuthorizationConfig(enabled=True, fail_closed=True, default_role="user")
+    monkeypatch.setattr("app.gateway.authz._get_route_authorization_config", lambda: config)
+    monkeypatch.setattr("app.gateway.authz._get_cached_route_provider", lambda c: provider)
+
+
+# ── Route-level wiring ────────────────────────────────────────────────────
+
+
+def test_me_lists_all_route_permissions_when_authorization_disabled(client):
+    _initialize_admin(client)
+    res = client.get("/api/v1/auth/me")
+    assert res.status_code == 200
+    assert res.json()["permissions"] == _ALL_PERMISSIONS
+
+
+def test_me_reuses_middleware_resolved_permissions(client, monkeypatch):
+    """One provider decision per registered permission — /me reads the
+    AuthContext AuthMiddleware already stamped instead of re-resolving."""
+    denied = {Permissions.THREADS_DELETE, Permissions.RUNS_CANCEL}
+    provider = _RecordingProvider(denied=denied)
+    _enable_authorization(monkeypatch, provider)
+    _initialize_admin(client)
+
+    res = client.get("/api/v1/auth/me")
+
+    assert res.status_code == 200
+    assert res.json()["permissions"] == [p for p in _ALL_PERMISSIONS if p not in denied]
+    assert len(provider.requests) == len(_ALL_PERMISSIONS)
+
+
+def test_initialize_response_leaves_permissions_unresolved(client):
+    """Credential-creation responses do not resolve permissions (None), so
+    they never advertise a misleading empty grant set."""
+    resp = _initialize_admin(client)
+    assert resp.json()["permissions"] is None
+
+
+def test_auth_disabled_me_includes_default_admin_permissions(monkeypatch, _setup_auth):
+    from app.gateway.app import create_app
+    from app.gateway.auth.config import AuthConfig, set_auth_config
+
+    monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
+    set_auth_config(AuthConfig(jwt_secret=_TEST_SECRET))
+    client = TestClient(create_app())
+
+    res = client.get("/api/v1/auth/me")
+
+    assert res.status_code == 200
+    assert res.json()["permissions"] == _ALL_PERMISSIONS
+
+
+# ── Middleware-less fallback ──────────────────────────────────────────────
+
+
+def _fallback_request(user: SimpleNamespace, auth_source: str) -> SimpleNamespace:
+    return SimpleNamespace(
+        state=SimpleNamespace(user=user, auth_source=auth_source),
+        cookies={},
+        headers={},
+    )
+
+
+def _stub_user() -> SimpleNamespace:
+    return SimpleNamespace(
+        id="user-123",
+        email="user@example.test",
+        system_role="user",
+        needs_setup=False,
+        oauth_provider=None,
+    )
+
+
+@pytest.mark.asyncio
+async def test_me_falls_back_to_fresh_resolution_without_middleware_context():
+    """Direct handler invocation (no AuthMiddleware) resolves permissions the
+    same way ``_authenticate`` would instead of reporting an empty grant."""
+    from app.gateway.routers.auth import get_me
+
+    response = await get_me(_fallback_request(_stub_user(), "session"))
+
+    assert response.permissions == _ALL_PERMISSIONS
+
+
+@pytest.mark.asyncio
+async def test_me_fallback_mirrors_authenticate_internal_caller_semantics(monkeypatch):
+    from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
+    from app.gateway.routers.auth import get_me
+
+    captured = {}
+
+    async def fake_resolve(user, *, is_internal):
+        captured["is_internal"] = is_internal
+        return ["threads:read"]
+
+    monkeypatch.setattr("app.gateway.authz.resolve_route_permissions", fake_resolve)
+
+    response = await get_me(_fallback_request(_stub_user(), AUTH_SOURCE_INTERNAL))
+
+    assert response.permissions == ["threads:read"]
+    assert captured["is_internal"] is True

frontend/tests/e2e-real-backend/auth-disabled-contract.spec.ts

@@ -6,6 +6,18 @@ const APP =
   process.env.E2E_APP_URL ??
   `http://localhost:${process.env.E2E_FRONTEND_PORT ?? "3000"}`;
 
+// /me also returns the caller's effective route permissions (RFC #4063
+// Phase 4). Auth-disabled mode grants the full registered set, in the order
+// of backend _ALL_PERMISSIONS (backend/app/gateway/authz.py).
+const AUTH_DISABLED_PERMISSIONS = [
+  "threads:read",
+  "threads:write",
+  "threads:delete",
+  "runs:create",
+  "runs:read",
+  "runs:cancel",
+];
+
 test.describe("auth-disabled contract (real backend)", () => {
   test("gateway /auth/me returns the frontend synthetic user without a cookie", async ({
     context,
@@ -13,6 +25,9 @@ test.describe("auth-disabled contract (real backend)", () => {
     const resp = await context.request.get(`${APP}/api/v1/auth/me`);
 
     expect(resp.status(), await resp.text()).toBe(200);
-    await expect(resp.json()).resolves.toEqual(AUTH_DISABLED_USER);
+    await expect(resp.json()).resolves.toEqual({
+      ...AUTH_DISABLED_USER,
+      permissions: AUTH_DISABLED_PERMISSIONS,
+    });
   });
 });