kyleandClaude Opus 5.5
d4e821799c
Admins (Microsoft 365 sign-in names in BACKUP_DASH_ADMINS) get an Admin link to /admin.html, where they can create, list and revoke the API keys collectors use, instead of running manage.py on the server. Backend - /api/v1/admin/keys (list, create, revoke), admins only (403 otherwise); the user comes from the headers nginx sets after sign-in - Changes need an X-Requested-By header, which another site's page can't add, so a signed-in admin's browser can't be used to forge them - A new key is returned once; only its hash is stored. Names are unique for good, so revoked keys' history stays unambiguous - Keys record who created/revoked them and the company they last reported for (schema v2; databases migrate automatically on start) - /api/v1/me reports is_admin; manage.py shares the key code and records itself as creator/revoker Frontend - admin.html / admin.js: name validation shown after interaction (:user-invalid, aria-invalid kept in sync), the one-time key with a copy button (falls back to selecting it if clipboard access is refused), a key table flagging keys quiet for over 24 hours, and revoking confirmed in a <dialog>; results announced in a status region - Shared el() and signed-in helpers moved to dom.js / signed-in.js - Mock mode (?mock=1) includes sample keys Tested: 15 new backend tests (access, forged requests, names, duplicates, revoking, last-used company, v1 -> v2 upgrade); in the browser, every admin flow in mock mode and at phone width; end to end through the nginx config with the real API (a key created on the API works for ingest; revoking it makes ingest 401; non-admins see an explanation and no Admin link). Co-Authored-By: Claude Opus 5.5 <[email protected]>
74 lines
2.7 KiB
Python
74 lines
2.7 KiB
Python
"""Authentication and authorisation.
|
|
|
|
Collectors (machines) authenticate with API keys, checked here. Dashboard *users* sign in with
|
|
Microsoft 365 in nginx (via oauth2-proxy), which then tells the app who they are in
|
|
X-Forwarded-* headers it sets itself (a browser can't supply them). The app only decides whether
|
|
that user is an admin.
|
|
"""
|
|
|
|
import hashlib
|
|
import secrets
|
|
import sqlite3
|
|
|
|
from fastapi import Depends, Header, HTTPException, Request
|
|
|
|
from .db import get_db
|
|
from .timeutil import to_db, utcnow
|
|
|
|
KEY_PREFIX = "bdk_"
|
|
|
|
|
|
def generate_key() -> str:
|
|
return KEY_PREFIX + secrets.token_urlsafe(32)
|
|
|
|
|
|
def hash_key(key: str) -> str:
|
|
# Keys are 256-bit random, so a fast unsalted hash is fine (no dictionary to attack).
|
|
return hashlib.sha256(key.encode()).hexdigest()
|
|
|
|
|
|
def current_user(request: Request) -> str | None:
|
|
"""The signed-in user's sign-in name, lowercase; None without Microsoft sign-in (e.g. dev)."""
|
|
user = request.headers.get("x-forwarded-email") or request.headers.get("x-forwarded-preferred-username")
|
|
return user.strip().lower() if user and user.strip() else None
|
|
|
|
|
|
def is_admin(request: Request) -> bool:
|
|
user = current_user(request)
|
|
return user is not None and user in request.app.state.settings.admins
|
|
|
|
|
|
def require_admin(request: Request) -> str:
|
|
if not is_admin(request):
|
|
raise HTTPException(403, "Admins only. Admins are listed in BACKUP_DASH_ADMINS on the server.")
|
|
return current_user(request)
|
|
|
|
|
|
# Changes made from the admin page must carry this header. A page on another site can't add a
|
|
# custom header to a request to the dashboard (the browser would first ask the API, which allows
|
|
# no other origins), so a signed-in admin's browser can't be used to forge them.
|
|
PAGE_HEADER = "X-Requested-By"
|
|
PAGE_HEADER_VALUE = "backup-dashboard"
|
|
|
|
|
|
def require_page_request(x_requested_by: str | None = Header(None)) -> None:
|
|
if x_requested_by != PAGE_HEADER_VALUE:
|
|
raise HTTPException(403, f"Missing {PAGE_HEADER} header")
|
|
|
|
|
|
def require_ingest_key(
|
|
authorization: str | None = Header(None),
|
|
conn: sqlite3.Connection = Depends(get_db),
|
|
) -> str:
|
|
scheme, _, token = (authorization or "").partition(" ")
|
|
token = token.strip()
|
|
if scheme.lower() != "bearer" or not token:
|
|
raise HTTPException(401, "Missing bearer token", headers={"WWW-Authenticate": "Bearer"})
|
|
row = conn.execute(
|
|
"SELECT id, name FROM api_keys WHERE key_hash = ? AND revoked_at IS NULL", (hash_key(token),)
|
|
).fetchone()
|
|
if row is None:
|
|
raise HTTPException(401, "Invalid or revoked API key", headers={"WWW-Authenticate": "Bearer"})
|
|
conn.execute("UPDATE api_keys SET last_used_at = ? WHERE id = ?", (to_db(utcnow()), row["id"]))
|
|
return row["name"]
|