jobe/nodeledge-mqtt-listener
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

🚧 WIP

Browse Source
This commit is contained in:
Jonas Bengtsson
2026-09-05 14:22:29 +02:00
parent f9d10364f8
commit 2117f96ae7
20 changed files with 1721 additions and 112 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/app.py
+638 -27
View File
@@ -12,10 +12,17 @@ resolves each device's TIME-VARYING attributes (offsets etc., from the
imei_attributes timeline) as of the message's device time and returns
them as a sibling "attributes" field.
Authentication: app tokens stored as SHA-256 digests in the app_tokens
table, passed either as the `X-App-Token` header or as an `app_token` query
parameter. Digests are compared in constant time; invalid tokens yield 401.
When the database is unavailable, authenticated endpoints return 503.
Authentication is layered:
- app tokens (SHA-256 digests in app_tokens, X-App-Token header) — machine
clients with unrestricted access
- interactive users (scrypt password hashes in the users table) logging
in via POST /auth/login to get an opaque Bearer session token; only its
SHA-256 digest is stored in sessions. Roles: admin (manages everything
incl. users), superuser (configures devices/configs/tokens and manages
day-to-day 'user' accounts), user (read-only viewer). App tokens keep
unrestricted access; session users carry their role per request and
writes are gated by require_level(). When the database is unavailable,
authenticated endpoints return 503.
"""
from __future__ import annotations
@@ -24,8 +31,10 @@ import hashlib
import hmac
import logging
import os
import secrets
import threading
from datetime import datetime, timezone
import time
from datetime import datetime, timedelta, timezone
from typing import Any
import psycopg
@@ -37,6 +46,40 @@ from listener.process import normalize_imei, validate_imei_doc, validate_imei_ma
logger = logging.getLogger("api")
# --- roles ---------------------------------------------------------------
# Privilege levels: user (read-only) < superuser (configure + manage 'user'
# accounts) < admin (everything). App tokens are unrestricted.
ROLE_LEVELS = {"user": 0, "superuser": 1, "admin": 2}
# Login brute-force protection (in-process, per username).
LOCKOUT_ATTEMPTS = 5
LOCKOUT_MINUTES = 15
# Session lifetime.
SESSION_TTL_DAYS = 30
def hash_password(password: str, salt: bytes | None = None) -> str:
"""scrypt password hash, stored as 'scrypt$<salthex>$<keyhex>'.
Only hashes are persisted; plaintext passwords never are."""
salt = salt if salt is not None else secrets.token_bytes(16)
key = hashlib.scrypt(
password.encode("utf-8"), salt=salt, n=2**14, r=8, p=1, dklen=32
)
return f"scrypt${salt.hex()}${key.hex()}"
def verify_password(password: str, stored: str) -> bool:
"""Constant-time check of a password against a stored scrypt hash."""
try:
algo, salt_hex, _key_hex = stored.split("$")
if algo != "scrypt":
return False
candidate = hash_password(password, salt=bytes.fromhex(salt_hex))
return hmac.compare_digest(candidate, stored)
except (ValueError, TypeError):
return False
def hash_token(token: str) -> str:
"""SHA-256 hex digest of an app token. Only digests are stored in the
@@ -395,6 +438,352 @@ class ImeiAttributesStore:
conn.close()
class UserStore:
"""Interactive users backed by the users table (scrypt hashes).
The first admin is seeded lazily from BOOTSTRAP_ADMIN_USER/
BOOTSTRAP_ADMIN_PASSWORD when the table is empty (same bootstrap
pattern as app_tokens). Sessions join users for the CURRENT role on
every request, so role changes apply immediately.
Capability caps are enforced HERE (not just in the endpoints), because
the stores are the single choke point every request funnels through:
- admin: any operation on any account
- superuser: create accounts, but only with role 'user'; modify/delete
only accounts whose current role is 'user'
- user: no user management
"""
def __init__(
self,
dsn: str,
bootstrap_user: str | None = None,
bootstrap_password: str | None = None,
) -> None:
self._dsn = dsn
self._bootstrap_user = bootstrap_user
self._bootstrap_password = bootstrap_password
self._bootstrap_lock = threading.Lock()
self._bootstrapped = False
def _ensure_bootstrapped(self) -> None:
"""Seed the first admin once, if configured and the table is empty.
Lazy like BOOTSTRAP_APP_TOKEN: a db outage at startup must not
block startup; a failed attempt retries on the next request."""
if self._bootstrapped:
return
with self._bootstrap_lock:
if self._bootstrapped:
return
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
if self._bootstrap_user and self._bootstrap_password:
cur.execute(
"""
INSERT INTO users (username, password_hash, role)
SELECT %s, %s, 'admin'
WHERE NOT EXISTS (SELECT 1 FROM users)
ON CONFLICT DO NOTHING
""",
(self._bootstrap_user, hash_password(self._bootstrap_password)),
)
if cur.rowcount:
logger.info(
"seeded bootstrap admin %r from BOOTSTRAP_ADMIN_USER",
self._bootstrap_user,
)
else:
cur.execute("SELECT count(*) FROM users")
if cur.fetchone()[0] == 0:
logger.warning(
"users is empty and BOOTSTRAP_ADMIN_USER/"
"BOOTSTRAP_ADMIN_PASSWORD are not set; the UI "
"cannot log in until a user is added"
)
finally:
conn.close()
self._bootstrapped = True
def verify(self, username: str, password: str) -> str | None:
"""Return the role on success, else None. Unknown usernames burn
the same scrypt time as wrong passwords (timing equalization)."""
self._ensure_bootstrapped()
dummy = hash_password("timing-equalizer")
conn = psycopg.connect(self._dsn)
try:
with conn.cursor() as cur:
cur.execute(
"SELECT role, password_hash FROM users WHERE username = %s",
(username,),
)
row = cur.fetchone()
if row is None:
verify_password(password, dummy) # equalize timing
return None
role, stored = row
if not verify_password(password, stored):
return None
return role
finally:
conn.close()
def list(self) -> list[dict[str, Any]]:
"""All users as [{username, role}] (passwords are unrecoverable)."""
self._ensure_bootstrapped()
conn = psycopg.connect(self._dsn)
try:
with conn.cursor() as cur:
cur.execute("SELECT username, role FROM users ORDER BY username")
return [
{"username": row[0], "role": row[1]} for row in cur.fetchall()
]
finally:
conn.close()
def get_role(self, username: str) -> str | None:
conn = psycopg.connect(self._dsn)
try:
with conn.cursor() as cur:
cur.execute("SELECT role FROM users WHERE username = %s", (username,))
row = cur.fetchone()
return row[0] if row is not None else None
finally:
conn.close()
def upsert(
self, username: str, password: str, role: str, actor_role: str
) -> None:
"""Create or fully update one user (password + role).
Caps (server-side, actor_role = the requesting user's role):
- superuser may only assign role 'user'; any other target role
raises PermissionError (they must never be able to escalate).
- admin may assign any role.
"""
if actor_role not in ROLE_LEVELS:
raise PermissionError(f"unknown actor role {actor_role!r}")
if ROLE_LEVELS[actor_role] < ROLE_LEVELS["admin"] and role != "user":
raise PermissionError(
f"{actor_role} may only create or assign the 'user' role"
)
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
cur.execute(
"""
INSERT INTO users (username, password_hash, role)
VALUES (%s, %s, %s)
ON CONFLICT (username) DO UPDATE
SET password_hash = EXCLUDED.password_hash,
role = EXCLUDED.role, updated_at = now()
""",
(username, hash_password(password), role),
)
finally:
conn.close()
def delete(self, username: str, actor_role: str) -> bool:
"""Delete one user (their sessions cascade).
Caps: superuser may only delete accounts whose current role is
'user'; admin any account except the LAST admin (refused, False).
Returns False when the username is unknown."""
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
cur.execute("SELECT role FROM users WHERE username = %s", (username,))
row = cur.fetchone()
if row is None:
return False
target_role = row[0]
if actor_role == "superuser" and target_role != "user":
raise PermissionError(
"superuser may only delete accounts with role 'user'"
)
if target_role == "admin":
cur.execute("SELECT count(*) FROM users WHERE role = 'admin'")
if cur.fetchone()[0] <= 1:
return False # never the last admin
cur.execute("DELETE FROM users WHERE username = %s", (username,))
return True
finally:
conn.close()
class SessionStore:
"""Login sessions backed by the sessions table: login creates an
opaque bearer token, only its SHA-256 digest is stored. resolve()
joins users so the CURRENT role is returned - role changes apply on
the very next request."""
def __init__(self, dsn: str, ttl_days: int = SESSION_TTL_DAYS) -> None:
self._dsn = dsn
self._ttl = timedelta(days=ttl_days)
def login(self, username: str) -> str:
"""Create a session and return the plaintext bearer token."""
token = secrets.token_urlsafe(32)
# interval as a string: psycopg cannot cast an integer parameter
# to interval directly (make_interval needs named args instead).
ttl = f"{int(self._ttl.total_seconds())} seconds"
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
# opportunistic cleanup of expired sessions
cur.execute("DELETE FROM sessions WHERE expires_at < now()")
cur.execute(
"""
INSERT INTO sessions (token_hash, username, expires_at)
VALUES (%s, %s, now() + %s::interval)
""",
(hash_token(token), username, ttl),
)
finally:
conn.close()
return token
def resolve(self, token: str) -> tuple[str, str] | None:
"""(username, CURRENT role) for a valid, unexpired session."""
conn = psycopg.connect(self._dsn)
try:
with conn.cursor() as cur:
cur.execute(
"""
SELECT u.username, u.role
FROM sessions s JOIN users u ON u.username = s.username
WHERE s.token_hash = %s AND s.expires_at > now()
""",
(hash_token(token),),
)
row = cur.fetchone()
return (row[0], row[1]) if row is not None else None
finally:
conn.close()
def logout(self, token: str) -> bool:
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM sessions WHERE token_hash = %s", (hash_token(token),)
)
return cur.rowcount > 0
finally:
conn.close()
class DeviceRenameStore:
"""Backfill-renames the device name on INGESTED messages. The IMEI is
the identity; names are labels — a rename just rewrites the label on
matched rows (both the device column and the payload key). No audit
trail: the caller (superuser/admin) decides; preview via dry_run."""
def __init__(self, dsn: str) -> None:
self._dsn = dsn
def rename(
self,
imei: str,
new_name: str,
from_time: datetime | None,
to_time: datetime | None,
payload_key: str = "device",
dry_run: bool = False,
) -> dict[str, Any]:
"""Rename matching rows' device column AND payload key. NULL bounds
mean unbounded (no framing on that end). Returns the row count."""
match_pred = """
SELECT count(*) FROM mqtt_messages
WHERE imei = %s
AND (%s::timestamptz IS NULL OR received_at >= %s::timestamptz)
AND (%s::timestamptz IS NULL OR received_at <= %s::timestamptz)
AND device IS DISTINCT FROM %s
"""
if dry_run:
# preview: count the matches with the exact predicate the
# update would use, but write nothing
conn = psycopg.connect(self._dsn)
try:
with conn.cursor() as cur:
cur.execute(
match_pred,
(imei, from_time, from_time, to_time, to_time, new_name),
)
return {"updated": cur.fetchone()[0], "dry_run": True}
finally:
conn.close()
# autocommit: a single-statement write commits immediately (a plain
# connection would roll back on close).
conn = psycopg.connect(self._dsn, autocommit=True)
try:
with conn.cursor() as cur:
# jsonb_set's path must be an array; ARRAY[%s]::text[] builds
# it from the key name safely (no string concatenation).
cur.execute(
"""
WITH matched AS (
SELECT id FROM mqtt_messages
WHERE imei = %s
AND (%s::timestamptz IS NULL OR received_at >= %s::timestamptz)
AND (%s::timestamptz IS NULL OR received_at <= %s::timestamptz)
AND device IS DISTINCT FROM %s
), updated AS (
UPDATE mqtt_messages m
SET device = %s,
payload = jsonb_set(m.payload, ARRAY[%s]::text[], %s::jsonb)
FROM matched
WHERE m.id = matched.id
RETURNING m.id
)
SELECT count(*) FROM updated
""",
(
imei,
from_time, from_time,
to_time, to_time,
new_name,
new_name,
payload_key,
Jsonb(new_name),
),
)
return {"updated": cur.fetchone()[0], "dry_run": False}
finally:
conn.close()
def _validate_rename_body(
data: dict[str, Any], source: str
) -> tuple[str, datetime | None, datetime | None, bool]:
"""Validate a rename body {"device", "from"?, "to"?, "dry_run"?} and
return (new_name, from, to, dry_run). Absent/null bounds are unbounded
(no framing on that end). Raises ValueError otherwise."""
if not isinstance(data, dict):
raise ValueError(
f'{source}: expected JSON on the form {{"device": "New name", "from"?: ..., "to"?: ...}}'
)
new_name = data.get("device")
if not isinstance(new_name, str) or not new_name.strip():
raise ValueError(f'{source}: "device" must be a non-empty string')
parsed: list[datetime | None] = []
for field in ("from", "to"):
raw = data.get(field)
if raw is None:
parsed.append(None)
continue
if not isinstance(raw, str):
raise ValueError(f'{source}: "{field}" must be an ISO 8601 string or null')
try:
parsed.append(_as_utc(datetime.fromisoformat(raw)))
except ValueError:
raise ValueError(
f'{source}: "{field}" is not a valid ISO 8601 timestamp'
) from None
dry_run = bool(data.get("dry_run", False))
return new_name.strip(), parsed[0], parsed[1], dry_run
def _validate_token_body(data: dict[str, Any], source: str) -> dict[str, str]:
"""Validate a {"tokens": {app: secret}} request body and return the
inner tokens mapping. Raises ValueError with a descriptive message
@@ -491,12 +880,16 @@ def create_app(
token_store: TokenStore | None = None,
imei_store: ImeiMapStore | None = None,
attributes_store: ImeiAttributesStore | None = None,
user_store: UserStore | None = None,
session_store: SessionStore | None = None,
rename_store: DeviceRenameStore | None = None,
) -> FastAPI:
"""Build the API application.
The stores are injectable so tests can run without a database; in
production the defaults are built from the environment (DB_DSN,
BOOTSTRAP_APP_TOKEN, BOOTSTRAP_APP_NAME).
BOOTSTRAP_APP_TOKEN, BOOTSTRAP_APP_NAME, BOOTSTRAP_ADMIN_USER,
BOOTSTRAP_ADMIN_PASSWORD).
"""
app = FastAPI(
title="nodeledge MQTT data API",
@@ -517,24 +910,69 @@ def create_app(
imei_store = ImeiMapStore(dsn)
if attributes_store is None:
attributes_store = ImeiAttributesStore(dsn)
if user_store is None:
user_store = UserStore(
dsn,
bootstrap_user=os.environ.get("BOOTSTRAP_ADMIN_USER") or None,
bootstrap_password=os.environ.get("BOOTSTRAP_ADMIN_PASSWORD") or None,
)
if session_store is None:
session_store = SessionStore(dsn)
if rename_store is None:
rename_store = DeviceRenameStore(dsn)
# Brute-force lockout state (in-process, per username).
login_attempts: dict[str, dict[str, float]] = {}
def authenticate(request: Request) -> None:
"""401 without valid credentials. X-App-Token keeps unrestricted
machine access; Bearer sessions carry the user's CURRENT role."""
token = request.headers.get("X-App-Token")
if not token:
token = request.query_params.get("app_token")
if not token:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="missing app token: pass the X-App-Token header",
)
try:
if not token_store.verify(token):
if token:
try:
if not token_store.verify(token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="invalid app token",
)
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
return # app token: unrestricted (machine clients)
bearer = request.headers.get("Authorization", "")
if bearer.startswith("Bearer "):
try:
session = session_store.resolve(bearer[7:].strip())
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
if session is None:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="invalid app token",
detail="invalid or expired session",
)
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
request.state.username = session[0]
request.state.role = session[1]
return
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="missing credentials: pass X-App-Token or a Bearer session token",
)
def require_level(minimum: str):
"""Dependency factory gating writes by role level. App tokens
(no role set on request.state) stay unrestricted; logged-in roles
below the minimum get 403."""
def check(request: Request) -> None:
role = getattr(request.state, "role", None)
if role is None:
return # app token: unrestricted per design decision
if ROLE_LEVELS.get(role, -1) < ROLE_LEVELS[minimum]:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"{minimum} role required",
)
return check
app.state.config = config
app.state.authenticate = authenticate
@@ -562,8 +1000,8 @@ def create_app(
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
@app.put("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)])
@app.post("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate)])
@app.put("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
@app.post("/configs/{name}", tags=["config"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def set_config(name: str, body: dict[str, Any]) -> dict[str, Any]:
try:
if name == "app_tokens":
@@ -619,7 +1057,7 @@ def create_app(
)
return _device_repr(normalized, doc)
@app.put("/devices/{imei}", tags=["devices"], dependencies=[Depends(authenticate)])
@app.put("/devices/{imei}", tags=["devices"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def set_device(imei: str, body: dict[str, Any]) -> dict[str, Any]:
normalized = normalize_imei(imei)
if normalized is None:
@@ -649,7 +1087,7 @@ def create_app(
logger.info("device %s upserted: %d setting(s)", normalized, len(doc))
return _device_repr(normalized, doc)
@app.delete("/devices/{imei}", tags=["devices"], dependencies=[Depends(authenticate)])
@app.delete("/devices/{imei}", tags=["devices"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def delete_device(imei: str) -> dict[str, Any]:
normalized = normalize_imei(imei)
if normalized is None:
@@ -728,8 +1166,8 @@ def create_app(
result["at"] = resolved_at.isoformat()
return result
@app.put("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)])
@app.post("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate)])
@app.put("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
@app.post("/devices/{imei}/attributes", tags=["devices"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def add_device_attribute(imei: str, body: dict[str, Any]) -> dict[str, Any]:
normalized = normalize_imei(imei)
if normalized is None:
@@ -764,7 +1202,7 @@ def create_app(
@app.delete(
"/devices/{imei}/attributes/{key}",
tags=["devices"],
dependencies=[Depends(authenticate)],
dependencies=[Depends(authenticate), Depends(require_level("superuser"))],
)
def delete_device_attribute(imei: str, key: str) -> dict[str, Any]:
normalized = normalize_imei(imei)
@@ -791,6 +1229,50 @@ def create_app(
logger.info("attribute %r latest version deleted for %s", name, normalized)
return {"status": "deleted", "imei": normalized, "key": name}
# --- Backfill rename of stored messages: the IMEI is the identity,
# names are labels — a rewrite of the label on matched rows (device
# column AND payload key), optionally framed on received_at. No audit
# trail, no undo: the caller decides after the dry_run preview.
@app.post(
"/devices/{imei}/rename",
tags=["devices"],
dependencies=[Depends(authenticate), Depends(require_level("superuser"))],
)
def rename_device_messages(imei: str, body: dict[str, Any]) -> dict[str, Any]:
normalized = normalize_imei(imei)
if normalized is None:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=f"{imei!r} is not a 15-digit IMEI",
)
try:
new_name, from_time, to_time, dry_run = _validate_rename_body(
body, "request body"
)
if from_time is not None and to_time is not None and from_time > to_time:
raise ValueError("'from' must be earlier than or equal to 'to'")
result = rename_store.rename(
normalized, new_name, from_time, to_time, dry_run=dry_run
)
except ValueError as exc:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail=str(exc),
) from exc
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
logger.info(
"rename for %s -> %r (from=%s to=%s dry_run=%s): %s",
normalized,
new_name,
from_time,
to_time,
dry_run,
result,
)
return {"imei": normalized, "device": new_name, **result}
# --- Token resources: add/rotate/remove ONE app token without
# touching the other apps' rows.
@@ -803,7 +1285,7 @@ def create_app(
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
@app.put("/tokens/{app_name}", tags=["tokens"], dependencies=[Depends(authenticate)])
@app.put("/tokens/{app_name}", tags=["tokens"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def set_token(app_name: str, body: dict[str, Any]) -> dict[str, Any]:
name = app_name.strip()
try:
@@ -822,7 +1304,7 @@ def create_app(
logger.info("token %r upserted", name)
return {"status": "updated", "app": name}
@app.delete("/tokens/{app_name}", tags=["tokens"], dependencies=[Depends(authenticate)])
@app.delete("/tokens/{app_name}", tags=["tokens"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def delete_token(app_name: str) -> dict[str, Any]:
name = app_name.strip()
if not name:
@@ -842,6 +1324,135 @@ def create_app(
logger.info("token %r deleted", name)
return {"status": "deleted", "app": name}
# --- Interactive authentication: username/password login issuing an
# opaque Bearer session token (only its SHA-256 digest is stored).
# POST /auth/login is anonymous; lockout is per username.
@app.post("/auth/login", tags=["auth"])
def login(body: dict[str, Any]) -> dict[str, Any]:
username = body.get("username") if isinstance(body, dict) else None
password = body.get("password") if isinstance(body, dict) else None
if not isinstance(username, str) or not username.strip():
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="username is required",
)
if not isinstance(password, str) or not password:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="password is required",
)
name = username.strip()
now = time.monotonic()
state = login_attempts.get(name, {"fails": 0.0, "locked_until": 0.0})
if state["locked_until"] > now:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="too many failed attempts; try again later",
)
try:
role = user_store.verify(name, password)
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
if role is None:
state["fails"] += 1
if state["fails"] >= LOCKOUT_ATTEMPTS:
state["locked_until"] = now + LOCKOUT_MINUTES * 60
state["fails"] = 0.0
logger.warning(
"login for %r locked out for %d minutes", name, LOCKOUT_MINUTES
)
login_attempts[name] = state
# Same response for unknown user and wrong password.
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="invalid username or password",
)
login_attempts.pop(name, None)
token = session_store.login(name)
logger.info("user %r logged in (role %s)", name, role)
return {"token": token, "username": name, "role": role}
@app.get("/auth/me", tags=["auth"], dependencies=[Depends(authenticate)])
def whoami(request: Request) -> dict[str, Any]:
return {
"username": getattr(request.state, "username", None),
"role": getattr(request.state, "role", None),
}
@app.post("/auth/logout", tags=["auth"], dependencies=[Depends(authenticate)])
def logout(request: Request) -> dict[str, Any]:
bearer = request.headers.get("Authorization", "")
if bearer.startswith("Bearer "):
try:
session_store.logout(bearer[7:].strip())
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
return {"status": "logged out"}
# --- User management. Add: superuser+ (a superuser may only assign
# role 'user' - no escalation). Modify/delete: superuser may only
# touch accounts whose current role is 'user'; admin anything except
# the last admin. Caps enforced in UserStore via the ACTOR's role.
@app.get("/users", tags=["users"], dependencies=[Depends(authenticate)])
def list_users() -> list[dict[str, Any]]:
try:
return user_store.list()
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
@app.put("/users/{username}", tags=["users"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def upsert_user(username: str, body: dict[str, Any], request: Request) -> dict[str, Any]:
name = username.strip()
role = "user"
if isinstance(body, dict) and body.get("role") in ROLE_LEVELS:
role = body["role"]
password = body.get("password") if isinstance(body, dict) else None
if not name or not isinstance(password, str) or not password:
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail="password (and optionally role) required",
)
# The actor's role comes from the session (set by authenticate);
# an app token has no role and is unrestricted.
actor_role = getattr(request.state, "role", None)
if actor_role not in ROLE_LEVELS:
actor_role = "admin"
try:
user_store.upsert(name, password, role, actor_role=actor_role)
except PermissionError as exc:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=str(exc),
) from exc
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
logger.info("user %r upserted (role %s)", name, role)
return {"status": "updated", "username": name, "role": role}
@app.delete("/users/{username}", tags=["users"], dependencies=[Depends(authenticate), Depends(require_level("superuser"))])
def delete_user(username: str, request: Request) -> dict[str, Any]:
actor_role = getattr(request.state, "role", None)
if actor_role not in ROLE_LEVELS:
actor_role = "admin" # app token: unrestricted
try:
deleted = user_store.delete(username.strip(), actor_role=actor_role)
except PermissionError as exc:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=str(exc),
) from exc
except psycopg.Error as exc:
raise _db_unavailable(exc) from exc
if not deleted:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"unknown user {username!r}, or the last admin cannot be deleted",
)
logger.info("user %r deleted", username.strip())
return {"status": "deleted", "username": username.strip()}
@app.get("/health", tags=["meta"])
def health() -> dict[str, Any]:
return {"status": "ok"}