from __future__ import annotations import hashlib import hmac import re import secrets import sqlite3 import threading import time from collections import defaultdict, deque from contextlib import closing from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any USERNAME_PATTERN = re.compile(r"^[A-Za-z0-9_.-]{3,64}$") PASSWORD_MIN_LENGTH = 12 PASSWORD_MAX_LENGTH = 256 SESSION_HOURS = 12 def utc_now() -> datetime: return datetime.now(timezone.utc) def iso_utc(value: datetime) -> str: return value.astimezone(timezone.utc).isoformat() def token_digest(value: str) -> str: return hashlib.sha256(value.encode("utf-8")).hexdigest() def derive_password(password: str, salt: bytes) -> bytes: return hashlib.scrypt( password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32, ) def validate_credentials(username: str, password: str) -> None: if not USERNAME_PATTERN.fullmatch(username): raise ValueError("Username must be 3-64 letters, numbers, dots, dashes, or underscores.") if not PASSWORD_MIN_LENGTH <= len(password) <= PASSWORD_MAX_LENGTH: raise ValueError( f"Password must be between {PASSWORD_MIN_LENGTH} and {PASSWORD_MAX_LENGTH} characters." ) class AuthStore: """Local password and server-side session storage. No plaintext secrets are stored.""" def __init__(self, database: Path, session_hours: int = SESSION_HOURS) -> None: self.database = Path(database) self.session_hours = max(1, min(int(session_hours), 168)) self._initialized = False self._init_lock = threading.Lock() def _connect(self) -> sqlite3.Connection: self._ensure_initialized() connection = sqlite3.connect(self.database, timeout=10) connection.row_factory = sqlite3.Row return connection def _ensure_initialized(self) -> None: if self._initialized: return with self._init_lock: if self._initialized: return self.database.parent.mkdir(parents=True, exist_ok=True) with closing(sqlite3.connect(self.database, timeout=10)) as connection: connection.executescript( """ CREATE TABLE IF NOT EXISTS auth_users ( username TEXT PRIMARY KEY, password_hash BLOB NOT NULL, password_salt BLOB NOT NULL, created_at TEXT NOT NULL, updated_at TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS auth_sessions ( token_hash TEXT PRIMARY KEY, username TEXT NOT NULL, csrf_hash TEXT NOT NULL, created_at TEXT NOT NULL, expires_at TEXT NOT NULL, FOREIGN KEY(username) REFERENCES auth_users(username) ON DELETE CASCADE ); CREATE INDEX IF NOT EXISTS auth_sessions_expiry ON auth_sessions(expires_at); """ ) connection.commit() self._initialized = True def configured(self) -> bool: with closing(self._connect()) as connection: row = connection.execute("SELECT 1 FROM auth_users LIMIT 1").fetchone() return row is not None def set_password(self, username: str, password: str) -> None: username = username.strip() validate_credentials(username, password) salt = secrets.token_bytes(16) password_hash = derive_password(password, salt) now = iso_utc(utc_now()) with closing(self._connect()) as connection: connection.execute("BEGIN IMMEDIATE") connection.execute( """ INSERT INTO auth_users(username, password_hash, password_salt, created_at, updated_at) VALUES (?, ?, ?, ?, ?) ON CONFLICT(username) DO UPDATE SET password_hash = excluded.password_hash, password_salt = excluded.password_salt, updated_at = excluded.updated_at """, (username, password_hash, salt, now, now), ) connection.execute("DELETE FROM auth_sessions WHERE username = ?", (username,)) connection.commit() def create_user(self, username: str, password: str) -> None: username = username.strip() validate_credentials(username, password) salt = secrets.token_bytes(16) password_hash = derive_password(password, salt) now = iso_utc(utc_now()) with closing(self._connect()) as connection: try: connection.execute( "INSERT INTO auth_users(username, password_hash, password_salt, created_at, updated_at) VALUES (?, ?, ?, ?, ?)", (username, password_hash, salt, now, now), ) connection.commit() except sqlite3.IntegrityError as exc: raise ValueError("That username is already in use.") from exc def verify_password(self, username: str, password: str) -> bool: if len(password) > PASSWORD_MAX_LENGTH: return False with closing(self._connect()) as connection: row = connection.execute( "SELECT password_hash, password_salt FROM auth_users WHERE username = ?", (username.strip(),), ).fetchone() if row is None: # Keep unknown-user requests computationally similar without exposing user existence. derive_password(password, b"\0" * 16) return False candidate = derive_password(password, bytes(row["password_salt"])) return hmac.compare_digest(candidate, bytes(row["password_hash"])) def create_session(self, username: str) -> dict[str, str]: token = secrets.token_urlsafe(32) csrf = secrets.token_urlsafe(32) csrf_value = token_digest(csrf) now = utc_now() expires = now + timedelta(hours=self.session_hours) with closing(self._connect()) as connection: connection.execute( """ INSERT INTO auth_sessions(token_hash, username, csrf_hash, created_at, expires_at) VALUES (?, ?, ?, ?, ?) """, (token_digest(token), username, csrf_value, iso_utc(now), iso_utc(expires)), ) connection.commit() return { "token": token, "csrf_token": csrf_value, "expires_at": iso_utc(expires), } def get_session(self, token: str | None) -> dict[str, str] | None: if not token or len(token) > 256: return None now = utc_now() with closing(self._connect()) as connection: connection.execute("DELETE FROM auth_sessions WHERE expires_at <= ?", (iso_utc(now),)) row = connection.execute( "SELECT username, csrf_hash, expires_at FROM auth_sessions WHERE token_hash = ?", (token_digest(token),), ).fetchone() connection.commit() if row is None: return None return { "username": str(row["username"]), "csrf_hash": str(row["csrf_hash"]), "expires_at": str(row["expires_at"]), } def valid_csrf(self, session: dict[str, str], csrf_token: str | None) -> bool: if not csrf_token or len(csrf_token) > 256: return False return hmac.compare_digest(session["csrf_hash"], csrf_token) def revoke_session(self, token: str | None) -> None: if not token or len(token) > 256: return with closing(self._connect()) as connection: connection.execute("DELETE FROM auth_sessions WHERE token_hash = ?", (token_digest(token),)) connection.commit() class LoginRateLimiter: """Small in-memory limiter for repeated login failures per client and username.""" def __init__( self, max_failures: int = 5, window_seconds: int = 300, lockout_seconds: int = 900, ) -> None: self.max_failures = max_failures self.window_seconds = window_seconds self.lockout_seconds = lockout_seconds self._failures: dict[str, deque[float]] = defaultdict(deque) self._blocked_until: dict[str, float] = {} self._lock = threading.Lock() @staticmethod def key(client: str, username: str) -> str: return f"{client}|{username.strip().lower()}" def retry_after(self, client: str, username: str, now: float | None = None) -> int: current = time.monotonic() if now is None else now key = self.key(client, username) with self._lock: blocked_until = self._blocked_until.get(key, 0.0) if blocked_until <= current: self._blocked_until.pop(key, None) return 0 return max(1, int(blocked_until - current)) def failure(self, client: str, username: str, now: float | None = None) -> int: current = time.monotonic() if now is None else now key = self.key(client, username) with self._lock: failures = self._failures[key] while failures and failures[0] <= current - self.window_seconds: failures.popleft() failures.append(current) if len(failures) >= self.max_failures: blocked_until = current + self.lockout_seconds self._blocked_until[key] = blocked_until failures.clear() return self.lockout_seconds return 0 def success(self, client: str, username: str) -> None: key = self.key(client, username) with self._lock: self._failures.pop(key, None) self._blocked_until.pop(key, None)