🚧 WIP

webui/auth.py

@@ -0,0 +1,174 @@
+"""Authentication for the web UI: password hashing, sessions, throttling.
+
+- Passwords are hashed with argon2id (passlib).
+- Session cookies carry a random token; only its SHA-256 is stored.
+- Sessions are persisted in Postgres, so a login in the web UI is also
+  valid against the API (shared/access.py resolves the same table).
+- Login is throttled per source IP (5 attempts per 15 minutes, then a
+  5-minute lockout) to blunt credential stuffing against the exposed
+  login form.
+- On startup, an admin account is created only if the users table is
+  empty: username/password from ADMIN_USERNAME/ADMIN_PASSWORD if set,
+  otherwise the documented default admin/admin. The default account is
+  flagged must_change_password so the UI forces a change on first login.
+"""
+
+import hmac
+import os
+import secrets
+from datetime import datetime, timedelta, timezone
+from typing import Any, Optional
+
+import psycopg
+from passlib.context import CryptContext
+
+from shared.access import COOKIE_NAME, hash_token
+
+# Fixed session lifetime; a sliding window would need a write per request,
+# so sessions simply expire after this span (users log in again).
+SESSION_TTL = timedelta(hours=12)
+
+LOCKOUT_ATTEMPTS = 5
+LOCKOUT_WINDOW = timedelta(minutes=15)
+LOCKOUT_DURATION = timedelta(minutes=5)
+
+_pwd: Optional[CryptContext] = None
+
+
+def _crypt() -> CryptContext:
+    """Lazy argon2 init: passlib's runtime self-check is slow (~1s)."""
+    global _pwd
+    if _pwd is None:
+        _pwd = CryptContext(schemes=["argon2"])
+    return _pwd
+
+
+def hash_password(password: str) -> str:
+    return _crypt().hash(password)
+
+
+def verify_password(password: str, password_hash: str) -> bool:
+    try:
+        return _crypt().verify(password, password_hash)
+    except Exception:  # noqa: BLE001 - malformed hash must not 500 the login
+        return False
+
+
+def _interval(td: timedelta) -> str:
+    return f"{int(td.total_seconds())} seconds"
+
+
+def _utcnow() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def create_session(conn: psycopg.Connection[Any], user_id: int) -> tuple[str, str]:
+    """Create a session; returns (cookie_token, csrf_token)."""
+    cookie_token = secrets.token_urlsafe(32)
+    csrf_token = secrets.token_urlsafe(32)
+    with conn.cursor() as cur:
+        cur.execute(
+            "INSERT INTO sessions (token_hash, user_id, csrf_token, expires_at)"
+            " VALUES (%s, %s, %s, now() + %s::interval)",
+            (hash_token(cookie_token), user_id, csrf_token, _interval(SESSION_TTL)),
+        )
+    return cookie_token, csrf_token
+
+
+def delete_session(conn: psycopg.Connection[Any], token: str) -> None:
+    with conn.cursor() as cur:
+        cur.execute("DELETE FROM sessions WHERE token_hash = %s", (hash_token(token),))
+
+
+def bootstrap_admin(conn: psycopg.Connection[Any]) -> None:
+    """Create the initial admin from ENV or default admin/admin.
+
+    Runs only when the users table is completely empty — restarting the
+    stack must never reset or resurrect an administrator.
+    """
+    with conn.cursor() as cur:
+        count = cur.execute("SELECT count(*) FROM users").fetchone()[0]
+        if count and count > 0:
+            return
+        username = os.getenv("ADMIN_USERNAME", "admin")
+        password = os.getenv("ADMIN_PASSWORD", "admin")
+        from_env = bool(os.getenv("ADMIN_PASSWORD"))
+        # A default (non-ENV) password must always be changed at first
+        # login; an ENV-provided one is treated as provisioned.
+        must_change = not from_env or username == "admin"
+        cur.execute(
+            "INSERT INTO users (username, password_hash, role, must_change_password)"
+            " VALUES (%s, %s, 'admin', %s)",
+            (username, hash_password(password), must_change),
+        )
+    print(
+        f"Bootstrapped initial admin '{username}'"
+        + (" (default credentials; change on first login)" if must_change else "")
+    )
+
+
+def _attempt_key(ip: str) -> str:
+    # login_attempts.ip is TEXT PRIMARY KEY; keep long IPv6 strings sane.
+    return ip[:64] if ip else "unknown"
+
+
+def check_throttle(conn: psycopg.Connection[Any], ip: str) -> tuple[bool, int]:
+    """Return (allowed, attempts). Locks out after LOCKOUT_ATTEMPTS tries."""
+    key = _attempt_key(ip)
+    with conn.cursor() as cur:
+        row = cur.execute(
+            "SELECT attempts, last_seen FROM login_attempts WHERE ip = %s", (key,)
+        ).fetchone()
+        if row is None:
+            return True, 0
+        attempts, last_seen = row
+        if attempts >= LOCKOUT_ATTEMPTS:
+            if last_seen < _utcnow() - LOCKOUT_DURATION:
+                # Lockout expired: start a fresh window.
+                cur.execute("DELETE FROM login_attempts WHERE ip = %s", (key,))
+                return True, 0
+            return False, attempts
+    return True, attempts
+
+
+def record_failed_login(conn: psycopg.Connection[Any], ip: str) -> None:
+    with conn.cursor() as cur:
+        cur.execute(
+            """
+            INSERT INTO login_attempts (ip, attempts, last_seen)
+            VALUES (%s, 1, now())
+            ON CONFLICT (ip) DO UPDATE
+            SET attempts = CASE
+                    WHEN login_attempts.last_seen < now() - %s::interval
+                    THEN 1
+                    ELSE login_attempts.attempts + 1
+                END,
+                last_seen = now()
+            """,
+            (_attempt_key(ip), _interval(LOCKOUT_WINDOW)),
+        )
+
+
+def clear_failed_logins(conn: psycopg.Connection[Any], ip: str) -> None:
+    with conn.cursor() as cur:
+        cur.execute("DELETE FROM login_attempts WHERE ip = %s", (_attempt_key(ip),))
+
+
+def set_cookie(response: Any, token: str) -> None:
+    """Attach the session cookie (HttpOnly; the webui is same-origin)."""
+    response.set_cookie(
+        COOKIE_NAME,
+        token,
+        httponly=True,
+        samesite="lax",
+        path="/",
+        max_age=int(SESSION_TTL.total_seconds()),
+    )
+
+
+def clear_cookie(response: Any) -> None:
+    response.delete_cookie(COOKIE_NAME, path="/")
+
+
+def constant_time_eq(a: str, b: str) -> bool:
+    return hmac.compare_digest(a.encode(), b.encode())