mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
* feat(authz): account disable/enable across every authentication surface (#4063 gap 3) Vertical slice 1 of the user-lifecycle work: an operator can suspend an account and every authentication surface rejects it, while the row — credentials, role, OAuth linkage — is retained so re-enabling restores exactly what was suspended. - users.disabled (NOT NULL, server default 0) via migration 0036_user_disabled; User/UserResponse carry and expose it. - Enforcement at all four surfaces: password login (no credential comparison happens), JWT session resolution (401 ACCOUNT_DISABLED — sessions die at the next request), PAT resolution (the generic 401, no oracle), OAuth provisioning (403 with a contact-admin message; the link is kept so re-enable restores access). - SQLiteUserRepository.set_disabled: single-column, serialized write (same idiom as update_system_role) with the last-ACTIVE-admin guard — disabled admins cannot authenticate, so they cannot restore management access; re-enabling is always allowed. Field-scoping completed both ways: the credential writer mirrors and never writes disabled, so a stale password-change snapshot cannot re-enable a suspended account. - PATCH /api/v1/admin/users/{id} becomes a partial account update (system_role and/or disabled), each field routed to its field-scoped writer; empty bodies 422. Tests 11->18: route (disable/re-enable, last-active-admin 409, empty 422, combined update), real-repo (field isolation both directions, disabled-admin-not-active, guard inside the serialized write, re-enable always), login rejection with the CORRECT password (only the disabled gate can reject) plus resolver-visible flag. Mutations verified red: guard removed (M19), login gate removed (M20 — after hardening the test with a real hash so the gate is load-bearing). * fix(authz): portable boolean default, alternate-authenticator suspension, active-admin role guard Review round 1 on #6566 — three P1s: - [PG default] sa.text("0") is a raw integer expression PostgreSQL rejects as a BOOLEAN default, breaking startup migration and fresh create_all bootstraps alike. Both the migration and the ORM model now use sa.false()/false() — dialect-rendered, portable. - [alternate authenticators] WebSockets bypass AuthMiddleware, so a disabled user's still-valid cookie could open a NEW browser/stream connection and drive the retained session; the standalone LangGraph authenticate callback accepted the same suspended JWT. Both now route through a shared post-lookup validator (validate_resolved_session_user in deps.py — token_version + disabled in one place) that the HTTP dependency also uses, so a lifecycle change lands everywhere at once. - [active-admin role guard] update_system_role counted every admin row, so 'disable B, then demote self' — or a combined {system_role, disabled} self-update — could strand the deployment with zero administrators able to authenticate or re-enable anyone. The demotion guard now counts ACTIVE admins only (disabled admins cannot authenticate), under the same serialization boundary. Tests 18->22: shared-validator verdicts, WS admission after suspension (real token, real repo), LangGraph-callback admission after suspension, disable-then-demote-self + combined self-update both rejected (fake-repo route level and real-repo update_system_role level). Mutations verified red: helper's disabled check removed (M21 — three admission tests fail), active-admin predicate reverted to all-admins (M22 — repo-level strand scenario fails). 366 tests green on the affected surface. * fix(migrations): chain 0036_user_disabled after upstream 0037 The maintainer main-merge brought 0037_project_document_summaries, which branched from 0035 independently of this PR's 0036 — alembic now sees two heads and every engine bootstrap fails. Chain 0036 after 0037 (single head; revision numbers are labels, the graph is what matters). * fix(migrations): renumber user_disabled to 0038 and follow the chain-test convention Upstream landed its own 0036 (run_idempotency_request) with 0037 chained after it while this PR was in review — my 0036_user_disabled collided on the number and my interim chain-fix (0036 revises 0037) left the graph in a state upstream's chain tests reject (heads == [0037] hardcoded). - 0036_user_disabled -> 0038_user_disabled, revises 0037_project_document_summaries: single head 0038. - Chain tests follow the repo convention (each migration PR appends the new head): 0032/0035 gain the DOC_SUMMARIES link and CURRENT_HEAD = 0038; 0037's single-head assertion moves to CURRENT_HEAD with its own-revision link check unchanged. * fix(migrations): pin the remaining chain tests to the 0038_user_disabled head The 0027/0028, 0029 and 0034 chain tests still pinned the single head to 0037_project_document_summaries (and 0034 asserted the head's parent was 0036_run_idempotency_request), so the renumbered lifecycle migration broke two unit-test shards. Follow the chain-test convention: every pinned head moves to 0038_user_disabled and 0034 walks the chain through 0037. * fix(authz): enforce account suspension on the internal-auth dispatch surface Round-3 review finding: the internal-auth principal is synthesized from trusted headers without a users-row lookup, so a suspended user with an active channel binding kept driving the agent, and scheduler-owned runs (their fabricated in-process requests skip AuthMiddleware entirely) kept launching. Two gates now enforce the lifecycle there: - AuthMiddleware rejects owner-bound internal calls whose users row is disabled (401 account_disabled) at the single gate every internal HTTP call passes through. Auth-disabled mode is exempt (no suspendable accounts, and its owner header carries the synthetic identity), and owner ids without a users row stay allowed — a nonexistent account cannot be suspended. - start_run re-asserts it at run admission after the owner is resolved for attribution: the scheduler records the rejection as the occurrence's launch error and no run record is created. Also addresses the remaining round-3 nits: drop the inline token_version check duplicated by validate_resolved_session_user, correct the local_provider comment (login never surfaces ACCOUNT_DISABLED), refresh the stale 0038 docstring header, and restore the deps module caches in the two suspension tests instead of leaking a closed engine. * fix(auth): carry the disabled field on the frontend synthetic users UserResponse now always includes disabled, so the auth-disabled contract e2e (gateway /auth/me deep-equals the frontend synthetic user) failed without the field on AUTH_DISABLED_USER. Add disabled to both synthetic users and declare it optional on the zod schema — absent still parses as not disabled for pre-gap-3 backends. * fix(authz): log when the internal-owner suspension check is skipped The middleware gate's users-lookup failure path silently degraded to allow; keep the fail-open behavior but leave a WARNING with the exception, so a store that cannot be read is observable in production logs instead of invisible. --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
hataa
GitHub
parent
29d654e86d
commit
47d8ff13b2
@ -22,6 +22,7 @@ class AuthErrorCode(StrEnum):
|
||||
NOT_AUTHENTICATED = "not_authenticated"
|
||||
SYSTEM_ALREADY_INITIALIZED = "system_already_initialized"
|
||||
REGISTRATION_DISABLED = "registration_disabled"
|
||||
ACCOUNT_DISABLED = "account_disabled"
|
||||
|
||||
|
||||
class TokenError(StrEnum):
|
||||
|
||||
@ -40,6 +40,14 @@ class LocalAuthProvider(AuthProvider):
|
||||
if user is None:
|
||||
return None
|
||||
|
||||
# Operator-disabled account (#3462 gap 3): no password is even
|
||||
# compared — the failure is indistinguishable from unknown-credentials
|
||||
# to the caller, so nothing about the account state leaks through the
|
||||
# login path (the distinct ACCOUNT_DISABLED code is surfaced only by
|
||||
# the session dependency once a token exists).
|
||||
if getattr(user, "disabled", False):
|
||||
return None
|
||||
|
||||
if user.password_hash is None:
|
||||
# OAuth user without local password
|
||||
return None
|
||||
|
||||
@ -34,6 +34,10 @@ class User(BaseModel):
|
||||
# Auth lifecycle
|
||||
needs_setup: bool = Field(default=False, description="True when a reset account must complete setup")
|
||||
token_version: int = Field(default=0, description="Incremented on password change to invalidate old JWTs")
|
||||
# Account lifecycle (RFC #4063 / #3462 gap 3): a disabled account is
|
||||
# rejected at every authentication surface; the row (credentials, role,
|
||||
# OAuth linkage) is retained so re-enabling restores what was suspended.
|
||||
disabled: bool = Field(default=False)
|
||||
|
||||
|
||||
class UserResponse(BaseModel):
|
||||
@ -42,6 +46,7 @@ class UserResponse(BaseModel):
|
||||
id: str
|
||||
email: str
|
||||
system_role: str = Field(..., min_length=1)
|
||||
disabled: bool = False
|
||||
needs_setup: bool = False
|
||||
oauth_provider: str | None = Field(None, description="OAuth/SSO provider ID if the user logged in via SSO (e.g. 'keycloak')")
|
||||
permissions: list[str] | None = Field(
|
||||
|
||||
@ -226,6 +226,11 @@ async def authenticate_pat(app: Any, authorization: str | None) -> tuple[Any, fr
|
||||
# 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")
|
||||
if getattr(user, "disabled", False):
|
||||
# Operator-disabled account (#3462 gap 3): the token stays valid on
|
||||
# disk but resolves nothing while the account is suspended; the same
|
||||
# generic 401 keeps the verdict from serving as an oracle.
|
||||
raise HTTPException(status_code=401, detail="Invalid token")
|
||||
await pat_repo.touch_last_used(str(record["id"]))
|
||||
return user, frozenset(record.get("scopes") or ())
|
||||
|
||||
|
||||
@ -14,6 +14,16 @@ class LastAdminRemainsError(RuntimeError):
|
||||
"""
|
||||
|
||||
|
||||
class LastActiveAdminError(RuntimeError):
|
||||
"""Raised when an account write would disable the last active admin.
|
||||
|
||||
"Active" = an enabled admin row: disabled admins cannot authenticate, so
|
||||
they cannot restore the deployment's management access. Raised inside the
|
||||
same serialized transaction that would perform the write (see
|
||||
``LastAdminRemainsError`` for the race this closes).
|
||||
"""
|
||||
|
||||
|
||||
class UserNotFoundError(LookupError):
|
||||
"""Raised when a user repository operation targets a non-existent row.
|
||||
|
||||
@ -90,6 +100,17 @@ class UserRepository(ABC):
|
||||
raise NotImplementedError
|
||||
|
||||
@abstractmethod
|
||||
async def set_disabled(self, user_id: str, disabled: bool) -> User:
|
||||
"""Enable/disable an account as a single-column, serialized write.
|
||||
|
||||
Origin: admin user-management surface (RFC #4063 / #3462 gap 3).
|
||||
Field-scoped like :meth:`update_system_role` — account-lifecycle
|
||||
state has exactly one writer, and credential/role writers never
|
||||
touch it. Disabling the last remaining ACTIVE admin raises
|
||||
:class:`LastActiveAdminError` (re-enabling is always allowed).
|
||||
"""
|
||||
raise NotImplementedError
|
||||
|
||||
async def update_system_role(self, user_id: str, system_role: str) -> User:
|
||||
"""Assign a user's system role as a single-column, serialized write.
|
||||
|
||||
|
||||
@ -21,7 +21,7 @@ from sqlalchemy.exc import IntegrityError
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from app.gateway.auth.models import User
|
||||
from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError, UserRepository
|
||||
from app.gateway.auth.repositories.base import LastActiveAdminError, LastAdminRemainsError, UserNotFoundError, UserRepository
|
||||
from deerflow.persistence.user.model import OAUTH_IDENTITY_INDEX_NAME, UserRow
|
||||
|
||||
# ``email`` is ``mapped_column(unique=True, index=True)``, which SQLAlchemy
|
||||
@ -161,6 +161,7 @@ class SQLiteUserRepository(UserRepository):
|
||||
oauth_id=row.oauth_id,
|
||||
needs_setup=row.needs_setup,
|
||||
token_version=row.token_version,
|
||||
disabled=row.disabled,
|
||||
)
|
||||
|
||||
@staticmethod
|
||||
@ -175,6 +176,7 @@ class SQLiteUserRepository(UserRepository):
|
||||
oauth_id=user.oauth_id,
|
||||
needs_setup=user.needs_setup,
|
||||
token_version=user.token_version,
|
||||
disabled=user.disabled,
|
||||
)
|
||||
|
||||
# ── CRUD ──────────────────────────────────────────────────────────
|
||||
@ -315,13 +317,34 @@ class SQLiteUserRepository(UserRepository):
|
||||
if row is None:
|
||||
raise UserNotFoundError(f"User {user_id} no longer exists")
|
||||
if row.system_role == "admin" and system_role != "admin":
|
||||
admin_count = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin"))
|
||||
if admin_count is None or admin_count <= 1:
|
||||
# Count ACTIVE admins only: a disabled admin cannot
|
||||
# authenticate, so counting them would let "disable B, then
|
||||
# demote self" (or a combined role+disable request) strand
|
||||
# the deployment with zero usable management credentials.
|
||||
active_admins = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin", UserRow.disabled.is_(False)))
|
||||
if active_admins is None or active_admins <= 1:
|
||||
raise LastAdminRemainsError("cannot demote the last remaining admin")
|
||||
row.system_role = system_role
|
||||
await session.commit()
|
||||
return self._row_to_user(row)
|
||||
|
||||
async def set_disabled(self, user_id: str, disabled: bool) -> User:
|
||||
"""Serialized single-column account enable/disable (see base contract)."""
|
||||
async with self._sf() as session:
|
||||
# Same serialization idiom as update_system_role: the
|
||||
# active-admin count and the column write share one transaction.
|
||||
await self._serialize_account_role_mutation(session)
|
||||
row = await session.get(UserRow, user_id)
|
||||
if row is None:
|
||||
raise UserNotFoundError(f"User {user_id} no longer exists")
|
||||
if disabled and not row.disabled and row.system_role == "admin":
|
||||
active_admins = await session.scalar(select(func.count()).select_from(UserRow).where(UserRow.system_role == "admin", UserRow.disabled.is_(False)))
|
||||
if active_admins is None or active_admins <= 1:
|
||||
raise LastActiveAdminError("cannot disable the last remaining active admin")
|
||||
row.disabled = disabled
|
||||
await session.commit()
|
||||
return self._row_to_user(row)
|
||||
|
||||
@staticmethod
|
||||
async def _serialize_account_role_mutation(session: AsyncSession) -> None:
|
||||
"""Serialize role mutations before the admin count is read.
|
||||
@ -375,6 +398,9 @@ class SQLiteUserRepository(UserRepository):
|
||||
# Role changes go through update_system_role, which likewise
|
||||
# never touches credentials.
|
||||
user.system_role = row.system_role
|
||||
# disabled is account-lifecycle state: only set_disabled writes
|
||||
# it (same field-scoping as the role column).
|
||||
user.disabled = row.disabled
|
||||
row.oauth_provider = user.oauth_provider
|
||||
row.oauth_id = user.oauth_id
|
||||
row.needs_setup = user.needs_setup
|
||||
|
||||
@ -38,6 +38,14 @@ async def get_or_provision_oidc_user(
|
||||
# 1. Existing OAuth link
|
||||
existing = await local_provider.get_user_by_oauth(provider_id, identity.subject)
|
||||
if existing:
|
||||
if getattr(existing, "disabled", False):
|
||||
# Operator-disabled account (#3462 gap 3): the IdP authenticated
|
||||
# the identity, but the account is suspended — the row stays
|
||||
# linked so re-enabling restores access unchanged.
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_403_FORBIDDEN,
|
||||
detail="This account has been disabled. Please contact your administrator.",
|
||||
)
|
||||
return {"user": existing, "created": False}
|
||||
|
||||
# 2. Verified email requirement
|
||||
|
||||
@ -9,6 +9,7 @@ owner filtering works automatically via the sentinel pattern.
|
||||
Fine-grained permission checks remain in authz.py decorators.
|
||||
"""
|
||||
|
||||
import logging
|
||||
from collections.abc import Callable
|
||||
|
||||
from fastapi import HTTPException, Request, Response
|
||||
@ -30,6 +31,8 @@ from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME, get_internal_us
|
||||
from app.gateway.request_path import get_request_route_path
|
||||
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# Paths that never require authentication.
|
||||
_PUBLIC_PATH_PREFIXES: tuple[str, ...] = (
|
||||
"/health",
|
||||
@ -104,6 +107,35 @@ class AuthMiddleware(BaseHTTPMiddleware):
|
||||
if owner_user_id:
|
||||
owner_user_id = owner_user_id.strip()
|
||||
internal_user = get_internal_user(owner_user_id=owner_user_id or None)
|
||||
if owner_user_id and not is_auth_disabled():
|
||||
# The internal principal is synthesized from trusted headers
|
||||
# without a users-row lookup, so suspension has no other chance
|
||||
# to bite on this surface (#3462 gap 3): enforce it here, the
|
||||
# one gate every owner-bound internal HTTP call (IM channel
|
||||
# dispatch above all) passes through. Auth-disabled mode is
|
||||
# exempt — its owner header carries the synthetic
|
||||
# auth-disabled identity and no suspendable account exists.
|
||||
# Owner ids without a users row stay allowed (a nonexistent
|
||||
# account cannot be suspended), and a store that cannot be
|
||||
# read cannot hold a suspension verdict either; run admission
|
||||
# re-asserts the check when the row is readable.
|
||||
from app.gateway.deps import get_local_provider
|
||||
|
||||
try:
|
||||
owner = await get_local_provider().get_user(owner_user_id)
|
||||
except Exception:
|
||||
logger.warning("Internal-owner suspension check skipped: users lookup failed", exc_info=True)
|
||||
owner = None
|
||||
if owner is not None and getattr(owner, "disabled", False):
|
||||
return JSONResponse(
|
||||
status_code=401,
|
||||
content={
|
||||
"detail": AuthErrorResponse(
|
||||
code=AuthErrorCode.ACCOUNT_DISABLED,
|
||||
message="Account disabled",
|
||||
).model_dump()
|
||||
},
|
||||
)
|
||||
|
||||
auth_source = AUTH_SOURCE_SESSION
|
||||
access_token = request.cookies.get("access_token")
|
||||
|
||||
@ -28,6 +28,7 @@ from typing import TYPE_CHECKING, Any, TypeVar, cast
|
||||
from fastapi import FastAPI, HTTPException, Request
|
||||
from langgraph.types import Checkpointer
|
||||
|
||||
from app.gateway.auth.errors import AuthErrorCode
|
||||
from deerflow.community.browser_automation.session import browser_multi_worker_error
|
||||
from deerflow.config.app_config import AppConfig, get_app_config
|
||||
from deerflow.config.deployment_config import multi_instance_declaration
|
||||
@ -1126,16 +1127,38 @@ async def get_current_user_from_request(request: Request):
|
||||
detail=AuthErrorResponse(code=AuthErrorCode.USER_NOT_FOUND, message="User not found").model_dump(),
|
||||
)
|
||||
|
||||
# Token version mismatch → password was changed, token is stale
|
||||
if user.token_version != payload.ver:
|
||||
# Operator-disabled account (#3462 gap 3) and stale token versions are
|
||||
# both verdicts of the shared post-lookup validator: the password and
|
||||
# PAT paths reject at their own surfaces; OAuth provisioning rejects at
|
||||
# resolve time.
|
||||
error = validate_resolved_session_user(user, payload)
|
||||
if error is not None:
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail=AuthErrorResponse(code=AuthErrorCode.TOKEN_INVALID, message="Token revoked (password changed)").model_dump(),
|
||||
detail=AuthErrorResponse(code=error, message="Account disabled" if error is AuthErrorCode.ACCOUNT_DISABLED else "Token revoked (password changed)").model_dump(),
|
||||
)
|
||||
|
||||
return user
|
||||
|
||||
|
||||
def validate_resolved_session_user(user, payload) -> AuthErrorCode | None:
|
||||
"""Shared post-lookup session validation for EVERY JWT authenticator.
|
||||
|
||||
The Gateway's HTTP dependency, the WebSocket authenticator (browser
|
||||
streaming bypasses AuthMiddleware), and the standalone LangGraph
|
||||
``authenticate`` callback all resolve cookie → JWT → user; this helper
|
||||
is the one place the post-lookup verdicts live so a lifecycle change
|
||||
(token_version bump, account suspension) lands everywhere at once.
|
||||
Returns the failure code, or ``None`` when the session is valid —
|
||||
callers map it to their own exception surface.
|
||||
"""
|
||||
if user.token_version != payload.ver:
|
||||
return AuthErrorCode.TOKEN_INVALID
|
||||
if getattr(user, "disabled", False):
|
||||
return AuthErrorCode.ACCOUNT_DISABLED
|
||||
return None
|
||||
|
||||
|
||||
async def is_admin_user(request: Request) -> bool:
|
||||
"""Return whether the authenticated caller is an admin user.
|
||||
|
||||
|
||||
@ -276,10 +276,17 @@ async def authenticate(request):
|
||||
status_code=401,
|
||||
detail="User not found",
|
||||
)
|
||||
if user.token_version != payload.ver:
|
||||
# Same post-lookup verdicts as every other JWT surface — including
|
||||
# account suspension, so a disabled user's still-valid cookie is
|
||||
# rejected here exactly as the Gateway's HTTP dependency rejects it.
|
||||
from app.gateway.auth.errors import AuthErrorCode
|
||||
from app.gateway.deps import validate_resolved_session_user
|
||||
|
||||
failure = validate_resolved_session_user(user, payload)
|
||||
if failure is not None:
|
||||
raise Auth.exceptions.HTTPException(
|
||||
status_code=401,
|
||||
detail="Token revoked (password changed)",
|
||||
detail="Token revoked (password changed)" if failure is AuthErrorCode.TOKEN_INVALID else "Account disabled",
|
||||
)
|
||||
|
||||
return payload.sub
|
||||
|
||||
@ -21,7 +21,7 @@ from uuid import UUID
|
||||
from fastapi import APIRouter, Depends, HTTPException, Request, status
|
||||
from pydantic import BaseModel, Field
|
||||
|
||||
from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError
|
||||
from app.gateway.auth.repositories.base import LastActiveAdminError, LastAdminRemainsError, UserNotFoundError
|
||||
from app.gateway.authz import assignable_role_names
|
||||
from app.gateway.deps import get_user_repository, require_admin_user
|
||||
|
||||
@ -33,10 +33,16 @@ from app.gateway.auth.models import UserResponse
|
||||
router = APIRouter(prefix="/api/v1/admin/users", tags=["admin-users"])
|
||||
|
||||
|
||||
class UserRoleUpdate(BaseModel):
|
||||
"""Assignment payload: a built-in or provider-configured role name."""
|
||||
class UserAccountUpdate(BaseModel):
|
||||
"""Partial account update: role and/or enable-disable.
|
||||
|
||||
system_role: str = Field(..., min_length=1, max_length=16, description="Built-in (admin/user) or a role configured in the authorization provider's roles mapping")
|
||||
Omitted fields are unchanged; each present field routes to its
|
||||
field-scoped writer (update_system_role / set_disabled) so role,
|
||||
credentials, and lifecycle state can never clobber one another.
|
||||
"""
|
||||
|
||||
system_role: str | None = Field(None, min_length=1, max_length=16, description="Built-in (admin/user) or a role configured in the authorization provider's roles mapping")
|
||||
disabled: bool | None = Field(None, description="Suspend (true) or restore (false) the account; every authentication surface rejects a disabled account")
|
||||
|
||||
|
||||
def _to_response(user) -> UserResponse:
|
||||
@ -46,6 +52,7 @@ def _to_response(user) -> UserResponse:
|
||||
system_role=str(user.system_role),
|
||||
needs_setup=bool(getattr(user, "needs_setup", False)),
|
||||
oauth_provider=getattr(user, "oauth_provider", None),
|
||||
disabled=bool(getattr(user, "disabled", False)),
|
||||
)
|
||||
|
||||
|
||||
@ -58,38 +65,55 @@ async def list_users(request: Request, repository: SQLiteUserRepository = Depend
|
||||
|
||||
|
||||
@router.patch("/{user_id}")
|
||||
async def update_user_role(
|
||||
async def update_user_account(
|
||||
request: Request,
|
||||
user_id: UUID,
|
||||
update: UserRoleUpdate,
|
||||
update: UserAccountUpdate,
|
||||
repository: SQLiteUserRepository = Depends(get_user_repository),
|
||||
) -> UserResponse:
|
||||
"""Assign a user's system role (admin-only).
|
||||
"""Update a user's account: assign a role and/or enable/disable (admin-only).
|
||||
|
||||
The role must be a built-in ("admin"/"user") or declared in the
|
||||
authorization provider's roles configuration — an unknown name would make
|
||||
the user's principal fall through to ``default_role`` on every decision,
|
||||
which is never what an operator intended. Demoting the last remaining
|
||||
admin is rejected so a deployment cannot lock itself out of management.
|
||||
admin, and disabling the last remaining ACTIVE admin, are both rejected
|
||||
so a deployment cannot lock itself out of management. Each present field
|
||||
routes to its field-scoped, serialized writer; enforcement of the
|
||||
disabled state happens at every authentication surface (password login,
|
||||
JWT sessions, PATs, OAuth provisioning), not here.
|
||||
"""
|
||||
await require_admin_user(request, detail="role assignment requires an admin credential")
|
||||
await require_admin_user(request, detail="account updates require an admin credential")
|
||||
|
||||
assignable = assignable_role_names()
|
||||
if update.system_role not in assignable:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"unknown role '{update.system_role}'; assignable roles: {sorted(assignable)}",
|
||||
)
|
||||
if update.system_role is None and update.disabled is None:
|
||||
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail="nothing to update: provide system_role and/or disabled")
|
||||
|
||||
if update.system_role is not None:
|
||||
assignable = assignable_role_names()
|
||||
if update.system_role not in assignable:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
||||
detail=f"unknown role '{update.system_role}'; assignable roles: {sorted(assignable)}",
|
||||
)
|
||||
|
||||
user: object | None = None
|
||||
try:
|
||||
# Single serialized column write: the last-admin count and the write
|
||||
# share one transaction (two concurrent demotions cannot both pass),
|
||||
# and credential writers can never race this — see
|
||||
# SQLiteUserRepository.update_system_role / update_user.
|
||||
user = await repository.update_system_role(str(user_id), update.system_role)
|
||||
if update.system_role is not None:
|
||||
# Single serialized column write: the last-admin count and the
|
||||
# write share one transaction (two concurrent demotions cannot
|
||||
# both pass), and credential writers can never race this — see
|
||||
# SQLiteUserRepository.update_system_role / update_user.
|
||||
user = await repository.update_system_role(str(user_id), update.system_role)
|
||||
if update.disabled is not None:
|
||||
# Same serialization for the active-admin guard; re-enabling is
|
||||
# always allowed.
|
||||
user = await repository.set_disabled(str(user_id), update.disabled)
|
||||
except UserNotFoundError:
|
||||
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="user not found") from None
|
||||
except LastAdminRemainsError:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="cannot demote the last remaining admin") from None
|
||||
except LastActiveAdminError:
|
||||
raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail="cannot disable the last remaining active admin") from None
|
||||
|
||||
assert user is not None
|
||||
return _to_response(user)
|
||||
|
||||
@ -133,8 +133,15 @@ async def _authenticate_ws(websocket: WebSocket):
|
||||
if not isinstance(payload, TokenError):
|
||||
provider = get_local_provider()
|
||||
user = await provider.get_user(payload.sub)
|
||||
if user is not None and user.token_version == payload.ver:
|
||||
return user
|
||||
if user is not None:
|
||||
# Same post-lookup verdicts as every other JWT surface —
|
||||
# including account suspension: WebSockets bypass
|
||||
# AuthMiddleware, so a disabled user's still-valid cookie
|
||||
# must not open a streaming/browser session.
|
||||
from app.gateway.deps import validate_resolved_session_user
|
||||
|
||||
if validate_resolved_session_user(user, payload) is None:
|
||||
return user
|
||||
if is_auth_disabled():
|
||||
return get_auth_disabled_user()
|
||||
return None
|
||||
|
||||
@ -2048,6 +2048,19 @@ async def start_run(
|
||||
internal_owner_user = await resolve_trusted_internal_owner_for_attribution(request, owner_user_id)
|
||||
if internal_owner_user is None and owner_user_id == AUTH_DISABLED_USER_ID and is_auth_disabled():
|
||||
internal_owner_user = get_auth_disabled_user()
|
||||
if getattr(internal_owner_user, "disabled", False):
|
||||
# The scheduler's in-process launches (and MCP task-event
|
||||
# delivery) fabricate requests that skip AuthMiddleware, where
|
||||
# the internal-auth suspension gate lives. Re-assert it at run
|
||||
# admission (#3462 gap 3): a disabled owner's occurrence fails
|
||||
# here instead of driving the agent. The scheduler records the
|
||||
# rejection as the occurrence's launch error.
|
||||
from app.gateway.auth.errors import AuthErrorCode, AuthErrorResponse
|
||||
|
||||
raise HTTPException(
|
||||
status_code=401,
|
||||
detail=AuthErrorResponse(code=AuthErrorCode.ACCOUNT_DISABLED, message="Account disabled").model_dump(),
|
||||
)
|
||||
inject_authenticated_user_context(
|
||||
config,
|
||||
request,
|
||||
|
||||
@ -0,0 +1,45 @@
|
||||
"""Account disabled flag on ``users``.
|
||||
|
||||
Revision ID: 0039_user_disabled
|
||||
Revises: 0038_channel_thread_bindings
|
||||
|
||||
Adds ``users.disabled`` (NOT NULL, server default ``0``): an
|
||||
operator-disabled account is rejected at every authentication surface while
|
||||
the row — credentials, role, OAuth linkage — is retained, so re-enabling
|
||||
restores exactly what was suspended (RFC #4063 / issue #3462 gap 3).
|
||||
|
||||
Column-guarded in both directions: an upgrade over a database that already
|
||||
has the column (an interrupted earlier attempt, or an empty-database
|
||||
bootstrap whose ``create_all`` saw the model first) completes, and a
|
||||
downgrade after the column is gone still moves the version row.
|
||||
"""
|
||||
|
||||
from collections.abc import Sequence
|
||||
|
||||
import sqlalchemy as sa
|
||||
from alembic import op
|
||||
|
||||
revision: str = "0039_user_disabled"
|
||||
down_revision: str | None = "0038_channel_thread_bindings"
|
||||
branch_labels: str | Sequence[str] | None = None
|
||||
depends_on: str | Sequence[str] | None = None
|
||||
|
||||
_TABLE = "users"
|
||||
_COLUMN = "disabled"
|
||||
|
||||
|
||||
def _has_column() -> bool:
|
||||
bind = op.get_bind()
|
||||
if not sa.inspect(bind).has_table(_TABLE):
|
||||
return False
|
||||
return _COLUMN in {column["name"] for column in sa.inspect(bind).get_columns(_TABLE)}
|
||||
|
||||
|
||||
def upgrade() -> None:
|
||||
if not _has_column():
|
||||
op.add_column(_TABLE, sa.Column(_COLUMN, sa.Boolean(), nullable=False, server_default=sa.false()))
|
||||
|
||||
|
||||
def downgrade() -> None:
|
||||
if _has_column():
|
||||
op.drop_column(_TABLE, _COLUMN)
|
||||
@ -13,7 +13,7 @@ from __future__ import annotations
|
||||
|
||||
from datetime import UTC, datetime
|
||||
|
||||
from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, String, text
|
||||
from sqlalchemy import JSON, Boolean, DateTime, ForeignKey, Index, String, false, text
|
||||
from sqlalchemy.orm import Mapped, mapped_column
|
||||
|
||||
from deerflow.persistence.base import Base
|
||||
@ -68,6 +68,12 @@ class UserRow(Base):
|
||||
needs_setup: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
|
||||
token_version: Mapped[int] = mapped_column(nullable=False, default=0)
|
||||
|
||||
# Account lifecycle (RFC #4063 / issue #3462 gap 3): an operator-disabled
|
||||
# account is rejected at every authentication surface (password login,
|
||||
# JWT session resolution, PAT, OAuth) — the row is retained with its
|
||||
# credentials and role so re-enabling restores exactly what was suspended.
|
||||
disabled: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False, server_default=false())
|
||||
|
||||
__table_args__ = (
|
||||
# sqlite_where alone is a SQLAlchemy dialect-specific kwarg -- it
|
||||
# does not apply on the postgresql dialect, so a table created
|
||||
|
||||
@ -50,6 +50,19 @@ class _FakeRepo:
|
||||
self.users[str(user.id)] = user
|
||||
return user
|
||||
|
||||
async def set_disabled(self, user_id: str, disabled: bool) -> User:
|
||||
from app.gateway.auth.repositories.base import LastActiveAdminError, UserNotFoundError
|
||||
|
||||
user = self.users.get(user_id)
|
||||
if user is None:
|
||||
raise UserNotFoundError(f"User {user_id} no longer exists")
|
||||
if disabled and not getattr(user, "disabled", False) and user.system_role == "admin":
|
||||
if sum(1 for u in self.users.values() if u.system_role == "admin" and not getattr(u, "disabled", False)) <= 1:
|
||||
raise LastActiveAdminError("cannot disable the last remaining active admin")
|
||||
user.disabled = disabled
|
||||
self.updates.append((user_id, f"disabled={disabled}"))
|
||||
return user
|
||||
|
||||
async def update_system_role(self, user_id: str, system_role: str) -> User:
|
||||
from app.gateway.auth.repositories.base import LastAdminRemainsError, UserNotFoundError
|
||||
|
||||
@ -57,7 +70,9 @@ class _FakeRepo:
|
||||
if user is None:
|
||||
raise UserNotFoundError(f"User {user_id} no longer exists")
|
||||
if user.system_role == "admin" and system_role != "admin":
|
||||
if await self.count_admin_users() <= 1:
|
||||
# Active admins only (disabled admins cannot authenticate).
|
||||
active = sum(1 for u in self.users.values() if u.system_role == "admin" and not getattr(u, "disabled", False))
|
||||
if active <= 1:
|
||||
raise LastAdminRemainsError("cannot demote the last remaining admin")
|
||||
user.system_role = system_role
|
||||
self.updates.append((user_id, system_role))
|
||||
@ -303,3 +318,277 @@ def test_assignment_reaches_principal_permissions():
|
||||
principal = build_principal_from_context({"user_id": "u1", "user_role": "guest"}, default_role="user")
|
||||
|
||||
assert principal.role == "guest"
|
||||
|
||||
|
||||
def test_admin_disables_and_reenables_account(monkeypatch):
|
||||
admin = _make_user(system_role="admin")
|
||||
target = _make_user(system_role="guest")
|
||||
repo = _FakeRepo([admin, target])
|
||||
client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user", "guest"})
|
||||
|
||||
disabled = client.patch(f"/api/v1/admin/users/{target.id}", json={"disabled": True})
|
||||
assert disabled.status_code == 200
|
||||
assert disabled.json()["disabled"] is True
|
||||
|
||||
reenabled = client.patch(f"/api/v1/admin/users/{target.id}", json={"disabled": False})
|
||||
assert reenabled.status_code == 200
|
||||
assert reenabled.json()["disabled"] is False
|
||||
|
||||
|
||||
def test_cannot_disable_last_active_admin(monkeypatch):
|
||||
admin = _make_user(system_role="admin")
|
||||
repo = _FakeRepo([admin])
|
||||
client = _make_client(monkeypatch, caller=admin, repo=repo)
|
||||
|
||||
response = client.patch(f"/api/v1/admin/users/{admin.id}", json={"disabled": True})
|
||||
|
||||
assert response.status_code == 409
|
||||
assert repo.users[str(admin.id)].disabled is False
|
||||
|
||||
|
||||
def test_empty_account_update_rejected(monkeypatch):
|
||||
admin = _make_user(system_role="admin")
|
||||
target = _make_user()
|
||||
repo = _FakeRepo([admin, target])
|
||||
client = _make_client(monkeypatch, caller=admin, repo=repo)
|
||||
|
||||
assert client.patch(f"/api/v1/admin/users/{target.id}", json={}).status_code == 422
|
||||
|
||||
|
||||
def test_combined_role_and_disable(monkeypatch):
|
||||
admin = _make_user(system_role="admin")
|
||||
other = _make_user(system_role="admin", email="b@example.com")
|
||||
target = _make_user()
|
||||
repo = _FakeRepo([admin, other, target])
|
||||
client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user", "guest"})
|
||||
|
||||
response = client.patch(f"/api/v1/admin/users/{target.id}", json={"system_role": "guest", "disabled": True})
|
||||
|
||||
assert response.status_code == 200
|
||||
body = response.json()
|
||||
assert body["system_role"] == "guest" and body["disabled"] is True
|
||||
|
||||
|
||||
def test_disable_then_demote_self_strands_no_admin(monkeypatch):
|
||||
"""[P1 regression] "Disable B, then demote self" and the combined
|
||||
{system_role, disabled} self-update must both be rejected: the role
|
||||
demotion counts ACTIVE admins only, so with B disabled the self-demotion
|
||||
would strand zero usable management credentials."""
|
||||
admin = _make_user(system_role="admin")
|
||||
other = _make_user(system_role="admin", email="b@example.com")
|
||||
repo = _FakeRepo([admin, other])
|
||||
client = _make_client(monkeypatch, caller=admin, repo=repo, roles={"admin", "user"})
|
||||
|
||||
# Disable B first.
|
||||
assert client.patch(f"/api/v1/admin/users/{other.id}", json={"disabled": True}).status_code == 200
|
||||
|
||||
# Self-demotion now strands: only A remains active.
|
||||
demote = client.patch(f"/api/v1/admin/users/{admin.id}", json={"system_role": "user"})
|
||||
assert demote.status_code == 409
|
||||
assert repo.users[str(admin.id)].system_role == "admin"
|
||||
|
||||
# Combined self demote+disable has the same outcome (role write first,
|
||||
# active-count guard fires).
|
||||
combined = client.patch(f"/api/v1/admin/users/{admin.id}", json={"system_role": "user", "disabled": True})
|
||||
assert combined.status_code == 409
|
||||
assert repo.users[str(admin.id)].system_role == "admin"
|
||||
|
||||
|
||||
def test_shared_session_validator_verdicts():
|
||||
"""The one post-lookup verdict helper every JWT surface routes through."""
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
from app.gateway.auth.errors import AuthErrorCode
|
||||
from app.gateway.deps import validate_resolved_session_user
|
||||
|
||||
user = NS(token_version=3, disabled=False)
|
||||
assert validate_resolved_session_user(user, NS(ver=3)) is None
|
||||
|
||||
stale = NS(token_version=4, disabled=False)
|
||||
assert validate_resolved_session_user(stale, NS(ver=3)) is AuthErrorCode.TOKEN_INVALID
|
||||
|
||||
suspended = NS(token_version=3, disabled=True)
|
||||
assert validate_resolved_session_user(suspended, NS(ver=3)) is AuthErrorCode.ACCOUNT_DISABLED
|
||||
|
||||
|
||||
def test_langgraph_authenticate_rejects_suspended_session():
|
||||
"""The standalone LangGraph authenticate callback rejects a still-valid
|
||||
cookie whose account was suspended — admission, not just login."""
|
||||
import asyncio
|
||||
import tempfile
|
||||
|
||||
import pytest
|
||||
|
||||
from app.gateway.auth.jwt import create_access_token
|
||||
from deerflow.persistence.engine import close_engine
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
repo = _make_sqlite_repo(tmpdir)
|
||||
|
||||
import app.gateway.deps as deps_module
|
||||
from app.gateway import langgraph_auth
|
||||
|
||||
# Point the cached provider at THIS test's repo (module-level
|
||||
# caches otherwise leak a previous test's closed engine), and put
|
||||
# the previous values back afterwards so later tests that call
|
||||
# get_local_provider() do not see this test's closed engine.
|
||||
saved_repo = deps_module._cached_repo
|
||||
saved_provider = deps_module._cached_local_provider
|
||||
deps_module._cached_repo = repo
|
||||
deps_module._cached_local_provider = None
|
||||
|
||||
async def _run():
|
||||
user = await repo.create_user(User(email="lg@example.com", system_role="user"))
|
||||
token = create_access_token(str(user.id), token_version=user.token_version)
|
||||
await repo.set_disabled(str(user.id), True)
|
||||
|
||||
request = NS_stub_request(token)
|
||||
|
||||
with pytest.raises(langgraph_auth.Auth.exceptions.HTTPException) as exc_info:
|
||||
await langgraph_auth.authenticate(request)
|
||||
assert exc_info.value.status_code == 401
|
||||
assert "disabled" in str(exc_info.value.detail).lower()
|
||||
|
||||
def NS_stub_request(token_value):
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
return NS(cookies={"access_token": token_value}, headers={}, method="GET", url=NS(path="/"), client=NS(host="test"))
|
||||
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
finally:
|
||||
asyncio.run(close_engine())
|
||||
deps_module._cached_repo = saved_repo
|
||||
deps_module._cached_local_provider = saved_provider
|
||||
|
||||
|
||||
def test_browser_ws_authenticator_rejects_suspended_session():
|
||||
"""The WebSocket authenticator (browser streaming bypasses
|
||||
AuthMiddleware) rejects a suspended account's still-valid cookie."""
|
||||
import asyncio
|
||||
import tempfile
|
||||
|
||||
from app.gateway.auth.jwt import create_access_token
|
||||
from deerflow.persistence.engine import close_engine
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
repo = _make_sqlite_repo(tmpdir)
|
||||
|
||||
from types import SimpleNamespace as NS
|
||||
|
||||
import app.gateway.deps as deps_module
|
||||
from app.gateway.routers.browser import _authenticate_ws
|
||||
|
||||
# Same cache hygiene as the authenticate test above: restore the
|
||||
# module-level caches so this test's closed engine does not leak.
|
||||
saved_repo = deps_module._cached_repo
|
||||
saved_provider = deps_module._cached_local_provider
|
||||
deps_module._cached_repo = repo
|
||||
deps_module._cached_local_provider = None
|
||||
|
||||
async def _run():
|
||||
user = await repo.create_user(User(email="ws@example.com", system_role="user"))
|
||||
token = create_access_token(str(user.id), token_version=user.token_version)
|
||||
await repo.set_disabled(str(user.id), True)
|
||||
|
||||
websocket = NS(cookies={"access_token": token}, headers={})
|
||||
assert await _authenticate_ws(websocket) is None
|
||||
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
finally:
|
||||
asyncio.run(close_engine())
|
||||
deps_module._cached_repo = saved_repo
|
||||
deps_module._cached_local_provider = saved_provider
|
||||
|
||||
|
||||
def test_repo_set_disabled_field_scoped_and_last_active_admin():
|
||||
"""Real-repo slice: set_disabled is a single-column write (role and
|
||||
credentials untouched; the credential writer never touches lifecycle
|
||||
state), and the last-ACTIVE-admin guard fires inside the serialized
|
||||
write — a disabled admin does not count as active, and re-enabling is
|
||||
always allowed."""
|
||||
import asyncio
|
||||
import tempfile
|
||||
|
||||
from app.gateway.auth.repositories.base import LastActiveAdminError
|
||||
from deerflow.persistence.engine import close_engine
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
repo = _make_sqlite_repo(tmpdir)
|
||||
|
||||
async def _run():
|
||||
first = await repo.create_user(User(email="a@example.com", system_role="admin", password_hash="h1"))
|
||||
second = await repo.create_user(User(email="b@example.com", system_role="admin", password_hash="h2"))
|
||||
|
||||
disabled_first = await repo.set_disabled(str(first.id), True)
|
||||
assert disabled_first.disabled is True
|
||||
assert disabled_first.password_hash == "h1"
|
||||
assert disabled_first.system_role == "admin"
|
||||
|
||||
# The credential writer never touches lifecycle state: a stale
|
||||
# snapshot with disabled=False does not re-enable the account.
|
||||
stale = User(id=first.id, email=first.email, password_hash="h3", system_role="admin")
|
||||
after = await repo.update_user(stale)
|
||||
assert after.password_hash == "h3"
|
||||
assert after.disabled is True
|
||||
|
||||
# Only one ACTIVE admin remains: disabling it raises, row intact.
|
||||
try:
|
||||
await repo.set_disabled(str(second.id), True)
|
||||
except LastActiveAdminError:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError("disabling the last active admin must raise")
|
||||
assert (await repo.get_user_by_id(str(second.id))).disabled is False
|
||||
|
||||
restored = await repo.set_disabled(str(first.id), False)
|
||||
assert restored.disabled is False
|
||||
|
||||
# The role-demotion guard counts ACTIVE admins only: disable
|
||||
# one of two, then demoting the remaining active admin raises.
|
||||
await repo.set_disabled(str(first.id), True)
|
||||
from app.gateway.auth.repositories.base import LastAdminRemainsError as _LAR
|
||||
|
||||
try:
|
||||
await repo.update_system_role(str(second.id), "user")
|
||||
except _LAR:
|
||||
pass
|
||||
else:
|
||||
raise AssertionError("demoting the last ACTIVE admin must raise")
|
||||
assert (await repo.get_user_by_id(str(second.id))).system_role == "admin"
|
||||
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
finally:
|
||||
asyncio.run(close_engine())
|
||||
|
||||
|
||||
def test_disabled_account_rejected_at_login_and_flagged_for_resolvers():
|
||||
"""Password login never compares credentials for a disabled account,
|
||||
and lookups return the flag so the JWT resolver can reject."""
|
||||
import asyncio
|
||||
import tempfile
|
||||
|
||||
from app.gateway.auth.local_provider import LocalAuthProvider
|
||||
from deerflow.persistence.engine import close_engine
|
||||
|
||||
with tempfile.TemporaryDirectory() as tmpdir:
|
||||
repo = _make_sqlite_repo(tmpdir)
|
||||
|
||||
async def _run():
|
||||
from app.gateway.auth.password import hash_password
|
||||
|
||||
user = await repo.create_user(User(email="d@example.com", system_role="user", password_hash=hash_password("correct-horse")))
|
||||
await repo.set_disabled(str(user.id), True)
|
||||
provider = LocalAuthProvider(repository=repo)
|
||||
|
||||
# The CORRECT password: only the disabled gate can reject it.
|
||||
assert await provider.authenticate({"email": "d@example.com", "password": "correct-horse"}) is None
|
||||
fetched = await provider.get_user(str(user.id))
|
||||
assert fetched is not None and fetched.disabled is True
|
||||
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
finally:
|
||||
asyncio.run(close_engine())
|
||||
|
||||
@ -447,6 +447,113 @@ def test_protected_post_with_internal_auth_header_passes():
|
||||
assert res.status_code == 200
|
||||
|
||||
|
||||
def test_internal_auth_owner_header_of_suspended_account_rejected(tmp_path):
|
||||
"""(#3462 gap 3) IM-channel dispatch surface: the internal principal is
|
||||
synthesized from trusted headers without a users-row lookup, so the
|
||||
middleware itself rejects an owner-bound internal call whose account was
|
||||
suspended — at the very next request, like every other surface."""
|
||||
import asyncio
|
||||
import json
|
||||
|
||||
from starlette.requests import Request as StarletteRequest
|
||||
from starlette.responses import PlainTextResponse
|
||||
|
||||
from app.gateway.auth.models import User
|
||||
from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
|
||||
from app.gateway.auth_middleware import AuthMiddleware
|
||||
from app.gateway.internal_auth import create_internal_auth_headers
|
||||
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
|
||||
|
||||
async def _dispatch(owner_user_id: str):
|
||||
headers = create_internal_auth_headers(owner_user_id=owner_user_id)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/api/threads/abc/runs/stream",
|
||||
"root_path": "",
|
||||
"query_string": b"",
|
||||
"headers": [(name.lower().encode(), value.encode()) for name, value in headers.items()],
|
||||
}
|
||||
|
||||
async def call_next(_request):
|
||||
return PlainTextResponse("route-reached")
|
||||
|
||||
response = await AuthMiddleware(None).dispatch(StarletteRequest(scope), call_next)
|
||||
body = json.loads(response.body) if response.status_code != 200 else {}
|
||||
return response.status_code, body
|
||||
|
||||
async def _run():
|
||||
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path}/users.db", sqlite_dir=str(tmp_path))
|
||||
repo = SQLiteUserRepository(get_session_factory())
|
||||
owner = await repo.create_user(User(email="channel-owner@example.com", system_role="user"))
|
||||
await repo.set_disabled(str(owner.id), True)
|
||||
|
||||
import app.gateway.deps as deps_module
|
||||
|
||||
saved_repo, saved_provider = deps_module._cached_repo, deps_module._cached_local_provider
|
||||
deps_module._cached_repo = repo
|
||||
deps_module._cached_local_provider = None
|
||||
try:
|
||||
status_code, body = await _dispatch(str(owner.id))
|
||||
assert status_code == 401
|
||||
assert body["detail"]["code"] == "account_disabled"
|
||||
|
||||
# Unknown owner ids stay allowed: an account that does not exist
|
||||
# in the users store cannot be suspended.
|
||||
status_code, body = await _dispatch("no-such-owner")
|
||||
assert status_code == 200
|
||||
|
||||
# Re-enabling restores the internal surface immediately.
|
||||
await repo.set_disabled(str(owner.id), False)
|
||||
status_code, body = await _dispatch(str(owner.id))
|
||||
assert status_code == 200
|
||||
finally:
|
||||
deps_module._cached_repo = saved_repo
|
||||
deps_module._cached_local_provider = saved_provider
|
||||
|
||||
try:
|
||||
asyncio.run(_run())
|
||||
finally:
|
||||
asyncio.run(close_engine())
|
||||
|
||||
|
||||
def test_internal_auth_owner_gate_skipped_in_auth_disabled_mode(monkeypatch):
|
||||
"""Auth-disabled mode sends the synthetic auth-disabled identity as the
|
||||
internal owner; deployments may have no users store at all there, so the
|
||||
suspension gate must not attempt a lookup (it would fail the request)."""
|
||||
import asyncio
|
||||
|
||||
from starlette.requests import Request as StarletteRequest
|
||||
from starlette.responses import PlainTextResponse
|
||||
|
||||
from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID
|
||||
from app.gateway.auth_middleware import AuthMiddleware
|
||||
from app.gateway.internal_auth import create_internal_auth_headers
|
||||
|
||||
monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1")
|
||||
|
||||
async def _run():
|
||||
headers = create_internal_auth_headers(owner_user_id=AUTH_DISABLED_USER_ID)
|
||||
scope = {
|
||||
"type": "http",
|
||||
"method": "POST",
|
||||
"path": "/api/threads/abc/runs/stream",
|
||||
"root_path": "",
|
||||
"query_string": b"",
|
||||
"headers": [(name.lower().encode(), value.encode()) for name, value in headers.items()],
|
||||
}
|
||||
|
||||
async def call_next(_request):
|
||||
return PlainTextResponse("route-reached")
|
||||
|
||||
# No engine / users store is configured on purpose: a lookup here
|
||||
# would raise and 500 the request.
|
||||
response = await AuthMiddleware(None).dispatch(StarletteRequest(scope), call_next)
|
||||
assert response.status_code == 200
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def test_protected_post_with_non_ascii_internal_auth_header_returns_401(client):
|
||||
from app.gateway.internal_auth import INTERNAL_AUTH_HEADER_NAME
|
||||
|
||||
|
||||
@ -24,16 +24,19 @@ def _stub_app_config():
|
||||
reset_app_config()
|
||||
|
||||
|
||||
def _make_start_run_request(run_manager, *, thread_store=None, auth_source=None):
|
||||
def _make_start_run_request(run_manager, *, thread_store=None, auth_source=None, state_user=None, headers=None):
|
||||
from langgraph.checkpoint.memory import InMemorySaver
|
||||
from langgraph.store.memory import InMemoryStore
|
||||
|
||||
from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore
|
||||
|
||||
store = InMemoryStore()
|
||||
state = SimpleNamespace(auth_source=auth_source)
|
||||
if state_user is not None:
|
||||
state.user = state_user
|
||||
return SimpleNamespace(
|
||||
headers={},
|
||||
state=SimpleNamespace(auth_source=auth_source),
|
||||
headers=dict(headers or {}),
|
||||
state=state,
|
||||
app=SimpleNamespace(
|
||||
state=SimpleNamespace(
|
||||
stream_bridge=SimpleNamespace(),
|
||||
@ -912,6 +915,68 @@ async def test_system_role_rejected_before_run_admission(_stub_app_config, auth_
|
||||
assert await request.app.state.checkpointer.aget_tuple({"configurable": {"thread_id": "synthetic-rejected-thread"}}) is None
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_run_rejects_suspended_internal_owner(_stub_app_config, tmp_path):
|
||||
"""(#3462 gap 3) Scheduler launches fabricate internal requests that skip
|
||||
AuthMiddleware (where the internal-auth suspension gate lives), so run
|
||||
admission re-asserts it: a suspended owner's occurrence fails before any
|
||||
run record exists, and re-enabling the owner admits the run again."""
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
from fastapi import HTTPException
|
||||
|
||||
import app.gateway.deps as deps_module
|
||||
from app.gateway.auth.models import User
|
||||
from app.gateway.auth.repositories.sqlite import SQLiteUserRepository
|
||||
from app.gateway.internal_auth import INTERNAL_OWNER_USER_ID_HEADER_NAME, get_internal_user
|
||||
from app.gateway.run_models import RunCreateRequest
|
||||
from app.gateway.services import start_run
|
||||
from deerflow.persistence.engine import close_engine, get_session_factory, init_engine
|
||||
from deerflow.runtime import RunManager
|
||||
from deerflow.runtime.runs.store.memory import MemoryRunStore
|
||||
|
||||
await init_engine("sqlite", url=f"sqlite+aiosqlite:///{tmp_path}/users.db", sqlite_dir=str(tmp_path))
|
||||
repo = SQLiteUserRepository(get_session_factory())
|
||||
owner = await repo.create_user(User(email="sched-owner@example.com", system_role="user"))
|
||||
await repo.set_disabled(str(owner.id), True)
|
||||
|
||||
saved_repo, saved_provider = deps_module._cached_repo, deps_module._cached_local_provider
|
||||
deps_module._cached_repo = repo
|
||||
deps_module._cached_local_provider = None
|
||||
|
||||
manager = RunManager(store=MemoryRunStore())
|
||||
|
||||
def _request():
|
||||
return _make_start_run_request(
|
||||
manager,
|
||||
auth_source=AUTH_SOURCE_INTERNAL,
|
||||
state_user=get_internal_user(),
|
||||
headers={INTERNAL_OWNER_USER_ID_HEADER_NAME: str(owner.id)},
|
||||
)
|
||||
|
||||
try:
|
||||
with patch("app.gateway.services.resolve_agent_factory", return_value=object()), patch("app.gateway.services.run_agent", new_callable=AsyncMock) as worker:
|
||||
with pytest.raises(HTTPException) as error:
|
||||
await start_run(RunCreateRequest(input={"messages": [{"role": "user", "content": "scheduled tick"}]}), "suspended-owner-thread", _request())
|
||||
worker.assert_not_awaited()
|
||||
assert error.value.status_code == 401
|
||||
assert error.value.detail["code"] == "account_disabled"
|
||||
assert await manager.list_by_thread("suspended-owner-thread", user_id=None) == []
|
||||
|
||||
# Re-enabling the owner re-admits internal launches (exact-restore
|
||||
# semantics of the lifecycle: suspension is not destructive). The
|
||||
# agent worker is scheduled as a task, so admission is proven by the
|
||||
# created run record rather than a synchronous await.
|
||||
await repo.set_disabled(str(owner.id), False)
|
||||
with patch("app.gateway.services.resolve_agent_factory", return_value=object()), patch("app.gateway.services.run_agent", new_callable=AsyncMock):
|
||||
record = await start_run(RunCreateRequest(input={"messages": [{"role": "user", "content": "scheduled tick"}]}), "suspended-owner-thread", _request())
|
||||
assert record.run_id
|
||||
finally:
|
||||
await close_engine()
|
||||
deps_module._cached_repo = saved_repo
|
||||
deps_module._cached_local_provider = saved_provider
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_start_run_preserves_system_messages_from_internal_auth(_stub_app_config):
|
||||
from langchain_core.messages import SystemMessage
|
||||
|
||||
@ -23,7 +23,7 @@ OUTBOX = "0027_notification_deliveries"
|
||||
PARKED = "0028_parked_attempts"
|
||||
SCHEDULER = "0029_scheduler_agent_tasks"
|
||||
CLAIM_TOKENS = "0030_notification_claim_tokens"
|
||||
HEAD = "0038_channel_thread_bindings"
|
||||
HEAD = "0039_user_disabled"
|
||||
PREVIOUS = "0026_mcp_task_lease_tokens"
|
||||
TABLE = "notification_deliveries"
|
||||
|
||||
|
||||
@ -23,7 +23,7 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
|
||||
from deerflow.persistence.scheduled_tasks import ScheduledTaskRepository
|
||||
|
||||
REVISION = "0029_scheduler_agent_tasks"
|
||||
CURRENT_HEAD = "0038_channel_thread_bindings"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
NEXT = "0030_notification_claim_tokens"
|
||||
PREVIOUS = "0028_parked_attempts"
|
||||
TASK_FIELDS = {"origin_thread_id", "goal_objective", "max_runs", "end_at", "standing_notes"}
|
||||
|
||||
@ -22,7 +22,9 @@ AFTER_NEXT = "0034_run_event_seq_watermark"
|
||||
LOGIN_THROTTLE = "0035_login_throttle"
|
||||
IDEMPOTENCY = "0036_run_idempotency_request"
|
||||
DOCUMENT_SUMMARIES = "0037_project_document_summaries"
|
||||
CURRENT_HEAD = "0038_channel_thread_bindings"
|
||||
CHANNEL_BINDINGS = "0038_channel_thread_bindings"
|
||||
USER_DISABLED = "0039_user_disabled"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
PREVIOUS = "0031_scheduled_streak_boundary"
|
||||
TABLES = {"thread_read_markers", "thread_read_versions", "scheduled_task_events"}
|
||||
INDEXES = {
|
||||
@ -42,7 +44,8 @@ async def test_0032_remains_in_the_single_migration_chain():
|
||||
assert script.get_revision(LOGIN_THROTTLE).down_revision == AFTER_NEXT
|
||||
assert script.get_revision(IDEMPOTENCY).down_revision == LOGIN_THROTTLE
|
||||
assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
|
||||
assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
|
||||
# alembic_version.version_num is VARCHAR(32).
|
||||
assert len(REVISION) <= 32
|
||||
|
||||
|
||||
@ -22,7 +22,9 @@ PREVIOUS = "0033_batch_result_artifact"
|
||||
LOGIN_THROTTLE = "0035_login_throttle"
|
||||
IDEMPOTENCY = "0036_run_idempotency_request"
|
||||
DOCUMENT_SUMMARIES = "0037_project_document_summaries"
|
||||
CURRENT_HEAD = "0038_channel_thread_bindings"
|
||||
CHANNEL_BINDINGS = "0038_channel_thread_bindings"
|
||||
USER_DISABLED = "0039_user_disabled"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
TABLE = "run_event_thread_seq"
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
@ -34,7 +36,8 @@ async def test_0034_remains_in_the_single_migration_chain():
|
||||
assert script.get_revision(LOGIN_THROTTLE).down_revision == REVISION
|
||||
assert script.get_revision(IDEMPOTENCY).down_revision == LOGIN_THROTTLE
|
||||
assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
|
||||
assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
|
||||
assert len(REVISION) <= 32
|
||||
|
||||
|
||||
|
||||
@ -23,7 +23,9 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
|
||||
REVISION = "0035_login_throttle"
|
||||
IDEMPOTENCY = "0036_run_idempotency_request"
|
||||
DOCUMENT_SUMMARIES = "0037_project_document_summaries"
|
||||
CURRENT_HEAD = "0038_channel_thread_bindings"
|
||||
CHANNEL_BINDINGS = "0038_channel_thread_bindings"
|
||||
USER_DISABLED = "0039_user_disabled"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
PREVIOUS = "0034_run_event_seq_watermark"
|
||||
TABLE = "login_throttle"
|
||||
COLUMNS = {"ip", "fail_count", "locked_at", "lock_duration_seconds", "updated_at"}
|
||||
@ -37,7 +39,8 @@ async def test_0035_remains_in_the_single_head_chain_after_0034():
|
||||
assert script.get_revision(REVISION).down_revision == PREVIOUS
|
||||
assert script.get_revision(IDEMPOTENCY).down_revision == REVISION
|
||||
assert script.get_revision(DOCUMENT_SUMMARIES).down_revision == IDEMPOTENCY
|
||||
assert script.get_revision(CURRENT_HEAD).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(CHANNEL_BINDINGS).down_revision == DOCUMENT_SUMMARIES
|
||||
assert script.get_revision(USER_DISABLED).down_revision == CHANNEL_BINDINGS
|
||||
# alembic_version.version_num is VARCHAR(32).
|
||||
assert len(REVISION) <= 32
|
||||
|
||||
|
||||
@ -18,14 +18,16 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
|
||||
|
||||
REVISION = "0037_project_document_summaries"
|
||||
PREVIOUS = "0036_run_idempotency_request"
|
||||
CURRENT_HEAD = "0038_channel_thread_bindings"
|
||||
CHANNEL_BINDINGS = "0038_channel_thread_bindings"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def test_0037_chains_after_0036_and_is_single_head():
|
||||
script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
|
||||
assert script.get_heads() == [CURRENT_HEAD]
|
||||
assert script.get_revision(CURRENT_HEAD).down_revision == REVISION
|
||||
assert script.get_revision(CHANNEL_BINDINGS).down_revision == REVISION
|
||||
assert script.get_revision(CURRENT_HEAD).down_revision == CHANNEL_BINDINGS
|
||||
assert script.get_revision(REVISION).down_revision == PREVIOUS
|
||||
# alembic_version.version_num is VARCHAR(32).
|
||||
assert len(REVISION) <= 32
|
||||
|
||||
@ -22,15 +22,16 @@ from deerflow.persistence.postgres_schema import build_asyncpg_connect_args
|
||||
|
||||
REVISION = "0038_channel_thread_bindings"
|
||||
PREVIOUS = "0037_project_document_summaries"
|
||||
CURRENT_HEAD = "0039_user_disabled"
|
||||
TABLE = "channel_thread_bindings"
|
||||
COLUMNS = {"key", "channel_name", "chat_id", "topic_id", "thread_id", "user_id", "created_at", "updated_at"}
|
||||
INDEX = "ix_channel_thread_bindings_channel_chat"
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
|
||||
async def test_0038_is_the_single_head_and_chains_after_0037():
|
||||
async def test_0038_remains_in_the_single_migration_chain():
|
||||
script = ScriptDirectory(str(bootstrap._MIGRATIONS_DIR))
|
||||
assert script.get_heads() == [REVISION]
|
||||
assert script.get_heads() == [CURRENT_HEAD]
|
||||
assert script.get_revision(REVISION).down_revision == PREVIOUS
|
||||
# alembic_version.version_num is VARCHAR(32).
|
||||
assert len(REVISION) <= 32
|
||||
@ -182,6 +183,8 @@ async def test_bootstrap_provisions_the_table_on_an_empty_database_and_is_idempo
|
||||
assert set(shape["columns"]) == COLUMNS
|
||||
assert INDEX in shape["indexes"]
|
||||
async with engine.connect() as conn:
|
||||
assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == REVISION
|
||||
# Bootstrap runs the chain to its head, which this test no longer
|
||||
# owns once later revisions chain after 0038.
|
||||
assert (await conn.execute(sa.text("SELECT version_num FROM alembic_version"))).scalar_one() == CURRENT_HEAD
|
||||
finally:
|
||||
await engine.dispose()
|
||||
|
||||
@ -5,6 +5,7 @@ export const AUTH_DISABLED_USER: User = {
|
||||
email: "default@test.local",
|
||||
system_role: "admin",
|
||||
needs_setup: false,
|
||||
disabled: false,
|
||||
oauth_provider: null,
|
||||
};
|
||||
|
||||
|
||||
@ -5,5 +5,6 @@ export const STATIC_WEBSITE_USER: User = {
|
||||
email: "static@example.local",
|
||||
system_role: "admin",
|
||||
needs_setup: false,
|
||||
disabled: false,
|
||||
oauth_provider: null,
|
||||
};
|
||||
|
||||
@ -12,6 +12,10 @@ export const userSchema = z.object({
|
||||
// other value is simply non-admin.
|
||||
system_role: z.string().min(1),
|
||||
needs_setup: z.boolean().optional().default(false),
|
||||
// Account lifecycle (RFC #4063 gap 3): true when the account is
|
||||
// operator-suspended. Optional so pre-gap-3 backends (and test literals)
|
||||
// parse unchanged — absent means not disabled.
|
||||
disabled: z.boolean().optional(),
|
||||
oauth_provider: z.string().nullable().optional().default(null),
|
||||
// Effective route permissions (RFC #4063 Phase 4). Optional + nullable:
|
||||
// absent = pre-Phase-4 backend, null = credential-creation response that
|
||||
|
||||
@ -200,6 +200,7 @@ describe("userSchema system_role", () => {
|
||||
email: "g@example.com",
|
||||
system_role: "",
|
||||
needs_setup: false,
|
||||
disabled: false,
|
||||
oauth_provider: null,
|
||||
permissions: null,
|
||||
}),
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user