spencermandClaude Opus 5.5
a82e056b6a
Admin: Departments (name and stations; add shows the three groups IT must create), Profiles (welder stamp, station limits, active) with a CSV import that previews and applies only when clean, Checklists (weld and plain), Security (session, lockout and rate-limit settings with bounds, alerts for the last 24 hours, the integrity result), Access log with filters and a guarded CSV, Who is in, Stamp check, Notifications, and the employeeID flags. Every Admin change is audited. Worker: ad-reconcile (nightly and on "refresh from AD now"), session-refresh (15 minutes), integrity (seals, asset hashes, missing and orphan files, the audit chain), backup-check, notify-send and notify-digest. Notifications route per event type to AD groups, users, mailboxes or the department Supervisor group, immediately or in a daily digest, over STARTTLS verified against the PESCO CA, with dry run on the test stack. Backups: scripts/backup/ps-backup.sh and ps-restore.sh with a systemd timer at 02:00 America/Denver. Restore rehearsed on the test stack 2026-10-21. CI: pip-audit and npm audit. pypdf 6.21.1 and certifi 2026.10.19. Migration 0010. Closes the app part of I6, M3, O1, O2 and S3. Co-Authored-By: Claude Opus 5.5 <[email protected]>
84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Every API error is JSON `{"detail": "<message>", "code": "<snake_code>"}`, with extra fields when useful."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from fastapi import FastAPI, Request
|
|
from fastapi.exceptions import RequestValidationError
|
|
from fastapi.responses import JSONResponse
|
|
from starlette.exceptions import HTTPException as StarletteHTTPException
|
|
|
|
HTTP_CODES = {
|
|
400: "bad_request",
|
|
401: "no_session",
|
|
403: "forbidden",
|
|
404: "not_found",
|
|
405: "method_not_allowed",
|
|
409: "conflict",
|
|
413: "too_large",
|
|
429: "rate_limited",
|
|
503: "unavailable",
|
|
}
|
|
|
|
|
|
class ApiError(Exception):
|
|
def __init__(
|
|
self,
|
|
status: int,
|
|
code: str,
|
|
detail: str,
|
|
*,
|
|
headers: dict[str, str] | None = None,
|
|
extra: dict[str, Any] | None = None,
|
|
):
|
|
super().__init__(detail)
|
|
self.status = status
|
|
self.code = code
|
|
self.detail = detail
|
|
self.headers = headers or {}
|
|
self.extra = extra or {}
|
|
|
|
def body(self) -> dict[str, Any]:
|
|
return {"detail": self.detail, "code": self.code, **self.extra}
|
|
|
|
|
|
def error_response(err: ApiError) -> JSONResponse:
|
|
return JSONResponse(err.body(), status_code=err.status, headers=err.headers)
|
|
|
|
|
|
def install(app: FastAPI) -> None:
|
|
@app.exception_handler(ApiError)
|
|
async def _api_error(request: Request, exc: ApiError) -> JSONResponse:
|
|
principal = getattr(request.state, "principal", None)
|
|
if exc.status == 403 and principal is not None:
|
|
# A refusal to a signed-in person is audited as policy.denied: method, path and code, never the body
|
|
from starlette.concurrency import run_in_threadpool
|
|
|
|
from ..domain.security_alerts import record_denied
|
|
|
|
await run_in_threadpool(
|
|
record_denied,
|
|
request.app.state.settings,
|
|
principal,
|
|
method=request.method,
|
|
path=request.url.path,
|
|
code=exc.code,
|
|
)
|
|
return error_response(exc)
|
|
|
|
@app.exception_handler(RequestValidationError)
|
|
async def _invalid(_: Request, exc: RequestValidationError) -> JSONResponse:
|
|
errors = [{"loc": [str(x) for x in e.get("loc", ())], "msg": str(e.get("msg", ""))} for e in exc.errors()]
|
|
where = ", ".join(".".join(e["loc"][1:]) or e["loc"][0] for e in errors if e["loc"]) or "request"
|
|
return JSONResponse(
|
|
{"detail": f"Invalid request: {where}", "code": "invalid_request", "errors": errors}, status_code=422
|
|
)
|
|
|
|
@app.exception_handler(StarletteHTTPException)
|
|
async def _http(_: Request, exc: StarletteHTTPException) -> JSONResponse:
|
|
code = HTTP_CODES.get(exc.status_code, "http_error")
|
|
return JSONResponse(
|
|
{"detail": str(exc.detail), "code": code}, status_code=exc.status_code, headers=exc.headers or None
|
|
)
|