328 lines
10 KiB
Python
328 lines
10 KiB
Python
# Copyright 2026 Umbra Tools (umbratools.com)
|
|
# SPDX-License-Identifier: Apache-2.0
|
|
|
|
"""Single-user authentication for the review UI.
|
|
|
|
Why this exists
|
|
---------------
|
|
Until now the review UI had **no authentication at all** - `active_project` is
|
|
an unsigned browser cookie, and the server binds ``0.0.0.0`` so every page is
|
|
reachable by anything on the LAN (and, via the Caddy reverse proxy, from the
|
|
internet). Spec 21/22 record this as an unauthenticated mutation surface.
|
|
|
|
This is the local stand-in. The intended end state is the controller on the
|
|
``oldbox`` machine authenticating users centrally; until that exists, this gate
|
|
keeps the UI closed. It is deliberately a separate module so it can be
|
|
replaced by controller-delegated identity without touching route code.
|
|
|
|
Design notes
|
|
------------
|
|
* **PBKDF2-HMAC-SHA256**, from :mod:`hashlib` in the stdlib, per the project
|
|
rule that stdlib comes before third-party. ``bcrypt``/``argon2`` would be
|
|
better algorithms but are not installed, and adding a dependency for a
|
|
single-user LAN gate is not worth it.
|
|
* The stored password is a **hash with a random per-user salt**. The bootstrap
|
|
password never lands on disk in plaintext.
|
|
* Sessions are **signed, not encrypted**, and carry no secrets: just the
|
|
username and an expiry. Verification is an HMAC compare against a server
|
|
secret, so a tampered cookie fails closed.
|
|
* Constant-time comparison everywhere (``hmac.compare_digest``).
|
|
* The credential store is written **atomically** (tmp + ``os.replace``) so a
|
|
crash mid-write cannot leave a truncated file that locks the user out.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import contextlib
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
from pathlib import Path
|
|
|
|
USERNAME = "bdumbra"
|
|
BOOTSTRAP_PASSWORD = "First321"
|
|
SESSION_TTL_SECONDS = 14 * 24 * 3600
|
|
PBKDF2_ITERATIONS = 240_000
|
|
MIN_PASSWORD_LENGTH = 7
|
|
|
|
AUTH_FILE_NAME = "auth.json"
|
|
SECRET_FILE_NAME = "session_secret"
|
|
SECRET_BYTES = 32
|
|
# A shorter file is treated as corrupt and regenerated rather than used as a key.
|
|
MIN_SECRET_BYTES = 16
|
|
|
|
|
|
def _store_dir() -> Path:
|
|
"""Where credentials live. Kept beside the app's other state."""
|
|
from umbraaudio.config import PROJECT_BASE
|
|
|
|
d = PROJECT_BASE / ".umbra"
|
|
d.mkdir(parents=True, exist_ok=True)
|
|
return d
|
|
|
|
|
|
def auth_file() -> Path:
|
|
return _store_dir() / AUTH_FILE_NAME
|
|
|
|
|
|
def _secret_file() -> Path:
|
|
return _store_dir() / SECRET_FILE_NAME
|
|
|
|
|
|
# ---------------------------------------------------------------- passwords
|
|
|
|
|
|
def validate_password(password: str) -> str | None:
|
|
"""Return an error message if ``password`` is unacceptable, else None.
|
|
|
|
Rule: **more than six characters, containing at least one letter and at
|
|
least one special (non-alphanumeric) character.**
|
|
"""
|
|
if len(password) <= 6:
|
|
return f"Password must be more than {MIN_PASSWORD_LENGTH - 1} characters."
|
|
if not any(c.isalpha() for c in password):
|
|
return "Password must contain at least one letter."
|
|
if not any(not c.isalnum() for c in password):
|
|
return "Password must contain at least one special character."
|
|
return None
|
|
|
|
|
|
def _hash_password(password: str, salt: bytes) -> str:
|
|
dk = hashlib.pbkdf2_hmac(
|
|
"sha256", password.encode("utf-8"), salt, PBKDF2_ITERATIONS
|
|
)
|
|
return base64.b64encode(dk).decode("ascii")
|
|
|
|
|
|
def _verify_password(password: str, salt_b64: str, expected_b64: str) -> bool:
|
|
try:
|
|
salt = base64.b64decode(salt_b64)
|
|
except Exception:
|
|
return False
|
|
actual = _hash_password(password, salt)
|
|
return hmac.compare_digest(actual, expected_b64)
|
|
|
|
|
|
# ------------------------------------------------------------- credential store
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class Credential:
|
|
username: str
|
|
salt_b64: str
|
|
hash_b64: str
|
|
must_change: bool
|
|
|
|
@property
|
|
def bootstrapped(self) -> bool:
|
|
"""True while the user is still on the one-time first-use password."""
|
|
return self.must_change
|
|
|
|
|
|
def _read_store() -> dict:
|
|
path = auth_file()
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
except (OSError, json.JSONDecodeError):
|
|
return {}
|
|
return data if isinstance(data, dict) else {}
|
|
|
|
|
|
def _write_store(data: dict) -> None:
|
|
"""Atomic write, so an interrupted save cannot lock the user out."""
|
|
path = auth_file()
|
|
tmp = path.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(data, indent=2))
|
|
tmp.chmod(0o600)
|
|
os.replace(tmp, path)
|
|
with contextlib.suppress(OSError):
|
|
path.chmod(0o600)
|
|
|
|
|
|
def ensure_credential() -> Credential:
|
|
"""Load the credential, seeding the one-time first-use password if absent.
|
|
|
|
The bootstrap password is stored **hashed**, and the record is marked
|
|
``must_change`` so the first successful login is forced through a password
|
|
change before anything else is reachable.
|
|
"""
|
|
data = _read_store()
|
|
if data.get("salt") and data.get("hash"):
|
|
return Credential(
|
|
username=data.get("username", USERNAME),
|
|
salt_b64=data["salt"],
|
|
hash_b64=data["hash"],
|
|
must_change=bool(data.get("must_change", False)),
|
|
)
|
|
|
|
salt = secrets.token_bytes(16)
|
|
_write_store(
|
|
{
|
|
"username": USERNAME,
|
|
"salt": base64.b64encode(salt).decode("ascii"),
|
|
"hash": _hash_password(BOOTSTRAP_PASSWORD, salt),
|
|
"must_change": True,
|
|
"created_at": int(time.time()),
|
|
"schema": 1,
|
|
}
|
|
)
|
|
return Credential(
|
|
username=USERNAME,
|
|
salt_b64=base64.b64encode(salt).decode("ascii"),
|
|
hash_b64=_read_store()["hash"],
|
|
must_change=True,
|
|
)
|
|
|
|
|
|
def check_login(username: str, password: str) -> Credential | None:
|
|
"""Return the credential on success, else None.
|
|
|
|
Compares the username with ``compare_digest`` too, so a wrong username and a
|
|
wrong password are indistinguishable from the outside.
|
|
"""
|
|
cred = ensure_credential()
|
|
user_ok = hmac.compare_digest(username or "", cred.username)
|
|
pass_ok = _verify_password(password or "", cred.salt_b64, cred.hash_b64)
|
|
if user_ok and pass_ok:
|
|
return cred
|
|
return None
|
|
|
|
|
|
def set_password(current_password: str, new_password: str) -> str | None:
|
|
"""Change the password after verifying the current one.
|
|
|
|
Returns an error message on failure, else None (and persists the change).
|
|
"""
|
|
cred = ensure_credential()
|
|
if not _verify_password(current_password or "", cred.salt_b64, cred.hash_b64):
|
|
return "Current password is incorrect."
|
|
problem = validate_password(new_password)
|
|
if problem:
|
|
return problem
|
|
if hmac.compare_digest(current_password or "", new_password or ""):
|
|
return "New password must be different from the current one."
|
|
|
|
salt = secrets.token_bytes(16)
|
|
data = _read_store()
|
|
data.update(
|
|
{
|
|
"salt": base64.b64encode(salt).decode("ascii"),
|
|
"hash": _hash_password(new_password, salt),
|
|
"must_change": False,
|
|
"changed_at": int(time.time()),
|
|
}
|
|
)
|
|
_write_store(data)
|
|
return None
|
|
|
|
|
|
def password_state() -> dict:
|
|
"""Non-secret view of the credential, for the UI and for tests."""
|
|
cred = ensure_credential()
|
|
return {
|
|
"username": cred.username,
|
|
"must_change": cred.must_change,
|
|
"on_bootstrap_password": cred.must_change,
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------ sessions
|
|
|
|
|
|
def _session_secret() -> bytes:
|
|
"""Server-side HMAC key, generated once and kept out of the repo."""
|
|
path = _secret_file()
|
|
if path.exists():
|
|
try:
|
|
# Read verbatim. ``bytes.strip()`` removes ASCII whitespace from both
|
|
# ends, so a secret that happened to start or end with one of those
|
|
# six bytes came back one byte shorter on every subsequent call than
|
|
# the value that had just been written - a different key, silently,
|
|
# for ~4% of installs. Tokens signed before that flip could never be
|
|
# verified again.
|
|
data = path.read_bytes()
|
|
if len(data) >= MIN_SECRET_BYTES:
|
|
return data
|
|
except OSError:
|
|
pass
|
|
secret = secrets.token_bytes(SECRET_BYTES)
|
|
path.write_bytes(secret)
|
|
with contextlib.suppress(OSError):
|
|
path.chmod(0o600)
|
|
return secret
|
|
|
|
|
|
def _b64e(raw: bytes) -> str:
|
|
return base64.urlsafe_b64encode(raw).decode("ascii").rstrip("=")
|
|
|
|
|
|
def _b64d(text: str) -> bytes:
|
|
pad = "=" * (-len(text) % 4)
|
|
return base64.urlsafe_b64decode(text + pad)
|
|
|
|
|
|
def issue_session(username: str, ttl: int = SESSION_TTL_SECONDS) -> str:
|
|
"""Mint a signed session token: ``payload.signature``.
|
|
|
|
The payload carries no secret - only the username and an expiry - and is
|
|
authenticated with HMAC-SHA256 over the server secret.
|
|
"""
|
|
payload = json.dumps(
|
|
{"u": username, "exp": int(time.time()) + ttl}, separators=(",", ":")
|
|
)
|
|
body = _b64e(payload.encode("utf-8"))
|
|
sig = _b64e(
|
|
hmac.new(_session_secret(), body.encode("ascii"), hashlib.sha256).digest()
|
|
)
|
|
return f"{body}.{sig}"
|
|
|
|
|
|
def verify_session(token: str | None) -> str | None:
|
|
"""Return the username if ``token`` is a valid, unexpired session, else None."""
|
|
if not token or "." not in token:
|
|
return None
|
|
body, _, sig = token.partition(".")
|
|
expected = _b64e(
|
|
hmac.new(_session_secret(), body.encode("ascii"), hashlib.sha256).digest()
|
|
)
|
|
if not hmac.compare_digest(sig, expected):
|
|
return None
|
|
try:
|
|
data = json.loads(_b64d(body))
|
|
except Exception:
|
|
return None
|
|
if not isinstance(data, dict):
|
|
return None
|
|
exp = data.get("exp")
|
|
if not isinstance(exp, int) or exp < int(time.time()):
|
|
return None
|
|
return data.get("u") or None
|
|
|
|
|
|
def session_max_age() -> int:
|
|
return SESSION_TTL_SECONDS
|
|
|
|
|
|
COOKIE_NAME = "umbra_session"
|
|
|
|
# Public paths reachable without a session. Everything else is gated.
|
|
# Static assets are needed by the login page itself.
|
|
PUBLIC_PATHS = frozenset(
|
|
{
|
|
"/login",
|
|
"/api/login",
|
|
"/api/auth/status",
|
|
"/health",
|
|
}
|
|
)
|
|
PUBLIC_PREFIXES = ("/static/",)
|