ARIA
bbd692ea2e
Add optional login protection for the web UI/API, intended for shared machines (e.g. AI servers). Dual mode: with no users configured the API and web UI are open (as before); once at least one user exists, every /api/* and /ws/* endpoint requires a valid session. - bcrypt password hashing: passwords stored as $2b$ hashes in /etc/nvcurve/users.json (0600, root-owned); plaintext never persisted. - 24-hour sessions: HttpOnly cookie for browsers, Authorization: Bearer token for CLI/scripts; in-memory, invalidated on server restart. - Multi-user: multiple named accounts (no shared-password mode). - New CLI: nvcurve user add|list|remove|set-password (root for mutating ops; password always prompted, never a CLI argument). - New endpoints: GET /api/ping (public), /api/auth/status|login|logout|users. - Web UI: sign-in screen when auth is enabled; status bar shows the signed-in user with sign-out; expired sessions (401) re-show sign-in. - Brute-force lockout: 10 failed logins/IP within 5 min -> 15 min lockout. - New dependency: bcrypt. Also: LSP config (pyrightconfig.json) pointing at the project .venv, and small error-handling cleanups in daemon.py/server.py.
290 lines
9.9 KiB
Python
290 lines
9.9 KiB
Python
"""Authentication: bcrypt user store, 24-hour sessions, dual-mode gating.
|
||
|
||
Dual mode
|
||
---------
|
||
- If the user store file exists and contains at least one user, the API
|
||
requires authentication for all ``/api/*`` and ``/ws/*`` endpoints
|
||
(except the public auth endpoints and ``/api/ping``).
|
||
- If the user store is absent or empty, the server runs with no
|
||
authentication at all — exactly like before.
|
||
|
||
Passwords are hashed with bcrypt (``$2b$``). Plaintext passwords are never
|
||
stored. Sessions are in-memory random 256-bit tokens valid for 24 hours;
|
||
they are presented either as an HttpOnly cookie (browser) or as an
|
||
``Authorization: Bearer <token>`` header (CLI / scripts).
|
||
|
||
A simple per-IP lockout slows down brute-force attempts against login.
|
||
"""
|
||
|
||
import json
|
||
import os
|
||
import secrets
|
||
import time
|
||
from typing import Any
|
||
|
||
import bcrypt
|
||
|
||
# ── Constants ─────────────────────────────────────────────────────────────────
|
||
|
||
SESSION_TTL_S = 24 * 60 * 60 # sessions live for 24 hours
|
||
COOKIE_NAME = "nvcurve_session"
|
||
BCRYPT_ROUNDS = 12
|
||
MAX_PASSWORD_BYTES = 72 # bcrypt only uses the first 72 bytes
|
||
|
||
# Brute-force lockout: MAX_FAILURES within WINDOW_S → locked for LOCKOUT_S.
|
||
_MAX_FAILURES = 10
|
||
_WINDOW_S = 300
|
||
_LOCKOUT_S = 900
|
||
|
||
# ── User store ────────────────────────────────────────────────────────────────
|
||
# File format (JSON): {"users": {"alice": "$2b$12$...", "bob": "$2b$12$..."}}
|
||
|
||
|
||
def _valid_username(name: str) -> bool:
|
||
if not name or len(name) > 32:
|
||
return False
|
||
return all(c.isalnum() or c in "_-." for c in name)
|
||
|
||
|
||
def load_users(path: str) -> dict[str, str]:
|
||
"""Load the user store. Returns {} if the file is absent or unreadable."""
|
||
try:
|
||
with open(path) as f:
|
||
data = json.load(f)
|
||
except (FileNotFoundError, json.JSONDecodeError, OSError):
|
||
return {}
|
||
users = data.get("users", {})
|
||
if not isinstance(users, dict):
|
||
return {}
|
||
return {k: v for k, v in users.items() if isinstance(k, str) and isinstance(v, str)}
|
||
|
||
|
||
def save_users(path: str, users: dict[str, str]) -> None:
|
||
"""Write the user store with restrictive permissions (0600, root-owned)."""
|
||
directory = os.path.dirname(path)
|
||
if directory:
|
||
try:
|
||
os.makedirs(directory, exist_ok=True)
|
||
except OSError as exc:
|
||
raise ValueError(
|
||
f"cannot create user store directory {directory!r}: {exc}"
|
||
) from exc
|
||
try:
|
||
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
|
||
except OSError as exc:
|
||
raise ValueError(f"cannot write user store {path!r}: {exc}") from exc
|
||
with os.fdopen(fd, "w") as f:
|
||
json.dump({"users": users}, f, indent=2)
|
||
f.write("\n")
|
||
|
||
|
||
def auth_enabled(path: str) -> bool:
|
||
"""True if at least one user is configured (→ authentication required)."""
|
||
return bool(load_users(path))
|
||
|
||
|
||
def list_users(path: str) -> list[str]:
|
||
return sorted(load_users(path))
|
||
|
||
|
||
def user_store_state(path: str) -> str:
|
||
"""Report the user store state without requiring read access.
|
||
|
||
Returns one of:
|
||
- "absent": no file (auth disabled)
|
||
- "empty": file exists but has no users (auth disabled)
|
||
- "unreadable": file exists but can't be read (e.g. non-root on a 0600
|
||
root-owned store) — auth is almost certainly enabled
|
||
- "enabled": file readable and has at least one user (auth enabled)
|
||
"""
|
||
if not os.path.exists(path):
|
||
return "absent"
|
||
try:
|
||
with open(path) as f:
|
||
data = json.load(f)
|
||
except (OSError, json.JSONDecodeError):
|
||
return "unreadable"
|
||
users = data.get("users", {}) if isinstance(data, dict) else {}
|
||
if not isinstance(users, dict) or not users:
|
||
return "empty"
|
||
return "enabled"
|
||
|
||
|
||
def add_user(path: str, username: str, password: str) -> None:
|
||
"""Create a new user with a bcrypt-hashed password."""
|
||
if not _valid_username(username):
|
||
raise ValueError(
|
||
f"invalid username {username!r} — use 1–32 chars: letters, digits, '_', '-', '.'"
|
||
)
|
||
if not password:
|
||
raise ValueError("password must not be empty")
|
||
if len(password.encode("utf-8")) > MAX_PASSWORD_BYTES:
|
||
raise ValueError(f"password too long (max {MAX_PASSWORD_BYTES} bytes)")
|
||
users = load_users(path)
|
||
if username in users:
|
||
raise ValueError(f"user {username!r} already exists")
|
||
users[username] = hash_password(password)
|
||
save_users(path, users)
|
||
|
||
|
||
def remove_user(path: str, username: str) -> bool:
|
||
"""Remove a user. Returns False if the user did not exist."""
|
||
users = load_users(path)
|
||
if username not in users:
|
||
return False
|
||
del users[username]
|
||
save_users(path, users)
|
||
return True
|
||
|
||
|
||
def set_password(path: str, username: str, password: str) -> None:
|
||
"""Replace the password of an existing user."""
|
||
if not password:
|
||
raise ValueError("password must not be empty")
|
||
if len(password.encode("utf-8")) > MAX_PASSWORD_BYTES:
|
||
raise ValueError(f"password too long (max {MAX_PASSWORD_BYTES} bytes)")
|
||
users = load_users(path)
|
||
if username not in users:
|
||
raise ValueError(f"user {username!r} does not exist")
|
||
users[username] = hash_password(password)
|
||
save_users(path, users)
|
||
|
||
|
||
# ── Password hashing ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def _pw_bytes(plain: str) -> bytes:
|
||
"""Encode a plaintext password, truncating to bcrypt's 72-byte limit.
|
||
|
||
bcrypt only uses the first 72 bytes and raises ValueError for longer
|
||
input; truncating here keeps the check total (no 500 on an over-long
|
||
login password) and matches what bcrypt does internally.
|
||
"""
|
||
return plain.encode("utf-8")[:MAX_PASSWORD_BYTES]
|
||
|
||
|
||
def hash_password(plain: str) -> str:
|
||
"""Hash a plaintext password with bcrypt ($2b$)."""
|
||
return bcrypt.hashpw(_pw_bytes(plain), bcrypt.gensalt(rounds=BCRYPT_ROUNDS)).decode(
|
||
"ascii"
|
||
)
|
||
|
||
|
||
def verify_password(plain: str, hashed: str) -> bool:
|
||
"""Constant-time check of a plaintext password against a bcrypt hash."""
|
||
try:
|
||
return bcrypt.checkpw(_pw_bytes(plain), hashed.encode("ascii"))
|
||
except (ValueError, TypeError):
|
||
return False
|
||
|
||
|
||
_dummy_hash: bytes | None = None
|
||
|
||
|
||
def check_credentials(users: dict[str, str], username: str, password: str) -> bool:
|
||
"""Verify username+password. Unknown users burn the same bcrypt time as
|
||
known ones so response timing does not reveal which usernames exist."""
|
||
global _dummy_hash
|
||
hashed = users.get(username)
|
||
if hashed is None:
|
||
if _dummy_hash is None:
|
||
_dummy_hash = bcrypt.hashpw(
|
||
b"nvcurve-timing-equalizer", bcrypt.gensalt(rounds=BCRYPT_ROUNDS)
|
||
)
|
||
bcrypt.checkpw(_pw_bytes(password), _dummy_hash)
|
||
return False
|
||
return verify_password(password, hashed)
|
||
|
||
|
||
# ── Sessions (in-memory, 24 h) ────────────────────────────────────────────────
|
||
|
||
_sessions: dict[str, dict[str, Any]] = {} # token -> {"username", "expires_at"}
|
||
|
||
|
||
def create_session(username: str) -> tuple[str, float]:
|
||
"""Create a session. Returns (token, expires_at_unix)."""
|
||
_purge_expired()
|
||
token = secrets.token_urlsafe(32)
|
||
expires_at = time.time() + SESSION_TTL_S
|
||
_sessions[token] = {"username": username, "expires_at": expires_at}
|
||
return token, expires_at
|
||
|
||
|
||
def get_session(token: str | None) -> str | None:
|
||
"""Return the username for a valid, unexpired token — else None."""
|
||
if not token:
|
||
return None
|
||
sess = _sessions.get(token)
|
||
if sess is None:
|
||
return None
|
||
if time.time() > sess["expires_at"]:
|
||
del _sessions[token]
|
||
return None
|
||
return sess["username"]
|
||
|
||
|
||
def get_session_expires_at(token: str | None) -> float | None:
|
||
if not token:
|
||
return None
|
||
sess = _sessions.get(token)
|
||
if sess is None:
|
||
return None
|
||
return sess["expires_at"]
|
||
|
||
|
||
def destroy_session(token: str | None) -> None:
|
||
if token:
|
||
_sessions.pop(token, None)
|
||
|
||
|
||
def _purge_expired() -> None:
|
||
now = time.time()
|
||
for t in [t for t, s in _sessions.items() if now > s["expires_at"]]:
|
||
del _sessions[t]
|
||
|
||
|
||
# ── Brute-force lockout (per client IP) ───────────────────────────────────────
|
||
|
||
_failures: dict[str, list[float]] = {} # ip -> recent failure timestamps
|
||
_lockouts: dict[str, float] = {} # ip -> unix time when lockout lifts
|
||
|
||
|
||
def is_locked_out(ip: str) -> bool:
|
||
until = _lockouts.get(ip, 0.0)
|
||
if until > time.time():
|
||
return True
|
||
if until:
|
||
del _lockouts[ip]
|
||
return False
|
||
|
||
|
||
def record_failure(ip: str) -> None:
|
||
now = time.time()
|
||
recent = [t for t in _failures.get(ip, []) if now - t < _WINDOW_S]
|
||
recent.append(now)
|
||
_failures[ip] = recent
|
||
if len(recent) >= _MAX_FAILURES:
|
||
_lockouts[ip] = now + _LOCKOUT_S
|
||
del _failures[ip]
|
||
|
||
|
||
def clear_failures(ip: str) -> None:
|
||
_failures.pop(ip, None)
|
||
|
||
|
||
# ── Token extraction ──────────────────────────────────────────────────────────
|
||
|
||
|
||
def extract_token(request: Any) -> str | None:
|
||
"""Pull the session token from an Authorization header or the cookie.
|
||
|
||
Works with both starlette Request and WebSocket objects (both expose
|
||
``.headers`` / ``.cookies``).
|
||
"""
|
||
auth_header = request.headers.get("authorization", "")
|
||
if auth_header.lower().startswith("bearer "):
|
||
token = auth_header[7:].strip()
|
||
if token:
|
||
return token
|
||
return request.cookies.get(COOKIE_NAME)
|