feat(08-04): create backend/api/admin/ package — users, quotas, ai, shared, init

- users.py: list_users, create_user, update_user_status, initiate_password_reset,
  update_ai_config, delete_user, create_system_topic (7 routes)
- quotas.py: get_user_quota, update_user_quota (2 routes)
- ai.py: get_ai_config_models, test_ai_connection, get_ai_config,
  update_system_ai_config (4 routes); validate_provider_id call-through (D-11)
- shared.py: _user_to_dict helper (T-02-27 / SEC-07 field whitelist)
- __init__.py: router aggregator with prefix=/api/admin; 13 routes total
- admin.py monolith renamed to admin_OLD_REMOVE_IN_TASK_3.py (deleted in task 3)
- All sub-routers have NO prefix (D-04)
- All handlers inject Depends(get_current_admin) (T-08-04-01)
This commit is contained in:
curo1305
2026-06-10 18:43:01 +02:00
parent 5117e2542a
commit 3ec198768d
6 changed files with 1900 additions and 0 deletions
+23
View File
@@ -0,0 +1,23 @@
"""Admin API package — router aggregator.
This file does ONLY router aggregation (Pitfall 2 prevention — T-08-04-04).
No helpers, no models, no logic belong here.
The prefix="/api/admin" is carried here so sub-routers have NO prefix (D-04):
users_router = APIRouter() -> /api/admin/users via parent
quotas_router = APIRouter() -> /api/admin/users/{id}/quota via parent
ai_router = APIRouter() -> /api/admin/ai-config via parent
main.py continues to use:
from api.admin import router as admin_router
app.include_router(admin_router)
"""
from fastapi import APIRouter
from api.admin.users import router as users_router
from api.admin.quotas import router as quotas_router
from api.admin.ai import router as ai_router
router = APIRouter(prefix="/api/admin", tags=["admin"])
router.include_router(users_router)
router.include_router(quotas_router)
router.include_router(ai_router)
+369
View File
@@ -0,0 +1,369 @@
"""Admin AI configuration endpoints.
Handles: get_ai_config_models, test_ai_connection, get_ai_config, update_system_ai_config.
All handlers require get_current_admin (SEC-07, T-08-04-01).
Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
_ai_config_to_dict is local to this module (only ai.py uses it — not in shared.py).
"""
from __future__ import annotations
from typing import Optional
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, ConfigDict, field_validator
from sqlalchemy import 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 SystemSettings, 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, validate_provider_id
from services.audit import write_audit_log
router = APIRouter() # NO prefix — parent __init__.py carries /api/admin (D-04)
# ── Local helper (only ai.py uses this — not in shared.py) ───────────────────
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,
}
# ── Request models ────────────────────────────────────────────────────────────
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:
return validate_provider_id(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:
return validate_provider_id(v)
# ── Endpoints ─────────────────────────────────────────────────────────────────
@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 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)
+115
View File
@@ -0,0 +1,115 @@
"""Admin quota management endpoints.
Handles: get_user_quota, update_user_quota.
All handlers require get_current_admin (SEC-07, T-08-04-01).
Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
"""
from __future__ import annotations
import uuid
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, field_validator
from sqlalchemy.ext.asyncio import AsyncSession
from db.models import Quota, User
from deps.auth import get_current_admin
from deps.db import get_db
from deps.utils import get_client_ip
from services.audit import write_audit_log
from api.admin.shared import _user_to_dict
router = APIRouter() # NO prefix — parent __init__.py carries /api/admin (D-04)
# ── Request models ────────────────────────────────────────────────────────────
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
# ── Endpoints ─────────────────────────────────────────────────────────────────
@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
+28
View File
@@ -0,0 +1,28 @@
"""Shared helpers for the admin API package.
These helpers are used by 2+ admin sub-modules and must not live in __init__.py
to avoid circular imports (T-08-04-04, RESEARCH.md Pitfall 2).
_ai_config_to_dict lives in ai.py (only ai.py uses it — local is correct).
"""
from __future__ import annotations
from db.models import User
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,
}
+431
View File
@@ -0,0 +1,431 @@
"""Admin user-management endpoints.
Handles: list_users, create_user, update_user_status, initiate_password_reset,
update_ai_config (per-user), delete_user, create_system_topic.
All handlers require get_current_admin (SEC-07, T-08-04-01).
Sub-router has NO prefix — parent __init__.py carries /api/admin (D-04).
"""
from __future__ import annotations
import time
import uuid
from typing import Optional
from config import settings
from fastapi import APIRouter, Depends, HTTPException, Request, status
from pydantic import BaseModel, EmailStr, Field, field_validator
from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession
from db.models import CloudConnection, Document, Quota, RefreshToken, Topic, User
from deps.auth import get_current_admin
from deps.db import get_db
from deps.utils import get_client_ip
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
from api.admin.shared import _user_to_dict
router = APIRouter() # NO prefix — parent __init__.py carries /api/admin (D-04)
# ── Constants ─────────────────────────────────────────────────────────────────
_DEFAULT_QUOTA_BYTES = 104857600 # 100 MB free-tier default (D-06)
# ── 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 UserAiConfigUpdate(BaseModel):
ai_provider: Optional[str] = None
ai_model: Optional[str] = None
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)
# ── 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.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
+934
View File
@@ -0,0 +1,934 @@
"""
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)