mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-09 13:39:26 +00:00
feat(auth): add personal access tokens for programmatic API access (#5041)
* feat(auth): add personal access tokens for programmatic API access (#4849) Backend-first implementation of the PAT contract from #4849: show-once dfp_ tokens bound to their owning user (AUTH_SOURCE_PAT, is_internal=false), digest-only storage (migration 0017), strict credential precedence (invalid Bearer is a 401, never cookie fallback), CSRF double-submit skipped only for Bearer requests while auth-endpoint origin checks still run, scopes intersecting the authz route permissions, session-auth-only PAT management and password changes, and throttled best-effort last_used_at stamps. * fix(auth): harden PAT scope boundary and schema parity from adversarial review Independent review of the initial draft found: (1) scopes only constrained the threads/runs permission axis while admin routes treated a PAT as its (possibly admin) owner — is_admin_user now rejects PAT callers outright since no scope grants admin capability; (2) the model declared a column UNIQUE constraint while migration 0017 created a named unique index, so downgrade failed on create_all-bootstrapped DBs — both now use the named unique index; (3) auth-disabled mode is an operator override and now stays ahead of the Bearer check so a stray Authorization header cannot 401 an E2E sandbox; plus wiring the previously-unused constants, bounding the last_used_at stamp cache, and four new tests (middleware-level expiry, expires_in_days, admin-capability rejection with session control, and the auth-disabled precedence). * docs(api): document personal access tokens for programmatic API access * fix(auth): close PAT security boundaries from review (default-deny routes, extension admin suppression) P1-1: scope intersection only constrains @require_permission routes, so undecorated mutation routes (DELETE /api/memory, POST /api/agents, Lark credential switching, channel config) accepted a PAT holding a single read scope. AuthMiddleware now enforces a default-deny route policy in auth/pat.py: PAT requests are admitted only to the thread/run lifecycle routes the v1 scopes govern; everything else answers 403 regardless of scopes. Session-cookie callers are unaffected. P1-2: the extension principal resolver projected is_admin/roles from the raw system_role, so an admin-owned PAT passed deerflow_extension_api.require_admin on contributed routes despite the documented no-admin guarantee. The projection is now PAT-aware and suppresses every admin signal for PAT callers, mirroring deps.is_admin_user. Both fixes carry regression tests (route outside policy 403 + session control; production resolver admin suppression), and API.md documents the default-deny boundary. * fix(auth): enforce PAT scopes on stateless run entry and harden decorator Follow-up hardening from an independent audit of the P1 fixes: - POST /api/runs/stream and /api/runs/wait were the only allowlisted run entrypoints without @require_permission, so a threads:read-only PAT could still start runs (same bug class as P1-1, now closed): both now carry @require_permission("runs", "create"). POST /api/threads and POST /api/threads/search gain threads:write / threads:read for the same reason. Authorization-disabled deployments see no change (the permission set resolves to all permissions). - require_permission now binds the wrapped signature to locate a positionally-passed request before injecting the test stub, fixing 'got multiple values for argument' on direct positional unit-test calls. - API.md: the intro PAT example used GET /api/models, which the new default-deny policy 403s — replaced with GET /api/threads; the default-deny route list now spells out method sets. Regression test: threads:read-only PAT is 403 on the decorated stateless entry while a runs:create PAT passes. * fix(auth): address review P2s (empty Authorization header, PAT name trimming, API example) - CSRFMiddleware treats an explicitly empty Authorization header as present (is None), so an invalid credential always reaches AuthMiddleware's uniform 401 instead of a CSRF 403 that varies by method/CSRF state. Regression: empty-header request dies at auth. - PATCreateRequest strips the name and rejects whitespace-only values before token generation; created names are stored trimmed. - API.md intro PAT example now uses the implemented POST /api/threads/search endpoint (GET /api/threads does not exist). - AGENTS.md trimmed back under the guidance soft budget after the upstream merge. * fix(auth): tighten PAT route policy to implemented methods only The allowlist admitted GET /api/threads, a method no router implements. Pre-authorizing a dead method weakens the default-deny boundary: a future GET collection route added without a permission decorator would become PAT-reachable without an explicit policy change. Restrict the rule to POST, fix the stale GET description in API.md's PAT constraints, and document the default-deny boundary accurately in the gateway AGENTS.md guidance (only the threads/runs allowlist is PAT-reachable; every other authenticated route 403s PAT callers). Audited every remaining rule against the mounted routers: all other method+path entries map to real routes. Regression: test_pat_policy_does_not_pre_authorize_unimplemented_methods. * test(auth): guarantee the negative digest test mutates the token token[:-1] + "X" is identical to the original whenever the generated token already ends in X (1/62), making the negative digest assertion fail intermittently. Choose the replacement character based on the existing tail so the mutated token always differs. * fix(auth): require runs:cancel for cancel-then-stream requests stream_existing_run is gated at runs:read so action-less stream joins work with read-only credentials, but its ?action=interrupt|rollback branch cancels the run — a separate permission. A runs:read-only PAT passed both the PAT route policy and the route decorator and could interrupt or roll back an active run, bypassing the runs:cancel scope. Decorators cannot express query-parameter-conditional permissions, so the check lives in require_cancel_permission_when_action(), applied at the top of the handler. Regression drives the real helper through the production middleware: runs:read-only PAT + action is 403, the same token joins action-less, runs:read+cancel passes, session control unaffected. * docs(changelog): add the PAT feature entry * docs(readme): add personal access tokens section Repo documentation-update policy requires user-facing features to update README.md in the same changeset; the PAT feature previously touched only backend/docs/API.md and the gateway AGENTS.md. * fix(auth): require runs:cancel for mutating multitask strategies All five run-creation entrypoints were gated only by runs:create, but RunCreateRequest.multitask_strategy accepts interrupt/rollback and start_run forwards it to create_or_reject, which terminates an already-active run. A runs:create-only PAT could therefore kill an existing run through a create request, bypassing runs:cancel. Decorators cannot express body-parameter-conditional permissions, and per-route checks leave the same hole for the next entrypoint, so the gate lives in start_run itself — the single choke point every run-creation path (HTTP routes and internal launchers) flows through. Regenerate launches pass multitask_strategy="reject" and are unaffected; requests without a stamped auth context (internal/test compositions) skip the gate. The check is the shared authz.require_cancel_permission_if primitive; require_cancel_permission_when_action now delegates to it, so every request dimension that carries cancel capability (query action, body strategy) flows through one gate. Regression drives the real middleware stack: runs:create-only PAT + interrupt/rollback is 403 with the exact detail, reject (explicit and default) stays available, runs:create+cancel passes, session control unaffected; a source anchor pins the gate inside start_run. * fix(runs): keep observer joins from applying creator cancel-on-disconnect sse_consumer's finally block applied the record's on_disconnect=cancel policy on ANY consumer's disconnect. The join surfaces (GET /join and the action-less GET/POST stream join) feed it the existing RunRecord, so anyone with thread read access — including a runs:read-only PAT — could cancel a locally-owned running run simply by closing the SSE connection, without runs:cancel. The policy expresses the creator's intent for their own connection; an observer's disconnect must never be read as that intent. sse_consumer gains apply_on_disconnect (default True). The two join surfaces pass False; the creating endpoints (thread-scoped and stateless create-and-stream) keep the creator semantics unchanged. wait_for_run_completion needs no change: its callers are creator-side or post-explicit-cancel paths only. Regression exercises a real generator close — the same machinery Starlette drives on client disconnect — against the production sse_consumer: creator stream disconnect cancels, observer join disconnect does not; a wiring anchor pins both join call sites and the creator defaults. API.md documents the cancel-capability constraint (this fix plus the action/strategy gates) in PAT Constraints. * test(auth): pin the multitask gate behaviorally; state wait invariant Independent adversarial review of the round-5 fixes found the P1-a regression only mirror-pinned: the source anchor could be satisfied by a comment, and deleting the gate from start_run would not fail the suite. This drives the production start_run directly — a create-only auth context gets 403 with the exact detail for interrupt, and a reject request with no cancel permission at all proceeds past the gate (never a permission 403). Also documents wait_for_run_completion's creator-side invariant (every caller is the creating endpoint or post-explicit-cancel) so a future observer wiring thinks twice before reusing it — the one-caller- away variant of the observer-disconnect P1. * docs(changelog): correct the PAT entry's digest and route-policy description The entry said HMAC digests (the implementation stores SHA-256 digests, as documented in API.md and pinned by the repository tests) and claimed the route policy admits 'implemented stateless endpoints' (it admits the thread/run lifecycle routes, narrowing further by scopes). Also notes the cancel-capability gate now covering action and multitask strategies. * fix(auth): enumerate the PAT runs route policy per implemented subroute The runs subtree rule was a GET|POST /runs(/.*)? wildcard — it pre-authorized every current and future subroute under /runs, including methods the router never implemented (e.g. GET /runs/stream), which is the same latent default-deny weakening the threads collection rule was tightened for: a future route added under /runs would become PAT-reachable without an explicit policy change. The wildcard is replaced with six segment-precise rules covering exactly the 14 implemented method+path combinations; the {run_id} slot necessarily matches any single segment, so the POST-only collection names (stream, wait, regenerate, edit-regenerate) are excluded from the GET run-id rule via negative lookahead — no dead method stays pre-authorized. Behavior for implemented routes is unchanged. test_pat_runs_policy_admits_exactly_the_mounted_routes derives the expected set from the mounted thread_runs router instead of a hand-maintained list: every implemented GET/POST route under /runs must be admitted, routes in this router outside the subtree stay denied, and representative unimplemented neighbors are denied — so adding a route under /runs now fails CI until it is explicitly allowlisted, and a removed route leaves a dead rule visible. API.md's PAT constraints list the enumerated routes and drops a feedback mention that belonged to the stateless /api/runs axis. * docs(migration): add the 0017 renumbering coordination note to 0017 The PR's migration-coordination comment states each migration file carries the note; the file did not. Adds it: numbering was generated against main head 0016 alongside #5078 and #4843; whoever merges first keeps the slot, the others renumber on rebase (revision/down_revision plus the bootstrap head assertions). * fix(auth): pad base62 tokens to a fixed 43-char width int.from_bytes discards leading zero bytes, so the unpadded encoder returned a variable-length body — empty for all-zero input, and shorter than 40 characters for any draw below 62**39 (~1 in 14.5M), leaving test_generate_pat_token_format probabilistically flaky and the token body without stable width (review round 6, P3). _base62 now left-pads with "0" to _base62_width(len(data)) — the exact integer digit count (62^43 > 2^256 > 62^42, so 43 for 32 bytes). The format test asserts the exact fixed width instead of a probabilistic floor, and a new unit test pins the all-zero, leading-zero-byte, and max-value edges deterministically.
This commit is contained in:
Sunshine
GitHub
parent
c6f6a01f56
commit
bf740ffa90
@ -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,
|
||||
|
||||
25
README.md
25
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.
|
||||
|
||||
@ -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`.
|
||||
|
||||
@ -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)
|
||||
|
||||
196
backend/app/gateway/auth/pat.py
Normal file
196
backend/app/gateway/auth/pat.py
Normal file
@ -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
|
||||
@ -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")
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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",
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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`
|
||||
|
||||
@ -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")
|
||||
@ -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",
|
||||
|
||||
@ -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"]
|
||||
@ -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)
|
||||
@ -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)
|
||||
@ -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",)
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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.
|
||||
|
||||
@ -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"
|
||||
|
||||
93
backend/tests/test_migration_0017_personal_access_tokens.py
Normal file
93
backend/tests/test_migration_0017_personal_access_tokens.py
Normal file
@ -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()
|
||||
700
backend/tests/test_pat_auth.py
Normal file
700
backend/tests/test_pat_auth.py
Normal file
@ -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"
|
||||
@ -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"
|
||||
|
||||
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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()
|
||||
|
||||
209
backend/tests/test_personal_access_tokens_repository.py
Normal file
209
backend/tests/test_personal_access_tokens_repository.py
Normal file
@ -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
|
||||
114
backend/tests/test_sse_observer_disconnect.py
Normal file
114
backend/tests/test_sse_observer_disconnect.py
Normal file
@ -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
|
||||
@ -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)),
|
||||
)
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user