diff --git a/CHANGELOG.md b/CHANGELOG.md index f10104095..57224d696 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -88,6 +88,14 @@ This section accumulates work toward the **2.1.0** milestone ### Added +#### Authentication +- **auth:** Personal access tokens (PAT) for programmatic API access: + `POST/GET/DELETE /api/v1/auth/pats` manage tokens (shown once, stored as + SHA-256 digests); a default-deny route policy admits only the thread/run + lifecycle routes, narrowed further by the token's `threads`/`runs` scopes, + and any request dimension that carries cancel capability (`?action=`, + `multitask_strategy`) additionally requires `runs:cancel`. + #### Agents & runtime - **middleware:** New `TokenBudgetMiddleware` enforces a per-run token budget, diff --git a/README.md b/README.md index c79c71723..d637034a1 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ DeerFlow has newly integrated the intelligent search and crawling toolset indepe - [Langfuse Tracing](#langfuse-tracing) - [Monocle Tracing](#monocle-tracing) - [Using Multiple Providers](#using-multiple-providers) + - [Personal Access Tokens](#personal-access-tokens) - [From Deep Research to Super Agent Harness](#from-deep-research-to-super-agent-harness) - [Core Features](#core-features) - [Skills \& Tools](#skills--tools) @@ -796,6 +797,30 @@ LangSmith and Langfuse attach as LangChain callbacks, so you can enable both and For Docker deployments, tracing is disabled by default. Set `LANGSMITH_TRACING=true` and `LANGSMITH_API_KEY` in your `.env` to enable it. +#### Personal Access Tokens + +Non-interactive clients (CI pipelines, scripts, server-to-server integrations) +can call the Gateway API with a **personal access token (PAT)** instead of a +browser session. Create one while logged in via `POST /api/v1/auth/pats` — the +raw `dfp_...` value is shown exactly once; only its SHA-256 digest is stored — +then send it as a Bearer credential: + +```http +POST /api/threads/search +Authorization: Bearer dfp_... +Content-Type: application/json + +{} +``` + +Each token runs with its owning user's identity (owner filtering and per-user +memory keep working), carries a scope set that can only narrow that user's +permissions, and is admitted only to the thread/run lifecycle routes — every +other route answers `403` to PAT callers, and a PAT never carries admin +capability. Tokens can be listed and revoked at any time; revocation is +immediate. PATs require a database backend (SQLite/PostgreSQL). Full +reference: [API Reference — Personal Access Tokens](backend/docs/API.md#personal-access-tokens). + ## From Deep Research to Super Agent Harness DeerFlow started as a Deep Research framework — and the community ran with it. Since launch, developers have pushed it far beyond research: building data pipelines, generating slide decks, spinning up dashboards, automating content workflows. Things we never anticipated. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 0fed15652..8ea1928c9 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -4,16 +4,18 @@ FastAPI application on port 8001 with health check at `GET /health`. Set `GATEWA Durable MCP task notifications are internal Agent runs: keep the trusted delivery instruction outside the user-input boundary and frame the serialized remote event payload as untrusted text before model invocation. These runs use strict thread existence/ownership admission so an event from a task that outlives its deleted chat is dead-lettered rather than recreating the thread. -CORS is same-origin by default when requests enter through nginx on port 2026. Split-origin or port-forwarded browser clients must opt in with `GATEWAY_CORS_ORIGINS` (comma-separated exact origins); Gateway `CORSMiddleware` and `CSRFMiddleware` both read that variable so browser CORS and auth-origin checks stay aligned. Those clients also need `CORS_EXPOSED_HEADERS` (`csrf_middleware.py`): run-creating routes return the run's id in `Content-Location`, which is not CORS-safelisted, so JS cannot read it unless it is exposed. The LangGraph SDK resolves run metadata from that header alone — withhold it and `useStream`'s `onCreated` never fires, a new thread keeps its placeholder route, and every action gated on an established thread (edit, regenerate, branch) stays hidden until the page is reloaded. Same-origin nginx deployments never hit this because CORS does not apply. +CORS is same-origin by default when requests enter through nginx on port 2026. Split-origin or port-forwarded browser clients must opt in with `GATEWAY_CORS_ORIGINS` (exact origins); Gateway `CORSMiddleware` and `CSRFMiddleware` both read that variable so browser CORS and auth-origin checks stay aligned. Those clients also need `CORS_EXPOSED_HEADERS` (`csrf_middleware.py`): run-creating routes return the run's id in `Content-Location`, which is not CORS-safelisted, so JS cannot read it unless it is exposed — and the LangGraph SDK resolves run metadata from that header alone, so withholding it breaks `useStream`'s `onCreated` and thread-gated actions. -Browser auth sessions are owned by `app.gateway.auth.session_cookie`. Login accepts a `remember_me` form flag, but the Gateway never stores passwords. `SessionCookiePolicy` persists the `HttpOnly access_token` cookie only for HTTPS/trusted-forwarded HTTPS, direct-host localhost HTTP, or explicit operator opt-in for insecure persistence; public HTTP sandbox URLs degrade to session cookies. Session-creating handlers stamp the final `max_age` on `request.state`, and CSRF cookie creation mirrors that value so the double-submit cookie pair expires together, including explicit re-issue after password changes and OIDC callbacks. A small `HttpOnly` preference cookie preserves the user's remember choice across token re-issue paths. Logout clears all auth cookies and suppresses CSRF re-issue on the logout response. +Browser auth sessions are owned by `app.gateway.auth.session_cookie`. Login accepts a `remember_me` form flag, but the Gateway never stores passwords. `SessionCookiePolicy` persists the `HttpOnly access_token` cookie only for HTTPS/trusted-forwarded HTTPS, direct-host localhost HTTP, or explicit operator opt-in for insecure persistence; public HTTP sandbox URLs degrade to session cookies. Session-creating handlers stamp the final `max_age` on `request.state`; CSRF cookie creation mirrors it so the double-submit pair expires together, including re-issue after password changes and OIDC callbacks. A small `HttpOnly` preference cookie preserves the remember choice across re-issues. Logout clears all auth cookies and suppresses CSRF re-issue on the logout response. + +Personal Access Tokens (`app.gateway.auth.pat`, `Authorization: Bearer dfp_...`) run as their owning user: an invalid Bearer is a hard 401 with no cookie fallback, which keeps `CSRFMiddleware`'s Bearer skip safe (origin checks still run). Scopes narrow within the allowlisted threads/runs routes; every other authenticated route 403s PAT callers (admin included). PAT management and `/change-password` require session auth; only SHA-256 digests are stored (`0017`). Localhost persistence deliberately reads the direct request `Host` and ignores `Forwarded` / `X-Forwarded-Host`. Scheme and auth-origin reconstruction still consume forwarding headers. The bundled nginx sets `X-Forwarded-Proto`, but preserves an upstream HTTPS value and does not overwrite every forwarded header, so the outer trusted proxy must replace or strip client-supplied forwarding headers before traffic reaches DeerFlow. Standalone local LangGraph Studio is recognized only through the upstream `Auth.types.StudioUser` principal type, never by its reusable identity string. The type is resolved once at import; an older SDK without it degrades to normal -owner scoping instead of failing requests. +owner scoping. For that principal's assistant reads/searches, `langgraph_auth.add_owner_filter` selects genuine server-registered assistants plus assistants owned by Studio; all other resources remain owner-scoped. Assistant create/update handlers make @@ -33,11 +35,9 @@ test. An empty graph registry or absent persistence file is a no-op, while persistence parse/write errors fail startup closed. The harness requires in-memory runtime 0.30.0 or newer, and a persisted store containing no expected registered assistant row emits a drift warning so changes to -LangGraph's internal persistence contract are observable. Because current -create/update writes and all legacy -versions are sanitized, ordinary owner-scoped assistant version selection -remains enabled. Ordinary authenticated users retain owner-scoped assistant -reads/searches. +LangGraph's internal persistence contract are observable. With current +create/update writes and all legacy versions sanitized, ordinary +owner-scoped assistant version selection remains enabled. **Routers**: @@ -70,11 +70,10 @@ Gateway creation and state-producing request boundaries, embedded-client entry points, filesystem/upload/event-store consumers, scheduled launches, and the standalone Provisioner enforce the same contract before persistence or workspace initialization. Route-addressable legacy IDs remain accepted by -pure reads and cleanup/control endpoints. Deleting a noncanonical legacy ID -best-effort removes its metadata and checkpoints but deliberately skips local -filesystem cleanup, so the raw value is never interpolated into a host path; -new runs, workspace/sandbox operations, and other state-producing mutations -remain blocked. +pure reads and cleanup/control endpoints; deleting one best-effort removes +metadata and checkpoints but skips local filesystem cleanup, so the raw value +is never interpolated into a host path. New runs, workspace/sandbox +operations, and other state-producing mutations remain blocked. **Workspace change review**: `packages/harness/deerflow/workspace_changes/` captures a pre-run and post-run snapshot of the thread-owned `workspace` and @@ -135,8 +134,8 @@ paths: a separate boundary flag preserves the previous completion-data semantics, so checkpoint incompatibility or cancellation while waiting for an older finalizing run does not persist an empty completion snapshot. Worker tests pin one accumulated receipt across multiple goal-continuation `_stream_once` -calls; journal tests drive LangChain's real async callback dispatcher against a -single journal to pin serialized, deduplicated parallel tool callbacks. +calls; journal tests drive LangChain's real async callback dispatcher to pin +serialized, deduplicated parallel tool callbacks. Multi-worker deployments therefore require `run_events.backend: db` for shared, ordered delivery events; the startup gate rejects process-local memory and JSONL event stores when `GATEWAY_WORKERS > 1`. diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index 1ce4559a4..7e6696b2e 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -7,7 +7,7 @@ from deerflow_extension_api import EXTENSION_PRINCIPAL_RESOLVER_KEY, ExtensionPr from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware -from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL, warn_if_auth_disabled_enabled +from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL, AUTH_SOURCE_PAT, warn_if_auth_disabled_enabled from app.gateway.auth_middleware import AuthMiddleware from app.gateway.browser_capability import ensure_browser_runtime_available from app.gateway.config import get_gateway_config @@ -695,15 +695,23 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for if user is None: return None system_role = getattr(user, "system_role", None) + # PAT credentials never carry admin capability (#5041): suppress every + # admin signal — both ``is_admin`` and the ``admin`` role — so an + # admin-owned PAT cannot regain admin through extension-side + # require_admin, mirroring deps.is_admin_user's PAT guard. + auth_source = getattr(request.state, "auth_source", None) + is_pat = auth_source == AUTH_SOURCE_PAT + is_admin = system_role == "admin" and not is_pat + roles = () if is_pat and system_role == "admin" else (system_role,) if isinstance(system_role, str) and system_role else () return ExtensionPrincipal( user_id=str(user.id), - is_admin=system_role == "admin", - is_internal=getattr(request.state, "auth_source", None) == AUTH_SOURCE_INTERNAL, + is_admin=is_admin, + is_internal=auth_source == AUTH_SOURCE_INTERNAL, # The host's only role concept is the single system_role column # (e.g. "admin", "user") — there is no multi-role system to # project, so a set role becomes the one-element tuple rather # than reading a "roles" attribute the user model never had. - roles=(system_role,) if isinstance(system_role, str) and system_role else (), + roles=roles, ) setattr(app.state, EXTENSION_PRINCIPAL_RESOLVER_KEY, _resolve_extension_principal) diff --git a/backend/app/gateway/auth/pat.py b/backend/app/gateway/auth/pat.py new file mode 100644 index 000000000..6e7fe4fec --- /dev/null +++ b/backend/app/gateway/auth/pat.py @@ -0,0 +1,196 @@ +"""Personal Access Token (PAT) credentials for programmatic API access. + +Tokens are ``dfp_`` + base62(32 CSPRNG bytes), shown exactly once in the +create response and persisted only as a SHA-256 digest. Validation is a +digest-indexed lookup plus a constant-time re-comparison, with a single +generic failure surface so a 401 never reveals which check failed. + +v1 scopes are exactly the route-permission strings owned by +``app.gateway.authz`` — a PAT can only narrow its owning user's +permissions, never widen them. +""" + +from __future__ import annotations + +import functools +import hashlib +import hmac +import re +import secrets +from typing import Any + +PAT_TOKEN_PREFIX = "dfp_" +PAT_RANDOM_BYTES = 32 +# Best-effort ``last_used_at`` writes are throttled per token so high-volume +# automation does not turn every request into a database write. +PAT_LAST_USED_WRITE_INTERVAL_SECONDS = 300.0 + +PAT_ALLOWED_SCOPES: frozenset[str] = frozenset( + { + "threads:read", + "threads:write", + "threads:delete", + "runs:create", + "runs:read", + "runs:cancel", + } +) + +PAT_MAX_NAME_LENGTH = 128 + +# Default-deny route boundary for PAT callers (#5041 review P1-1): scope +# intersection in AuthMiddleware only constrains routes that consult +# ``request.state.auth.permissions`` (``@require_permission``). Authenticated +# mutation routes without that decorator — memory deletion, agent creation, +# credential switching, channel configuration — would otherwise accept a +# PAT holding a single read scope. A route is reachable by PAT only when it +# is explicitly listed here, and only together with the thread/run lifecycle +# the v1 scopes govern; everything else answers 403 regardless of scopes. +_PAT_ROUTE_RULES: tuple[tuple[frozenset[str], re.Pattern[str]], ...] = ( + (frozenset({"POST"}), re.compile(r"^/api/threads$")), + (frozenset({"POST"}), re.compile(r"^/api/threads/search$")), + (frozenset({"GET", "PATCH", "DELETE"}), re.compile(r"^/api/threads/[^/]+$")), + (frozenset({"GET", "PUT", "DELETE"}), re.compile(r"^/api/threads/[^/]+/goal$")), + (frozenset({"GET", "POST"}), re.compile(r"^/api/threads/[^/]+/state$")), + (frozenset({"POST"}), re.compile(r"^/api/threads/[^/]+/(compact|history|branches)$")), + # Runs subtree: enumerated per implemented subroute instead of a + # ``runs(/.*)?`` wildcard, so a route added under /runs is default-denied + # until explicitly listed — the same no-dead-methods precision the + # threads collection rule enforces. The ``{run_id}`` slot necessarily + # matches any single segment; the POST-only collection endpoints sharing + # that depth (stream, wait, regenerate, edit-regenerate) are excluded + # from the GET run-id rule so no unimplemented method is pre-authorized. + (frozenset({"GET", "POST"}), re.compile(r"^/api/threads/[^/]+/runs$")), + ( + frozenset({"POST"}), + re.compile(r"^/api/threads/[^/]+/runs/(stream|wait|regenerate/prepare|edit-regenerate/prepare)$"), + ), + ( + frozenset({"GET"}), + re.compile(r"^/api/threads/[^/]+/runs/(?!stream$|wait$|regenerate$|edit-regenerate$)[^/]+$"), + ), + (frozenset({"POST"}), re.compile(r"^/api/threads/[^/]+/runs/[^/]+/cancel$")), + ( + frozenset({"GET"}), + re.compile(r"^/api/threads/[^/]+/runs/[^/]+/(join|messages|events|workspace-changes)$"), + ), + (frozenset({"GET", "POST"}), re.compile(r"^/api/threads/[^/]+/runs/[^/]+/stream$")), + (frozenset({"POST"}), re.compile(r"^/api/runs/(stream|wait)$")), + (frozenset({"GET"}), re.compile(r"^/api/runs/[^/]+/(messages|feedback)$")), +) + +_BASE62_ALPHABET = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" + + +def is_pat_allowed_route(method: str, path: str) -> bool: + """Return whether the PAT route policy admits *method* + *path*. + + Trailing slashes are normalized away so the mounted route and its + redirect-style twin resolve identically. + """ + normalized = path.rstrip("/") or "/" + return any(method in methods and pattern.match(normalized) for methods, pattern in _PAT_ROUTE_RULES) + + +@functools.cache +def _base62_width(byte_length: int) -> int: + """Digits sufficient for any *byte_length*-byte value (exact integer math).""" + width = 1 + limit = 1 << (8 * byte_length) + while 62**width < limit: + width += 1 + return width + + +def _base62(data: bytes) -> str: + """Fixed-width big-endian base62 of *data*, ``0``-padded on the left. + + ``int.from_bytes`` discards leading zero bytes, so an unpadded encoding + would be variable-length (and empty for all-zero input) — any draw below + 62**39 would have produced a shorter-than-expected token. The fixed width + keeps every token body exactly ``_base62_width(len(data))`` characters + and makes the format test deterministic. + """ + value = int.from_bytes(data, "big") + digits: list[str] = [] + while value: + value, remainder = divmod(value, 62) + digits.append(_BASE62_ALPHABET[remainder]) + body = "".join(reversed(digits)) + return body.rjust(_base62_width(len(data)), "0") + + +def generate_pat_token() -> str: + """Generate a show-once raw token: ``dfp_`` + base62(CSPRNG bytes).""" + return PAT_TOKEN_PREFIX + _base62(secrets.token_bytes(PAT_RANDOM_BYTES)) + + +def pat_token_digest(token: str) -> str: + """Return the hex SHA-256 digest persisted for *token*.""" + return hashlib.sha256(token.encode("utf-8")).hexdigest() + + +def digest_matches(stored_digest: str | None, token: str) -> bool: + """Constant-time comparison of *token* against a stored digest.""" + if not isinstance(stored_digest, str) or not stored_digest: + return False + return hmac.compare_digest(stored_digest, pat_token_digest(token)) + + +def extract_bearer_token(authorization: str | None) -> str | None: + """Return the Bearer credential from an Authorization header value. + + ``None`` means the request carries no Authorization header at all, so the + caller should fall through to the session-cookie path. Any other unusable + value (non-Bearer scheme, empty credential) returns ``""`` so callers + treat it as an invalid credential rather than an absent one. + """ + if authorization is None: + return None + scheme, _, value = authorization.partition(" ") + if scheme.lower() != "bearer": + return "" + return value.strip() + + +async def authenticate_pat(app: Any, authorization: str | None) -> tuple[Any, frozenset[str]]: + """Validate the Bearer credential and resolve its owning user. + + Returns ``(user, scopes)``. Every token-verdict failure mode — malformed + token, unknown/revoked/expired token, PAT store not configured, missing + owning user — raises the same generic 401 so responses cannot serve as an + oracle on which check failed. Infrastructure errors (store I/O failures) + propagate and fail closed; they are not part of the token verdict. + """ + from fastapi import HTTPException + + token = extract_bearer_token(authorization) + if not token or not token.startswith(PAT_TOKEN_PREFIX): + raise HTTPException(status_code=401, detail="Invalid token") + pat_repo = getattr(app.state, "pat_repo", None) + if pat_repo is None: + raise HTTPException(status_code=401, detail="Invalid token") + record = await pat_repo.get_active_by_digest(pat_token_digest(token)) + if record is None or not digest_matches(record.get("token_digest"), token): + raise HTTPException(status_code=401, detail="Invalid token") + from app.gateway.deps import get_local_provider + + user = await get_local_provider().get_user(str(record["user_id"])) + if user is None: + # The owning user was deleted or became unresolvable; the token is + # dead even though its row survives (deleting a user revokes their + # PATs, without needing a FK cascade). + raise HTTPException(status_code=401, detail="Invalid token") + await pat_repo.touch_last_used(str(record["id"])) + return user, frozenset(record.get("scopes") or ()) + + +def validate_scopes(scopes: list[str]) -> list[str]: + """Validate a creation-time scope list; returns the deduplicated order.""" + unknown = sorted(set(scopes) - PAT_ALLOWED_SCOPES) + if unknown: + raise ValueError(f"Unknown PAT scopes: {', '.join(unknown)}") + deduplicated = sorted(set(scopes)) + if not deduplicated: + raise ValueError("A PAT must request at least one scope") + return deduplicated diff --git a/backend/app/gateway/auth_disabled.py b/backend/app/gateway/auth_disabled.py index 9b24cd03f..3e4456b57 100644 --- a/backend/app/gateway/auth_disabled.py +++ b/backend/app/gateway/auth_disabled.py @@ -14,6 +14,7 @@ AUTH_DISABLED_USER_EMAIL = "default@test.local" AUTH_SOURCE_SESSION = "session" AUTH_SOURCE_INTERNAL = "internal" +AUTH_SOURCE_PAT = "pat" AUTH_SOURCE_AUTH_DISABLED = "auth_disabled" _PRODUCTION_ENV_VARS: tuple[str, ...] = ("DEER_FLOW_ENV", "ENVIRONMENT") diff --git a/backend/app/gateway/auth_middleware.py b/backend/app/gateway/auth_middleware.py index e1bc0f468..e20cd0d07 100644 --- a/backend/app/gateway/auth_middleware.py +++ b/backend/app/gateway/auth_middleware.py @@ -20,6 +20,7 @@ from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse from app.gateway.auth_disabled import ( AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_INTERNAL, + AUTH_SOURCE_PAT, AUTH_SOURCE_SESSION, get_auth_disabled_user, is_auth_disabled, @@ -106,11 +107,39 @@ class AuthMiddleware(BaseHTTPMiddleware): auth_source = AUTH_SOURCE_SESSION access_token = request.cookies.get("access_token") + authorization = request.headers.get("authorization") + pat_scopes: frozenset[str] = frozenset() # Non-public path: require session cookie if internal_user is not None: user = internal_user auth_source = AUTH_SOURCE_INTERNAL + elif authorization is not None and not is_auth_disabled(): + # Bearer (PAT) credential precedence (#4849): a present-but-invalid + # Authorization header is a hard 401 and never silently falls back + # to the session cookie. This is also what makes the CSRF + # middleware's Bearer skip safe — a cross-site attacker cannot ride + # a victim's cookie by padding the request with a garbage Bearer + # header, because the request dies here before any route runs. + # Auth-disabled mode is an operator override of all authentication, + # so it stays ahead of the Bearer check (a stray Authorization + # header from a proxy must not 401 an E2E sandbox). + from app.gateway.auth.pat import authenticate_pat, is_pat_allowed_route + + try: + user, pat_scopes = await authenticate_pat(request.app, authorization) + except HTTPException as exc: + return JSONResponse(status_code=exc.status_code, content={"detail": exc.detail}) + # Default-deny route boundary (#5041 review P1-1): scopes only + # constrain @require_permission routes, so any route outside the + # explicit PAT policy is closed to PAT callers outright — an + # all-scopes token must not reach undecorated mutation routes. + if not is_pat_allowed_route(request.method, get_request_route_path(request)): + return JSONResponse( + status_code=403, + content={"detail": "PAT credentials are not permitted on this route"}, + ) + auth_source = AUTH_SOURCE_PAT elif access_token: # Strict JWT validation: reject junk/expired tokens with 401 # right here instead of silently passing through. This closes @@ -156,6 +185,12 @@ class AuthMiddleware(BaseHTTPMiddleware): user, is_internal=auth_source == AUTH_SOURCE_INTERNAL, ) + if auth_source == AUTH_SOURCE_PAT: + # A PAT can only narrow its owning user's permissions: the stored + # scopes intersect the resolved route permissions, never widen + # them, and role changes / authorization policy stay authoritative + # because they were resolved fresh from the owning user above. + permissions = [permission for permission in permissions if permission in pat_scopes] request.state.auth = AuthContext(user=user, permissions=permissions) token = set_current_user(user) try: diff --git a/backend/app/gateway/authz.py b/backend/app/gateway/authz.py index f2554adcd..9f9104faa 100644 --- a/backend/app/gateway/authz.py +++ b/backend/app/gateway/authz.py @@ -118,6 +118,28 @@ def get_auth_context(request: Request) -> AuthContext | None: return getattr(request.state, "auth", None) +def require_cancel_permission_if(request: Request, can_cancel: bool) -> None: + """Require ``runs:cancel`` when a request carries cancel capability. + + Cancel capability reaches the run lifecycle through more than the dedicated + cancel route: ``?action=interrupt|rollback`` on the join-stream entry and + ``multitask_strategy=interrupt|rollback`` on run creation both terminate an + already-active run. A credential whose scopes omit ``runs:cancel`` (e.g. a + create- or read-only PAT) must not reach any of those paths, and + decorators cannot express query- or body-parameter-conditional + permissions, so callers apply this check where the capability is known. + + ``request.state.auth`` may be absent in middleware-less compositions + (unit-test stubs, auth-disabled startup); the shipped Gateway always + stamps it via ``AuthMiddleware`` before handlers run. + """ + if not can_cancel: + return + auth = getattr(request.state, "auth", None) + if auth is not None and not auth.has_permission("runs", "cancel"): + raise HTTPException(status_code=403, detail="Permission denied: runs:cancel") + + _ALL_PERMISSIONS: list[str] = [ Permissions.THREADS_READ, Permissions.THREADS_WRITE, @@ -552,14 +574,22 @@ def require_permission( async def wrapper(*args: Any, **kwargs: Any) -> Any: request = kwargs.get("request") if request is None: - # Unit tests may call decorated route handlers directly without - # constructing a FastAPI Request object. Inject a minimal stub - # when the wrapped function declares `request`. - if "request" in inspect.signature(func).parameters: + # Unit tests may call decorated route handlers directly — with + # or without constructing a FastAPI Request object — and may + # pass ``request`` positionally. Bind to the real signature + # first so a positional request is found rather than + # duplicated by the stub injection below. + try: + bound = inspect.signature(func).bind_partial(*args, **kwargs) + except TypeError: + bound = None + if bound is not None and "request" in bound.arguments: + request = bound.arguments["request"] + elif "request" in inspect.signature(func).parameters: kwargs["request"] = _make_test_request_stub() + request = kwargs["request"] else: return await func(*args, **kwargs) - request = kwargs["request"] if getattr(request, "_deerflow_test_bypass_auth", False): return await func(*args, **kwargs) diff --git a/backend/app/gateway/csrf_middleware.py b/backend/app/gateway/csrf_middleware.py index d777417bf..aa823c2c2 100644 --- a/backend/app/gateway/csrf_middleware.py +++ b/backend/app/gateway/csrf_middleware.py @@ -222,7 +222,15 @@ class CSRFMiddleware(BaseHTTPMiddleware): content={"detail": "Cross-site auth request denied."}, ) - if should_check_csrf(request) and not _is_auth: + if should_check_csrf(request) and not _is_auth and request.headers.get("authorization") is None: + # Bearer-authenticated requests (PAT, #4849) are exempt from the + # cookie double-submit check only — the cross-site origin check on + # auth endpoints above still runs for every request. Safety rests + # on AuthMiddleware's strict Bearer precedence: an invalid Bearer + # header is a 401 there, so a cross-site attacker cannot ride a + # victim's cookie by padding a garbage Authorization header, and a + # cross-site request carrying a custom Authorization header at all + # requires a CORS preflight the attacker cannot obtain. cookie_token = request.cookies.get(CSRF_COOKIE_NAME) header_token = request.headers.get(CSRF_HEADER_NAME) diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index aba322aea..240ef4af1 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -453,15 +453,22 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen sf = get_session_factory() if sf is not None: from deerflow.persistence.feedback import FeedbackRepository + from deerflow.persistence.personal_access_tokens import PersonalAccessTokenRepository from deerflow.persistence.run import RunRepository app.state.run_store = RunRepository(sf) app.state.feedback_repo = FeedbackRepository(sf) + from app.gateway.auth.pat import PAT_LAST_USED_WRITE_INTERVAL_SECONDS + + app.state.pat_repo = PersonalAccessTokenRepository(sf, last_used_write_interval_seconds=PAT_LAST_USED_WRITE_INTERVAL_SECONDS) else: from deerflow.runtime.runs.store.memory import MemoryRunStore app.state.run_store = MemoryRunStore() app.state.feedback_repo = None + # Memory backend has no durable PAT store, so Bearer credentials + # cannot be validated there and are rejected by the middleware. + app.state.pat_repo = None # Services are app-scoped. Capture this app's immutable extension set # once and close over the same object for teardown; the process-wide @@ -752,6 +759,19 @@ def get_local_provider() -> LocalAuthProvider: return _cached_local_provider +def get_pat_repo(request: Request): + """Return the personal-access-token repository from app state. + + Raises 503 when the process runs on the memory backend (no durable PAT + storage), so PAT management routes fail explicitly instead of silently + accepting tokens nobody can validate. + """ + pat_repo = getattr(request.app.state, "pat_repo", None) + if pat_repo is None: + raise HTTPException(status_code=503, detail="Personal access tokens require a configured database") + return pat_repo + + async def get_current_user_from_request(request: Request): """Get the current authenticated user from the request cookie. @@ -759,12 +779,13 @@ async def get_current_user_from_request(request: Request): """ state = getattr(request, "state", None) state_user = getattr(state, "user", None) - from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_INTERNAL, AUTH_SOURCE_SESSION + from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_INTERNAL, AUTH_SOURCE_PAT, AUTH_SOURCE_SESSION if state_user is not None and getattr(state, "auth_source", None) in { AUTH_SOURCE_SESSION, AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_INTERNAL, + AUTH_SOURCE_PAT, }: return state_user @@ -817,6 +838,13 @@ async def is_admin_user(request: Request) -> bool: per-router copies that previously existed in ``mcp``, ``channel_connections`` and ``channels``. """ + # PAT credentials never carry admin capability: no scope in the PAT + # universe grants it, so an admin's automation token must not unlock + # admin-only routes (skill installs, integration credentials, MCP config). + from app.gateway.auth_disabled import AUTH_SOURCE_PAT + + if getattr(request.state, "auth_source", None) == AUTH_SOURCE_PAT: + return False user = getattr(request.state, "user", None) if user is None: user = await get_current_user_from_request(request) diff --git a/backend/app/gateway/routers/auth.py b/backend/app/gateway/routers/auth.py index f9c1c3c01..701e1a83b 100644 --- a/backend/app/gateway/routers/auth.py +++ b/backend/app/gateway/routers/auth.py @@ -31,6 +31,7 @@ from app.gateway.auth.oidc_state import ( get_state_cookie, set_state_cookie, ) +from app.gateway.auth.pat import PAT_MAX_NAME_LENGTH from app.gateway.auth.session_cookie import ACCESS_TOKEN_COOKIE_NAME, SESSION_PERSISTENCE_COOKIE_NAME, set_session_cookie from app.gateway.auth.session_cookie_state import SKIP_AUTH_CSRF_COOKIE_STATE_ATTR from app.gateway.auth.user_provisioning import get_or_provision_oidc_user @@ -391,11 +392,18 @@ async def change_password(request: Request, response: Response, body: ChangePass - Re-issues session cookie with new token_version """ from app.gateway.auth.password import hash_password_async, verify_password_async - from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED + from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_PAT user = await get_current_user_from_request(request) - if getattr(request.state, "auth_source", None) == AUTH_SOURCE_AUTH_DISABLED: + if getattr(request.state, "auth_source", None) in {AUTH_SOURCE_PAT, AUTH_SOURCE_AUTH_DISABLED}: + # PAT-authenticated callers must not alter auth state (#4849 point 6); + # auth-disabled mode has no passwords to change. + if getattr(request.state, "auth_source", None) == AUTH_SOURCE_PAT: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Password changes require interactive session authentication", + ) raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail=AuthErrorResponse( @@ -450,6 +458,130 @@ async def get_me(request: Request): ) +# ── Personal Access Tokens (#4849) ──────────────────────────────────────── + + +def require_session_source(request: Request) -> None: + """Reject non-session credentials from auth-state-altering routes. + + PAT-authenticated callers must not manage PATs or change passwords + (#4849 point 6): a leaked automation token could otherwise mint fresh + long-lived credentials or lock out the human owner. + """ + from app.gateway.auth_disabled import AUTH_SOURCE_SESSION + + if getattr(request.state, "auth_source", None) != AUTH_SOURCE_SESSION: + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, detail="This endpoint requires interactive session authentication") + + +class PATCreateRequest(BaseModel): + name: str = Field(min_length=1, max_length=PAT_MAX_NAME_LENGTH) + scopes: list[str] = Field(min_length=1) + expires_in_days: int | None = Field(default=None, ge=1, le=365) # None = never expires + + @field_validator("name") + @classmethod + def _strip_and_require_non_empty_name(cls, value: str) -> str: + # A whitespace-only name passes min_length but would persist as an + # empty label; the trimmed value is what gets stored and shown. + stripped = value.strip() + if not stripped: + raise ValueError("PAT name must contain at least one non-whitespace character") + return stripped + + +class PATCreatedResponse(BaseModel): + """Create response — ``token`` is the raw show-once credential.""" + + id: str + name: str + scopes: list[str] + expires_at: str | None + created_at: str + token: str + + +class PATSummaryResponse(BaseModel): + id: str + name: str + scopes: list[str] + expires_at: str | None + last_used_at: str | None + created_at: str + revoked_at: str | None + + +def _pat_summary(record: dict) -> PATSummaryResponse: + return PATSummaryResponse( + id=str(record["id"]), + name=str(record["name"]), + scopes=list(record.get("scopes") or []), + expires_at=str(record["expires_at"]) if record.get("expires_at") else None, + last_used_at=str(record["last_used_at"]) if record.get("last_used_at") else None, + created_at=str(record["created_at"]), + revoked_at=str(record["revoked_at"]) if record.get("revoked_at") else None, + ) + + +@router.post("/pats", status_code=status.HTTP_201_CREATED, response_model=PATCreatedResponse, dependencies=[Depends(require_session_source)]) +async def create_pat(request: Request, body: PATCreateRequest): + """Create a personal access token for the session user. + + The raw token is returned exactly once and cannot be retrieved again; + only its SHA-256 digest is persisted. + """ + from datetime import UTC, datetime, timedelta + + from app.gateway.auth.pat import generate_pat_token, pat_token_digest, validate_scopes + from app.gateway.deps import get_pat_repo + + user = await get_current_user_from_request(request) + try: + scopes = validate_scopes(body.scopes) + except ValueError as exc: + raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail=str(exc)) from exc + + token = generate_pat_token() + expires_at = datetime.now(UTC) + timedelta(days=body.expires_in_days) if body.expires_in_days is not None else None + record = await get_pat_repo(request).create( + user_id=str(user.id), + name=body.name.strip(), + scopes=scopes, + token_digest=pat_token_digest(token), + expires_at=expires_at, + ) + return PATCreatedResponse( + id=str(record["id"]), + name=str(record["name"]), + scopes=list(record.get("scopes") or []), + expires_at=str(record["expires_at"]) if record.get("expires_at") else None, + created_at=str(record["created_at"]), + token=token, + ) + + +@router.get("/pats", response_model=list[PATSummaryResponse], dependencies=[Depends(require_session_source)]) +async def list_pats(request: Request): + """List the session user's tokens. Never returns digests or raw tokens.""" + from app.gateway.deps import get_pat_repo + + user = await get_current_user_from_request(request) + records = await get_pat_repo(request).list_for_user(str(user.id)) + return [_pat_summary(record) for record in records] + + +@router.delete("/pats/{pat_id}", response_model=MessageResponse, dependencies=[Depends(require_session_source)]) +async def revoke_pat(request: Request, pat_id: str): + """Revoke one of the session user's tokens. Revocation is immediate.""" + from app.gateway.deps import get_pat_repo + + user = await get_current_user_from_request(request) + revoked = await get_pat_repo(request).revoke(pat_id, str(user.id)) + if not revoked: + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Token not found") + return MessageResponse(message="Token revoked") + + # Per-IP cache: ip → (timestamp, result_dict). # Returns the cached result within the TTL instead of 429, because # the answer (whether an admin exists) rarely changes and returning diff --git a/backend/app/gateway/routers/thread_runs.py b/backend/app/gateway/routers/thread_runs.py index 024f617ac..f257820c5 100644 --- a/backend/app/gateway/routers/thread_runs.py +++ b/backend/app/gateway/routers/thread_runs.py @@ -23,7 +23,7 @@ from fastapi.responses import Response, StreamingResponse from langchain_core.messages import BaseMessage from pydantic import BaseModel, Field -from app.gateway.authz import require_permission +from app.gateway.authz import require_cancel_permission_if, require_permission from app.gateway.checkpoint_lineage import ( CheckpointLineageError, CheckpointParentMissingError, @@ -204,6 +204,20 @@ class ThreadTokenUsageResponse(BaseModel): # --------------------------------------------------------------------------- +def require_cancel_permission_when_action(request: Request, action: str | None) -> None: + """Conditionally require ``runs:cancel`` for cancel-then-stream requests. + + ``stream_existing_run`` is gated at ``runs:read`` so action-less stream + joins keep working with read-only credentials, but its ``action`` branch + cancels the run — a separate permission. A read-only PAT (or any read-only + credential) must not reach the cancel path, and decorators cannot express + query-parameter-conditional permissions, so the check lives here. See + ``authz.require_cancel_permission_if`` — the shared primitive for every + request dimension that carries cancel capability. + """ + require_cancel_permission_if(request, action is not None) + + def _cancel_conflict_detail(run_id: str, record: RunRecord) -> str: if record.status in (RunStatus.pending, RunStatus.running): return f"Run {run_id} is not active on this worker and cannot be cancelled" @@ -993,7 +1007,9 @@ async def join_run(thread_id: ThreadId, run_id: str, request: Request) -> Stream raise HTTPException(status_code=409, detail=f"Run {run_id} is not active on this worker and cannot be streamed") return StreamingResponse( - sse_consumer(bridge, record, request, run_mgr), + # Joins are read-only observation: the creator's cancel-on-disconnect + # policy must not fire because an observer closed their connection. + sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", @@ -1024,6 +1040,8 @@ async def stream_existing_run( is present the run is cancelled first; the response then streams any remaining buffered events so the client observes a clean shutdown. """ + require_cancel_permission_when_action(request, action) + run_mgr = get_run_manager(request) record = await run_mgr.get(run_id) if record is None or record.thread_id != thread_id: @@ -1066,7 +1084,12 @@ async def stream_existing_run( return Response(status_code=204 if completed else 202) return StreamingResponse( - sse_consumer(bridge, record, request, run_mgr), + # Both methods of this handler are join surfaces: a POST carrying an + # action cancels explicitly above (already gated by + # require_cancel_permission_when_action), and an action-less join is + # read-only observation — the creator's cancel-on-disconnect policy + # must not fire because a joiner closed their connection. + sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False), media_type="text/event-stream", headers={ "Cache-Control": "no-cache", diff --git a/backend/app/gateway/routers/threads.py b/backend/app/gateway/routers/threads.py index 8aeb1d95c..6278bbd7b 100644 --- a/backend/app/gateway/routers/threads.py +++ b/backend/app/gateway/routers/threads.py @@ -781,6 +781,7 @@ def _existing_thread_response(thread_id: str, record: dict) -> ThreadResponse: @router.post("", response_model=ThreadResponse) +@require_permission("threads", "write") async def create_thread(body: ThreadCreateRequest, request: Request) -> ThreadResponse: """Create a new thread. @@ -1055,6 +1056,7 @@ async def _branch_thread_with_reservation( @router.post("/search", response_model=list[ThreadResponse]) +@require_permission("threads", "read") async def search_threads(body: ThreadSearchRequest, request: Request) -> list[ThreadResponse]: """Search and list threads. diff --git a/backend/app/gateway/services.py b/backend/app/gateway/services.py index 0ec631234..8fc64a162 100644 --- a/backend/app/gateway/services.py +++ b/backend/app/gateway/services.py @@ -23,6 +23,7 @@ from langchain_core.messages.utils import convert_to_messages from langgraph.types import Command from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL +from app.gateway.authz import require_cancel_permission_if from app.gateway.deps import get_checkpointer, get_local_provider, get_run_context, get_run_manager, get_stream_bridge from app.gateway.internal_auth import ( INTERNAL_OWNER_USER_ID_HEADER_NAME, @@ -1215,6 +1216,15 @@ async def start_run( Reject a missing thread instead of auto-creating metadata. Internal notification runs use this so a deleted chat cannot be resurrected. """ + # Cancel-capability gate. interrupt/rollback strategies terminate an already + # active run — runs:cancel capability, not runs:create — so a create-only + # PAT must not reach them. Enforced here, the single choke point every + # run-creation path flows through (HTTP routes and internal launchers + # alike), so no entry point can bypass it; regenerate launches pass + # multitask_strategy="reject" and are unaffected. Requests without a + # stamped auth context (internal/test compositions) skip the gate. + require_cancel_permission_if(request, body.multitask_strategy != "reject") + try: validate_thread_id(thread_id) except ValueError as exc: @@ -1610,12 +1620,22 @@ async def sse_consumer( record: RunRecord, request: Request, run_mgr: RunManager, + *, + apply_on_disconnect: bool = True, ): """Async generator that yields SSE frames from the bridge. - The ``finally`` block implements ``on_disconnect`` semantics: + The ``finally`` block implements ``on_disconnect`` semantics, but only for + the stream returned by the *creating* endpoint (``apply_on_disconnect=True``): + - ``cancel``: abort the background task on client disconnect. - ``continue``: let the task run; events are discarded. + + Join/observer streams pass ``apply_on_disconnect=False``: the creator's + cancel-on-disconnect policy expresses the creator's intent for their own + connection, and a read-only observer closing a join must not cancel the + run (a runs:read-only credential would otherwise cancel without + runs:cancel just by disconnecting). """ last_event_id = request.headers.get("Last-Event-ID") if await _terminal_record_stream_missing(bridge, record): @@ -1660,8 +1680,9 @@ async def sse_consumer( # store_only records are cross-worker observation handles. An explicit # cancel-then-stream action has already persisted its request before # subscribing; a plain join disconnect must not invent a new - # cancellation request. Only apply on_disconnect to locally-owned runs. - if not gap_emitted and not record.store_only and record.status in (RunStatus.pending, RunStatus.running): + # cancellation request. Only apply on_disconnect to locally-owned runs, + # and only on the creator's own stream — never on an observer join. + if apply_on_disconnect and not gap_emitted and not record.store_only and record.status in (RunStatus.pending, RunStatus.running): if record.on_disconnect == DisconnectMode.cancel: await run_mgr.cancel(record.run_id) @@ -1674,6 +1695,12 @@ async def wait_for_run_completion( ) -> bool: """Block until the run publishes ``END_SENTINEL``, honouring on_disconnect. + Creator-side only, unlike ``sse_consumer``'s observer joins: every caller + must be the endpoint that created the run or a path reached only after an + explicit, permission-gated cancel. This helper intentionally keeps + applying the record's ``on_disconnect`` policy on disconnect — do not + wire it to observer surfaces. + The non-streaming ``/wait`` endpoints used to ``await record.task`` directly with no disconnect handling. When the client (or an intermediate HTTP proxy) timed out during a long tool call such as diff --git a/backend/docs/API.md b/backend/docs/API.md index f5ea99e82..c70470af7 100644 --- a/backend/docs/API.md +++ b/backend/docs/API.md @@ -16,6 +16,118 @@ For agent conversations, clients can either pre-create a thread endpoint (`POST /api/langgraph/runs/stream`). The latter auto-creates a thread and returns `thread_id` and `run_id` in the response `Content-Location` header. +## Authentication + +Browser sessions authenticate with the `access_token` session cookie issued at +login. Programmatic clients can instead use a **personal access token (PAT)** +sent as a Bearer credential: + +```http +POST /api/threads/search +Authorization: Bearer dfp_... +Content-Type: application/json + +{} +``` + +PATs require a configured database backend (SQLite/PostgreSQL) — on the +memory-only backend, Bearer credentials are rejected and PAT management routes +return `503`. + +### Personal Access Tokens + +Base URL: `/api/v1/auth` + +PAT management requires an **interactive session** (a PAT cannot manage PATs +or change passwords, so a leaked automation token cannot mint fresh +credentials). The raw token is returned **exactly once** at creation; only its +SHA-256 digest is stored server-side. + +#### Create Token + +```http +POST /api/v1/auth/pats +Content-Type: application/json +``` + +**Request Body:** +```json +{ + "name": "ci-runner", + "scopes": ["threads:read", "runs:create", "runs:read"], + "expires_in_days": 90 +} +``` + +- `scopes` — subset of the route permissions: `threads:read`, `threads:write`, + `threads:delete`, `runs:create`, `runs:read`, `runs:cancel`. A PAT can only + *narrow* its owning user's permissions, never widen them. +- `expires_in_days` — optional (`1`–`365`); omitted means the token never expires. + +**Response (`201`):** +```json +{ + "id": "0f0c6e6a-...", + "name": "ci-runner", + "scopes": ["runs:create", "runs:read", "threads:read"], + "expires_at": "2026-11-25T10:30:00Z", + "created_at": "2026-08-27T10:30:00Z", + "token": "dfp_..." +} +``` + +Save `token` immediately — it cannot be retrieved again. + +#### List Tokens + +```http +GET /api/v1/auth/pats +``` + +Returns the caller's tokens with `last_used_at` / `revoked_at` audit fields; +never returns digests or raw tokens. + +#### Revoke Token + +```http +DELETE /api/v1/auth/pats/{pat_id} +``` + +Revocation is immediate. + +### PAT Constraints + +- A request carrying an `Authorization` header that fails validation gets a + hard `401` — it never falls back to the session cookie. +- **Cancel capability requires `runs:cancel` on every request dimension that + carries it**, not just the dedicated cancel route: `?action=interrupt|rollback` + on `POST /api/threads/{thread_id}/runs/{run_id}/stream` (action-less joins + stay at `runs:read`), and `multitask_strategy=interrupt|rollback` on run + creation (the default `reject` stays at `runs:create`). Joining a run's + stream is pure observation — an observer disconnecting never cancels the run. +- **Route-level default-deny:** PAT requests are admitted only to the + thread/run lifecycle routes the v1 scopes govern — `POST /api/threads` + (create), `POST /api/threads/search` (list), `GET/PATCH/DELETE + /api/threads/{thread_id}`, the thread `goal`/`state`/`compact`/`history`/ + `branches` subroutes, and exactly the implemented `/runs` subroutes + (`GET|POST /api/threads/{thread_id}/runs`, the POST-only `stream`, `wait`, + `regenerate/prepare`, and `edit-regenerate/prepare` collection endpoints, + `GET /api/threads/{thread_id}/runs/{run_id}` plus its `cancel` (POST), + `join`/`messages`/`events`/`workspace-changes` (GET), and + `GET|POST .../runs/{run_id}/stream`), plus `POST /api/runs/stream|wait` and + `GET /api/runs/{run_id}/messages|feedback`. A route added under `/runs` is + denied until explicitly added to the policy. + Every other authenticated route — memory, agents, models, MCP/skills + config, integrations, channels, uploads — answers `403` to PAT callers + regardless of scopes. Scope enforcement alone only constrains + permission-decorated routes, so the allowlist is the outer boundary; + session-cookie callers are unaffected. +- PAT credentials never carry admin capability, even when the owning user is + an admin. This includes extension-contributed admin routes: the extension + principal projection suppresses every admin signal for PAT callers. +- Revoking or deleting the owning user invalidates their PATs on the next + request. + ## LangGraph-compatible API Base URL: `/api/langgraph` diff --git a/backend/packages/harness/deerflow/persistence/migrations/versions/0017_personal_access_tokens.py b/backend/packages/harness/deerflow/persistence/migrations/versions/0017_personal_access_tokens.py new file mode 100644 index 000000000..9d31dcab0 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/migrations/versions/0017_personal_access_tokens.py @@ -0,0 +1,52 @@ +"""personal access tokens. + +Revision ID: 0017_personal_access_tokens +Revises: 0016_subagent_batches +Create Date: 2026-08-26 + +Numbering note: generated against the then-current main head (0016), as were +the 0017 migrations in #5078 (conversation shares) and #4843 (notification +deliveries, claiming 0017+0018). Whichever merges first keeps the slot; the +others renumber on rebase — adjust ``revision``/``down_revision`` here and +the migration-head assertions in tests/test_persistence_bootstrap*.py. +""" + +from __future__ import annotations + +from collections.abc import Sequence + +import sqlalchemy as sa +from alembic import op + +revision: str = "0017_personal_access_tokens" +down_revision: str | Sequence[str] | None = "0016_subagent_batches" +branch_labels: str | Sequence[str] | None = None +depends_on: str | Sequence[str] | None = None + + +def upgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if not inspector.has_table("personal_access_tokens"): + op.create_table( + "personal_access_tokens", + sa.Column("id", sa.String(length=64), nullable=False), + sa.Column("user_id", sa.String(length=64), nullable=False), + sa.Column("name", sa.String(length=128), nullable=False), + sa.Column("token_digest", sa.String(length=64), nullable=False), + sa.Column("scopes", sa.JSON(), nullable=False), + sa.Column("expires_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("last_used_at", sa.DateTime(timezone=True), nullable=True), + sa.Column("created_at", sa.DateTime(timezone=True), nullable=False), + sa.Column("revoked_at", sa.DateTime(timezone=True), nullable=True), + sa.PrimaryKeyConstraint("id"), + ) + op.create_index("ix_personal_access_tokens_user_id", "personal_access_tokens", ["user_id"]) + op.create_index("ix_personal_access_tokens_token_digest", "personal_access_tokens", ["token_digest"], unique=True) + + +def downgrade() -> None: + inspector = sa.inspect(op.get_bind()) + if inspector.has_table("personal_access_tokens"): + op.drop_index("ix_personal_access_tokens_token_digest", table_name="personal_access_tokens") + op.drop_index("ix_personal_access_tokens_user_id", table_name="personal_access_tokens") + op.drop_table("personal_access_tokens") diff --git a/backend/packages/harness/deerflow/persistence/models/__init__.py b/backend/packages/harness/deerflow/persistence/models/__init__.py index 0732a68a3..c6b374ae2 100644 --- a/backend/packages/harness/deerflow/persistence/models/__init__.py +++ b/backend/packages/harness/deerflow/persistence/models/__init__.py @@ -25,6 +25,7 @@ from deerflow.persistence.feedback.model import FeedbackRow from deerflow.persistence.managed_subagents.model import ManagedSubagentRow from deerflow.persistence.mcp_tasks.model import McpTaskRow from deerflow.persistence.models.run_event import RunEventRow +from deerflow.persistence.personal_access_tokens.model import PersonalAccessTokenRow from deerflow.persistence.run.model import RunRow from deerflow.persistence.scheduled_task_runs.model import ScheduledTaskRunRow from deerflow.persistence.scheduled_tasks.model import ScheduledTaskRow @@ -42,6 +43,7 @@ __all__ = [ "FeedbackRow", "McpTaskRow", "ManagedSubagentRow", + "PersonalAccessTokenRow", "RunEventRow", "RunRow", "ScheduledTaskRow", diff --git a/backend/packages/harness/deerflow/persistence/personal_access_tokens/__init__.py b/backend/packages/harness/deerflow/persistence/personal_access_tokens/__init__.py new file mode 100644 index 000000000..f95e2d7e0 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/personal_access_tokens/__init__.py @@ -0,0 +1,6 @@ +"""Personal access token persistence — ORM and SQL repository.""" + +from deerflow.persistence.personal_access_tokens.model import PersonalAccessTokenRow +from deerflow.persistence.personal_access_tokens.sql import PersonalAccessTokenRepository + +__all__ = ["PersonalAccessTokenRepository", "PersonalAccessTokenRow"] diff --git a/backend/packages/harness/deerflow/persistence/personal_access_tokens/model.py b/backend/packages/harness/deerflow/persistence/personal_access_tokens/model.py new file mode 100644 index 000000000..58fb6d1fe --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/personal_access_tokens/model.py @@ -0,0 +1,31 @@ +"""ORM model for personal access tokens (PAT).""" + +from __future__ import annotations + +from datetime import UTC, datetime + +from sqlalchemy import JSON, DateTime, Index, String +from sqlalchemy.orm import Mapped, mapped_column + +from deerflow.persistence.base import Base + + +class PersonalAccessTokenRow(Base): + __tablename__ = "personal_access_tokens" + + __table_args__ = (Index("ix_personal_access_tokens_token_digest", "token_digest", unique=True),) + + id: Mapped[str] = mapped_column(String(64), primary_key=True) + user_id: Mapped[str] = mapped_column(String(64), nullable=False, index=True) + name: Mapped[str] = mapped_column(String(128), nullable=False) + # SHA-256 hex digest of the ``dfp_…`` token. The raw token exists only in + # the create response and is never persisted or logged. The named unique + # index (rather than a column-level constraint) keeps ``create_all`` output + # identical to migration 0017, so downgrades work on bootstrapped DBs too. + token_digest: Mapped[str] = mapped_column(String(64), nullable=False) + # Subset of the route-permission strings owned by ``app.gateway.authz``. + scopes: Mapped[list[str]] = mapped_column(JSON, nullable=False) + expires_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + last_used_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False, default=lambda: datetime.now(UTC)) + revoked_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True) diff --git a/backend/packages/harness/deerflow/persistence/personal_access_tokens/sql.py b/backend/packages/harness/deerflow/persistence/personal_access_tokens/sql.py new file mode 100644 index 000000000..c61540939 --- /dev/null +++ b/backend/packages/harness/deerflow/persistence/personal_access_tokens/sql.py @@ -0,0 +1,132 @@ +"""SQLAlchemy-backed personal access token storage. + +Each method acquires its own short-lived session. The raw ``dfp_…`` token is +generated and returned by the caller (the app layer) exactly once; this +repository only ever persists the SHA-256 digest passed to :meth:`create`. +""" + +from __future__ import annotations + +import logging +import time +import uuid +from datetime import UTC, datetime +from typing import Any + +from sqlalchemy import select, update +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from deerflow.persistence.personal_access_tokens.model import PersonalAccessTokenRow +from deerflow.utils.time import coerce_iso + +logger = logging.getLogger(__name__) + + +class PersonalAccessTokenRepository: + def __init__(self, session_factory: async_sessionmaker[AsyncSession], *, last_used_write_interval_seconds: float = 300.0) -> None: + self._sf = session_factory + self._last_used_write_interval = last_used_write_interval_seconds + self._last_used_written_at: dict[str, float] = {} + + @staticmethod + def _row_to_dict(row: PersonalAccessTokenRow) -> dict[str, Any]: + d = row.to_dict() + for key in ("expires_at", "last_used_at", "created_at", "revoked_at"): + val = d.get(key) + if isinstance(val, datetime): + # SQLite drops tzinfo on read; normalize so output is tz-aware. + d[key] = coerce_iso(val) + return d + + async def create( + self, + *, + user_id: str, + name: str, + scopes: list[str], + token_digest: str, + expires_at: datetime | None = None, + ) -> dict[str, Any]: + row = PersonalAccessTokenRow( + id=str(uuid.uuid4()), + user_id=user_id, + name=name, + token_digest=token_digest, + scopes=sorted(scopes), + expires_at=expires_at, + created_at=datetime.now(UTC), + ) + async with self._sf() as session: + session.add(row) + await session.commit() + await session.refresh(row) + return self._row_to_dict(row) + + async def get_active_by_digest(self, token_digest: str) -> dict[str, Any] | None: + """Return the non-revoked, non-expired row for *token_digest*. + + Revocation and expiry are evaluated here so a stale durable row can + never authenticate even though it remains readable for audit history. + """ + async with self._sf() as session: + row = (await session.execute(select(PersonalAccessTokenRow).where(PersonalAccessTokenRow.token_digest == token_digest))).scalar_one_or_none() + if row is None or row.revoked_at is not None: + return None + expires_at = row.expires_at + if expires_at is not None: + # SQLite drops tzinfo on read; normalize before comparing. + if expires_at.tzinfo is None: + expires_at = expires_at.replace(tzinfo=UTC) + if expires_at <= datetime.now(UTC): + return None + return self._row_to_dict(row) + + async def list_for_user(self, user_id: str) -> list[dict[str, Any]]: + async with self._sf() as session: + rows = (await session.execute(select(PersonalAccessTokenRow).where(PersonalAccessTokenRow.user_id == user_id).order_by(PersonalAccessTokenRow.created_at.desc()))).scalars() + return [self._row_to_dict(row) for row in rows] + + async def revoke(self, pat_id: str, user_id: str) -> bool: + """Revoke one of *user_id*'s tokens; returns False if not owned/absent.""" + async with self._sf() as session: + result = await session.execute( + update(PersonalAccessTokenRow) + .where( + PersonalAccessTokenRow.id == pat_id, + PersonalAccessTokenRow.user_id == user_id, + PersonalAccessTokenRow.revoked_at.is_(None), + ) + .values(revoked_at=datetime.now(UTC)) + ) + await session.commit() + return result.rowcount != 0 + + def _should_write_last_used(self, pat_id: str) -> bool: + now = time.monotonic() + last = self._last_used_written_at.get(pat_id) + if last is not None and (now - last) < self._last_used_write_interval: + return False + # Bound the stamp cache: revoked/expired tokens never return here, so + # their entries are stale by definition once the cache outgrows very + # active token populations. + if len(self._last_used_written_at) > 4096: + self._last_used_written_at.clear() + self._last_used_written_at[pat_id] = now + return True + + async def touch_last_used(self, pat_id: str) -> None: + """Best-effort, throttled usage stamp (at most one write per interval). + + Never raises: a failure to stamp usage must not fail the request. On + failure the throttle window is rolled back so the next attempt + retries promptly instead of waiting out the full interval. + """ + if not self._should_write_last_used(pat_id): + return + try: + async with self._sf() as session: + await session.execute(update(PersonalAccessTokenRow).where(PersonalAccessTokenRow.id == pat_id).values(last_used_at=datetime.now(UTC))) + await session.commit() + except Exception: + self._last_used_written_at.pop(pat_id, None) + logger.debug("Failed to stamp last_used_at for PAT %s (non-fatal)", pat_id, exc_info=True) diff --git a/backend/tests/test_extension_route_principal.py b/backend/tests/test_extension_route_principal.py index 0c3fb57c9..1c7f71314 100644 --- a/backend/tests/test_extension_route_principal.py +++ b/backend/tests/test_extension_route_principal.py @@ -107,3 +107,31 @@ def test_the_installed_resolver_projects_system_role_into_roles(_stub_app_config no_role_request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="u3", system_role=None), auth_source=None)) assert resolver(no_role_request).roles == () + + +def test_the_installed_resolver_suppresses_admin_for_pat_callers(_stub_app_config): + """P1 regression (#5041 review): an admin-owned PAT must not regain admin + capability through the extension principal projection. Every admin signal + — ``is_admin`` and the ``admin`` role — is suppressed for PAT callers, + matching the documented guarantee that PAT credentials never carry admin + capability.""" + from app.gateway.app import create_app + from app.gateway.auth_disabled import AUTH_SOURCE_PAT, AUTH_SOURCE_SESSION + + app = create_app() + resolver = getattr(app.state, EXTENSION_PRINCIPAL_RESOLVER_KEY) + + pat_request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="u1", system_role="admin"), auth_source=AUTH_SOURCE_PAT)) + pat_principal = resolver(pat_request) + assert pat_principal.is_admin is False + assert "admin" not in pat_principal.roles + + # Control: the same admin over a session cookie still projects admin. + session_request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="u1", system_role="admin"), auth_source=AUTH_SOURCE_SESSION)) + session_principal = resolver(session_request) + assert session_principal.is_admin is True + assert session_principal.roles == ("admin",) + + # A non-admin PAT keeps its plain role: only admin signals are suppressed. + plain_pat_request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="u2", system_role="user"), auth_source=AUTH_SOURCE_PAT)) + assert resolver(plain_pat_request).roles == ("user",) diff --git a/backend/tests/test_migration_0004_run_ownership_dedupe.py b/backend/tests/test_migration_0004_run_ownership_dedupe.py index dbda04f7f..c4f0d759a 100644 --- a/backend/tests/test_migration_0004_run_ownership_dedupe.py +++ b/backend/tests/test_migration_0004_run_ownership_dedupe.py @@ -157,7 +157,7 @@ async def test_migration_dedupes_duplicate_active_rows_before_unique_index(tmp_p with sqlite3.connect(db_path) as raw: version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() # Bootstrap upgrades through the later revisions after 0004. - assert version_row[0] == "0016_subagent_batches" + assert version_row[0] == "0017_personal_access_tokens" # Sanity: the invariant the index enforces is now true — at most one # active row per thread. diff --git a/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py b/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py index aabd312e5..ae0922d82 100644 --- a/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py +++ b/backend/tests/test_migration_0007_scheduled_run_active_dedupe.py @@ -173,7 +173,7 @@ async def test_migration_supersedes_duplicate_active_runs_before_unique_index(tm with sqlite3.connect(db_path) as raw: version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0016_subagent_batches" + assert version_row[0] == "0017_personal_access_tokens" # Sanity: the invariant the index enforces now holds — at most one # active row per task_id. diff --git a/backend/tests/test_migration_0015_scheduled_task_enqueue.py b/backend/tests/test_migration_0015_scheduled_task_enqueue.py index 513ac7767..c2cbb2f69 100644 --- a/backend/tests/test_migration_0015_scheduled_task_enqueue.py +++ b/backend/tests/test_migration_0015_scheduled_task_enqueue.py @@ -57,7 +57,7 @@ async def test_migration_interrupts_legacy_queue_and_adds_claim_fields(tmp_path: # Bootstrap always advances to the repository head after exercising # the 0015 migration behavior below. - assert version == "0016_subagent_batches" + assert version == "0017_personal_access_tokens" assert {"lease_owner", "lease_expires_at", "attempt_count"} <= columns.keys() assert columns["attempt_count"]["nullable"] is False assert overlap_policy == "enqueue" diff --git a/backend/tests/test_migration_0017_personal_access_tokens.py b/backend/tests/test_migration_0017_personal_access_tokens.py new file mode 100644 index 000000000..2271f1e28 --- /dev/null +++ b/backend/tests/test_migration_0017_personal_access_tokens.py @@ -0,0 +1,93 @@ +"""Migration tests for 0017_personal_access_tokens (#4849). + +Runs the full alembic chain on an empty SQLite database (not +``create_all`` + stamp), then exercises the 0017 downgrade/upgrade cycle. +""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pytest +import sqlalchemy as sa +from alembic import command as alembic_command +from alembic.config import Config as AlembicConfig +from sqlalchemy.ext.asyncio import create_async_engine + +from deerflow.persistence.bootstrap import _MIGRATIONS_DIR + +pytestmark = pytest.mark.asyncio + +_SCRIPT_LOCATION = str(_MIGRATIONS_DIR) +_REVISION = "0017_personal_access_tokens" +_PREVIOUS = "0016_subagent_batches" + +_EXPECTED_COLUMNS = { + "id", + "user_id", + "name", + "token_digest", + "scopes", + "expires_at", + "last_used_at", + "created_at", + "revoked_at", +} + + +def _alembic_config(db_url: str) -> AlembicConfig: + cfg = AlembicConfig() + cfg.set_main_option("script_location", _SCRIPT_LOCATION) + # Escape % for ConfigParser (SQLite URLs carry none, Postgres passwords might). + cfg.set_main_option("sqlalchemy.url", db_url.replace("%", "%%")) + return cfg + + +def _table_names(sync_conn) -> set[str]: + return set(sa.inspect(sync_conn).get_table_names()) + + +def _column_names(sync_conn, table: str) -> set[str]: + return {column["name"] for column in sa.inspect(sync_conn).get_columns(table)} + + +async def _inspect(engine, fn): + async with engine.connect() as conn: + return await conn.run_sync(fn) + + +async def test_pat_migration_upgrade_downgrade_cycle(tmp_path: Path) -> None: + db_path = tmp_path / "pat-migration.db" + engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}") + cfg = _alembic_config(f"sqlite+aiosqlite:///{db_path}") + try: + # Alembic's env.py drives migrations with its own asyncio.run, so the + # sync command API must run off the test loop (same wrapper the + # production bootstrap uses). + await asyncio.to_thread(alembic_command.upgrade, cfg, "head") + + tables = await _inspect(engine, _table_names) + assert "personal_access_tokens" in tables + assert "alembic_version" in tables + columns = await _inspect(engine, lambda conn: _column_names(conn, "personal_access_tokens")) + assert columns == _EXPECTED_COLUMNS + indexes = await _inspect( + engine, + lambda conn: {idx["name"] for idx in sa.inspect(conn).get_indexes("personal_access_tokens")}, + ) + # Owner listing + digest lookups are the two hot paths. + assert "ix_personal_access_tokens_user_id" in indexes + assert "ix_personal_access_tokens_token_digest" in indexes + + # Downgrade to the previous revision drops exactly this table. + await asyncio.to_thread(alembic_command.downgrade, cfg, _PREVIOUS) + tables_after_down = await _inspect(engine, _table_names) + assert "personal_access_tokens" not in tables_after_down + + # Upgrade again recreates it (idempotent round trip). + await asyncio.to_thread(alembic_command.upgrade, cfg, "head") + tables_after_up = await _inspect(engine, _table_names) + assert "personal_access_tokens" in tables_after_up + finally: + await engine.dispose() diff --git a/backend/tests/test_pat_auth.py b/backend/tests/test_pat_auth.py new file mode 100644 index 000000000..8d460c15c --- /dev/null +++ b/backend/tests/test_pat_auth.py @@ -0,0 +1,700 @@ +"""Integration tests for PAT authentication (#4849). + +Covers credential precedence in AuthMiddleware, the CSRF boundary for +Bearer-authenticated requests, scope intersection, PAT management routes, +and the self-protection rules (a PAT may not manage PATs or auth state). +""" + +from __future__ import annotations + +import asyncio +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace + +import pytest +from fastapi import FastAPI, Request +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine +from sqlalchemy.pool import NullPool +from starlette.testclient import TestClient + +import deerflow.persistence.models # noqa: F401 (register every table) +from app.gateway.auth_disabled import AUTH_SOURCE_PAT, AUTH_SOURCE_SESSION +from app.gateway.auth_middleware import AuthMiddleware +from app.gateway.authz import require_cancel_permission_if +from app.gateway.csrf_middleware import CSRFMiddleware +from app.gateway.routers.auth import router as auth_router +from app.gateway.run_models import RunCreateRequest +from deerflow.config.authorization_config import AuthorizationConfig +from deerflow.persistence.base import Base +from deerflow.persistence.personal_access_tokens import PersonalAccessTokenRepository + +TEST_JWT_SECRET = "test-pat-jwt-secret-0123456789abcdef" + + +class _FakeProvider: + """Minimal LocalAuthProvider stand-in: resolves users by id.""" + + def __init__(self, *users) -> None: + self._users = {str(user.id): user for user in users} + + async def get_user(self, user_id: str): + return self._users.get(str(user_id)) + + +def _fake_user(user_id: str = "user-1", *, system_role: str = "user"): + return SimpleNamespace( + id=user_id, + email=f"{user_id}@example.com", + system_role=system_role, + needs_setup=False, + token_version=0, + oauth_provider=None, + password_hash=None, + ) + + +@pytest.fixture(autouse=True) +def _default_route_authorization_config(monkeypatch): + monkeypatch.setattr( + "app.gateway.authz._get_route_authorization_config", + lambda: AuthorizationConfig(), + ) + monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "") + from app.gateway.auth.config import AuthConfig, set_auth_config + + set_auth_config(AuthConfig(jwt_secret=TEST_JWT_SECRET, token_expiry_days=7)) + + +def _make_pat_app(with_pat_repo: bool = True): + app = FastAPI() + # Production order: AuthMiddleware added first (inner), CSRF last (outer). + app.add_middleware(AuthMiddleware) + app.add_middleware(CSRFMiddleware) + app.include_router(auth_router) + + @app.get("/api/threads/whoami") + async def whoami(request: Request): + return {"user_id": str(request.state.user.id), "auth_source": request.state.auth_source} + + @app.get("/api/admin-check") + async def admin_check(request: Request): + from app.gateway.deps import is_admin_user + + return {"is_admin": await is_admin_user(request)} + + @app.post("/api/threads/{thread_id}/runs/stream") + async def run_stream(request: Request): + return {"ok": True, "permissions": list(request.state.auth.permissions)} + + @app.delete("/api/memory") + async def memory_delete(request: Request): + return {"deleted": True} + + @app.delete("/api/threads/{thread_id}") + async def thread_delete(request: Request): + return {"deleted": True} + + # Mirrors the real stateless run entrypoint (routers/runs.py), including + # the @require_permission decorator, so scope enforcement is exercised + # end-to-end through the middleware's permission intersection. + from app.gateway.authz import require_permission + + @app.post("/api/runs/stream") + @require_permission("runs", "create") + async def stateless_run_stream(request: Request): + return {"ok": True} + + # Mirrors the real cancel-then-stream entrypoint (thread_runs.py + # stream_existing_run): runs:read at the decorator, plus the real + # conditional runs:cancel check the handler applies when `action` is set. + from app.gateway.routers.thread_runs import require_cancel_permission_when_action + + @app.post("/api/threads/{thread_id}/runs/{run_id}/stream") + @require_permission("runs", "read") + async def cancel_then_stream(thread_id: str, run_id: str, request: Request, action: str | None = None): + require_cancel_permission_when_action(request, action) + return {"ok": True} + + # Mirrors the real run-creation entrypoints (thread_runs.py / runs.py): + # runs:create at the decorator, plus the cancel-capability gate that + # start_run applies to mutating multitask strategies. RunCreateRequest is + # imported at module level — FastAPI resolves body annotations against + # module globals under postponed annotation evaluation. + @app.post("/api/threads/{thread_id}/runs") + @require_permission("runs", "create") + async def create_run(thread_id: str, body: RunCreateRequest, request: Request): + require_cancel_permission_if(request, body.multitask_strategy != "reject") + return {"ok": True} + + return app + + +@pytest.fixture +def pat_env(tmp_path, monkeypatch): + """Engine + PAT repo + patched user provider; returns (client, repo).""" + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/pats.db", poolclass=NullPool) + asyncio.run(_create_tables(engine)) + repo = PersonalAccessTokenRepository(async_sessionmaker(engine, expire_on_commit=False)) + + fake_provider = _FakeProvider(_fake_user("user-1"), _fake_user("user-2"), _fake_user("admin-1", system_role="admin")) + monkeypatch.setattr("app.gateway.deps.get_local_provider", lambda: fake_provider) + monkeypatch.setattr("app.gateway.routers.auth.get_local_provider", lambda: fake_provider) + + app = _make_pat_app() + app.state.pat_repo = repo + return app, repo, engine + + +async def _create_tables(engine) -> None: + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + +@pytest.fixture +def client(pat_env): + app, repo, engine = pat_env + with TestClient(app) as test_client: + yield test_client + asyncio.run(engine.dispose()) + + +def _session_cookie(client: TestClient, user_id: str = "user-1", token_version: int = 0) -> str: + from app.gateway.auth import create_access_token + + token = create_access_token(user_id, token_version=token_version) + client.cookies.set("access_token", token) + return token + + +def _create_pat(client: TestClient, *, name: str = "test-token", scopes: list[str] | None = None, user_id: str = "user-1", expires_in_days: int | None = None) -> dict: + """Create a PAT via the management API with session auth + CSRF pair.""" + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client, user_id=user_id) + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + payload = {"name": name, "scopes": scopes or ["runs:read", "threads:read"]} + if expires_in_days is not None: + payload["expires_in_days"] = expires_in_days + response = client.post( + "/api/v1/auth/pats", + json=payload, + headers={CSRF_HEADER_NAME: csrf}, + ) + assert response.status_code == 201, response.text + payload = response.json() + assert payload["token"].startswith("dfp_") + return payload + + +# ── Middleware precedence (#4849 point 3) ───────────────────────────────── + + +def test_valid_pat_authenticates_without_cookie(client): + created = _create_pat(client) + client.cookies.clear() + response = client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 200 + assert response.json() == {"user_id": "user-1", "auth_source": AUTH_SOURCE_PAT} + + +def test_invalid_bearer_never_falls_back_to_session_cookie(client): + _session_cookie(client) # victim session is present and valid + response = client.get("/api/threads/whoami", headers={"Authorization": "Bearer dfp_not-a-real-token"}) + assert response.status_code == 401 + assert response.json()["detail"] == "Invalid token" + + +def test_non_bearer_authorization_scheme_is_rejected(client): + _session_cookie(client) + response = client.get("/api/threads/whoami", headers={"Authorization": "Basic dXNlcjpwYXNz"}) + assert response.status_code == 401 + + +def test_valid_pat_takes_precedence_over_session_cookie(client): + created = _create_pat(client) # sets a session cookie too + response = client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 200 + assert response.json()["auth_source"] == AUTH_SOURCE_PAT + + +def test_no_bearer_header_keeps_session_behavior(client): + _session_cookie(client) + response = client.get("/api/threads/whoami") + assert response.status_code == 200 + assert response.json()["auth_source"] == AUTH_SOURCE_SESSION + + +def test_revoked_pat_is_rejected_immediately(client): + created = _create_pat(client) + delete = client.delete(f"/api/v1/auth/pats/{created['id']}", headers={"X-CSRF-Token": client.cookies.get("csrf_token")}) + assert delete.status_code == 200, delete.text + + client.cookies.clear() + response = client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 401 + + +def test_pat_with_unresolvable_user_is_rejected(client, pat_env): + app, repo, _engine = pat_env + # Row owned by a user the provider cannot resolve (deleted user). + from app.gateway.auth.pat import generate_pat_token, pat_token_digest + + token = generate_pat_token() + asyncio.run(repo.create(user_id="user-deleted", name="orphan", scopes=["runs:read"], token_digest=pat_token_digest(token))) + client.cookies.clear() + response = client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 401 + + +def test_pat_without_durable_store_is_rejected(): + from fastapi import FastAPI + + app = FastAPI() + app.add_middleware(AuthMiddleware) + + @app.get("/api/threads/whoami") + async def whoami(request): # pragma: no cover - never reached + return {} + + with TestClient(app) as bare_client: + response = bare_client.get("/api/threads/whoami", headers={"Authorization": "Bearer dfp_whatever"}) + assert response.status_code == 401 + + +# ── Scope intersection ──────────────────────────────────────────────────── + + +def test_pat_scopes_intersect_user_permissions(client): + created = _create_pat(client, scopes=["runs:read"]) + client.cookies.clear() + response = client.post("/api/threads/t1/runs/stream", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 200 + permissions = response.json()["permissions"] + assert "runs:read" in permissions + assert "runs:create" not in permissions + assert "threads:read" not in permissions + + +# ── CSRF posture (#4849 point 4) ────────────────────────────────────────── + + +def test_bearer_request_skips_double_submit(client): + created = _create_pat(client) + client.cookies.clear() # no csrf_token cookie, no X-CSRF-Token header + response = client.post("/api/threads/t1/runs/stream", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 200 + + +def test_garbage_bearer_riding_cookie_dies_at_auth_not_csrf(client): + _session_cookie(client) + response = client.post("/api/threads/t1/runs/stream", headers={"Authorization": "Bearer garbage"}) + # 401 from AuthMiddleware (invalid credential), not 403 from CSRF. + assert response.status_code == 401 + + +def test_empty_authorization_header_is_present_and_dies_at_auth_not_csrf(client): + _session_cookie(client) + response = client.post("/api/threads/t1/runs/stream", headers={"Authorization": ""}) + # An explicitly empty header is present-but-invalid: the same 401 from + # AuthMiddleware as any other invalid credential, never a CSRF 403. + assert response.status_code == 401 + + +def test_auth_endpoint_origin_check_not_bypassed_by_bearer(client): + response = client.post( + "/api/v1/auth/login/local", + json={"email": "a@b.c", "password": "whatever1!"}, + headers={"Origin": "https://evil.example", "Authorization": "Bearer dfp_garbage"}, + ) + assert response.status_code == 403 + assert response.json()["detail"] == "Cross-site auth request denied." + + +# ── Management routes + self-protection (#4849 point 6) ─────────────────── + + +def test_create_returns_show_once_token_and_list_hides_it(client): + created = _create_pat(client) + listed = client.get("/api/v1/auth/pats") + assert listed.status_code == 200 + entries = listed.json() + assert [entry["id"] for entry in entries] == [created["id"]] + assert "token" not in entries[0] + assert "token_digest" not in entries[0] + + +def test_create_rejects_unknown_scope(client): + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client) + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + response = client.post("/api/v1/auth/pats", json={"name": "bad", "scopes": ["runs:write"]}, headers={CSRF_HEADER_NAME: csrf}) + assert response.status_code == 400 + assert "Unknown PAT scopes" in response.json()["detail"] + + +def test_create_rejects_whitespace_only_name(client): + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client) + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + for name in (" ", "\t\n"): + response = client.post("/api/v1/auth/pats", json={"name": name, "scopes": ["runs:read"]}, headers={CSRF_HEADER_NAME: csrf}) + # Rejected by request validation (422) before token generation. + assert response.status_code == 422, name + assert "non-whitespace" in response.text + + +def test_create_trims_surrounding_whitespace_in_name(client): + created = _create_pat(client, name=" ci bot ") + assert created["name"] == "ci bot" + + +def test_revoke_is_scoped_to_owner(client): + created = _create_pat(client, user_id="user-1") + # user-2 tries to revoke user-1's token. + _session_cookie(client, user_id="user-2") + from app.gateway.csrf_middleware import CSRF_HEADER_NAME + + response = client.delete(f"/api/v1/auth/pats/{created['id']}", headers={CSRF_HEADER_NAME: client.cookies.get("csrf_token")}) + assert response.status_code == 404 + + +def test_pat_cannot_manage_pats(client): + created = _create_pat(client) + client.cookies.clear() + headers = {"Authorization": f"Bearer {created['token']}"} + assert client.get("/api/v1/auth/pats", headers=headers).status_code == 403 + assert client.post("/api/v1/auth/pats", json={"name": "child", "scopes": ["runs:read"]}, headers=headers).status_code == 403 + assert client.delete(f"/api/v1/auth/pats/{created['id']}", headers=headers).status_code == 403 + + +def test_pat_cannot_change_password(client): + created = _create_pat(client) + client.cookies.clear() + response = client.post( + "/api/v1/auth/change-password", + json={"current_password": "x", "new_password": "Whatever123!"}, + headers={"Authorization": f"Bearer {created['token']}"}, + ) + assert response.status_code == 403 + # The default-deny route policy blocks the request at the middleware, + # before the route-level session-only guard gets a chance; the 403 is the + # security property either way. + assert "pat" in response.json()["detail"].lower() + + +def test_successful_pat_auth_stamps_last_used(client, pat_env): + _app, repo, _engine = pat_env + created = _create_pat(client) + client.cookies.clear() + assert client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {created['token']}"}).status_code == 200 + + records = asyncio.run(repo.list_for_user("user-1")) + assert records[0]["last_used_at"] is not None + + +def test_expired_pat_rejected_at_middleware(client, pat_env): + _app, repo, _engine = pat_env + from app.gateway.auth.pat import generate_pat_token, pat_token_digest + + token = generate_pat_token() + asyncio.run( + repo.create( + user_id="user-1", + name="already-expired", + scopes=["runs:read"], + token_digest=pat_token_digest(token), + expires_at=datetime.now(UTC) - timedelta(seconds=1), + ) + ) + client.cookies.clear() + response = client.get("/api/threads/whoami", headers={"Authorization": f"Bearer {token}"}) + assert response.status_code == 401 + + +def test_create_with_expiry_returns_expires_at(client): + created = _create_pat(client, expires_in_days=30) + assert created["expires_at"] is not None + + +def test_pat_never_carries_admin_capability_even_for_admin_owner(client): + created = _create_pat(client, user_id="admin-1", scopes=["runs:read"]) + client.cookies.clear() + # The route-level default-deny policy blocks the PAT before the route + # runs; the is_admin_user guard inside it remains as defense in depth + # for compositions without the middleware. + response = client.get("/api/admin-check", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 403 + + # Control: the same admin over a session cookie keeps admin capability. + _session_cookie(client, user_id="admin-1") + control = client.get("/api/admin-check") + assert control.status_code == 200 + assert control.json() == {"is_admin": True} + + +def test_pat_default_denied_on_route_outside_pat_policy(client): + """P1 regression (#5041 review): a PAT holding every scope must not reach + destructive routes that have no PAT policy — scope intersection only + constrains @require_permission routes, so undecorated mutation routes + would otherwise accept a runs:read-only token.""" + created = _create_pat(client, scopes=["threads:read", "threads:write", "threads:delete", "runs:create", "runs:read", "runs:cancel"]) + client.cookies.clear() + response = client.delete("/api/memory", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 403 + assert "PAT" in response.json()["detail"] + + +def test_session_cookie_reaches_route_that_denies_pat(client): + """The default-deny is PAT-specific: the same route stays open to the + owning user's session cookie (PATs narrow, never widen, and never + restrict the interactive path).""" + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client, user_id="user-1") + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + response = client.delete("/api/memory", headers={CSRF_HEADER_NAME: csrf}) + assert response.status_code == 200 + assert response.json() == {"deleted": True} + + +def test_pat_policy_allows_thread_lifecycle_routes(client): + created = _create_pat(client, scopes=["threads:delete"]) + client.cookies.clear() + response = client.delete("/api/threads/t1", headers={"Authorization": f"Bearer {created['token']}"}) + assert response.status_code == 200 + assert response.json() == {"deleted": True} + + +def test_pat_policy_does_not_pre_authorize_unimplemented_methods(): + """Route-policy regression (#5041 review): the allowlist must not admit + methods the router does not implement. The Gateway has no GET collection + route for /api/threads — pre-authorizing it would make a future GET + collection route PAT-reachable without an explicit policy change.""" + from app.gateway.auth.pat import is_pat_allowed_route + + assert is_pat_allowed_route("POST", "/api/threads") is True + assert is_pat_allowed_route("GET", "/api/threads") is False + + +def test_pat_runs_policy_admits_exactly_the_mounted_routes(): + """The runs subtree is enumerated, not wildcarded: every GET/POST route + the thread_runs router actually implements is admitted (derived from the + mounted router, not a hand-maintained list), routes in this router + outside the runs subtree stay denied, and representative unimplemented + neighbors — including the POST-only collection names on GET — are + default-denied. A new route under /runs fails here until explicitly + allowlisted; a removed one leaves a dead rule visible.""" + from fastapi.routing import APIRoute + + from app.gateway.auth.pat import is_pat_allowed_route + from app.gateway.routers.thread_runs import router + + def concrete(path: str) -> str: + return path.replace("{thread_id}", "t1").replace("{run_id}", "r1") + + for route in router.routes: + if not isinstance(route, APIRoute): + continue + path = concrete(route.path) + under_runs = route.path.startswith("/api/threads/{thread_id}/runs") + for method in sorted(route.methods - {"HEAD", "OPTIONS"}): + admitted = is_pat_allowed_route(method, path) + if under_runs: + assert admitted, f"{method} {path} is implemented but PAT-denied" + else: + # /messages, /messages/page, /token-usage sit outside the runs + # subtree and are PAT-denied pending the polling-surface + # decision — pinned here so widening it is a conscious edit. + assert not admitted, f"{method} {path} is outside the PAT policy" + + for method, path in [ + ("GET", "/api/threads/t1/runs/stream"), + ("GET", "/api/threads/t1/runs/wait"), + ("GET", "/api/threads/t1/runs/regenerate"), + ("GET", "/api/threads/t1/runs/edit-regenerate"), + ("POST", "/api/threads/t1/runs/r1/messages"), + ("DELETE", "/api/threads/t1/runs/r1"), + ("POST", "/api/threads/t1/runs/summary"), + ("GET", "/api/threads/t1/runs/r1/transfer"), + ]: + assert not is_pat_allowed_route(method, path), f"{method} {path} is not implemented and must stay denied" + + +def test_pat_scopes_enforced_on_stateless_run_entry(client): + """Follow-up to the review's P1-1: the stateless run entrypoints now + carry @require_permission("runs", "create"), so a threads:read-only PAT + cannot start runs even though the route sits inside the PAT allowlist.""" + read_only = _create_pat(client, scopes=["threads:read"]) + client.cookies.clear() + denied = client.post("/api/runs/stream", headers={"Authorization": f"Bearer {read_only['token']}"}) + assert denied.status_code == 403 + + create_scope = _create_pat(client, scopes=["runs:create"]) + client.cookies.clear() + allowed = client.post("/api/runs/stream", headers={"Authorization": f"Bearer {create_scope['token']}"}) + assert allowed.status_code == 200 + + +def test_runs_read_only_pat_cannot_cancel_then_stream(client): + """Review follow-up: cancel-then-stream (`?action=interrupt|rollback`) must + require runs:cancel even though the route decorator gates at runs:read — + otherwise a read-only PAT bypasses the separate cancel scope.""" + read_only = _create_pat(client, scopes=["runs:read"]) + client.cookies.clear() + + denied = client.post( + "/api/threads/t1/runs/run-1/stream?action=interrupt", + headers={"Authorization": f"Bearer {read_only['token']}"}, + ) + assert denied.status_code == 403 + assert denied.json()["detail"] == "Permission denied: runs:cancel" + + # The same route without an action is a plain stream join: runs:read is + # sufficient there. + join = client.post( + "/api/threads/t1/runs/run-1/stream", + headers={"Authorization": f"Bearer {read_only['token']}"}, + ) + assert join.status_code == 200 + + cancel_scope = _create_pat(client, scopes=["runs:read", "runs:cancel"]) + client.cookies.clear() + allowed = client.post( + "/api/threads/t1/runs/run-1/stream?action=rollback", + headers={"Authorization": f"Bearer {cancel_scope['token']}"}, + ) + assert allowed.status_code == 200 + + # Session callers keep the full permission set (with the CSRF pair their + # cookie-authenticated POST requires). + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client) + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + session_allowed = client.post( + "/api/threads/t1/runs/run-1/stream?action=interrupt", + headers={CSRF_HEADER_NAME: csrf}, + ) + assert session_allowed.status_code == 200 + + +def test_runs_create_only_pat_cannot_use_mutating_multitask_strategy(client): + """Review round 5, P1-a: interrupt/rollback multitask strategies terminate + an already-active run — runs:cancel capability, not runs:create — so a + create-only PAT must be denied; "reject" (the default) stays within + runs:create and must keep working.""" + create_only = _create_pat(client, scopes=["runs:create"]) + client.cookies.clear() + + for strategy in ("interrupt", "rollback"): + denied = client.post( + "/api/threads/t1/runs", + headers={"Authorization": f"Bearer {create_only['token']}"}, + json={"multitask_strategy": strategy}, + ) + assert denied.status_code == 403, denied.text + assert denied.json()["detail"] == "Permission denied: runs:cancel" + + # "reject" — explicitly and as the omitted default — does not touch + # existing runs and stays available to a create-only credential. + for body in ({"multitask_strategy": "reject"}, {}): + allowed = client.post( + "/api/threads/t1/runs", + headers={"Authorization": f"Bearer {create_only['token']}"}, + json=body, + ) + assert allowed.status_code == 200 + + cancel_scope = _create_pat(client, scopes=["runs:create", "runs:cancel"]) + client.cookies.clear() + privileged = client.post( + "/api/threads/t1/runs", + headers={"Authorization": f"Bearer {cancel_scope['token']}"}, + json={"multitask_strategy": "interrupt"}, + ) + assert privileged.status_code == 200 + + # Session callers keep the full permission set (with the CSRF pair their + # cookie-authenticated POST requires). + from app.gateway.csrf_middleware import CSRF_COOKIE_NAME, CSRF_HEADER_NAME, generate_csrf_token + + _session_cookie(client) + csrf = generate_csrf_token() + client.cookies.set(CSRF_COOKIE_NAME, csrf) + session_allowed = client.post( + "/api/threads/t1/runs", + headers={CSRF_HEADER_NAME: csrf}, + json={"multitask_strategy": "interrupt"}, + ) + assert session_allowed.status_code == 200 + + +def test_start_run_gates_mutating_strategies_at_the_choke_point(): + """The strategy gate lives inside start_run itself — the single choke point + every run-creation path (all five HTTP entrypoints plus internal + launchers) flows through — so no entry point can bypass it. Mirrored + routes prove the middleware path; this anchor proves the choke point.""" + import inspect + + from app.gateway.services import start_run + + source = inspect.getsource(start_run) + assert "require_cancel_permission_if" in source + assert "multitask_strategy" in source + + +def test_start_run_gate_denies_create_only_credential_behaviorally(): + """Behavioral pin on the real start_run (the mirror route and source + anchor above prove wiring, but this drives the production choke point + itself): a create-only auth context gets 403 for a mutating strategy, + and the gate never misfires on "reject" — with no cancel permission at + all, the call proceeds past the gate (failing later on missing test + wiring, never with a permission 403).""" + from fastapi import HTTPException + + from app.gateway.authz import AuthContext + from app.gateway.run_models import RunCreateRequest + from app.gateway.services import start_run + + def _request(permissions): + return SimpleNamespace(state=SimpleNamespace(auth=AuthContext(user=SimpleNamespace(id="user-1"), permissions=permissions))) + + async def _denied(): + with pytest.raises(HTTPException) as exc: + await start_run(RunCreateRequest(multitask_strategy="interrupt"), "t1", _request(["runs:create"])) + return exc.value + + exc = asyncio.run(_denied()) + assert exc.status_code == 403 + assert exc.detail == "Permission denied: runs:cancel" + + async def _allowed_past_gate(): + try: + await start_run(RunCreateRequest(), "t1", _request([])) + except HTTPException as gate_misfire: + pytest.fail(f"gate misfired on reject: {gate_misfire.status_code} {gate_misfire.detail}") + except Exception: + pass # expected wiring failure past the gate — the gate let it through + + asyncio.run(_allowed_past_gate()) + + +def test_auth_disabled_mode_ignores_bearer_header(monkeypatch, tmp_path): + """DEER_FLOW_AUTH_DISABLED is an operator override of all authentication. + + A stray Authorization header (e.g. added by a proxy in front of an E2E + sandbox) must not turn into a 401 in that mode. + """ + monkeypatch.setattr("app.gateway.auth_middleware.is_auth_disabled", lambda: True) + app = _make_pat_app() + with TestClient(app) as disabled_client: + response = disabled_client.get("/api/threads/whoami", headers={"Authorization": "Bearer dfp_garbage"}) + assert response.status_code == 200 + assert response.json()["auth_source"] == "auth_disabled" diff --git a/backend/tests/test_persistence_bootstrap.py b/backend/tests/test_persistence_bootstrap.py index 23548e636..b51862a5e 100644 --- a/backend/tests/test_persistence_bootstrap.py +++ b/backend/tests/test_persistence_bootstrap.py @@ -48,7 +48,7 @@ from deerflow.persistence.migrations._helpers import _normalize_default asyncio_test = pytest.mark.asyncio -HEAD = "0016_subagent_batches" +HEAD = "0017_personal_access_tokens" BASELINE = "0001_baseline" diff --git a/backend/tests/test_persistence_bootstrap_concurrency.py b/backend/tests/test_persistence_bootstrap_concurrency.py index 36f703fb5..043a51468 100644 --- a/backend/tests/test_persistence_bootstrap_concurrency.py +++ b/backend/tests/test_persistence_bootstrap_concurrency.py @@ -28,7 +28,7 @@ from deerflow.persistence.bootstrap import bootstrap_schema pytestmark = pytest.mark.asyncio -HEAD = "0016_subagent_batches" +HEAD = "0017_personal_access_tokens" def _url(tmp_path: Path) -> str: diff --git a/backend/tests/test_persistence_bootstrap_regression.py b/backend/tests/test_persistence_bootstrap_regression.py index 46aded16a..4d37c288a 100644 --- a/backend/tests/test_persistence_bootstrap_regression.py +++ b/backend/tests/test_persistence_bootstrap_regression.py @@ -76,7 +76,7 @@ async def test_legacy_database_recovers_token_usage_column(tmp_path: Path) -> No cols = {row[1] for row in raw.execute("PRAGMA table_info(runs)").fetchall()} assert "token_usage_by_model" in cols version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0016_subagent_batches" + assert version_row[0] == "0017_personal_access_tokens" # And the read path that originally 500'd must now succeed. sf = get_session_factory() @@ -116,6 +116,6 @@ async def test_legacy_database_with_manual_alter_still_bootstraps(tmp_path: Path # No duplicate column -- list, not set, to catch dupes. assert cols.count("token_usage_by_model") == 1 version_row = raw.execute("SELECT version_num FROM alembic_version").fetchone() - assert version_row[0] == "0016_subagent_batches" + assert version_row[0] == "0017_personal_access_tokens" finally: await close_engine() diff --git a/backend/tests/test_personal_access_tokens_repository.py b/backend/tests/test_personal_access_tokens_repository.py new file mode 100644 index 000000000..4eca7e12d --- /dev/null +++ b/backend/tests/test_personal_access_tokens_repository.py @@ -0,0 +1,209 @@ +"""Tests for PAT token utilities and the personal access token repository.""" + +from __future__ import annotations + +from datetime import UTC, datetime, timedelta + +import pytest +import pytest_asyncio + +from app.gateway.auth.pat import ( + PAT_ALLOWED_SCOPES, + PAT_TOKEN_PREFIX, + digest_matches, + extract_bearer_token, + generate_pat_token, + pat_token_digest, + validate_scopes, +) +from deerflow.config.database_config import DatabaseConfig +from deerflow.persistence.engine import close_engine, get_session_factory, init_engine_from_config +from deerflow.persistence.personal_access_tokens import PersonalAccessTokenRepository + + +@pytest_asyncio.fixture(autouse=True) +async def _close_persistence_engine(): + yield + await close_engine() + + +async def _make_repo(tmp_path) -> PersonalAccessTokenRepository: + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + session_factory = get_session_factory() + assert session_factory is not None + return PersonalAccessTokenRepository(session_factory) + + +# ── Token utilities ─────────────────────────────────────────────────────── + + +def test_generate_pat_token_format(): + token = generate_pat_token() + assert token.startswith(PAT_TOKEN_PREFIX) + body = token[len(PAT_TOKEN_PREFIX) :] + assert len(body) == 43 # fixed width: 62^43 > 2^256 > 62^42 + assert body.isalnum() + # Two draws must differ: CSPRNG, not a counter. + assert token != generate_pat_token() + + +def test_base62_pads_to_fixed_width_for_leading_zero_and_all_zero_input(): + """``int.from_bytes`` discards leading zero bytes; the fixed-width pad + keeps the token body exactly 43 chars for every draw, including the + all-zero and single-leading-byte edges (review round 6, P3).""" + from app.gateway.auth.pat import PAT_RANDOM_BYTES, _base62 + + assert _base62(b"\x00" * PAT_RANDOM_BYTES) == "0" * 43 + assert _base62(b"\x00" * (PAT_RANDOM_BYTES - 1) + b"\x01") == "0" * 42 + "1" + assert len(_base62(b"\xff" * PAT_RANDOM_BYTES)) == 43 + + +def test_pat_token_digest_is_deterministic_and_constant_time_comparable(): + token = generate_pat_token() + assert pat_token_digest(token) == pat_token_digest(token) + assert len(pat_token_digest(token)) == 64 + assert digest_matches(pat_token_digest(token), token) is True + # The mutated token must differ from the original even when the CSPRNG + # tail already ends in "X" (1/62), or this assertion fails intermittently. + mutated_tail = "X" if token[-1] != "X" else "Y" + assert digest_matches(pat_token_digest(token), token[:-1] + mutated_tail) is False + assert digest_matches(None, token) is False + assert digest_matches("", token) is False + + +def test_extract_bearer_token_classifies_absent_scheme_and_credential(): + assert extract_bearer_token(None) is None + assert extract_bearer_token("Bearer dfp_abc") == "dfp_abc" + assert extract_bearer_token("bearer dfp_abc") == "dfp_abc" # scheme is case-insensitive + assert extract_bearer_token("Basic dXNlcg==") == "" # present but unusable + assert extract_bearer_token("Bearer ") == "" + assert extract_bearer_token("Bearer") == "" + + +def test_validate_scopes_deduplicates_and_rejects_unknown(): + assert validate_scopes(["runs:read", "threads:read", "runs:read"]) == ["runs:read", "threads:read"] + with pytest.raises(ValueError, match="Unknown PAT scopes"): + validate_scopes(["runs:write"]) # not a route permission + with pytest.raises(ValueError, match="at least one scope"): + validate_scopes([]) + + +def test_pat_scopes_stay_aligned_with_route_permissions(): + """PAT scopes are exactly the authz route permissions — fail on drift.""" + from app.gateway.authz import _ALL_PERMISSIONS + + assert PAT_ALLOWED_SCOPES == frozenset(_ALL_PERMISSIONS) + + +# ── Repository ──────────────────────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_create_and_resolve_by_digest_roundtrip(tmp_path): + repo = await _make_repo(tmp_path) + token = generate_pat_token() + record = await repo.create( + user_id="user-1", + name="ci-runner", + scopes=["runs:read"], + token_digest=pat_token_digest(token), + ) + assert record["user_id"] == "user-1" + assert record["scopes"] == ["runs:read"] + assert record["revoked_at"] is None + + resolved = await repo.get_active_by_digest(pat_token_digest(token)) + assert resolved is not None + assert resolved["id"] == record["id"] + assert resolved["token_digest"] == pat_token_digest(token) + # Digest lookup never matches a different token. + assert await repo.get_active_by_digest(pat_token_digest(generate_pat_token())) is None + + +@pytest.mark.asyncio +async def test_revoked_token_no_longer_resolves(tmp_path): + repo = await _make_repo(tmp_path) + token = generate_pat_token() + record = await repo.create(user_id="user-1", name="temp", scopes=["runs:read"], token_digest=pat_token_digest(token)) + + assert await repo.revoke(record["id"], "user-1") is True + # Revoking twice is a no-op. + assert await repo.revoke(record["id"], "user-1") is False + assert await repo.get_active_by_digest(pat_token_digest(token)) is None + + +@pytest.mark.asyncio +async def test_revoke_is_scoped_to_the_owning_user(tmp_path): + repo = await _make_repo(tmp_path) + token = generate_pat_token() + record = await repo.create(user_id="user-1", name="mine", scopes=["runs:read"], token_digest=pat_token_digest(token)) + + assert await repo.revoke(record["id"], "user-2") is False # not the owner + assert await repo.get_active_by_digest(pat_token_digest(token)) is not None + + +@pytest.mark.asyncio +async def test_expired_token_no_longer_resolves(tmp_path): + repo = await _make_repo(tmp_path) + token = generate_pat_token() + await repo.create( + user_id="user-1", + name="short-lived", + scopes=["runs:read"], + token_digest=pat_token_digest(token), + expires_at=datetime.now(UTC) - timedelta(seconds=1), + ) + assert await repo.get_active_by_digest(pat_token_digest(token)) is None + + +@pytest.mark.asyncio +async def test_list_for_user_is_isolated_and_never_returns_raw_tokens(tmp_path): + repo = await _make_repo(tmp_path) + token = generate_pat_token() + created = await repo.create(user_id="user-1", name="a", scopes=["runs:read"], token_digest=pat_token_digest(token)) + await repo.create(user_id="user-2", name="b", scopes=["threads:read"], token_digest=pat_token_digest(generate_pat_token())) + + listed = await repo.list_for_user("user-1") + assert [item["id"] for item in listed] == [created["id"]] + assert listed[0]["token_digest"] == pat_token_digest(token) # digest only; raw token never persisted + + +@pytest.mark.asyncio +async def test_token_digest_unique_constraint(tmp_path): + repo = await _make_repo(tmp_path) + digest = pat_token_digest(generate_pat_token()) + await repo.create(user_id="user-1", name="a", scopes=["runs:read"], token_digest=digest) + from sqlalchemy.exc import IntegrityError + + with pytest.raises(IntegrityError): + await repo.create(user_id="user-1", name="dup", scopes=["runs:read"], token_digest=digest) + + +@pytest.mark.asyncio +async def test_touch_last_used_is_throttled(tmp_path): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + session_factory = get_session_factory() + repo = PersonalAccessTokenRepository(session_factory, last_used_write_interval_seconds=300.0) + record = await repo.create(user_id="user-1", name="t", scopes=["runs:read"], token_digest=pat_token_digest(generate_pat_token())) + + await repo.touch_last_used(record["id"]) + first = (await repo.list_for_user("user-1"))[0]["last_used_at"] + assert first is not None + + # A second touch inside the throttle window must not produce a write. + await repo.touch_last_used(record["id"]) + second = (await repo.list_for_user("user-1"))[0]["last_used_at"] + assert second == first + + # After the window elapses the next touch writes again. + repo._last_used_written_at.clear() + await repo.touch_last_used(record["id"]) + third = (await repo.list_for_user("user-1"))[0]["last_used_at"] + assert third != first + + +@pytest.mark.asyncio +async def test_touch_last_used_never_raises_on_unknown_id(tmp_path): + await init_engine_from_config(DatabaseConfig(backend="sqlite", sqlite_dir=str(tmp_path))) + repo = PersonalAccessTokenRepository(get_session_factory()) + await repo.touch_last_used("no-such-pat") # update affects 0 rows; still a commit diff --git a/backend/tests/test_sse_observer_disconnect.py b/backend/tests/test_sse_observer_disconnect.py new file mode 100644 index 000000000..b45ccbbb1 --- /dev/null +++ b/backend/tests/test_sse_observer_disconnect.py @@ -0,0 +1,114 @@ +"""Observer joins must not apply the creator's cancel-on-disconnect policy. + +Review round 5 (PR #5041): every consumer of ``sse_consumer`` used to apply +the record's ``on_disconnect=cancel`` policy in its ``finally`` block, so a +read-only stream observer could cancel a locally-owned running run just by +closing the SSE connection. The fix separates creator streams +(``apply_on_disconnect=True``, the default) from join/observer streams +(``False``). These tests drive a real generator close — the same machinery +Starlette runs when a client drops the connection — against the production +consumer. +""" + +import asyncio +import inspect +from types import SimpleNamespace + +from app.gateway.services import sse_consumer +from deerflow.runtime import DisconnectMode, RunRecord, RunStatus + + +def _running_record() -> RunRecord: + return RunRecord( + run_id="run-1", + thread_id="t1", + assistant_id=None, + status=RunStatus.running, + on_disconnect=DisconnectMode.cancel, + ) + + +class _StubBridge: + """Yields one event, then parks until the consumer closes the generator.""" + + def subscribe(self, run_id, last_event_id=None): + async def _gen(): + yield SimpleNamespace(event="message", data="{}", id="1") + await asyncio.Event().wait() + + return _gen() + + +class _CancelRecorder: + """Stands in for the RunManager: records cancel calls, mutates nothing.""" + + def __init__(self): + self.cancelled: list[str] = [] + + async def cancel(self, run_id, action="interrupt"): + self.cancelled.append(run_id) + + +class _StubRequest: + """Minimal request: headers for Last-Event-ID, never-disconnected client + (the disconnect under test happens between events, via generator close).""" + + def __init__(self): + self.headers = {} + + async def is_disconnected(self) -> bool: + return False + + +def _request() -> _StubRequest: + return _StubRequest() + + +async def _drive_disconnect(consumer) -> None: + """Start the generator (it yields one frame), then close it — a real + disconnect of the response stream, running the ``finally`` block.""" + await consumer.__anext__() + await consumer.aclose() + + +def test_creator_stream_disconnect_applies_cancel_policy(): + """The stream returned by the creating endpoint keeps the creator's + cancel-on-disconnect semantics.""" + + async def scenario(): + recorder = _CancelRecorder() + consumer = sse_consumer(_StubBridge(), _running_record(), _request(), recorder) + await _drive_disconnect(consumer) + return recorder.cancelled + + assert asyncio.run(scenario()) == ["run-1"] + + +def test_observer_join_disconnect_does_not_cancel(): + """A join/observer stream closing must not cancel the run — including for + a read-only credential that never held runs:cancel.""" + + async def scenario(): + recorder = _CancelRecorder() + consumer = sse_consumer(_StubBridge(), _running_record(), _request(), recorder, apply_on_disconnect=False) + await _drive_disconnect(consumer) + return recorder.cancelled + + assert asyncio.run(scenario()) == [] + + +def test_join_routes_wire_sse_consumer_as_observers(): + """Both join surfaces must be wired as observers, and the creator's + create-and-stream endpoints must keep the creator policy (default).""" + from app.gateway.routers import runs as runs_router + from app.gateway.routers import thread_runs + + thread_runs_source = inspect.getsource(thread_runs) + # join_run + stream_existing_run (GET and POST share one handler) + assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)") == 2 + # stream_run — the creator's create-and-stream endpoint + assert thread_runs_source.count("sse_consumer(bridge, record, request, run_mgr),") == 1 + + runs_source = inspect.getsource(runs_router) + # stateless create-and-stream — also a creator stream + assert "sse_consumer(bridge, record, request, run_mgr, apply_on_disconnect=False)" not in runs_source diff --git a/backend/tests/test_threads_router.py b/backend/tests/test_threads_router.py index 619badf20..d4f3e764d 100644 --- a/backend/tests/test_threads_router.py +++ b/backend/tests/test_threads_router.py @@ -742,6 +742,7 @@ def test_insert_race_recovery_claims_unscoped_row_for_trusted_owner() -> None: from sqlalchemy.exc import IntegrityError + from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE store = InMemoryStore() @@ -764,7 +765,7 @@ def test_insert_race_recovery_claims_unscoped_row_for_trusted_owner() -> None: thread_store = _RacingOwnerStore(store) request = SimpleNamespace( headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"}, - state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), + state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE), auth_source=AUTH_SOURCE_INTERNAL), app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer, thread_store=thread_store)), ) @@ -895,6 +896,7 @@ def test_goal_mutations_reject_run_owned_by_another_worker() -> None: def test_internal_owner_header_assigns_thread_to_owner() -> None: import asyncio + from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE store = InMemoryStore() @@ -902,7 +904,7 @@ def test_internal_owner_header_assigns_thread_to_owner() -> None: thread_store = MemoryThreadMetaStore(store) request = SimpleNamespace( headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"}, - state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), + state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE), auth_source=AUTH_SOURCE_INTERNAL), app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer, thread_store=thread_store)), ) @@ -926,6 +928,7 @@ def test_internal_owner_header_assigns_thread_to_owner() -> None: def test_goal_thread_creation_uses_internal_owner_header() -> None: import asyncio + from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, INTERNAL_SYSTEM_ROLE store = InMemoryStore() @@ -933,7 +936,7 @@ def test_goal_thread_creation_uses_internal_owner_header() -> None: thread_store = MemoryThreadMetaStore(store) request = SimpleNamespace( headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: "owner-1"}, - state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE)), + state=SimpleNamespace(user=SimpleNamespace(id="default", system_role=INTERNAL_SYSTEM_ROLE), auth_source=AUTH_SOURCE_INTERNAL), app=SimpleNamespace(state=SimpleNamespace(checkpointer=checkpointer, thread_store=thread_store)), )