feat: add autonomous fiscal control panel

central/app/human_auth.py

@@ -0,0 +1,75 @@
+"""Separate human authentication backed by hashed passwords and sessions."""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import os
+import secrets
+from datetime import datetime, timedelta, timezone
+
+from fastapi import HTTPException, Request
+from sqlalchemy import select
+
+from .config import settings
+from .models import AdminUser, WebSession
+
+
+def _digest(value: str) -> str:
+    return hashlib.sha256(value.encode()).hexdigest()
+
+
+def hash_password(password: str) -> str:
+    salt = os.urandom(16)
+    encoded = hashlib.pbkdf2_hmac("sha256", password.encode(), salt, 240_000)
+    return "pbkdf2_sha256$240000$%s$%s" % (base64.urlsafe_b64encode(salt).decode(), base64.urlsafe_b64encode(encoded).decode())
+
+
+def verify_password(password: str, stored: str) -> bool:
+    try:
+        algorithm, rounds, salt, digest = stored.split("$", 3)
+        candidate = hashlib.pbkdf2_hmac(algorithm.removeprefix("pbkdf2_"), password.encode(), base64.urlsafe_b64decode(salt), int(rounds))
+        return hmac.compare_digest(base64.urlsafe_b64encode(candidate).decode(), digest)
+    except (ValueError, TypeError):
+        return False
+
+
+def create_admin(session, username: str, password: str):
+    username = username.strip().lower()
+    if len(username) < 3 or len(password) < 12:
+        raise ValueError("username must have 3 characters and password 12 characters")
+    if session.scalar(select(AdminUser).where(AdminUser.username == username)):
+        raise ValueError("username already exists")
+    admin = AdminUser(username=username, password_hash=hash_password(password))
+    session.add(admin); session.commit(); return admin
+
+
+def create_session(session, admin: AdminUser, request: Request):
+    raw_token = secrets.token_urlsafe(32)
+    raw_csrf = secrets.token_urlsafe(32)
+    now = datetime.now(timezone.utc)
+    row = WebSession(admin_user_id=admin.id, token_hash=_digest(raw_token), csrf_hash=_digest(raw_csrf), expires_at=now + timedelta(seconds=settings.web_session_seconds), user_agent=request.headers.get("user-agent", "")[:512], remote_addr=request.client.host if request.client else None)
+    session.add(row); admin.last_login_at = now; session.commit()
+    return raw_token, raw_csrf
+
+
+def current_session(request: Request, session):
+    token = request.cookies.get(settings.web_cookie_name)
+    if not token: return None, None
+    row = session.scalar(select(WebSession).where(WebSession.token_hash == _digest(token)))
+    if not row or row.expires_at.replace(tzinfo=timezone.utc) < datetime.now(timezone.utc): return None, None
+    admin = session.get(AdminUser, row.admin_user_id)
+    if not admin or not admin.is_active: return None, None
+    return row, admin
+
+
+def require_human(request: Request, session):
+    web_session, admin = current_session(request, session)
+    if not admin: raise HTTPException(401, "human login required")
+    return web_session, admin
+
+
+def require_csrf(web_session: WebSession, csrf_token: str | None):
+    if not csrf_token or not hmac.compare_digest(web_session.csrf_hash, _digest(csrf_token)):
+        raise HTTPException(403, "invalid CSRF token")