""" 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 uuid from datetime import datetime from typing import Optional 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) 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)