1from __future__ import annotations23from fastapi import APIRouter, Depends, Request, Response4from findata_core.config import get_settings5from findata_core.db import connection6from pydantic import BaseModel, EmailStr78from findata_api import auth as A9from findata_api.errors import ApiError1011router = APIRouter(prefix="/auth", tags=["auth"])121314class LoginIn(BaseModel):15 email: EmailStr16 password: str171819class CodeIn(BaseModel):20 code: str212223def _set_cookie(resp: Response, sid: str) -> None:24 s = get_settings()25 resp.set_cookie(A.SESSION_COOKIE, sid, httponly=True, secure=s.is_production or s.public_url.startswith("https"),26 samesite="strict", max_age=int(A.SESSION_TTL.total_seconds()), path="/")272829@router.post("/login")30def login(body: LoginIn, request: Request, response: Response):31 with connection() as conn:32 u = conn.execute("SELECT * FROM users WHERE email=%s", (body.email.lower(),)).fetchone()33 ok, _ = __import__("findata_api.ratelimit", fromlist=["bucket"]).bucket.allow(f"login:{A.client_ip(request) or '?'}", 10)34 if not ok:35 raise ApiError(429, "rate_limited", "Too many login attempts")36 if not u or u["disabled"] or not A.verify_password(u["password_hash"], body.password):37 A.audit(request, "auth.login", body.email, outcome="denied")38 raise ApiError(401, "invalid_credentials", "Invalid e-mail or password")39 sid, csrf, mfa_passed = A.create_session(u, A.client_ip(request), request.headers.get("user-agent"))40 _set_cookie(response, sid)41 request.state.principal = A.Principal("user", str(u["id"]), u["email"], u["role"])42 A.audit(request, "auth.login", u["email"], mfa_required=not mfa_passed)43 return {"ok": True, "mfa_required": not mfa_passed, "csrf_token": csrf, "email": u["email"], "role": u["role"]}444546@router.post("/mfa")47def mfa(body: CodeIn, request: Request, p: A.Principal = Depends(A.require_user_session)):48 if p.mfa_passed:49 return {"ok": True, "csrf_token": p.csrf_token}50 if request.headers.get("x-csrf-token") != p.csrf_token:51 raise ApiError(403, "csrf", "Missing or invalid CSRF token")52 if not A.totp_verify(p.id, body.code):53 A.audit(request, "auth.mfa", p.email, outcome="denied")54 raise ApiError(401, "invalid_code", "Invalid one-time code")55 A.mark_mfa_passed(p.session_id)56 A.audit(request, "auth.mfa", p.email)57 return {"ok": True, "csrf_token": p.csrf_token}585960@router.post("/logout")61def logout(request: Request, response: Response, p: A.Principal = Depends(A.require_user_session)):62 A.revoke_session(p.session_id)63 response.delete_cookie(A.SESSION_COOKIE, path="/")64 A.audit(request, "auth.logout", p.email)65 return {"ok": True}666768@router.get("/me")69def me(request: Request, p: A.Principal = Depends(A.get_principal)):70 if p.kind == "anonymous":71 raise ApiError(401, "unauthenticated", "Not signed in")72 totp_enabled = None73 if p.kind == "user":74 with connection() as conn:75 u = conn.execute("SELECT totp_enabled FROM users WHERE id=%s", (p.id,)).fetchone()76 totp_enabled = bool(u and u["totp_enabled"])77 return {"kind": p.kind, "email": p.email, "role": p.role, "scopes": sorted(p.scopes), "mfa_passed": p.mfa_passed,78 "totp_enabled": totp_enabled, "csrf_token": p.csrf_token}798081@router.post("/totp/setup")82def totp_setup(request: Request, p: A.Principal = Depends(A.require("admin"))):83 out = A.totp_setup(p.id, p.email)84 A.audit(request, "auth.totp_setup", p.email)85 return {"otpauth_uri": out["otpauth_uri"], "qr_png_data_url": out["qr_png_data_url"], "secret": out["secret"]}868788@router.post("/totp/verify")89def totp_verify(body: CodeIn, request: Request, p: A.Principal = Depends(A.require("admin"))):90 if not A.totp_verify(p.id, body.code, enable=True):91 raise ApiError(400, "invalid_code", "Code did not match; try again")92 A.audit(request, "auth.totp_enabled", p.email)93 return {"ok": True, "totp_enabled": True}94