"""Login / logout / me + user admin API.""" from __future__ import annotations from pathlib import Path import uuid from typing import Any, Optional from fastapi import APIRouter, File, HTTPException, Request, UploadFile from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse from fastapi.templating import Jinja2Templates from pydantic import BaseModel, Field from app import auth_users router = APIRouter(tags=["auth"]) api_router = APIRouter(prefix="/api/auth", tags=["auth"]) users_api = APIRouter(prefix="/api/settings/users", tags=["users"]) BASE_DIR = Path(__file__).resolve().parent.parent.parent templates = Jinja2Templates(directory=str(BASE_DIR / "templates")) AVATAR_DIR = BASE_DIR / "static" / "uploads" / "avatars" AVATAR_DIR.mkdir(parents=True, exist_ok=True) ALLOWED_AVATAR = {".jpg", ".jpeg", ".png", ".webp", ".gif"} async def _save_avatar_file(user_id: int, file: UploadFile) -> str: raw_name = (file.filename or "avatar.jpg").lower() ext = Path(raw_name).suffix if ext not in ALLOWED_AVATAR: raise HTTPException(status_code=400, detail="Alleen JPG, PNG, WEBP of GIF") data = await file.read() if len(data) > 3 * 1024 * 1024: raise HTTPException(status_code=400, detail="Max 3 MB") # strip old files for this user for old in AVATAR_DIR.glob(f"user_{user_id}_*"): try: old.unlink() except OSError: pass fname = f"user_{user_id}_{uuid.uuid4().hex[:10]}{ext}" path = AVATAR_DIR / fname path.write_bytes(data) return f"/static/uploads/avatars/{fname}" class LoginBody(BaseModel): username: str password: str class UserCreateBody(BaseModel): username: str password: str display_name: Optional[str] = None role: str = "user" permissions: dict[str, Any] = Field(default_factory=dict) class UserUpdateBody(BaseModel): display_name: Optional[str] = None role: Optional[str] = None is_active: Optional[bool] = None password: Optional[str] = None permissions: Optional[dict[str, Any]] = None def _session_user(request: Request) -> dict[str, Any] | None: return getattr(request.state, "user", None) @router.get("/login", response_class=HTMLResponse) async def login_page(request: Request, next: str = "/"): if _session_user(request): return RedirectResponse(next or "/", status_code=303) return templates.TemplateResponse( "login.html", {"request": request, "page_title": "Inloggen", "next": next or "/", "error": None}, ) @router.post("/login") async def login_submit(request: Request): form = await request.form() username = str(form.get("username") or "").strip() password = str(form.get("password") or "") next_url = str(form.get("next") or "/") if not next_url.startswith("/"): next_url = "/" user = auth_users.authenticate(username, password) if not user: return templates.TemplateResponse( "login.html", { "request": request, "page_title": "Inloggen", "next": next_url, "error": "Ongeldige gebruikersnaam of wachtwoord", "username": username, }, status_code=401, ) request.session.clear() request.session["user_id"] = user["id"] request.session["username"] = user["username"] return RedirectResponse(next_url, status_code=303) @router.post("/logout") @router.get("/logout") async def logout(request: Request): request.session.clear() return RedirectResponse("/login", status_code=303) @api_router.post("/login") async def api_login(request: Request, body: LoginBody): user = auth_users.authenticate(body.username, body.password) if not user: raise HTTPException(status_code=401, detail="Ongeldige login") request.session.clear() request.session["user_id"] = user["id"] request.session["username"] = user["username"] return {"ok": True, "user": user} @api_router.post("/logout") async def api_logout(request: Request): request.session.clear() return {"ok": True} @api_router.get("/me") async def api_me(request: Request): user = _session_user(request) if not user: raise HTTPException(status_code=401, detail="Niet ingelogd") return { "user": user, "pages": auth_users.PAGE_CATALOG, "can_manage_users": auth_users.can_manage_users(user), } def _require_user_admin(request: Request) -> dict[str, Any]: user = _session_user(request) if not user: raise HTTPException(status_code=401, detail="Niet ingelogd") if not auth_users.can_manage_users(user): raise HTTPException(status_code=403, detail="Geen rechten om gebruikers te beheren") return user @users_api.get("") async def users_list(request: Request): _require_user_admin(request) return {"users": auth_users.list_users(), "pages": auth_users.PAGE_CATALOG} @users_api.post("") async def users_create(request: Request, body: UserCreateBody): _require_user_admin(request) try: user = auth_users.create_user( body.username, body.password, body.display_name, body.role, body.permissions or {} ) except Exception as e: msg = str(e) if "unique" in msg.lower() or "duplicate" in msg.lower(): raise HTTPException(status_code=400, detail="Gebruikersnaam bestaat al") from e raise HTTPException(status_code=400, detail=msg) from e return {"user": user} @users_api.put("/{user_id}") async def users_update(request: Request, user_id: int, body: UserUpdateBody): _require_user_admin(request) try: user = auth_users.update_user( user_id, display_name=body.display_name, role=body.role, is_active=body.is_active, permissions=body.permissions, password=body.password or None, ) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e return {"user": user} @users_api.delete("/{user_id}") async def users_delete(request: Request, user_id: int): actor = _require_user_admin(request) try: auth_users.delete_user(user_id, actor_id=actor["id"]) except ValueError as e: raise HTTPException(status_code=400, detail=str(e)) from e return {"ok": True} @api_router.post("/me/avatar") async def upload_my_avatar(request: Request, file: UploadFile = File(...)): user = _session_user(request) if not user: raise HTTPException(status_code=401, detail="Niet ingelogd") url = await _save_avatar_file(user["id"], file) updated = auth_users.set_avatar(user["id"], url) request.session["username"] = updated["username"] return {"user": updated, "avatar_url": url} @users_api.post("/{user_id}/avatar") async def upload_user_avatar(request: Request, user_id: int, file: UploadFile = File(...)): _require_user_admin(request) url = await _save_avatar_file(user_id, file) updated = auth_users.set_avatar(user_id, url) return {"user": updated, "avatar_url": url} @users_api.delete("/{user_id}/avatar") async def delete_user_avatar(request: Request, user_id: int): actor = _require_user_admin(request) for old in AVATAR_DIR.glob(f"user_{user_id}_*"): try: old.unlink() except OSError: pass updated = auth_users.set_avatar(user_id, None) return {"user": updated}