4a2afc68bf
Eliminat zgomotul de trasabilitate (US-xxx, PRD x.x, Rn, OV-x, Tn, decizii/naratiune istorica) din 41 fisiere app/ + template-uri. Pastrate comentariile care documenteaza invarianti si logica ne-evidenta (idempotenta/hash, reconciliere anti-duplicat, RAR 500 esec definitiv, creds per cont, WAF User-Agent, 422 fara echo de parola, scope NULL->1), curatate doar de tokeni. Verificare: pentru cele 27 module .py curatate, structura de cod (tokeni non-comentariu/ non-string) e IDENTICA fata de HEAD -> doar comentarii/docstring-uri schimbate. Singura schimbare de cod e in tests/test_web_responsive.py (scos 3 assert pe markeri US-006/007/008, inlocuite de asertiunile structurale alaturate). 0 tokeni US/PRD reziduali in app/. Regresie: 896 passed, 1 deselected. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
100 lines
3.2 KiB
Python
100 lines
3.2 KiB
Python
"""Helper-e sesiune web.
|
|
|
|
Mecanism require_login: NU un dependency FastAPI care intoarce RedirectResponse
|
|
(acela nu scurtcircuiteaza handler-ul — FastAPI continua executia). In schimb:
|
|
- require_login() RIDICA LoginRequired
|
|
- app.main inregistreaza @app.exception_handler(LoginRequired) care intoarce
|
|
RedirectResponse('/login', 303)
|
|
Astfel handler-ul e intrerupt imediat la raise, independent de logica FastAPI.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from starlette.requests import Request
|
|
|
|
from ..config import get_settings
|
|
from ..mapping import DEFAULT_ACCOUNT_ID
|
|
|
|
|
|
class LoginRequired(Exception):
|
|
"""Ridica pentru a redirectiona la /login (prinsa de exception_handler in main.py)."""
|
|
|
|
|
|
class AdminRequired(Exception):
|
|
"""Ridica cand contul sesiunii nu are rol admin (prinsa de exception_handler in main.py)."""
|
|
|
|
|
|
def current_account(request: Request) -> int | None:
|
|
"""account_id din sesiune sau None daca nu e logat."""
|
|
val = request.session.get("account_id")
|
|
return int(val) if val is not None else None
|
|
|
|
|
|
def current_user_id(request: Request) -> int | None:
|
|
"""user_id din sesiune sau None (leaga import_attestations.confirmed_by)."""
|
|
val = request.session.get("user_id")
|
|
return int(val) if val is not None else None
|
|
|
|
|
|
def web_account(request: Request) -> int | None:
|
|
"""account_id pentru rutele web de CITIRE.
|
|
|
|
- sesiune activa -> contul sesiunii
|
|
- fara sesiune + web_auth_required=False (dev) -> DEFAULT_ACCOUNT_ID (cont 1, back-compat)
|
|
- fara sesiune + web_auth_required=True (prod) -> None
|
|
|
|
Rutele de SCRIERE trebuie sa foloseasca require_login() direct, nu web_account(),
|
|
ca sa nu cada niciodata tacit pe contul 1 in prod.
|
|
"""
|
|
aid = current_account(request)
|
|
if aid is not None:
|
|
return aid
|
|
settings = get_settings()
|
|
if not settings.web_auth_required:
|
|
return DEFAULT_ACCOUNT_ID
|
|
return None
|
|
|
|
|
|
def require_login(request: Request) -> int:
|
|
"""Verifica sesiunea activa; ridica LoginRequired daca nu.
|
|
|
|
Intoarce account_id la succes. Aruncatorul (exception_handler din main.py)
|
|
intercepteaza LoginRequired si intoarce RedirectResponse('/login', 303).
|
|
"""
|
|
aid = web_account(request)
|
|
if aid is None:
|
|
raise LoginRequired()
|
|
return aid
|
|
|
|
|
|
def require_admin(request: Request) -> int:
|
|
"""Verifica ca userul logat are rol admin pe contul sesiunii.
|
|
|
|
Intai cheama require_login (nelogat -> LoginRequired -> /login redirect).
|
|
Daca e logat dar nu e admin -> ridica AdminRequired.
|
|
Intoarce account_id la succes.
|
|
"""
|
|
account_id = require_login(request)
|
|
from ..db import get_connection
|
|
from ..users import is_account_admin
|
|
conn = get_connection()
|
|
try:
|
|
admin = is_account_admin(conn, account_id)
|
|
finally:
|
|
conn.close()
|
|
if not admin:
|
|
raise AdminRequired()
|
|
return account_id
|
|
|
|
|
|
def set_session(request: Request, account_id: int, user_id: int) -> None:
|
|
"""Seteaza sesiunea dupa login. Curata mai intai (anti-fixare sesiune)."""
|
|
request.session.clear()
|
|
request.session["account_id"] = account_id
|
|
request.session["user_id"] = user_id
|
|
|
|
|
|
def clear_session(request: Request) -> None:
|
|
"""Sterge sesiunea (logout)."""
|
|
request.session.clear()
|