curo/kite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(08-04): delete admin.py monolith after URL regression — CODE-01 CODE-08

Browse Source 
Task 3: 54 admin+cloud+ai-config tests pass with package; admin.py and
admin_OLD_REMOVE_IN_TASK_3.py removed. admin/ package is the sole definition.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
curo1305
2026-06-12 10:16:38 +02:00
co-authored by Claude Sonnet 4.6
parent 7e99b6ecc1
commit f01fb0e6d5
4 changed files with 0 additions and 3518 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/api/admin.py
-934
View File
@@ -1,934 +0,0 @@
"""
Admin API endpoints for DocuVault.
All handlers require get_current_admin (SEC-07, D-08) — no handler uses
get_current_user alone.
Implements:
GET /api/admin/users — list all users (ADMIN-01)
POST /api/admin/users — create user (ADMIN-01)
PATCH /api/admin/users/{id}/status — deactivate/reactivate (ADMIN-02)
POST /api/admin/users/{id}/password-reset — initiate reset email (ADMIN-03)
GET /api/admin/users/{id}/quota — view quota (ADMIN-04)
PATCH /api/admin/users/{id}/quota — adjust quota (ADMIN-04)
PATCH /api/admin/users/{id}/ai-config — assign AI provider/model (ADMIN-05)
Security invariants:
- Every handler injects Depends(get_current_admin) — verified by grep count
- _user_to_dict() whitelist helper prevents accidental field leakage (T-02-27)
- No impersonation endpoint — ADMIN-07 enforced by omission (T-02-28)
- Admin-created users: password_must_change=True (ADMIN-01, T-02-32)
- Deactivation of sole admin prevented (T-02-29)
- Password reset sends email via Celery; does not return token (T-02-30)
"""
from __future__ import annotations
import time
import uuid
from datetime import datetime
from typing import Optional
from config import settings
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from ai import get_provider
from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
from db.models import CloudConnection, Document, Quota, RefreshToken, SystemSettings, Topic, User
from deps.auth import get_current_admin
from deps.db import get_db
from deps.utils import get_client_ip
from services.ai_config import encrypt_api_key, load_provider_config_by_id
from services.audit import write_audit_log
from services.auth import hash_password, revoke_all_refresh_tokens, validate_password_strength, verify_password
from storage import get_storage_backend, get_storage_backend_for_document
router = APIRouter(prefix="/api/admin", tags=["admin"])
# ── Constants ─────────────────────────────────────────────────────────────────
_DEFAULT_QUOTA_BYTES = 104857600 # 100 MB free-tier default (D-06)
# ── Safe response helpers ─────────────────────────────────────────────────────
def _ai_config_to_dict(row: SystemSettings) -> dict:
"""Return a safe subset of SystemSettings fields — explicitly excludes api_key_enc.
has_api_key is the ONLY indicator that a key is stored (T-07-01 mitigated).
The raw encrypted value and any decrypted plaintext are NEVER returned.
"""
return {
"provider_id": row.provider_id,
"base_url": row.base_url,
"model_name": row.model_name,
"context_chars": row.context_chars,
"is_active": row.is_active,
"has_api_key": row.api_key_enc is not None,
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
}
def _user_to_dict(user: User) -> dict:
"""Return a safe subset of User fields — never includes password_hash,
credentials_enc, totp_secret, or any document content (T-02-27, SEC-07).
"""
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"is_active": user.is_active,
"totp_enabled": user.totp_enabled,
"ai_provider": user.ai_provider,
"ai_model": user.ai_model,
"password_must_change": user.password_must_change,
"created_at": user.created_at.isoformat() if user.created_at else None,
}
# ── Request models ────────────────────────────────────────────────────────────
class UserCreate(BaseModel):
handle: str
email: EmailStr
password: str
role: str = "user"
@field_validator("password")
@classmethod
def password_strength(cls, v: str) -> str:
validate_password_strength(v)
return v
class UserStatusUpdate(BaseModel):
is_active: bool
class QuotaUpdate(BaseModel):
limit_bytes: int
@field_validator("limit_bytes")
@classmethod
def must_be_positive(cls, v: int) -> int:
if v <= 0:
raise ValueError("limit_bytes must be greater than 0")
return v
class UserAiConfigUpdate(BaseModel):
ai_provider: Optional[str] = None
ai_model: Optional[str] = None
class SystemAiConfigUpdate(BaseModel):
"""Request model for PUT /api/admin/ai-config (system-level provider configuration).
Security: extra="forbid" prevents mass-assignment of unexpected fields (T-07-13).
provider_id is validated against PROVIDER_DEFAULTS keys (T-07-13).
api_key is write-only: when None the existing api_key_enc is left untouched,
when "" the api_key_enc is cleared, when a non-empty string it is encrypted.
"""
model_config = ConfigDict(extra="forbid")
provider_id: str
api_key: Optional[str] = None
base_url: Optional[str] = None
model_name: Optional[str] = None
context_chars: Optional[int] = None
is_active: Optional[bool] = None
@field_validator("provider_id")
@classmethod
def provider_must_be_known(cls, v: str) -> str:
if v not in PROVIDER_DEFAULTS:
raise ValueError(
f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
)
return v
class TestConnectionRequest(BaseModel):
"""Request body for POST /api/admin/ai-config/test-connection.
Unsaved form values (api_key, base_url, model_name) override the DB row so
admins can verify credentials before saving. All override fields are optional;
omitting them falls back to whatever is stored in system_settings.
"""
model_config = ConfigDict(extra="forbid")
provider_id: str
api_key: Optional[str] = None # If non-empty, used instead of stored api_key_enc
base_url: Optional[str] = None # If non-None, overrides DB base_url
model_name: Optional[str] = None # If non-empty, overrides DB model_name
@field_validator("provider_id")
@classmethod
def provider_must_be_known(cls, v: str) -> str:
if v not in PROVIDER_DEFAULTS:
raise ValueError(
f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
)
return v
class SystemTopicCreate(BaseModel):
"""Request model for admin system topic creation (D-09)."""
name: str
description: str = ""
color: str = "#6366f1"
class UserDeleteConfirm(BaseModel):
"""Admin password confirmation required before hard-deleting a user (ADMIN-02, T-05-11-01)."""
admin_password: str = Field(..., min_length=1)
# ── SEC-08: Safe CloudConnection response model ───────────────────────────────
class CloudConnectionOut(BaseModel):
"""SEC-08: credentials_enc deliberately excluded from this response model.
Any admin or user endpoint returning CloudConnection ORM objects MUST use
this model to prevent accidental exposure of encrypted credentials.
Safe-by-default: whitelist of allowed fields (not blacklist).
Note: id is declared as str and coerced via validator so UUID ORM values
serialize correctly without json_encoders (Rule 1 fix — T-05-06 test suite).
"""
id: str
provider: str
display_name: str
status: str
connected_at: datetime
server_url: Optional[str] = None
connection_username: Optional[str] = None
model_config = {"from_attributes": True}
@field_validator("id", mode="before")
@classmethod
def coerce_id_to_str(cls, v) -> str:
"""Coerce UUID objects to str so the model validates from ORM instances."""
return str(v)
# ── Endpoints ─────────────────────────────────────────────────────────────────
@router.get("/users")
async def list_users(
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""List all users, ordered by created_at DESC.
Response shape: { items: [...safe user fields...] }
Never includes password_hash, credentials_enc, or document content (T-02-27).
"""
result = await session.execute(
select(User).order_by(User.created_at.desc())
)
users = result.scalars().all()
return {"items": [_user_to_dict(u) for u in users]}
@router.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(
request: Request,
body: UserCreate,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Admin creates a new user account (ADMIN-01).
- password_must_change=True forces the user to change their password on
first login (T-02-32, D-06).
- Quota row initialized at 100 MB (D-06).
- Returns 409 if email or handle is already taken.
"""
# Check uniqueness
existing_email = await session.execute(
select(User).where(User.email == str(body.email))
)
if existing_email.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already registered",
)
existing_handle = await session.execute(
select(User).where(User.handle == body.handle)
)
if existing_handle.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Handle already taken",
)
new_user = User(
id=uuid.uuid4(),
handle=body.handle,
email=str(body.email),
password_hash=hash_password(body.password),
role=body.role,
is_active=True,
totp_enabled=False,
password_must_change=True, # ADMIN-01: force password change on first login
)
session.add(new_user)
quota = Quota(
user_id=new_user.id,
limit_bytes=_DEFAULT_QUOTA_BYTES,
used_bytes=0,
)
session.add(quota)
await session.flush() # persist User + Quota before audit_log FK references them
# D-13: admin user created event
_ip_addr = get_client_ip(request)
await write_audit_log(
session,
event_type="admin.user_created",
user_id=new_user.id,
actor_id=_admin.id,
resource_id=new_user.id,
ip_address=_ip_addr,
)
await session.commit()
return {
"id": str(new_user.id),
"handle": new_user.handle,
"email": new_user.email,
"role": new_user.role,
"created_at": new_user.created_at.isoformat() if new_user.created_at else None,
}
@router.patch("/users/{user_id}/status")
async def update_user_status(
user_id: uuid.UUID,
body: UserStatusUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Deactivate or reactivate a user account (ADMIN-02).
- Prevents deactivating the last active admin (T-02-29).
- On deactivation: all refresh tokens are revoked (family revocation).
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# Guard: cannot deactivate the only remaining active admin (T-02-29)
if not body.is_active and user.role == "admin":
count_result = await session.execute(
select(func.count(User.id)).where(
User.role == "admin",
User.is_active.is_(True),
)
)
active_admin_count = count_result.scalar_one()
if active_admin_count <= 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot deactivate the only admin",
)
_ip_addr = get_client_ip(request)
user.is_active = body.is_active
if not body.is_active:
# Revoke all refresh tokens on deactivation
await revoke_all_refresh_tokens(session, user.id)
# Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
session.add(user)
# D-13: user deactivated/activated event
_event = "admin.user_deactivated" if not body.is_active else "admin.user_activated"
await write_audit_log(
session,
event_type=_event,
user_id=user.id,
actor_id=_admin.id,
resource_id=user.id,
ip_address=_ip_addr,
)
await session.commit()
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"is_active": user.is_active,
}
@router.post("/users/{user_id}/password-reset", status_code=status.HTTP_202_ACCEPTED)
async def initiate_password_reset(
user_id: uuid.UUID,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Admin initiates a password reset for a user (ADMIN-03).
Sends the reset email via Celery. Does NOT:
- return a reset token (T-02-30)
- grant admin access to the account
- log in as the target user (ADMIN-07 — no impersonation)
Returns 202 immediately regardless of email delivery status.
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
from services.auth import create_password_reset_token # noqa: PLC0415
from config import settings as _settings # noqa: PLC0415
reset_token = create_password_reset_token(str(user.id))
reset_link = f"{_settings.frontend_url}/password-reset/confirm?token={reset_token}"
# Deferred import to avoid circular imports (same pattern as document_tasks)
from tasks.email_tasks import send_reset_email # noqa: PLC0415
send_reset_email.delay(user.email, reset_link)
return {"message": "Password reset email sent"}
@router.get("/users/{user_id}/quota")
async def get_user_quota(
user_id: uuid.UUID,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return quota details for a user (ADMIN-04).
Quota info is admin-visible operational data — no PII, no document content
(T-02-31 disposition: accept).
"""
quota = await session.get(Quota, user_id)
if quota is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
return {
"user_id": str(quota.user_id),
"limit_bytes": quota.limit_bytes,
"used_bytes": quota.used_bytes,
"limit_mb": quota.limit_bytes // 1048576,
"used_mb": quota.used_bytes // 1048576,
}
@router.patch("/users/{user_id}/quota")
async def update_user_quota(
user_id: uuid.UUID,
body: QuotaUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Adjust a user's storage quota (ADMIN-04).
If the new limit is below current usage, still applies the change but
returns warning=True with an explanatory message. Uploads will be blocked
but existing documents are preserved.
"""
quota = await session.get(Quota, user_id)
if quota is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
warning = body.limit_bytes < quota.used_bytes
warning_message = (
"New limit is below current usage. Uploads will be blocked but existing documents are preserved."
if warning
else None
)
_ip_addr = get_client_ip(request)
old_limit = quota.limit_bytes
quota.limit_bytes = body.limit_bytes
session.add(quota)
# D-13: quota changed event
await write_audit_log(
session,
event_type="admin.quota_changed",
user_id=user_id,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"old_bytes": old_limit, "new_bytes": body.limit_bytes},
)
await session.commit()
response: dict = {
"user_id": str(quota.user_id),
"limit_bytes": quota.limit_bytes,
"used_bytes": quota.used_bytes,
"warning": warning,
}
if warning_message:
response["message"] = warning_message
return response
@router.patch("/users/{user_id}/ai-config")
async def update_ai_config(
user_id: uuid.UUID,
body: UserAiConfigUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Assign AI provider and model for a user (ADMIN-05).
Users cannot change their own AI provider or model (PROJECT.md Key Decision).
Only admins have this capability.
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
_ip_addr = get_client_ip(request)
user.ai_provider = body.ai_provider
user.ai_model = body.ai_model
session.add(user)
# D-13: AI provider assigned event
await write_audit_log(
session,
event_type="admin.ai_provider_assigned",
user_id=user_id,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"provider": body.ai_provider, "model": body.ai_model},
)
await session.commit()
return {
"id": str(user.id),
"email": user.email,
"ai_provider": user.ai_provider,
"ai_model": user.ai_model,
}
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
user_id: uuid.UUID,
body: UserDeleteConfirm,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> None:
"""Delete a user account and clean up all their MinIO objects (SEC-09, D-19).
Security invariants:
- Admin password verified via Argon2 before any deletion (T-05-11-01)
- Cannot delete admin accounts (T-04-07-04)
- MinIO objects are deleted BEFORE DB records are removed (SEC-09)
- MinIO deletion is best-effort (try/except) — DB row is deleted regardless
- Audit log written with event_type="admin.user_deleted"
"""
# T-05-11-01: Verify admin password before performing any destructive action.
# Fail fast — no DB reads for the target user until the admin is confirmed.
if not verify_password(body.admin_password, _admin.password_hash):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid admin password",
)
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# T-04-07-04: Cannot delete admin accounts
if user.role == "admin":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot delete admin accounts",
)
_ip_addr = get_client_ip(request)
# SEC-09 (cloud): purge cloud-stored documents and credentials BEFORE DB delete.
# Must run before MinIO cleanup so that credentials are still available to build
# the cloud backend instances for delete_object calls.
cloud_conns_result = await session.execute(
select(CloudConnection).where(CloudConnection.user_id == user_id)
)
cloud_conns = cloud_conns_result.scalars().all()
for conn in cloud_conns:
# Delete cloud objects stored in this provider for this user
cloud_docs_result = await session.execute(
select(Document).where(
Document.user_id == user_id,
Document.storage_backend == conn.provider,
)
)
for doc in cloud_docs_result.scalars().all():
try:
backend = await get_storage_backend_for_document(doc, user, session)
await backend.delete_object(doc.object_key)
except Exception:
pass # Best-effort cloud object cleanup; deletion proceeds regardless
# Purge the credentials row (FK cascade would also remove it, but explicit
# deletion here guarantees credentials_enc is gone before commit — SEC-09)
await session.delete(conn)
if cloud_conns:
await session.flush() # Flush connection deletes before user delete
await write_audit_log(
session,
event_type="cloud.credentials_purged",
user_id=user_id,
actor_id=_admin.id,
resource_id=user_id,
ip_address=_ip_addr,
metadata_={"providers": [c.provider for c in cloud_conns]},
)
# SEC-09 (minio): collect all user documents and delete MinIO objects BEFORE DB delete
docs_result = await session.execute(
select(Document).where(Document.user_id == user_id)
)
user_docs = docs_result.scalars().all()
storage = get_storage_backend()
for doc in user_docs:
try:
await storage.delete_object(doc.object_key)
except Exception:
pass # Best-effort MinIO cleanup; DB deletion proceeds regardless
# D-13: audit log BEFORE deleting the user row (user FK still valid at flush time)
await write_audit_log(
session,
event_type="admin.user_deleted",
user_id=user_id,
actor_id=_admin.id,
resource_id=user_id,
ip_address=_ip_addr,
)
await session.flush()
# Delete user record (CASCADE removes quota, documents, refresh_tokens, etc.)
await session.delete(user)
await session.commit()
@router.post("/topics", status_code=status.HTTP_201_CREATED)
async def create_system_topic(
body: SystemTopicCreate,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Create a system topic visible to all users (D-09, DOC-04).
System topics have user_id = NULL, making them visible to every user as
defaults in their topic namespace. Only admins can create system topics.
Regular users create per-user topics via POST /api/topics.
Deduplication: case-insensitive match within the system namespace (user_id IS NULL).
Returns the existing system topic if one with the same name already exists.
"""
from services import storage # noqa: PLC0415
topic = await storage.create_topic(
session, body.name, body.description, body.color, user_id=None
)
return topic
# ── System AI Provider Configuration (D-08, D-15) ────────────────────────────
@router.get("/ai-config/models")
async def get_ai_config_models(
provider_id: str,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return the list of model IDs available from a provider's API (D-08).
Calls the provider's standard GET /models endpoint using the stored
config (base_url + api_key from system_settings). Always returns 200
with {"models": [...]} — never 5xx on provider failure (returns empty list).
Security: requires get_current_admin; provider_id from query param only;
decrypted api_key never appears in the response.
"""
import httpx # noqa: PLC0415 — local import keeps admin.py startup fast
config = await load_provider_config_by_id(session, provider_id)
# Resolve base_url: prefer DB row, fall back to PROVIDER_DEFAULTS
if config and config.base_url:
base_url = config.base_url.rstrip("/")
else:
base_url = (PROVIDER_DEFAULTS.get(provider_id, {}).get("base_url") or "").rstrip("/")
if not base_url:
return {"models": [], "provider_id": provider_id}
api_key = config.api_key if config else ""
# Build request headers — Anthropic uses x-api-key; all others use Bearer
if provider_id == "anthropic":
headers = {
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
}
models_url = "https://api.anthropic.com/v1/models"
else:
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
models_url = f"{base_url}/models"
try:
async with httpx.AsyncClient(timeout=8.0) as client:
resp = await client.get(models_url, headers=headers)
resp.raise_for_status()
data = resp.json()
# Standard OpenAI-compat shape: {"data": [{"id": "...", ...}, ...]}
# Anthropic shape: {"data": [{"id": "...", ...}, ...]}
# Ollama OpenAI-compat: same shape
raw_list = data.get("data") or data.get("models") or []
model_ids: list[str] = sorted(
{
item["id"] if isinstance(item, dict) else str(item)
for item in raw_list
if item
}
)
return {"models": model_ids, "provider_id": provider_id}
except Exception as exc:
return {"models": [], "provider_id": provider_id, "error": str(exc)[:120]}
@router.post("/ai-config/test-connection")
async def test_ai_connection(
body: TestConnectionRequest,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Test connectivity for an AI provider, optionally with unsaved form values (D-08).
Loads the stored system_settings row for body.provider_id, then overlays any
non-empty values from the request body so admins can verify credentials before
saving them to the database.
Override priority (highest → lowest):
1. body.api_key / base_url / model_name (unsaved form values)
2. system_settings DB row (previously saved config)
3. PROVIDER_DEFAULTS (built-in fallback)
Returns {"ok": true/false, "provider_id": str} — never raises 5xx for
provider-side failures; surfaces as ok=False so the UI shows a clear status.
Security: requires get_current_admin; api_key from body is used only for the
in-flight health_check() call and is never stored or logged.
"""
provider_id = body.provider_id
stored = await load_provider_config_by_id(session, provider_id)
defaults = PROVIDER_DEFAULTS.get(provider_id, {})
# Resolve effective values: body overrides DB, DB overrides PROVIDER_DEFAULTS
effective_api_key = (
body.api_key
if body.api_key
else (stored.api_key if stored else "")
)
effective_base_url = (
body.base_url
if body.base_url is not None
else (stored.base_url if stored else defaults.get("base_url"))
)
effective_model = (
body.model_name
if body.model_name
else (stored.model if stored else defaults.get("model", ""))
)
effective_config = ProviderConfig(
provider_id=provider_id,
api_key=effective_api_key,
base_url=effective_base_url,
model=effective_model,
)
try:
provider = get_provider(effective_config)
ok = await provider.health_check()
return {"ok": ok, "provider_id": provider_id}
except Exception:
return {"ok": False, "provider_id": provider_id, "reason": "health_check_failed"}
@router.get("/ai-config")
async def get_ai_config(
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return all AI provider configurations for the admin panel (D-08).
Includes DB rows for providers that have been saved AND synthesised stubs
for providers that only exist in PROVIDER_DEFAULTS (so the admin UI always
shows all 10 providers even before any have been configured).
Security invariant: api_key_enc is NEVER returned (T-07-01).
Use has_api_key (bool) as the only indicator that a key is stored.
"""
result = await session.execute(select(SystemSettings))
db_rows = result.scalars().all()
# Build a lookup for DB rows
db_by_provider: dict[str, SystemSettings] = {r.provider_id: r for r in db_rows}
providers_out = []
for pid in PROVIDER_DEFAULTS:
if pid in db_by_provider:
providers_out.append(_ai_config_to_dict(db_by_provider[pid]))
else:
# Synthesise a stub entry for providers with no DB row yet
defaults = PROVIDER_DEFAULTS[pid]
providers_out.append({
"provider_id": pid,
"base_url": defaults.get("base_url"),
"model_name": defaults.get("model", ""),
"context_chars": defaults.get("context_chars", 8000),
"is_active": False,
"has_api_key": False,
"updated_at": None,
})
return {"providers": providers_out}
@router.put("/ai-config")
async def update_system_ai_config(
body: SystemAiConfigUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Create or update a system-level AI provider configuration (D-08, D-15).
Upsert semantics: if no row exists for body.provider_id, one is created using
PROVIDER_DEFAULTS for any omitted fields.
API key handling (T-07-01 mitigated):
- body.api_key is None → leave existing api_key_enc untouched
- body.api_key == "" → clear api_key_enc (set to NULL)
- body.api_key is a non-empty string → HKDF-encrypt and store
is_active=True handling (T-07-03 mitigated):
When body.is_active is True, a single atomic UPDATE flips all rows:
SET is_active = (provider_id = :target_id)
This guarantees COUNT(WHERE is_active) == 1 with no read-then-write race.
Audit log (T-07-14 mitigated):
metadata_ contains only provider_id + fields_changed list — never the
api_key value itself.
"""
from config import settings as _settings # noqa: PLC0415
# Load existing row or create a new one from PROVIDER_DEFAULTS
stmt = select(SystemSettings).where(SystemSettings.provider_id == body.provider_id)
result = await session.execute(stmt)
row = result.scalar_one_or_none()
is_new = row is None
if is_new:
defaults = PROVIDER_DEFAULTS[body.provider_id]
row = SystemSettings(
provider_id=body.provider_id,
model_name=defaults.get("model", ""),
context_chars=defaults.get("context_chars", 8000),
base_url=defaults.get("base_url"),
is_active=False,
api_key_enc=None,
)
# Track which fields the caller explicitly set (for audit log — never api_key value)
fields_changed: list[str] = []
# Apply provided fields
if body.api_key is not None:
fields_changed.append("api_key")
if body.api_key == "":
row.api_key_enc = None
else:
master_key_str = _settings.cloud_creds_key
master_key_bytes = (
master_key_str.encode("utf-8")
if isinstance(master_key_str, str)
else master_key_str
)
row.api_key_enc = encrypt_api_key(master_key_bytes, body.provider_id, body.api_key)
if body.base_url is not None:
row.base_url = body.base_url
fields_changed.append("base_url")
if body.model_name is not None:
row.model_name = body.model_name
fields_changed.append("model_name")
if body.context_chars is not None:
row.context_chars = body.context_chars
fields_changed.append("context_chars")
if body.is_active is not None:
fields_changed.append("is_active")
if is_new:
session.add(row)
await session.flush() # ensure row has an id before UPDATE
# Atomic is_active flip: SET is_active = (provider_id = :target) on ALL rows.
# Single UPDATE statement prevents dual-active race condition (T-07-03).
if body.is_active is True:
await session.execute(
update(SystemSettings).values(
is_active=(SystemSettings.provider_id == body.provider_id)
)
)
# Reflect the flip on the in-memory row
row.is_active = True
_ip_addr = get_client_ip(request)
await write_audit_log(
session,
event_type="admin.ai_config_changed",
user_id=None,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"provider_id": body.provider_id, "fields_changed": fields_changed},
)
await session.commit()
# Reload to pick up DB-generated updated_at after commit
await session.refresh(row)
return _ai_config_to_dict(row)
backend/api/admin_OLD_REMOVE_IN_TASK_3.py
-934
View File
@@ -1,934 +0,0 @@
"""
Admin API endpoints for DocuVault.
All handlers require get_current_admin (SEC-07, D-08) — no handler uses
get_current_user alone.
Implements:
GET /api/admin/users — list all users (ADMIN-01)
POST /api/admin/users — create user (ADMIN-01)
PATCH /api/admin/users/{id}/status — deactivate/reactivate (ADMIN-02)
POST /api/admin/users/{id}/password-reset — initiate reset email (ADMIN-03)
GET /api/admin/users/{id}/quota — view quota (ADMIN-04)
PATCH /api/admin/users/{id}/quota — adjust quota (ADMIN-04)
PATCH /api/admin/users/{id}/ai-config — assign AI provider/model (ADMIN-05)
Security invariants:
- Every handler injects Depends(get_current_admin) — verified by grep count
- _user_to_dict() whitelist helper prevents accidental field leakage (T-02-27)
- No impersonation endpoint — ADMIN-07 enforced by omission (T-02-28)
- Admin-created users: password_must_change=True (ADMIN-01, T-02-32)
- Deactivation of sole admin prevented (T-02-29)
- Password reset sends email via Celery; does not return token (T-02-30)
"""
from __future__ import annotations
import time
import uuid
from datetime import datetime
from typing import Optional
from config import settings
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
from sqlalchemy import func, select, update
from sqlalchemy.ext.asyncio import AsyncSession
from ai import get_provider
from ai.provider_config import ProviderConfig, PROVIDER_DEFAULTS
from db.models import CloudConnection, Document, Quota, RefreshToken, SystemSettings, Topic, User
from deps.auth import get_current_admin
from deps.db import get_db
from deps.utils import get_client_ip
from services.ai_config import encrypt_api_key, load_provider_config_by_id
from services.audit import write_audit_log
from services.auth import hash_password, revoke_all_refresh_tokens, validate_password_strength, verify_password
from storage import get_storage_backend, get_storage_backend_for_document
router = APIRouter(prefix="/api/admin", tags=["admin"])
# ── Constants ─────────────────────────────────────────────────────────────────
_DEFAULT_QUOTA_BYTES = 104857600 # 100 MB free-tier default (D-06)
# ── Safe response helpers ─────────────────────────────────────────────────────
def _ai_config_to_dict(row: SystemSettings) -> dict:
"""Return a safe subset of SystemSettings fields — explicitly excludes api_key_enc.
has_api_key is the ONLY indicator that a key is stored (T-07-01 mitigated).
The raw encrypted value and any decrypted plaintext are NEVER returned.
"""
return {
"provider_id": row.provider_id,
"base_url": row.base_url,
"model_name": row.model_name,
"context_chars": row.context_chars,
"is_active": row.is_active,
"has_api_key": row.api_key_enc is not None,
"updated_at": row.updated_at.isoformat() if row.updated_at else None,
}
def _user_to_dict(user: User) -> dict:
"""Return a safe subset of User fields — never includes password_hash,
credentials_enc, totp_secret, or any document content (T-02-27, SEC-07).
"""
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"is_active": user.is_active,
"totp_enabled": user.totp_enabled,
"ai_provider": user.ai_provider,
"ai_model": user.ai_model,
"password_must_change": user.password_must_change,
"created_at": user.created_at.isoformat() if user.created_at else None,
}
# ── Request models ────────────────────────────────────────────────────────────
class UserCreate(BaseModel):
handle: str
email: EmailStr
password: str
role: str = "user"
@field_validator("password")
@classmethod
def password_strength(cls, v: str) -> str:
validate_password_strength(v)
return v
class UserStatusUpdate(BaseModel):
is_active: bool
class QuotaUpdate(BaseModel):
limit_bytes: int
@field_validator("limit_bytes")
@classmethod
def must_be_positive(cls, v: int) -> int:
if v <= 0:
raise ValueError("limit_bytes must be greater than 0")
return v
class UserAiConfigUpdate(BaseModel):
ai_provider: Optional[str] = None
ai_model: Optional[str] = None
class SystemAiConfigUpdate(BaseModel):
"""Request model for PUT /api/admin/ai-config (system-level provider configuration).
Security: extra="forbid" prevents mass-assignment of unexpected fields (T-07-13).
provider_id is validated against PROVIDER_DEFAULTS keys (T-07-13).
api_key is write-only: when None the existing api_key_enc is left untouched,
when "" the api_key_enc is cleared, when a non-empty string it is encrypted.
"""
model_config = ConfigDict(extra="forbid")
provider_id: str
api_key: Optional[str] = None
base_url: Optional[str] = None
model_name: Optional[str] = None
context_chars: Optional[int] = None
is_active: Optional[bool] = None
@field_validator("provider_id")
@classmethod
def provider_must_be_known(cls, v: str) -> str:
if v not in PROVIDER_DEFAULTS:
raise ValueError(
f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
)
return v
class TestConnectionRequest(BaseModel):
"""Request body for POST /api/admin/ai-config/test-connection.
Unsaved form values (api_key, base_url, model_name) override the DB row so
admins can verify credentials before saving. All override fields are optional;
omitting them falls back to whatever is stored in system_settings.
"""
model_config = ConfigDict(extra="forbid")
provider_id: str
api_key: Optional[str] = None # If non-empty, used instead of stored api_key_enc
base_url: Optional[str] = None # If non-None, overrides DB base_url
model_name: Optional[str] = None # If non-empty, overrides DB model_name
@field_validator("provider_id")
@classmethod
def provider_must_be_known(cls, v: str) -> str:
if v not in PROVIDER_DEFAULTS:
raise ValueError(
f"Unknown provider_id {v!r}. Must be one of: {list(PROVIDER_DEFAULTS.keys())}"
)
return v
class SystemTopicCreate(BaseModel):
"""Request model for admin system topic creation (D-09)."""
name: str
description: str = ""
color: str = "#6366f1"
class UserDeleteConfirm(BaseModel):
"""Admin password confirmation required before hard-deleting a user (ADMIN-02, T-05-11-01)."""
admin_password: str = Field(..., min_length=1)
# ── SEC-08: Safe CloudConnection response model ───────────────────────────────
class CloudConnectionOut(BaseModel):
"""SEC-08: credentials_enc deliberately excluded from this response model.
Any admin or user endpoint returning CloudConnection ORM objects MUST use
this model to prevent accidental exposure of encrypted credentials.
Safe-by-default: whitelist of allowed fields (not blacklist).
Note: id is declared as str and coerced via validator so UUID ORM values
serialize correctly without json_encoders (Rule 1 fix — T-05-06 test suite).
"""
id: str
provider: str
display_name: str
status: str
connected_at: datetime
server_url: Optional[str] = None
connection_username: Optional[str] = None
model_config = {"from_attributes": True}
@field_validator("id", mode="before")
@classmethod
def coerce_id_to_str(cls, v) -> str:
"""Coerce UUID objects to str so the model validates from ORM instances."""
return str(v)
# ── Endpoints ─────────────────────────────────────────────────────────────────
@router.get("/users")
async def list_users(
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""List all users, ordered by created_at DESC.
Response shape: { items: [...safe user fields...] }
Never includes password_hash, credentials_enc, or document content (T-02-27).
"""
result = await session.execute(
select(User).order_by(User.created_at.desc())
)
users = result.scalars().all()
return {"items": [_user_to_dict(u) for u in users]}
@router.post("/users", status_code=status.HTTP_201_CREATED)
async def create_user(
request: Request,
body: UserCreate,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Admin creates a new user account (ADMIN-01).
- password_must_change=True forces the user to change their password on
first login (T-02-32, D-06).
- Quota row initialized at 100 MB (D-06).
- Returns 409 if email or handle is already taken.
"""
# Check uniqueness
existing_email = await session.execute(
select(User).where(User.email == str(body.email))
)
if existing_email.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email already registered",
)
existing_handle = await session.execute(
select(User).where(User.handle == body.handle)
)
if existing_handle.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Handle already taken",
)
new_user = User(
id=uuid.uuid4(),
handle=body.handle,
email=str(body.email),
password_hash=hash_password(body.password),
role=body.role,
is_active=True,
totp_enabled=False,
password_must_change=True, # ADMIN-01: force password change on first login
)
session.add(new_user)
quota = Quota(
user_id=new_user.id,
limit_bytes=_DEFAULT_QUOTA_BYTES,
used_bytes=0,
)
session.add(quota)
await session.flush() # persist User + Quota before audit_log FK references them
# D-13: admin user created event
_ip_addr = get_client_ip(request)
await write_audit_log(
session,
event_type="admin.user_created",
user_id=new_user.id,
actor_id=_admin.id,
resource_id=new_user.id,
ip_address=_ip_addr,
)
await session.commit()
return {
"id": str(new_user.id),
"handle": new_user.handle,
"email": new_user.email,
"role": new_user.role,
"created_at": new_user.created_at.isoformat() if new_user.created_at else None,
}
@router.patch("/users/{user_id}/status")
async def update_user_status(
user_id: uuid.UUID,
body: UserStatusUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Deactivate or reactivate a user account (ADMIN-02).
- Prevents deactivating the last active admin (T-02-29).
- On deactivation: all refresh tokens are revoked (family revocation).
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# Guard: cannot deactivate the only remaining active admin (T-02-29)
if not body.is_active and user.role == "admin":
count_result = await session.execute(
select(func.count(User.id)).where(
User.role == "admin",
User.is_active.is_(True),
)
)
active_admin_count = count_result.scalar_one()
if active_admin_count <= 1:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot deactivate the only admin",
)
_ip_addr = get_client_ip(request)
user.is_active = body.is_active
if not body.is_active:
# Revoke all refresh tokens on deactivation
await revoke_all_refresh_tokens(session, user.id)
# Revoke any pre-deactivation access tokens still within their TTL (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
session.add(user)
# D-13: user deactivated/activated event
_event = "admin.user_deactivated" if not body.is_active else "admin.user_activated"
await write_audit_log(
session,
event_type=_event,
user_id=user.id,
actor_id=_admin.id,
resource_id=user.id,
ip_address=_ip_addr,
)
await session.commit()
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"is_active": user.is_active,
}
@router.post("/users/{user_id}/password-reset", status_code=status.HTTP_202_ACCEPTED)
async def initiate_password_reset(
user_id: uuid.UUID,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Admin initiates a password reset for a user (ADMIN-03).
Sends the reset email via Celery. Does NOT:
- return a reset token (T-02-30)
- grant admin access to the account
- log in as the target user (ADMIN-07 — no impersonation)
Returns 202 immediately regardless of email delivery status.
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
from services.auth import create_password_reset_token # noqa: PLC0415
from config import settings as _settings # noqa: PLC0415
reset_token = create_password_reset_token(str(user.id))
reset_link = f"{_settings.frontend_url}/password-reset/confirm?token={reset_token}"
# Deferred import to avoid circular imports (same pattern as document_tasks)
from tasks.email_tasks import send_reset_email # noqa: PLC0415
send_reset_email.delay(user.email, reset_link)
return {"message": "Password reset email sent"}
@router.get("/users/{user_id}/quota")
async def get_user_quota(
user_id: uuid.UUID,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return quota details for a user (ADMIN-04).
Quota info is admin-visible operational data — no PII, no document content
(T-02-31 disposition: accept).
"""
quota = await session.get(Quota, user_id)
if quota is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
return {
"user_id": str(quota.user_id),
"limit_bytes": quota.limit_bytes,
"used_bytes": quota.used_bytes,
"limit_mb": quota.limit_bytes // 1048576,
"used_mb": quota.used_bytes // 1048576,
}
@router.patch("/users/{user_id}/quota")
async def update_user_quota(
user_id: uuid.UUID,
body: QuotaUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Adjust a user's storage quota (ADMIN-04).
If the new limit is below current usage, still applies the change but
returns warning=True with an explanatory message. Uploads will be blocked
but existing documents are preserved.
"""
quota = await session.get(Quota, user_id)
if quota is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Quota not found")
warning = body.limit_bytes < quota.used_bytes
warning_message = (
"New limit is below current usage. Uploads will be blocked but existing documents are preserved."
if warning
else None
)
_ip_addr = get_client_ip(request)
old_limit = quota.limit_bytes
quota.limit_bytes = body.limit_bytes
session.add(quota)
# D-13: quota changed event
await write_audit_log(
session,
event_type="admin.quota_changed",
user_id=user_id,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"old_bytes": old_limit, "new_bytes": body.limit_bytes},
)
await session.commit()
response: dict = {
"user_id": str(quota.user_id),
"limit_bytes": quota.limit_bytes,
"used_bytes": quota.used_bytes,
"warning": warning,
}
if warning_message:
response["message"] = warning_message
return response
@router.patch("/users/{user_id}/ai-config")
async def update_ai_config(
user_id: uuid.UUID,
body: UserAiConfigUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Assign AI provider and model for a user (ADMIN-05).
Users cannot change their own AI provider or model (PROJECT.md Key Decision).
Only admins have this capability.
"""
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
_ip_addr = get_client_ip(request)
user.ai_provider = body.ai_provider
user.ai_model = body.ai_model
session.add(user)
# D-13: AI provider assigned event
await write_audit_log(
session,
event_type="admin.ai_provider_assigned",
user_id=user_id,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"provider": body.ai_provider, "model": body.ai_model},
)
await session.commit()
return {
"id": str(user.id),
"email": user.email,
"ai_provider": user.ai_provider,
"ai_model": user.ai_model,
}
@router.delete("/users/{user_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_user(
user_id: uuid.UUID,
body: UserDeleteConfirm,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> None:
"""Delete a user account and clean up all their MinIO objects (SEC-09, D-19).
Security invariants:
- Admin password verified via Argon2 before any deletion (T-05-11-01)
- Cannot delete admin accounts (T-04-07-04)
- MinIO objects are deleted BEFORE DB records are removed (SEC-09)
- MinIO deletion is best-effort (try/except) — DB row is deleted regardless
- Audit log written with event_type="admin.user_deleted"
"""
# T-05-11-01: Verify admin password before performing any destructive action.
# Fail fast — no DB reads for the target user until the admin is confirmed.
if not verify_password(body.admin_password, _admin.password_hash):
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Invalid admin password",
)
user = await session.get(User, user_id)
if user is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found")
# T-04-07-04: Cannot delete admin accounts
if user.role == "admin":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Cannot delete admin accounts",
)
_ip_addr = get_client_ip(request)
# SEC-09 (cloud): purge cloud-stored documents and credentials BEFORE DB delete.
# Must run before MinIO cleanup so that credentials are still available to build
# the cloud backend instances for delete_object calls.
cloud_conns_result = await session.execute(
select(CloudConnection).where(CloudConnection.user_id == user_id)
)
cloud_conns = cloud_conns_result.scalars().all()
for conn in cloud_conns:
# Delete cloud objects stored in this provider for this user
cloud_docs_result = await session.execute(
select(Document).where(
Document.user_id == user_id,
Document.storage_backend == conn.provider,
)
)
for doc in cloud_docs_result.scalars().all():
try:
backend = await get_storage_backend_for_document(doc, user, session)
await backend.delete_object(doc.object_key)
except Exception:
pass # Best-effort cloud object cleanup; deletion proceeds regardless
# Purge the credentials row (FK cascade would also remove it, but explicit
# deletion here guarantees credentials_enc is gone before commit — SEC-09)
await session.delete(conn)
if cloud_conns:
await session.flush() # Flush connection deletes before user delete
await write_audit_log(
session,
event_type="cloud.credentials_purged",
user_id=user_id,
actor_id=_admin.id,
resource_id=user_id,
ip_address=_ip_addr,
metadata_={"providers": [c.provider for c in cloud_conns]},
)
# SEC-09 (minio): collect all user documents and delete MinIO objects BEFORE DB delete
docs_result = await session.execute(
select(Document).where(Document.user_id == user_id)
)
user_docs = docs_result.scalars().all()
storage = get_storage_backend()
for doc in user_docs:
try:
await storage.delete_object(doc.object_key)
except Exception:
pass # Best-effort MinIO cleanup; DB deletion proceeds regardless
# D-13: audit log BEFORE deleting the user row (user FK still valid at flush time)
await write_audit_log(
session,
event_type="admin.user_deleted",
user_id=user_id,
actor_id=_admin.id,
resource_id=user_id,
ip_address=_ip_addr,
)
await session.flush()
# Delete user record (CASCADE removes quota, documents, refresh_tokens, etc.)
await session.delete(user)
await session.commit()
@router.post("/topics", status_code=status.HTTP_201_CREATED)
async def create_system_topic(
body: SystemTopicCreate,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Create a system topic visible to all users (D-09, DOC-04).
System topics have user_id = NULL, making them visible to every user as
defaults in their topic namespace. Only admins can create system topics.
Regular users create per-user topics via POST /api/topics.
Deduplication: case-insensitive match within the system namespace (user_id IS NULL).
Returns the existing system topic if one with the same name already exists.
"""
from services import storage # noqa: PLC0415
topic = await storage.create_topic(
session, body.name, body.description, body.color, user_id=None
)
return topic
# ── System AI Provider Configuration (D-08, D-15) ────────────────────────────
@router.get("/ai-config/models")
async def get_ai_config_models(
provider_id: str,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return the list of model IDs available from a provider's API (D-08).
Calls the provider's standard GET /models endpoint using the stored
config (base_url + api_key from system_settings). Always returns 200
with {"models": [...]} — never 5xx on provider failure (returns empty list).
Security: requires get_current_admin; provider_id from query param only;
decrypted api_key never appears in the response.
"""
import httpx # noqa: PLC0415 — local import keeps admin.py startup fast
config = await load_provider_config_by_id(session, provider_id)
# Resolve base_url: prefer DB row, fall back to PROVIDER_DEFAULTS
if config and config.base_url:
base_url = config.base_url.rstrip("/")
else:
base_url = (PROVIDER_DEFAULTS.get(provider_id, {}).get("base_url") or "").rstrip("/")
if not base_url:
return {"models": [], "provider_id": provider_id}
api_key = config.api_key if config else ""
# Build request headers — Anthropic uses x-api-key; all others use Bearer
if provider_id == "anthropic":
headers = {
"x-api-key": api_key,
"anthropic-version": "2023-06-01",
}
models_url = "https://api.anthropic.com/v1/models"
else:
headers = {"Authorization": f"Bearer {api_key}"} if api_key else {}
models_url = f"{base_url}/models"
try:
async with httpx.AsyncClient(timeout=8.0) as client:
resp = await client.get(models_url, headers=headers)
resp.raise_for_status()
data = resp.json()
# Standard OpenAI-compat shape: {"data": [{"id": "...", ...}, ...]}
# Anthropic shape: {"data": [{"id": "...", ...}, ...]}
# Ollama OpenAI-compat: same shape
raw_list = data.get("data") or data.get("models") or []
model_ids: list[str] = sorted(
{
item["id"] if isinstance(item, dict) else str(item)
for item in raw_list
if item
}
)
return {"models": model_ids, "provider_id": provider_id}
except Exception as exc:
return {"models": [], "provider_id": provider_id, "error": str(exc)[:120]}
@router.post("/ai-config/test-connection")
async def test_ai_connection(
body: TestConnectionRequest,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Test connectivity for an AI provider, optionally with unsaved form values (D-08).
Loads the stored system_settings row for body.provider_id, then overlays any
non-empty values from the request body so admins can verify credentials before
saving them to the database.
Override priority (highest → lowest):
1. body.api_key / base_url / model_name (unsaved form values)
2. system_settings DB row (previously saved config)
3. PROVIDER_DEFAULTS (built-in fallback)
Returns {"ok": true/false, "provider_id": str} — never raises 5xx for
provider-side failures; surfaces as ok=False so the UI shows a clear status.
Security: requires get_current_admin; api_key from body is used only for the
in-flight health_check() call and is never stored or logged.
"""
provider_id = body.provider_id
stored = await load_provider_config_by_id(session, provider_id)
defaults = PROVIDER_DEFAULTS.get(provider_id, {})
# Resolve effective values: body overrides DB, DB overrides PROVIDER_DEFAULTS
effective_api_key = (
body.api_key
if body.api_key
else (stored.api_key if stored else "")
)
effective_base_url = (
body.base_url
if body.base_url is not None
else (stored.base_url if stored else defaults.get("base_url"))
)
effective_model = (
body.model_name
if body.model_name
else (stored.model if stored else defaults.get("model", ""))
)
effective_config = ProviderConfig(
provider_id=provider_id,
api_key=effective_api_key,
base_url=effective_base_url,
model=effective_model,
)
try:
provider = get_provider(effective_config)
ok = await provider.health_check()
return {"ok": ok, "provider_id": provider_id}
except Exception:
return {"ok": False, "provider_id": provider_id, "reason": "health_check_failed"}
@router.get("/ai-config")
async def get_ai_config(
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Return all AI provider configurations for the admin panel (D-08).
Includes DB rows for providers that have been saved AND synthesised stubs
for providers that only exist in PROVIDER_DEFAULTS (so the admin UI always
shows all 10 providers even before any have been configured).
Security invariant: api_key_enc is NEVER returned (T-07-01).
Use has_api_key (bool) as the only indicator that a key is stored.
"""
result = await session.execute(select(SystemSettings))
db_rows = result.scalars().all()
# Build a lookup for DB rows
db_by_provider: dict[str, SystemSettings] = {r.provider_id: r for r in db_rows}
providers_out = []
for pid in PROVIDER_DEFAULTS:
if pid in db_by_provider:
providers_out.append(_ai_config_to_dict(db_by_provider[pid]))
else:
# Synthesise a stub entry for providers with no DB row yet
defaults = PROVIDER_DEFAULTS[pid]
providers_out.append({
"provider_id": pid,
"base_url": defaults.get("base_url"),
"model_name": defaults.get("model", ""),
"context_chars": defaults.get("context_chars", 8000),
"is_active": False,
"has_api_key": False,
"updated_at": None,
})
return {"providers": providers_out}
@router.put("/ai-config")
async def update_system_ai_config(
body: SystemAiConfigUpdate,
request: Request,
session: AsyncSession = Depends(get_db),
_admin: User = Depends(get_current_admin),
) -> dict:
"""Create or update a system-level AI provider configuration (D-08, D-15).
Upsert semantics: if no row exists for body.provider_id, one is created using
PROVIDER_DEFAULTS for any omitted fields.
API key handling (T-07-01 mitigated):
- body.api_key is None → leave existing api_key_enc untouched
- body.api_key == "" → clear api_key_enc (set to NULL)
- body.api_key is a non-empty string → HKDF-encrypt and store
is_active=True handling (T-07-03 mitigated):
When body.is_active is True, a single atomic UPDATE flips all rows:
SET is_active = (provider_id = :target_id)
This guarantees COUNT(WHERE is_active) == 1 with no read-then-write race.
Audit log (T-07-14 mitigated):
metadata_ contains only provider_id + fields_changed list — never the
api_key value itself.
"""
from config import settings as _settings # noqa: PLC0415
# Load existing row or create a new one from PROVIDER_DEFAULTS
stmt = select(SystemSettings).where(SystemSettings.provider_id == body.provider_id)
result = await session.execute(stmt)
row = result.scalar_one_or_none()
is_new = row is None
if is_new:
defaults = PROVIDER_DEFAULTS[body.provider_id]
row = SystemSettings(
provider_id=body.provider_id,
model_name=defaults.get("model", ""),
context_chars=defaults.get("context_chars", 8000),
base_url=defaults.get("base_url"),
is_active=False,
api_key_enc=None,
)
# Track which fields the caller explicitly set (for audit log — never api_key value)
fields_changed: list[str] = []
# Apply provided fields
if body.api_key is not None:
fields_changed.append("api_key")
if body.api_key == "":
row.api_key_enc = None
else:
master_key_str = _settings.cloud_creds_key
master_key_bytes = (
master_key_str.encode("utf-8")
if isinstance(master_key_str, str)
else master_key_str
)
row.api_key_enc = encrypt_api_key(master_key_bytes, body.provider_id, body.api_key)
if body.base_url is not None:
row.base_url = body.base_url
fields_changed.append("base_url")
if body.model_name is not None:
row.model_name = body.model_name
fields_changed.append("model_name")
if body.context_chars is not None:
row.context_chars = body.context_chars
fields_changed.append("context_chars")
if body.is_active is not None:
fields_changed.append("is_active")
if is_new:
session.add(row)
await session.flush() # ensure row has an id before UPDATE
# Atomic is_active flip: SET is_active = (provider_id = :target) on ALL rows.
# Single UPDATE statement prevents dual-active race condition (T-07-03).
if body.is_active is True:
await session.execute(
update(SystemSettings).values(
is_active=(SystemSettings.provider_id == body.provider_id)
)
)
# Reflect the flip on the in-memory row
row.is_active = True
_ip_addr = get_client_ip(request)
await write_audit_log(
session,
event_type="admin.ai_config_changed",
user_id=None,
actor_id=_admin.id,
resource_id=None,
ip_address=_ip_addr,
metadata_={"provider_id": body.provider_id, "fields_changed": fields_changed},
)
await session.commit()
# Reload to pick up DB-generated updated_at after commit
await session.refresh(row)
return _ai_config_to_dict(row)
backend/api/auth.py
-825
View File
@@ -1,825 +0,0 @@
"""
Auth API endpoints for DocuVault.
Implements:
POST /api/auth/register — new user registration with HIBP check
POST /api/auth/login — login with optional TOTP/backup-code second factor
POST /api/auth/refresh — rotate refresh token (httpOnly cookie in/out)
POST /api/auth/logout — revoke current refresh token, clear cookie
GET /api/auth/me — return current user profile
POST /api/auth/change-password — update password (requires current password)
Security invariants:
- Per-account rate limit: 10 login attempts per email per 15 minutes (SEC-02)
- HTTP 429 returned before any DB lookup when the counter is exceeded
- httpOnly Secure SameSite=Strict refresh cookie (CLAUDE.md constraint)
- HIBP breach check on register and change-password (SEC-03)
- TOTP takes precedence over backup_code when both fields are provided
- password_must_change=True: returns requires_password_change without tokens
"""
from __future__ import annotations
import hashlib
import time
import uuid
from typing import Literal, Optional
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel, EmailStr
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from config import settings
from db.models import BackupCode, Quota, RefreshToken, User
from deps.auth import get_current_user
from deps.db import get_db
from deps.utils import get_client_ip
from services import auth as auth_service
from services.audit import write_audit_log
from slowapi import Limiter
from sqlalchemy import delete
router = APIRouter(prefix="/api/auth", tags=["auth"])
# IP-level rate limiter (SEC-02 — 10 req/min on register/login/refresh)
limiter = Limiter(key_func=get_client_ip)
# ── Request models ────────────────────────────────────────────────────────────
class RegisterRequest(BaseModel):
handle: str
email: EmailStr
password: str
class LoginRequest(BaseModel):
email: EmailStr
password: str
totp_code: Optional[str] = None
backup_code: Optional[str] = None
remember_me: bool = False
class ChangePasswordRequest(BaseModel):
current_password: str
new_password: str
# ── Helper: set httpOnly refresh cookie ──────────────────────────────────────
def _set_refresh_cookie(
response: Response, raw_token: str, remember_me: bool = False
) -> None:
"""Set the httpOnly Secure SameSite=Strict refresh cookie (CLAUDE.md constraint).
remember_me=False (default): Max-Age = refresh_token_expire_hours * 3600 (16h, D-11, RM-03)
remember_me=True: Max-Age = refresh_token_expire_days * 86400 (30d, D-11, RM-03)
"""
max_age = (
settings.refresh_token_expire_days * 86400
if remember_me
else settings.refresh_token_expire_hours * 3600
)
response.set_cookie(
key="refresh_token",
value=raw_token,
httponly=True,
secure=True,
samesite="strict",
path="/api/auth/refresh",
max_age=max_age,
)
def _user_dict(user: User) -> dict:
"""Return serialisable user metadata (no password_hash, no credentials_enc)."""
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
"created_at": user.created_at.isoformat() if user.created_at else None,
}
# ── POST /api/auth/register ───────────────────────────────────────────────────
@router.post("/register", status_code=status.HTTP_201_CREATED)
@limiter.limit("10/minute")
async def register(
request: Request,
body: RegisterRequest,
session: AsyncSession = Depends(get_db),
):
"""Register a new user account.
- Validates password strength (min 12 chars, upper, lower, digit, special)
- Checks HIBP k-anonymity API for breached passwords
- Hashes password with Argon2
- Inserts User + Quota rows in a single transaction
"""
# Password strength check
try:
auth_service.validate_password_strength(body.password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# HIBP breach check
if await auth_service.check_hibp(body.password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Duplicate email/handle check
result = await session.execute(
select(User).where(
(User.email == str(body.email)) | (User.handle == body.handle)
)
)
if result.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email or handle already in use",
)
# Create user and quota
user_id = uuid.uuid4()
new_user = User(
id=user_id,
handle=body.handle,
email=str(body.email),
password_hash=auth_service.hash_password(body.password),
role="user",
is_active=True,
password_must_change=False,
)
quota = Quota(
user_id=user_id,
limit_bytes=104857600, # 100 MB default (STORE-01)
used_bytes=0,
)
try:
session.add(new_user)
await session.flush() # persist User before Quota FK
session.add(quota)
await session.commit()
await session.refresh(new_user)
except IntegrityError:
await session.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email or handle already in use",
)
return {
"id": str(new_user.id),
"handle": new_user.handle,
"email": new_user.email,
"role": new_user.role,
"totp_enabled": new_user.totp_enabled,
"created_at": new_user.created_at.isoformat() if new_user.created_at else None,
}
# ── POST /api/auth/login ──────────────────────────────────────────────────────
@router.post("/login")
@limiter.limit("10/minute")
async def login(
request: Request,
body: LoginRequest,
response: Response,
session: AsyncSession = Depends(get_db),
):
"""Authenticate a user and issue tokens.
Per-account rate limiting (SEC-02): checks Redis counter keyed by email
BEFORE any DB lookup to prevent enumeration timing attacks.
Three login flows:
1. No TOTP enabled: password → tokens
2. TOTP enabled, no code provided: requires_totp = True (challenge)
3. TOTP enabled, totp_code provided: verify TOTP → tokens
4. TOTP enabled, backup_code provided (no totp_code): verify backup → tokens
"""
# Per-account rate limiting (SEC-02)
redis_client = request.app.state.redis
rate_key = f"login_attempts:{body.email}"
count = await redis_client.incr(rate_key)
if count == 1:
# Set TTL only on first increment (15-minute window)
await redis_client.expire(rate_key, 900)
if count > 10:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Too many login attempts. Try again in 15 minutes.",
)
# Look up user by email
result = await session.execute(select(User).where(User.email == str(body.email)))
user: Optional[User] = result.scalar_one_or_none()
# IP extraction for audit log (used in both success and failure paths)
_ip = get_client_ip(request)
# Verify password (anti-enumeration: same error regardless of whether user exists)
if user is None or not auth_service.verify_password(body.password, user.password_hash):
await write_audit_log(
session,
event_type="auth.login_failed",
user_id=user.id if user else None,
actor_id=user.id if user else None,
resource_id=None,
ip_address=_ip,
metadata_={"attempted_email_hash": hashlib.sha256(str(body.email).encode()).hexdigest()[:16]},
)
await session.commit()
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
)
# Active check
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Account deactivated",
)
# Password must change: return challenge without issuing tokens (T-02-16)
if user.password_must_change:
return {"requires_password_change": True, "user_id": str(user.id)}
# TOTP second-factor dispatch
if user.totp_enabled:
if body.totp_code is None and body.backup_code is None:
# Challenge: prompt for second factor
return {"requires_totp": True}
if body.totp_code is not None:
# TOTP path takes precedence (even if backup_code also provided)
ok = await auth_service.verify_totp(session, user.id, body.totp_code, redis_client)
if not ok:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect code",
)
else:
# Backup code path (body.backup_code is not None and body.totp_code is None)
ok = await auth_service.verify_backup_code(session, user.id, body.backup_code)
if not ok:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or already used code",
)
# D-13: backup code used event
await write_audit_log(
session,
event_type="auth.backup_code_used",
user_id=user.id,
actor_id=user.id,
resource_id=None,
ip_address=_ip,
)
# Issue tokens
access_token = auth_service.create_access_token(
str(user.id),
user.role,
user_agent=request.headers.get("User-Agent", ""),
accept_lang=request.headers.get("Accept-Language", ""),
)
raw_refresh = await auth_service.create_refresh_token(session, user.id, remember_me=body.remember_me)
_set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me)
# D-13: login success event
await write_audit_log(
session,
event_type="auth.login",
user_id=user.id,
actor_id=user.id,
resource_id=None,
ip_address=_ip,
metadata_={"totp_used": user.totp_enabled and body.totp_code is not None},
)
await session.commit()
return {
"access_token": access_token,
"user": {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
},
}
# ── POST /api/auth/refresh ────────────────────────────────────────────────────
@router.post("/refresh")
@limiter.limit("10/minute")
async def refresh_token(
request: Request,
response: Response,
session: AsyncSession = Depends(get_db),
):
"""Rotate the refresh token.
Reads the refresh_token httpOnly cookie; on success issues a new access
token and rotates the refresh cookie.
On token reuse (revoked token presented), revokes entire family and raises 401.
"""
raw_token = request.cookies.get("refresh_token")
if not raw_token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="No refresh token",
)
try:
new_raw, user_id_str = await auth_service.rotate_refresh_token(session, raw_token)
except ValueError as exc:
if "token_family_revoked" in str(exc):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session revoked",
) from exc
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired refresh token",
) from exc
# Look up user for response body
user = await session.get(User, uuid.UUID(user_id_str))
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or deactivated",
)
# Set new refresh cookie
_set_refresh_cookie(response, new_raw)
access_token = auth_service.create_access_token(
user_id_str,
user.role,
user_agent=request.headers.get("User-Agent", ""),
accept_lang=request.headers.get("Accept-Language", ""),
)
return {
"access_token": access_token,
"user": {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
},
}
# ── POST /api/auth/logout ─────────────────────────────────────────────────────
@router.post("/logout")
async def logout(request: Request, response: Response, session: AsyncSession = Depends(get_db)):
"""Revoke current refresh token and clear the cookie."""
import hashlib as _hashlib
_ip = get_client_ip(request)
raw_token = request.cookies.get("refresh_token")
_logout_user_id = None
if raw_token:
token_hash = _hashlib.sha256(raw_token.encode()).hexdigest()
result = await session.execute(
select(RefreshToken).where(RefreshToken.token_hash == token_hash)
)
row: Optional[RefreshToken] = result.scalar_one_or_none()
if row is not None:
_logout_user_id = row.user_id
row.revoked = True
# D-13: logout event (written before commit, within same transaction)
await write_audit_log(
session,
event_type="auth.logout",
user_id=_logout_user_id,
actor_id=_logout_user_id,
resource_id=None,
ip_address=_ip,
)
await session.commit()
response.delete_cookie("refresh_token", path="/api/auth/refresh")
return {"message": "Logged out"}
# ── POST /api/auth/logout-all ─────────────────────────────────────────────────
@router.post("/logout-all")
async def logout_all(
request: Request,
response: Response,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Sign out of all devices: revoke all refresh tokens for current user."""
_ip = get_client_ip(request)
count = await auth_service.revoke_all_refresh_tokens(session, current_user.id)
# D-13: sign-out-all event
await write_audit_log(
session,
event_type="auth.sign_out_all",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": count},
)
await session.commit()
response.delete_cookie("refresh_token", path="/api/auth/refresh")
return {"message": f"Signed out of {count} session(s)"}
# ── GET /api/auth/me ──────────────────────────────────────────────────────────
@router.get("/me")
async def get_me(current_user: User = Depends(get_current_user)):
"""Return the current user's profile (requires valid Bearer token)."""
return _user_dict(current_user)
# ── GET /api/auth/me/quota ────────────────────────────────────────────────────
@router.get("/me/quota")
async def get_my_quota(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_db),
):
"""Return the current user's quota usage (STORE-04).
Returns {"used_bytes": int, "limit_bytes": int} for the sidebar quota bar.
Quota row is created at registration (100 MB default — STORE-01).
"""
q = await session.get(Quota, current_user.id)
if q is None:
raise HTTPException(status_code=404, detail="Quota not found")
return {"used_bytes": q.used_bytes, "limit_bytes": q.limit_bytes}
# ── POST /api/auth/change-password ───────────────────────────────────────────
@router.post("/change-password")
async def change_password(
request: Request,
body: ChangePasswordRequest,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update the current user's password.
Checks:
1. current_password matches stored hash
2. new_password has not appeared in HIBP (SEC-03)
3. new_password meets strength requirements (AUTH-01)
"""
# Verify current password
if not auth_service.verify_password(body.current_password, current_user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Current password is incorrect",
)
# HIBP breach check on new password (SEC-03)
if await auth_service.check_hibp(body.new_password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Password strength check
try:
auth_service.validate_password_strength(body.new_password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# Update password
_ip = get_client_ip(request)
user = await session.get(User, current_user.id)
user.password_hash = auth_service.hash_password(body.new_password)
# Revoke other sessions; keep current one alive via skip_token_hash (CR-01)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: password changed event (flush within same transaction before commit)
await write_audit_log(
session,
event_type="auth.password_changed",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-change access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"message": "Password updated", "sessions_revoked": revoked}
# ── Request models for new endpoints ─────────────────────────────────────────
class TotpEnableRequest(BaseModel):
code: str
class PasswordResetRequest(BaseModel):
email: EmailStr
class PasswordResetConfirmRequest(BaseModel):
token: str
new_password: str
# ── GET /api/auth/totp/setup ──────────────────────────────────────────────────
@router.get("/totp/setup")
async def totp_setup(
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Provision a TOTP secret for the current user.
If TOTP is already enabled, returns 400.
Returns { provisioning_uri, secret } — the provisioning_uri is suitable
for QR code generation. The secret is the base32-encoded TOTP secret.
"""
if current_user.totp_enabled:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="TOTP already enabled",
)
secret, provisioning_uri = await auth_service.provision_totp(session, current_user.id)
return {"provisioning_uri": provisioning_uri, "secret": secret}
# ── POST /api/auth/totp/enable ───────────────────────────────────────────────
@router.post("/totp/enable")
@limiter.limit("10/minute")
async def enable_totp(
request: Request,
body: TotpEnableRequest,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Enable TOTP for the current user.
Rate-limited to 10 attempts/minute per IP (SEC-02 / T-02-25).
Verifies the submitted 6-digit code (with Redis replay prevention, AUTH-08).
On success: marks TOTP enabled, generates and returns 10 one-time backup codes.
The backup codes are ONLY returned here — they are stored as Argon2 hashes
in the DB and never returned again (T-02-19).
"""
redis_client = request.app.state.redis
ok = await auth_service.verify_totp(session, current_user.id, body.code, redis_client)
if not ok:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Incorrect or expired code",
)
# Mark TOTP as enabled
user = await session.get(User, current_user.id)
user.totp_enabled = True
await session.flush()
# Generate and store 10 backup codes; return plaintext to user (one-time, T-02-19)
plain_codes = auth_service.generate_backup_codes(10)
await auth_service.store_backup_codes(session, current_user.id, plain_codes)
# Revoke other sessions; keep current one alive via skip_token_hash (CR-02)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: TOTP enrolled event
_ip = get_client_ip(request)
await write_audit_log(
session,
event_type="auth.totp_enrolled",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-enroll access tokens still within their TTL window (T-7.2-01)
await redis_client.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"backup_codes": plain_codes, "sessions_revoked": revoked}
# ── DELETE /api/auth/totp ─────────────────────────────────────────────────────
@router.delete("/totp")
async def disable_totp(
request: Request,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Disable TOTP for the current user.
Clears totp_secret, sets totp_enabled=False, and deletes all backup codes.
"""
_ip = get_client_ip(request)
user = await session.get(User, current_user.id)
user.totp_enabled = False
user.totp_secret = None
# Delete all backup codes for this user (including unused ones)
await session.execute(delete(BackupCode).where(BackupCode.user_id == current_user.id))
# Revoke other sessions; keep current one alive via skip_token_hash (CR-03)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: TOTP revoked event
await write_audit_log(
session,
event_type="auth.totp_revoked",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-revoke access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"message": "TOTP disabled", "sessions_revoked": revoked}
# ── POST /api/auth/password-reset ─────────────────────────────────────────────
@router.post("/password-reset", status_code=status.HTTP_202_ACCEPTED)
@limiter.limit("5/hour")
async def password_reset_request(
request: Request,
body: PasswordResetRequest,
session: AsyncSession = Depends(get_db),
):
"""Request a password reset email.
Always returns 202 regardless of whether the email exists (anti-enumeration, T-02-22).
If the user is found, a signed reset token (1-hour JWT) is generated and a Celery
task is enqueued to send the email (D-02, D-03).
"""
from sqlalchemy import select as _select # noqa: PLC0415 (already imported above)
result = await session.execute(_select(User).where(User.email == str(body.email)))
user: Optional[User] = result.scalar_one_or_none()
if user is not None:
token = auth_service.create_password_reset_token(str(user.id))
reset_link = f"{settings.frontend_url}/password-reset/confirm?token={token}"
# Deferred import to avoid circular import; Celery task is fire-and-forget
from tasks.email_tasks import send_reset_email # noqa: PLC0415
send_reset_email.delay(user.email, reset_link)
# Always return 202 (anti-enumeration — never reveal whether email exists)
return {
"message": (
"If an account exists for that email, you will receive a reset link shortly."
)
}
# ── POST /api/auth/password-reset/confirm ────────────────────────────────────
@router.post("/password-reset/confirm")
async def password_reset_confirm(
request: Request,
body: PasswordResetConfirmRequest,
session: AsyncSession = Depends(get_db),
):
"""Confirm a password reset using the token from the email link.
Validates the reset token, enforces password strength + HIBP check, updates
the password, and revokes all refresh tokens. Does NOT issue new tokens —
the user must sign in again through /login (AUTH-05, T-02-21).
"""
try:
user_id_str = auth_service.decode_password_reset_token(body.token)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid or expired reset link",
)
# Password strength validation
try:
auth_service.validate_password_strength(body.new_password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# HIBP breach check (SEC-03)
if await auth_service.check_hibp(body.new_password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Load user
user = await session.get(User, uuid.UUID(user_id_str))
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid or expired reset link",
)
# Update password and revoke all sessions (forces re-auth through TOTP if enabled)
user.password_hash = auth_service.hash_password(body.new_password)
await auth_service.revoke_all_refresh_tokens(session, user.id)
# Revoke any pre-reset access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
# Do NOT issue tokens (AUTH-05 — user must pass TOTP gate on next login)
return {"message": "Password updated. Please sign in."}
# ── Preferences models ────────────────────────────────────────────────────────
class PreferencesUpdate(BaseModel):
"""Request body for PATCH /api/auth/me/preferences.
Validates pdf_open_mode strictly via Literal (T-04-05-05 — no mass assignment).
"""
pdf_open_mode: Literal["in_app", "new_tab"]
# ── GET /api/auth/me/preferences ─────────────────────────────────────────────
@router.get("/me/preferences")
async def get_my_preferences(
current_user: User = Depends(get_current_user),
):
"""Return the current user's PDF open mode preference (D-10).
Both regular users and admins can read their own preferences.
Falls back to 'in_app' if the column is absent (migration not yet run).
"""
try:
pdf_open_mode = current_user.pdf_open_mode
except AttributeError:
pdf_open_mode = "in_app"
return {"pdf_open_mode": pdf_open_mode}
# ── PATCH /api/auth/me/preferences ───────────────────────────────────────────
@router.patch("/me/preferences")
async def update_my_preferences(
body: PreferencesUpdate,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update the current user's PDF open mode preference (D-10).
Both regular users and admins can update their own preferences.
Pydantic Literal["in_app", "new_tab"] enforces strict allowlist (T-04-05-05).
"""
user = await session.get(User, current_user.id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
user.pdf_open_mode = body.pdf_open_mode
session.add(user)
await session.commit()
return {"pdf_open_mode": user.pdf_open_mode}
backend/api/auth_OLD_REMOVE_IN_TASK_2.py
-825
View File
@@ -1,825 +0,0 @@
"""
Auth API endpoints for DocuVault.
Implements:
POST /api/auth/register — new user registration with HIBP check
POST /api/auth/login — login with optional TOTP/backup-code second factor
POST /api/auth/refresh — rotate refresh token (httpOnly cookie in/out)
POST /api/auth/logout — revoke current refresh token, clear cookie
GET /api/auth/me — return current user profile
POST /api/auth/change-password — update password (requires current password)
Security invariants:
- Per-account rate limit: 10 login attempts per email per 15 minutes (SEC-02)
- HTTP 429 returned before any DB lookup when the counter is exceeded
- httpOnly Secure SameSite=Strict refresh cookie (CLAUDE.md constraint)
- HIBP breach check on register and change-password (SEC-03)
- TOTP takes precedence over backup_code when both fields are provided
- password_must_change=True: returns requires_password_change without tokens
"""
from __future__ import annotations
import hashlib
import time
import uuid
from typing import Literal, Optional
from fastapi import APIRouter, Depends, HTTPException, Request, Response, status
from pydantic import BaseModel, EmailStr
from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession
from config import settings
from db.models import BackupCode, Quota, RefreshToken, User
from deps.auth import get_current_user
from deps.db import get_db
from deps.utils import get_client_ip
from services import auth as auth_service
from services.audit import write_audit_log
from slowapi import Limiter
from sqlalchemy import delete
router = APIRouter(prefix="/api/auth", tags=["auth"])
# IP-level rate limiter (SEC-02 — 10 req/min on register/login/refresh)
limiter = Limiter(key_func=get_client_ip)
# ── Request models ────────────────────────────────────────────────────────────
class RegisterRequest(BaseModel):
handle: str
email: EmailStr
password: str
class LoginRequest(BaseModel):
email: EmailStr
password: str
totp_code: Optional[str] = None
backup_code: Optional[str] = None
remember_me: bool = False
class ChangePasswordRequest(BaseModel):
current_password: str
new_password: str
# ── Helper: set httpOnly refresh cookie ──────────────────────────────────────
def _set_refresh_cookie(
response: Response, raw_token: str, remember_me: bool = False
) -> None:
"""Set the httpOnly Secure SameSite=Strict refresh cookie (CLAUDE.md constraint).
remember_me=False (default): Max-Age = refresh_token_expire_hours * 3600 (16h, D-11, RM-03)
remember_me=True: Max-Age = refresh_token_expire_days * 86400 (30d, D-11, RM-03)
"""
max_age = (
settings.refresh_token_expire_days * 86400
if remember_me
else settings.refresh_token_expire_hours * 3600
)
response.set_cookie(
key="refresh_token",
value=raw_token,
httponly=True,
secure=True,
samesite="strict",
path="/api/auth/refresh",
max_age=max_age,
)
def _user_dict(user: User) -> dict:
"""Return serialisable user metadata (no password_hash, no credentials_enc)."""
return {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
"created_at": user.created_at.isoformat() if user.created_at else None,
}
# ── POST /api/auth/register ───────────────────────────────────────────────────
@router.post("/register", status_code=status.HTTP_201_CREATED)
@limiter.limit("10/minute")
async def register(
request: Request,
body: RegisterRequest,
session: AsyncSession = Depends(get_db),
):
"""Register a new user account.
- Validates password strength (min 12 chars, upper, lower, digit, special)
- Checks HIBP k-anonymity API for breached passwords
- Hashes password with Argon2
- Inserts User + Quota rows in a single transaction
"""
# Password strength check
try:
auth_service.validate_password_strength(body.password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# HIBP breach check
if await auth_service.check_hibp(body.password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Duplicate email/handle check
result = await session.execute(
select(User).where(
(User.email == str(body.email)) | (User.handle == body.handle)
)
)
if result.scalar_one_or_none() is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email or handle already in use",
)
# Create user and quota
user_id = uuid.uuid4()
new_user = User(
id=user_id,
handle=body.handle,
email=str(body.email),
password_hash=auth_service.hash_password(body.password),
role="user",
is_active=True,
password_must_change=False,
)
quota = Quota(
user_id=user_id,
limit_bytes=104857600, # 100 MB default (STORE-01)
used_bytes=0,
)
try:
session.add(new_user)
await session.flush() # persist User before Quota FK
session.add(quota)
await session.commit()
await session.refresh(new_user)
except IntegrityError:
await session.rollback()
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Email or handle already in use",
)
return {
"id": str(new_user.id),
"handle": new_user.handle,
"email": new_user.email,
"role": new_user.role,
"totp_enabled": new_user.totp_enabled,
"created_at": new_user.created_at.isoformat() if new_user.created_at else None,
}
# ── POST /api/auth/login ──────────────────────────────────────────────────────
@router.post("/login")
@limiter.limit("10/minute")
async def login(
request: Request,
body: LoginRequest,
response: Response,
session: AsyncSession = Depends(get_db),
):
"""Authenticate a user and issue tokens.
Per-account rate limiting (SEC-02): checks Redis counter keyed by email
BEFORE any DB lookup to prevent enumeration timing attacks.
Three login flows:
1. No TOTP enabled: password → tokens
2. TOTP enabled, no code provided: requires_totp = True (challenge)
3. TOTP enabled, totp_code provided: verify TOTP → tokens
4. TOTP enabled, backup_code provided (no totp_code): verify backup → tokens
"""
# Per-account rate limiting (SEC-02)
redis_client = request.app.state.redis
rate_key = f"login_attempts:{body.email}"
count = await redis_client.incr(rate_key)
if count == 1:
# Set TTL only on first increment (15-minute window)
await redis_client.expire(rate_key, 900)
if count > 10:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="Too many login attempts. Try again in 15 minutes.",
)
# Look up user by email
result = await session.execute(select(User).where(User.email == str(body.email)))
user: Optional[User] = result.scalar_one_or_none()
# IP extraction for audit log (used in both success and failure paths)
_ip = get_client_ip(request)
# Verify password (anti-enumeration: same error regardless of whether user exists)
if user is None or not auth_service.verify_password(body.password, user.password_hash):
await write_audit_log(
session,
event_type="auth.login_failed",
user_id=user.id if user else None,
actor_id=user.id if user else None,
resource_id=None,
ip_address=_ip,
metadata_={"attempted_email_hash": hashlib.sha256(str(body.email).encode()).hexdigest()[:16]},
)
await session.commit()
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
)
# Active check
if not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Account deactivated",
)
# Password must change: return challenge without issuing tokens (T-02-16)
if user.password_must_change:
return {"requires_password_change": True, "user_id": str(user.id)}
# TOTP second-factor dispatch
if user.totp_enabled:
if body.totp_code is None and body.backup_code is None:
# Challenge: prompt for second factor
return {"requires_totp": True}
if body.totp_code is not None:
# TOTP path takes precedence (even if backup_code also provided)
ok = await auth_service.verify_totp(session, user.id, body.totp_code, redis_client)
if not ok:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect code",
)
else:
# Backup code path (body.backup_code is not None and body.totp_code is None)
ok = await auth_service.verify_backup_code(session, user.id, body.backup_code)
if not ok:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or already used code",
)
# D-13: backup code used event
await write_audit_log(
session,
event_type="auth.backup_code_used",
user_id=user.id,
actor_id=user.id,
resource_id=None,
ip_address=_ip,
)
# Issue tokens
access_token = auth_service.create_access_token(
str(user.id),
user.role,
user_agent=request.headers.get("User-Agent", ""),
accept_lang=request.headers.get("Accept-Language", ""),
)
raw_refresh = await auth_service.create_refresh_token(session, user.id, remember_me=body.remember_me)
_set_refresh_cookie(response, raw_refresh, remember_me=body.remember_me)
# D-13: login success event
await write_audit_log(
session,
event_type="auth.login",
user_id=user.id,
actor_id=user.id,
resource_id=None,
ip_address=_ip,
metadata_={"totp_used": user.totp_enabled and body.totp_code is not None},
)
await session.commit()
return {
"access_token": access_token,
"user": {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
},
}
# ── POST /api/auth/refresh ────────────────────────────────────────────────────
@router.post("/refresh")
@limiter.limit("10/minute")
async def refresh_token(
request: Request,
response: Response,
session: AsyncSession = Depends(get_db),
):
"""Rotate the refresh token.
Reads the refresh_token httpOnly cookie; on success issues a new access
token and rotates the refresh cookie.
On token reuse (revoked token presented), revokes entire family and raises 401.
"""
raw_token = request.cookies.get("refresh_token")
if not raw_token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="No refresh token",
)
try:
new_raw, user_id_str = await auth_service.rotate_refresh_token(session, raw_token)
except ValueError as exc:
if "token_family_revoked" in str(exc):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session revoked",
) from exc
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired refresh token",
) from exc
# Look up user for response body
user = await session.get(User, uuid.UUID(user_id_str))
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found or deactivated",
)
# Set new refresh cookie
_set_refresh_cookie(response, new_raw)
access_token = auth_service.create_access_token(
user_id_str,
user.role,
user_agent=request.headers.get("User-Agent", ""),
accept_lang=request.headers.get("Accept-Language", ""),
)
return {
"access_token": access_token,
"user": {
"id": str(user.id),
"handle": user.handle,
"email": user.email,
"role": user.role,
"totp_enabled": user.totp_enabled,
},
}
# ── POST /api/auth/logout ─────────────────────────────────────────────────────
@router.post("/logout")
async def logout(request: Request, response: Response, session: AsyncSession = Depends(get_db)):
"""Revoke current refresh token and clear the cookie."""
import hashlib as _hashlib
_ip = get_client_ip(request)
raw_token = request.cookies.get("refresh_token")
_logout_user_id = None
if raw_token:
token_hash = _hashlib.sha256(raw_token.encode()).hexdigest()
result = await session.execute(
select(RefreshToken).where(RefreshToken.token_hash == token_hash)
)
row: Optional[RefreshToken] = result.scalar_one_or_none()
if row is not None:
_logout_user_id = row.user_id
row.revoked = True
# D-13: logout event (written before commit, within same transaction)
await write_audit_log(
session,
event_type="auth.logout",
user_id=_logout_user_id,
actor_id=_logout_user_id,
resource_id=None,
ip_address=_ip,
)
await session.commit()
response.delete_cookie("refresh_token", path="/api/auth/refresh")
return {"message": "Logged out"}
# ── POST /api/auth/logout-all ─────────────────────────────────────────────────
@router.post("/logout-all")
async def logout_all(
request: Request,
response: Response,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Sign out of all devices: revoke all refresh tokens for current user."""
_ip = get_client_ip(request)
count = await auth_service.revoke_all_refresh_tokens(session, current_user.id)
# D-13: sign-out-all event
await write_audit_log(
session,
event_type="auth.sign_out_all",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": count},
)
await session.commit()
response.delete_cookie("refresh_token", path="/api/auth/refresh")
return {"message": f"Signed out of {count} session(s)"}
# ── GET /api/auth/me ──────────────────────────────────────────────────────────
@router.get("/me")
async def get_me(current_user: User = Depends(get_current_user)):
"""Return the current user's profile (requires valid Bearer token)."""
return _user_dict(current_user)
# ── GET /api/auth/me/quota ────────────────────────────────────────────────────
@router.get("/me/quota")
async def get_my_quota(
current_user: User = Depends(get_current_user),
session: AsyncSession = Depends(get_db),
):
"""Return the current user's quota usage (STORE-04).
Returns {"used_bytes": int, "limit_bytes": int} for the sidebar quota bar.
Quota row is created at registration (100 MB default — STORE-01).
"""
q = await session.get(Quota, current_user.id)
if q is None:
raise HTTPException(status_code=404, detail="Quota not found")
return {"used_bytes": q.used_bytes, "limit_bytes": q.limit_bytes}
# ── POST /api/auth/change-password ───────────────────────────────────────────
@router.post("/change-password")
async def change_password(
request: Request,
body: ChangePasswordRequest,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update the current user's password.
Checks:
1. current_password matches stored hash
2. new_password has not appeared in HIBP (SEC-03)
3. new_password meets strength requirements (AUTH-01)
"""
# Verify current password
if not auth_service.verify_password(body.current_password, current_user.password_hash):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Current password is incorrect",
)
# HIBP breach check on new password (SEC-03)
if await auth_service.check_hibp(body.new_password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Password strength check
try:
auth_service.validate_password_strength(body.new_password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# Update password
_ip = get_client_ip(request)
user = await session.get(User, current_user.id)
user.password_hash = auth_service.hash_password(body.new_password)
# Revoke other sessions; keep current one alive via skip_token_hash (CR-01)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: password changed event (flush within same transaction before commit)
await write_audit_log(
session,
event_type="auth.password_changed",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-change access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"message": "Password updated", "sessions_revoked": revoked}
# ── Request models for new endpoints ─────────────────────────────────────────
class TotpEnableRequest(BaseModel):
code: str
class PasswordResetRequest(BaseModel):
email: EmailStr
class PasswordResetConfirmRequest(BaseModel):
token: str
new_password: str
# ── GET /api/auth/totp/setup ──────────────────────────────────────────────────
@router.get("/totp/setup")
async def totp_setup(
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Provision a TOTP secret for the current user.
If TOTP is already enabled, returns 400.
Returns { provisioning_uri, secret } — the provisioning_uri is suitable
for QR code generation. The secret is the base32-encoded TOTP secret.
"""
if current_user.totp_enabled:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="TOTP already enabled",
)
secret, provisioning_uri = await auth_service.provision_totp(session, current_user.id)
return {"provisioning_uri": provisioning_uri, "secret": secret}
# ── POST /api/auth/totp/enable ───────────────────────────────────────────────
@router.post("/totp/enable")
@limiter.limit("10/minute")
async def enable_totp(
request: Request,
body: TotpEnableRequest,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Enable TOTP for the current user.
Rate-limited to 10 attempts/minute per IP (SEC-02 / T-02-25).
Verifies the submitted 6-digit code (with Redis replay prevention, AUTH-08).
On success: marks TOTP enabled, generates and returns 10 one-time backup codes.
The backup codes are ONLY returned here — they are stored as Argon2 hashes
in the DB and never returned again (T-02-19).
"""
redis_client = request.app.state.redis
ok = await auth_service.verify_totp(session, current_user.id, body.code, redis_client)
if not ok:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Incorrect or expired code",
)
# Mark TOTP as enabled
user = await session.get(User, current_user.id)
user.totp_enabled = True
await session.flush()
# Generate and store 10 backup codes; return plaintext to user (one-time, T-02-19)
plain_codes = auth_service.generate_backup_codes(10)
await auth_service.store_backup_codes(session, current_user.id, plain_codes)
# Revoke other sessions; keep current one alive via skip_token_hash (CR-02)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: TOTP enrolled event
_ip = get_client_ip(request)
await write_audit_log(
session,
event_type="auth.totp_enrolled",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-enroll access tokens still within their TTL window (T-7.2-01)
await redis_client.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"backup_codes": plain_codes, "sessions_revoked": revoked}
# ── DELETE /api/auth/totp ─────────────────────────────────────────────────────
@router.delete("/totp")
async def disable_totp(
request: Request,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Disable TOTP for the current user.
Clears totp_secret, sets totp_enabled=False, and deletes all backup codes.
"""
_ip = get_client_ip(request)
user = await session.get(User, current_user.id)
user.totp_enabled = False
user.totp_secret = None
# Delete all backup codes for this user (including unused ones)
await session.execute(delete(BackupCode).where(BackupCode.user_id == current_user.id))
# Revoke other sessions; keep current one alive via skip_token_hash (CR-03)
raw_cookie = request.cookies.get("refresh_token")
skip_hash = hashlib.sha256(raw_cookie.encode()).hexdigest() if raw_cookie else None
revoked = await auth_service.revoke_all_refresh_tokens(session, current_user.id, skip_token_hash=skip_hash)
# D-13: TOTP revoked event
await write_audit_log(
session,
event_type="auth.totp_revoked",
user_id=current_user.id,
actor_id=current_user.id,
resource_id=None,
ip_address=_ip,
metadata_={"sessions_revoked": revoked},
)
# Revoke any pre-revoke access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{current_user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
return {"message": "TOTP disabled", "sessions_revoked": revoked}
# ── POST /api/auth/password-reset ─────────────────────────────────────────────
@router.post("/password-reset", status_code=status.HTTP_202_ACCEPTED)
@limiter.limit("5/hour")
async def password_reset_request(
request: Request,
body: PasswordResetRequest,
session: AsyncSession = Depends(get_db),
):
"""Request a password reset email.
Always returns 202 regardless of whether the email exists (anti-enumeration, T-02-22).
If the user is found, a signed reset token (1-hour JWT) is generated and a Celery
task is enqueued to send the email (D-02, D-03).
"""
from sqlalchemy import select as _select # noqa: PLC0415 (already imported above)
result = await session.execute(_select(User).where(User.email == str(body.email)))
user: Optional[User] = result.scalar_one_or_none()
if user is not None:
token = auth_service.create_password_reset_token(str(user.id))
reset_link = f"{settings.frontend_url}/password-reset/confirm?token={token}"
# Deferred import to avoid circular import; Celery task is fire-and-forget
from tasks.email_tasks import send_reset_email # noqa: PLC0415
send_reset_email.delay(user.email, reset_link)
# Always return 202 (anti-enumeration — never reveal whether email exists)
return {
"message": (
"If an account exists for that email, you will receive a reset link shortly."
)
}
# ── POST /api/auth/password-reset/confirm ────────────────────────────────────
@router.post("/password-reset/confirm")
async def password_reset_confirm(
request: Request,
body: PasswordResetConfirmRequest,
session: AsyncSession = Depends(get_db),
):
"""Confirm a password reset using the token from the email link.
Validates the reset token, enforces password strength + HIBP check, updates
the password, and revokes all refresh tokens. Does NOT issue new tokens —
the user must sign in again through /login (AUTH-05, T-02-21).
"""
try:
user_id_str = auth_service.decode_password_reset_token(body.token)
except ValueError:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid or expired reset link",
)
# Password strength validation
try:
auth_service.validate_password_strength(body.new_password)
except ValueError as exc:
raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY, detail=str(exc))
# HIBP breach check (SEC-03)
if await auth_service.check_hibp(body.new_password):
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="This password has appeared in a data breach. Choose a different password.",
)
# Load user
user = await session.get(User, uuid.UUID(user_id_str))
if user is None or not user.is_active:
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail="Invalid or expired reset link",
)
# Update password and revoke all sessions (forces re-auth through TOTP if enabled)
user.password_hash = auth_service.hash_password(body.new_password)
await auth_service.revoke_all_refresh_tokens(session, user.id)
# Revoke any pre-reset access tokens still within their TTL window (T-7.2-01)
await request.app.state.redis.set(
f"user_nbf:{user.id}",
int(time.time()),
ex=settings.access_token_expire_minutes * 60,
)
await session.commit()
# Do NOT issue tokens (AUTH-05 — user must pass TOTP gate on next login)
return {"message": "Password updated. Please sign in."}
# ── Preferences models ────────────────────────────────────────────────────────
class PreferencesUpdate(BaseModel):
"""Request body for PATCH /api/auth/me/preferences.
Validates pdf_open_mode strictly via Literal (T-04-05-05 — no mass assignment).
"""
pdf_open_mode: Literal["in_app", "new_tab"]
# ── GET /api/auth/me/preferences ─────────────────────────────────────────────
@router.get("/me/preferences")
async def get_my_preferences(
current_user: User = Depends(get_current_user),
):
"""Return the current user's PDF open mode preference (D-10).
Both regular users and admins can read their own preferences.
Falls back to 'in_app' if the column is absent (migration not yet run).
"""
try:
pdf_open_mode = current_user.pdf_open_mode
except AttributeError:
pdf_open_mode = "in_app"
return {"pdf_open_mode": pdf_open_mode}
# ── PATCH /api/auth/me/preferences ───────────────────────────────────────────
@router.patch("/me/preferences")
async def update_my_preferences(
body: PreferencesUpdate,
session: AsyncSession = Depends(get_db),
current_user: User = Depends(get_current_user),
):
"""Update the current user's PDF open mode preference (D-10).
Both regular users and admins can update their own preferences.
Pydantic Literal["in_app", "new_tab"] enforces strict allowlist (T-04-05-05).
"""
user = await session.get(User, current_user.id)
if user is None:
raise HTTPException(status_code=404, detail="User not found")
user.pdf_open_mode = body.pdf_open_mode
session.add(user)
await session.commit()
return {"pdf_open_mode": user.pdf_open_mode}