# backend/operator_auth.py """Operator authentication: the deny-by-default gate, session cookie, login + lockout, and the small data helpers shared by the routes and the CLI. Reuses the argon2 hasher (auth_passwords) and the HMAC signer (auth_cookies). The flag and SessionLocal are read as module globals at call time so tests can monkeypatch them.""" import os import time import uuid from datetime import datetime, timedelta from backend.models import OperatorUser from backend.auth_passwords import hash_password, verify_password, generate_password from backend.auth_cookies import sign, read, COOKIE_SECURE # Feature flag — OFF by default. When off, the gate and require_role both pass # everything through and the app behaves exactly as it does today. OPERATOR_AUTH_ENABLED = os.getenv("OPERATOR_AUTH_ENABLED", "false").lower() in ("1", "true", "yes") COOKIE_NAME = "tv_session" COOKIE_MAX_AGE = 60 * 60 * 24 * 30 # 30 days ("remember this device") MAX_LOGIN_FAILURES = 5 LOCK_MINUTES = 15 # Role ladder — a rank map so checks read naturally and 'operator' slots in later. _ROLE_RANK = {"operator": 10, "admin": 20, "superadmin": 30} def role_at_least(role: str, minimum: str) -> bool: """True iff `role` ranks at or above `minimum`. Unknown roles rank as 0.""" return _ROLE_RANK.get(role, 0) >= _ROLE_RANK[minimum] def _norm_email(email: str) -> str: return (email or "").strip().lower() def make_operator_cookie(uid: str, iat: int = None) -> str: """Sign a tv_session value for a user id. iat defaults to now; pass an explicit iat when you bump sessions_valid_from to that same instant (change-password).""" return sign({"uid": uid, "iat": int(iat if iat is not None else time.time())}) def current_operator(request, db): """Resolve the OperatorUser for a request's tv_session cookie, or None. Re-validated against the DB every call: a disabled / locked / password-changed user drops on the next request. Used by the gate middleware (with its own session) — does not raise.""" data = read(request.cookies.get(COOKIE_NAME), COOKIE_MAX_AGE) if not data: return None uid, iat = data.get("uid"), data.get("iat") if not uid or not isinstance(iat, (int, float)): return None user = db.query(OperatorUser).filter_by(id=uid).first() if not user or not user.active: return None if user.locked_until and user.locked_until > datetime.utcnow(): return None if user.sessions_valid_from and datetime.utcfromtimestamp(int(iat)) < user.sessions_valid_from: return None return user