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) _DEFAULT_QUOTA_BYTES = 104857600 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): name: str description: str = "" color: str = "#6366f1" class UserDeleteConfirm(BaseModel): admin_password: str = Field(..., min_length=1) async def _get_user_or_404(session: AsyncSession, user_id: uuid.UUID) -> User: user = await session.get(User, user_id) if user is None: raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="User not found") return user @router.get("/users") async def list_users( session: AsyncSession = Depends(get_db), _admin: User = Depends(get_current_admin), ) -> dict: 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: 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, ) 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() _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: user = await _get_user_or_404(session, user_id) 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: await revoke_all_refresh_tokens(session, user.id) await request.app.state.redis.set( f"user_nbf:{user.id}", int(time.time()), ex=settings.access_token_expire_minutes * 60, ) session.add(user) _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: user = await _get_user_or_404(session, user_id) 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: user = await _get_user_or_404(session, user_id) _ip_addr = get_client_ip(request) user.ai_provider = body.ai_provider user.ai_model = body.ai_model session.add(user) 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: if not verify_password(body.admin_password, _admin.password_hash): raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="Invalid admin password", ) user = await _get_user_or_404(session, user_id) if user.role == "admin": raise HTTPException( status_code=status.HTTP_400_BAD_REQUEST, detail="Cannot delete admin accounts", ) _ip_addr = get_client_ip(request) 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: 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 await session.delete(conn) if cloud_conns: await session.flush() 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]}, ) 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 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() 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: from services import storage # noqa: PLC0415 topic = await storage.create_topic( session, body.name, body.description, body.color, user_id=None ) return topic