"""Single-user login gate. The app sits on a public hostname, so it needs a door. One shared password from the environment buys a signed, expiring cookie — no user table, no password reset flow, nothing to administer. Set PARTS_AUTH=off for LAN-only use. """ import base64 import hashlib import hmac import os import secrets import threading import time from fastapi import Cookie, Depends, HTTPException, Request from . import db COOKIE_NAME = "parts_session" DIGEST_SIZE = hashlib.sha256().digest_size def _env(name: str, default: str = "") -> str: return os.environ.get(name, default).strip() def _env_int(name: str, default: int, minimum: int = 1) -> int: try: return max(minimum, int(_env(name, str(default)))) except ValueError: return default def auth_enabled() -> bool: return _env("PARTS_AUTH", "on").lower() not in ("off", "0", "false", "no") def _secret() -> bytes: secret = _env("PARTS_SECRET") if not secret: # Without a configured secret, derive one from the password so sessions # are still unforgeable — they just don't survive a password change. secret = "fallback:" + _env("PARTS_PASSWORD") return hashlib.sha256(secret.encode()).digest() def session_days() -> int: return _env_int("PARTS_SESSION_DAYS", 30) # --- stored password -------------------------------------------------------- # # The password lives in the database, not the environment, so it can be changed # from the UI and survives a container rebuild (it is in the /data volume). # PARTS_PASSWORD is only the bootstrap credential: it works until a password has # been set through the app, and is ignored from then on — otherwise "changing" # the password would leave the old one working. PASSWORD_KEY = "password_hash" EPOCH_KEY = "session_epoch" MIN_PASSWORD_LENGTH = 8 _SCRYPT_N = 1 << 14 _SCRYPT_R = 8 _SCRYPT_P = 1 _SCRYPT_MAXMEM = 64 * 1024 * 1024 def hash_password(password: str) -> str: salt = secrets.token_bytes(16) digest = hashlib.scrypt( password.encode(), salt=salt, n=_SCRYPT_N, r=_SCRYPT_R, p=_SCRYPT_P, dklen=32, maxmem=_SCRYPT_MAXMEM, ) return "scrypt${}${}${}${}${}".format( _SCRYPT_N, _SCRYPT_R, _SCRYPT_P, base64.b64encode(salt).decode(), base64.b64encode(digest).decode(), ) def verify_hash(password: str, stored: str) -> bool: try: scheme, n, r, pp, salt_b64, digest_b64 = stored.split("$") if scheme != "scrypt": return False digest = hashlib.scrypt( password.encode(), salt=base64.b64decode(salt_b64), n=int(n), r=int(r), p=int(pp), dklen=len(base64.b64decode(digest_b64)), maxmem=_SCRYPT_MAXMEM, ) return hmac.compare_digest(digest, base64.b64decode(digest_b64)) except Exception: return False def stored_hash(conn) -> str | None: return db.get_setting(conn, PASSWORD_KEY) def using_bootstrap_password(conn) -> bool: """True while the env-provided password is still the live one.""" return stored_hash(conn) is None def check_password(conn, candidate: str) -> bool: stored = stored_hash(conn) if stored: return verify_hash(candidate, stored) expected = _env("PARTS_PASSWORD") if not expected: return False return hmac.compare_digest(candidate.encode(), expected.encode()) def set_password(conn, new_password: str): """Store a new password and invalidate every outstanding session.""" db.set_setting(conn, PASSWORD_KEY, hash_password(new_password)) bump_epoch(conn) def password_problem(new_password: str) -> str | None: if len(new_password) < MIN_PASSWORD_LENGTH: return f"Password must be at least {MIN_PASSWORD_LENGTH} characters" if not new_password.strip(): return "Password must not be blank" return None # --- session epoch ---------------------------------------------------------- # # Baked into every token. Bumping it invalidates all outstanding cookies at # once, which is what makes "change my password" and "sign out everywhere" # work without having to rotate PARTS_SECRET by hand on the host. def current_epoch(conn) -> int: try: return int(db.get_setting(conn, EPOCH_KEY, "0") or "0") except ValueError: return 0 def bump_epoch(conn) -> int: epoch = current_epoch(conn) + 1 db.set_setting(conn, EPOCH_KEY, epoch) return epoch # --- session tokens --------------------------------------------------------- # # The signature is raw HMAC bytes, which can contain any byte value including # the one for ".". An earlier version joined payload and signature with "." and # split on the last occurrence, so roughly 12% of issued tokens (1 - (255/256)^32) # split in the wrong place and failed their own validator. The digest is a fixed # 32 bytes, so slice by length instead of looking for a delimiter. def issue_token(conn) -> str: expires = int(time.time()) + session_days() * 86400 payload = f"{expires}:{current_epoch(conn)}".encode() sig = hmac.new(_secret(), payload, hashlib.sha256).digest() return base64.urlsafe_b64encode(payload + sig).decode() def token_valid(token: str, conn=None) -> bool: try: raw = base64.urlsafe_b64decode(token.encode()) if len(raw) <= DIGEST_SIZE: return False payload, sig = raw[:-DIGEST_SIZE], raw[-DIGEST_SIZE:] expected = hmac.new(_secret(), payload, hashlib.sha256).digest() if not hmac.compare_digest(sig, expected): return False expires_s, _, epoch_s = payload.decode().partition(":") if int(expires_s) <= time.time(): return False if conn is not None and int(epoch_s or 0) != current_epoch(conn): return False return True except Exception: return False # --- failed-login throttling ------------------------------------------------ # # Deliberately a GLOBAL window rather than per-IP. Caddy appends to # X-Forwarded-For rather than replacing it, so the client-supplied end of that # header is attacker-controlled and a per-IP bucket would be trivially evaded by # forging it. There is exactly one legitimate user, so a global cap costs # nothing in practice and cannot be side-stepped by rotating source addresses. # # The trade-off: someone spraying guesses can lock the login form for the # cooldown. Existing sessions keep working (they carry a valid cookie), so that # is an annoyance rather than a lockout. _failures: list[float] = [] _failures_lock = threading.Lock() def _window() -> int: return _env_int("PARTS_LOGIN_WINDOW", 300) def _max_failures() -> int: return _env_int("PARTS_LOGIN_MAX_FAILURES", 10) def login_retry_after() -> int: """Seconds until the next login attempt is allowed, or 0 if allowed now.""" now = time.time() window = _window() with _failures_lock: recent = [t for t in _failures if now - t < window] _failures[:] = recent if len(recent) < _max_failures(): return 0 return max(1, int(window - (now - recent[0]))) def record_failure(): with _failures_lock: _failures.append(time.time()) def clear_failures(): with _failures_lock: _failures.clear() # --- request helpers -------------------------------------------------------- def is_https(request: Request) -> bool: """Whether the *original* client request used TLS. Caddy terminates TLS and proxies to this app over plain HTTP, so `request.url.scheme` is always "http" in production and can't decide whether the session cookie may carry the Secure flag. X-Forwarded-Proto is what the proxy records the real scheme in. """ override = _env("PARTS_SECURE_COOKIE", "auto").lower() if override in ("on", "1", "true", "yes"): return True if override in ("off", "0", "false", "no"): return False forwarded = request.headers.get("x-forwarded-proto", "") if forwarded: return forwarded.split(",")[0].strip().lower() == "https" return request.url.scheme == "https" def require_auth(parts_session: str | None = Cookie(default=None), conn=Depends(db.get_db)): """FastAPI dependency guarding every data route. Takes the same request-scoped connection the route uses (FastAPI caches dependency results per request), so checking the session epoch costs one indexed lookup rather than a second connection. """ if not auth_enabled(): return True if parts_session and token_valid(parts_session, conn): return True raise HTTPException(status_code=401, detail="Not authenticated")