""" 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)