398 lines
16 KiB
Python
398 lines
16 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 datetime as dt
|
|
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, the "is auth on?" check,
|
|
# and the forgot-password code flow (which by definition happens while locked out).
|
|
OPEN_PATHS = {"/api/health", "/api/auth/login", "/api/auth/status",
|
|
"/api/auth/request-code", "/api/auth/verify-code",
|
|
"/api/auth/reset-password"}
|
|
|
|
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"):
|
|
# Identity is attached whenever a valid token is present — including on open
|
|
# paths, so e.g. a signed-in password-code request knows who is asking.
|
|
user = _user_from_request(request)
|
|
request.state.user = user
|
|
if user is None and path not in OPEN_PATHS and auth_required():
|
|
return JSONResponse({"detail": "Not signed in (or the session expired). "
|
|
"Sign in to continue."}, status_code=401)
|
|
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, and whether email codes work."""
|
|
from ..config import email_enabled
|
|
return {"auth_required": auth_required(), "email_enabled": email_enabled()}
|
|
|
|
|
|
@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}
|
|
|
|
|
|
# ------------------------------------------------------------- emailed password codes
|
|
# Usernames ARE email addresses, so the code goes to the account's own address. The code
|
|
# is stored as an HMAC (never plaintext), lives 10 minutes, works once, and the account
|
|
# locks the flow after 5 wrong attempts (request a fresh code to retry).
|
|
CODE_TTL_MINUTES = 10
|
|
CODE_MAX_ATTEMPTS = 5
|
|
_GENERIC_CODE_MSG = ("If that account exists, a code has been emailed to it. "
|
|
"It expires in 10 minutes.")
|
|
|
|
|
|
def _hash_code(code: str) -> str:
|
|
return hmac.new(_SECRET, f"pwcode:{code}".encode(), hashlib.sha256).hexdigest()
|
|
|
|
|
|
class RequestCodeIn(BaseModel):
|
|
username: str = "" # optional when signed in (defaults to the session's account)
|
|
|
|
|
|
@router.post("/request-code")
|
|
def request_password_code(body: RequestCodeIn, request: Request,
|
|
db: OrmSession = Depends(db_dep)) -> dict:
|
|
"""Email a 6-digit password code to the account's address.
|
|
|
|
The response never reveals whether the username exists — same message either way, so
|
|
the login screen can't be used to enumerate accounts."""
|
|
from ..config import email_enabled
|
|
from ..services.mailer import MailerError, send_password_code
|
|
if not email_enabled():
|
|
raise HTTPException(503, "Email is not set up on this server — ask the "
|
|
"administrator to reset your password instead.")
|
|
me_user = current_user(request)
|
|
username = (me_user.username if me_user else body.username).strip().lower()
|
|
if not username:
|
|
raise HTTPException(400, "Enter your username (email address).")
|
|
|
|
user = db.query(models.User).filter(models.User.username == username).first()
|
|
if user is None or not user.is_active:
|
|
logger.info("password code requested for unknown/inactive account: %s", username)
|
|
return {"sent": True, "detail": _GENERIC_CODE_MSG}
|
|
|
|
# Light resend throttle: one code per minute (a resend invalidates the previous code).
|
|
now = dt.datetime.utcnow()
|
|
if user.reset_code_expires:
|
|
issued_at = user.reset_code_expires - dt.timedelta(minutes=CODE_TTL_MINUTES)
|
|
if now - issued_at < dt.timedelta(seconds=60):
|
|
raise HTTPException(429, "A code was just sent — check your inbox, or try "
|
|
"again in a minute.")
|
|
|
|
code = f"{secrets.randbelow(1_000_000):06d}"
|
|
user.reset_code_hash = _hash_code(code)
|
|
user.reset_code_expires = now + dt.timedelta(minutes=CODE_TTL_MINUTES)
|
|
user.reset_code_attempts = 0
|
|
db.commit()
|
|
try:
|
|
send_password_code(user.username, code, CODE_TTL_MINUTES)
|
|
except MailerError as e:
|
|
# Roll the code back — a code nobody received must not stay live.
|
|
user.reset_code_hash = ""
|
|
user.reset_code_expires = None
|
|
db.commit()
|
|
raise HTTPException(502, f"{e} Ask the administrator to reset your password.")
|
|
return {"sent": True, "detail": _GENERIC_CODE_MSG}
|
|
|
|
|
|
def _user_with_valid_code(db: OrmSession, username: str, code: str) -> models.User:
|
|
"""The account IF the code is currently valid — one generic error otherwise (never
|
|
confirms which part was wrong). A wrong code counts toward the attempt lockout."""
|
|
generic = HTTPException(400, "That code is wrong, expired, or already used — "
|
|
"request a fresh one.")
|
|
if not username or not code.strip():
|
|
raise generic
|
|
user = db.query(models.User).filter(models.User.username == username).first()
|
|
now = dt.datetime.utcnow()
|
|
if (user is None or not user.is_active or not user.reset_code_hash
|
|
or not user.reset_code_expires or user.reset_code_expires < now
|
|
or user.reset_code_attempts >= CODE_MAX_ATTEMPTS):
|
|
raise generic
|
|
if not hmac.compare_digest(_hash_code(code.strip()), user.reset_code_hash):
|
|
user.reset_code_attempts += 1
|
|
db.commit()
|
|
raise generic
|
|
return user
|
|
|
|
|
|
class VerifyCodeIn(BaseModel):
|
|
username: str = "" # optional when signed in
|
|
code: str
|
|
|
|
|
|
@router.post("/verify-code")
|
|
def verify_password_code(body: VerifyCodeIn, request: Request,
|
|
db: OrmSession = Depends(db_dep)) -> dict:
|
|
"""Step check for the reset UI: is this code valid? Does NOT consume the code — the
|
|
reset itself re-validates and burns it. Wrong guesses still count toward lockout."""
|
|
me_user = current_user(request)
|
|
username = (me_user.username if me_user else body.username).strip().lower()
|
|
_user_with_valid_code(db, username, body.code)
|
|
return {"valid": True}
|
|
|
|
|
|
class ResetPasswordIn(BaseModel):
|
|
username: str = "" # optional when signed in
|
|
code: str
|
|
new_password: str
|
|
|
|
|
|
@router.post("/reset-password")
|
|
def reset_password_with_code(body: ResetPasswordIn, request: Request,
|
|
db: OrmSession = Depends(db_dep)) -> dict:
|
|
"""Set a new password using the emailed code (works signed-in and from the login
|
|
screen). One generic failure message — never confirms which part was wrong."""
|
|
me_user = current_user(request)
|
|
username = (me_user.username if me_user else body.username).strip().lower()
|
|
if len(body.new_password) < 8:
|
|
raise HTTPException(400, "The new password must be at least 8 characters.")
|
|
user = _user_with_valid_code(db, username, body.code)
|
|
|
|
user.password_hash = hash_password(body.new_password)
|
|
user.reset_code_hash = "" # single use
|
|
user.reset_code_expires = None
|
|
user.reset_code_attempts = 0
|
|
db.commit()
|
|
logger.info("password reset via email code: %s", user.username)
|
|
return {"changed": True}
|
|
|
|
|
|
class ChangePasswordIn(BaseModel):
|
|
current_password: str
|
|
new_password: str
|
|
|
|
|
|
@router.post("/change-password")
|
|
def change_password(body: ChangePasswordIn, request: Request,
|
|
db: OrmSession = Depends(db_dep)) -> dict:
|
|
"""Signed-in users change their own password (admins reset others via manage.py).
|
|
|
|
Requires the current password so a walked-away-from session can't be hijacked into a
|
|
permanent account takeover. Existing tokens stay valid until their normal expiry."""
|
|
user = current_user(request)
|
|
if user is None:
|
|
raise HTTPException(401, "Sign in to change your password.")
|
|
row = db.get(models.User, user.id)
|
|
if row is None or not row.is_active:
|
|
raise HTTPException(401, "Account not found or deactivated.")
|
|
if not verify_password(body.current_password, row.password_hash):
|
|
raise HTTPException(400, "The current password is wrong.")
|
|
if len(body.new_password) < 8:
|
|
raise HTTPException(400, "The new password must be at least 8 characters.")
|
|
if body.new_password == body.current_password:
|
|
raise HTTPException(400, "The new password must be different from the current one.")
|
|
row.password_hash = hash_password(body.new_password)
|
|
db.commit()
|
|
logger.info("password changed: %s", row.username)
|
|
return {"changed": True}
|