239 lines
8.5 KiB
Python
239 lines
8.5 KiB
Python
"""
|
|
Authentication: per-user login with signed bearer tokens. Stdlib only — no new deps.
|
|
|
|
Design (deliberately minimal for a ~5-user internal finance tool):
|
|
* Passwords: hashlib.scrypt (OpenSSL), per-user random salt, constant-time compare.
|
|
* Tokens: HMAC-SHA256-signed JSON (user id, username, display name, expiry) — the
|
|
same shape as a JWT but without the dependency. Signed with AR_SECRET_KEY;
|
|
when unset, an ephemeral key is generated and a warning logged (every
|
|
restart then logs everyone out — fine on a laptop, wrong on a server).
|
|
* Enforcement: an HTTP middleware guards every /api/* route except the open set below.
|
|
AR_AUTH=auto (default) requires login as soon as at least one user exists,
|
|
so a fresh dev checkout and the test suite run without ceremony while
|
|
creating the first real user turns authentication on by itself.
|
|
* Identity: the verified display name feeds reviewed_by / approved_by / confirmed_by
|
|
via actor_name(), replacing free-text name fields.
|
|
|
|
Users are created with `python manage.py add-user` — there is no self-signup endpoint.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import hashlib
|
|
import hmac
|
|
import json
|
|
import logging
|
|
import secrets
|
|
import time
|
|
from dataclasses import dataclass
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, Request
|
|
from fastapi.responses import JSONResponse
|
|
from pydantic import BaseModel
|
|
from sqlalchemy.orm import Session as OrmSession
|
|
|
|
from ..config import AUTH_MODE, AUTH_TOKEN_HOURS, SECRET_KEY
|
|
from ..db import models
|
|
from ..db.database import SessionLocal
|
|
from .deps import db_dep
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
router = APIRouter(prefix="/api/auth", tags=["auth"])
|
|
|
|
# Paths reachable without a token: health probes, login itself, and the "is auth on?"
|
|
# check the frontend makes before deciding whether to show the login screen.
|
|
OPEN_PATHS = {"/api/health", "/api/auth/login", "/api/auth/status"}
|
|
|
|
if SECRET_KEY:
|
|
_SECRET = SECRET_KEY.encode()
|
|
else:
|
|
_SECRET = secrets.token_bytes(32)
|
|
logger.warning(
|
|
"AR_SECRET_KEY is not set — using an ephemeral signing key. Login sessions will "
|
|
"not survive a restart. Set AR_SECRET_KEY in production."
|
|
)
|
|
|
|
|
|
# --------------------------------------------------------------------- password hashing
|
|
_SCRYPT_N, _SCRYPT_R, _SCRYPT_P = 16384, 8, 1
|
|
|
|
|
|
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)
|
|
return (f"scrypt${_SCRYPT_N}${_SCRYPT_R}${_SCRYPT_P}"
|
|
f"${salt.hex()}${digest.hex()}")
|
|
|
|
|
|
def verify_password(password: str, stored: str) -> bool:
|
|
try:
|
|
algo, n, r, p, salt_hex, hash_hex = stored.split("$")
|
|
if algo != "scrypt":
|
|
return False
|
|
digest = hashlib.scrypt(password.encode(), salt=bytes.fromhex(salt_hex),
|
|
n=int(n), r=int(r), p=int(p),
|
|
dklen=len(bytes.fromhex(hash_hex)))
|
|
return hmac.compare_digest(digest, bytes.fromhex(hash_hex))
|
|
except (ValueError, TypeError):
|
|
return False
|
|
|
|
|
|
# ----------------------------------------------------------------------------- tokens
|
|
def _b64(data: bytes) -> str:
|
|
return base64.urlsafe_b64encode(data).rstrip(b"=").decode()
|
|
|
|
|
|
def _unb64(data: str) -> bytes:
|
|
return base64.urlsafe_b64decode(data + "=" * (-len(data) % 4))
|
|
|
|
|
|
def create_token(user: models.User) -> str:
|
|
payload = json.dumps({
|
|
"uid": user.id, "u": user.username, "dn": user.display_name,
|
|
"exp": int(time.time()) + AUTH_TOKEN_HOURS * 3600,
|
|
}, separators=(",", ":")).encode()
|
|
sig = hmac.new(_SECRET, payload, hashlib.sha256).digest()
|
|
return f"{_b64(payload)}.{_b64(sig)}"
|
|
|
|
|
|
def parse_token(token: str) -> dict | None:
|
|
"""The signed payload, or None if the token is malformed, forged, or expired."""
|
|
try:
|
|
payload_b64, sig_b64 = token.split(".")
|
|
payload = _unb64(payload_b64)
|
|
expected = hmac.new(_SECRET, payload, hashlib.sha256).digest()
|
|
if not hmac.compare_digest(expected, _unb64(sig_b64)):
|
|
return None
|
|
data = json.loads(payload)
|
|
if data.get("exp", 0) < time.time():
|
|
return None
|
|
return data
|
|
except (ValueError, TypeError, json.JSONDecodeError):
|
|
return None
|
|
|
|
|
|
# ------------------------------------------------------------------------- enforcement
|
|
@dataclass
|
|
class AuthUser:
|
|
id: int
|
|
username: str
|
|
display_name: str
|
|
|
|
|
|
# auto mode asks "do any users exist?" — cached briefly so it isn't a query per request.
|
|
_users_exist_cache: tuple[float, bool] = (0.0, False)
|
|
_USERS_CACHE_TTL_S = 10.0
|
|
|
|
|
|
def _users_exist() -> bool:
|
|
global _users_exist_cache
|
|
ts, val = _users_exist_cache
|
|
now = time.time()
|
|
if now - ts < _USERS_CACHE_TTL_S:
|
|
return val
|
|
db = SessionLocal()
|
|
try:
|
|
val = db.query(models.User.id).filter(
|
|
models.User.is_active == True).first() is not None # noqa: E712
|
|
except Exception: # noqa: BLE001 — table may not exist mid-migration; fail open once
|
|
val = False
|
|
finally:
|
|
db.close()
|
|
_users_exist_cache = (now, val)
|
|
return val
|
|
|
|
|
|
def invalidate_users_cache() -> None:
|
|
global _users_exist_cache
|
|
_users_exist_cache = (0.0, False)
|
|
|
|
|
|
def auth_required() -> bool:
|
|
if AUTH_MODE == "off":
|
|
return False
|
|
if AUTH_MODE == "on":
|
|
return True
|
|
return _users_exist() # auto
|
|
|
|
|
|
def _user_from_request(request: Request) -> AuthUser | None:
|
|
header = request.headers.get("Authorization", "")
|
|
if not header.startswith("Bearer "):
|
|
return None
|
|
data = parse_token(header[7:].strip())
|
|
if data is None:
|
|
return None
|
|
return AuthUser(id=data["uid"], username=data["u"], display_name=data["dn"])
|
|
|
|
|
|
async def auth_middleware(request: Request, call_next):
|
|
"""Guards every /api/* route except OPEN_PATHS. Registered in api/main.py."""
|
|
path = request.url.path.rstrip("/") or "/"
|
|
if path.startswith("/api") and path not in OPEN_PATHS:
|
|
user = _user_from_request(request)
|
|
if user is not None:
|
|
request.state.user = user
|
|
elif auth_required():
|
|
return JSONResponse({"detail": "Not signed in (or the session expired). "
|
|
"Sign in to continue."}, status_code=401)
|
|
else:
|
|
request.state.user = None
|
|
return await call_next(request)
|
|
|
|
|
|
def current_user(request: Request) -> AuthUser | None:
|
|
"""The signed-in user, or None when auth is off/auto-without-users (dev, tests)."""
|
|
return getattr(request.state, "user", None)
|
|
|
|
|
|
def actor_name(request: Request, provided: str = "") -> str:
|
|
"""The name that lands in accountability fields (reviewed_by / approved_by / …).
|
|
|
|
The verified identity always wins; the body-provided name is only honoured when no
|
|
one is signed in (auth off / auto without users), which keeps dev and tests working."""
|
|
user = current_user(request)
|
|
if user is not None and user.display_name:
|
|
return user.display_name
|
|
return (provided or "").strip()
|
|
|
|
|
|
# ----------------------------------------------------------------------------- routes
|
|
class LoginIn(BaseModel):
|
|
username: str
|
|
password: str
|
|
|
|
|
|
@router.get("/status")
|
|
def auth_status() -> dict:
|
|
"""Whether the frontend must show a login screen."""
|
|
return {"auth_required": auth_required()}
|
|
|
|
|
|
@router.post("/login")
|
|
def login(body: LoginIn, db: OrmSession = Depends(db_dep)) -> dict:
|
|
user = db.query(models.User).filter(
|
|
models.User.username == body.username.strip().lower()).first()
|
|
if (user is None or not user.is_active
|
|
or not verify_password(body.password, user.password_hash)):
|
|
# One message for both wrong-user and wrong-password: don't confirm usernames.
|
|
raise HTTPException(401, "Wrong username or password.")
|
|
logger.info("login: %s", user.username)
|
|
return {
|
|
"token": create_token(user),
|
|
"user": {"username": user.username, "display_name": user.display_name},
|
|
"expires_in_hours": AUTH_TOKEN_HOURS,
|
|
}
|
|
|
|
|
|
@router.get("/me")
|
|
def me(request: Request) -> dict:
|
|
user = current_user(request)
|
|
if user is None:
|
|
if auth_required():
|
|
raise HTTPException(401, "Not signed in.")
|
|
return {"authenticated": False, "auth_required": False}
|
|
return {"authenticated": True, "auth_required": True,
|
|
"username": user.username, "display_name": user.display_name}
|