spencermandClaude Opus 5.5
70a97398f1
- C1 (issue 1): over target follows VISUAL ticket hours, not the timer. - C2 (issue 24): a Q.C. reject always needs a reason; accept needs an NCR number after a reject without one. - C3 (issue 25): an incomplete unit ITP blocks filing. - C4 (issue 27): prints open the print dialog on the page, no new tab. - C5 (issue 31): filing is blocked while VISUAL shows a work order open; the feed sets vclosed on VISUAL packets. - C6 (issue 8): the data book leaves out the history pages. - C7 (issue 18): the supervisor approves each received short; accept and filing wait for it. - C8 (issue 23): per-department checklists, edited by the department supervisor or Admin, frozen at START. - C9 (issue 32): Q.C. and Clerical may run the stamp check. - C10 (issue 33): the PM screen and reports are for PS-ProjectMgmt and Admin only. - C11 (issue 36): the Stall screen lists the packets assigned to the welder, at the supervisor's stall. Migration 0011. Docs: PLAN (also issue 9 confirmed, issue 4 asked), POLICY, SECURITY, SCHEMA, FEED, README, CHANGELOG. Co-Authored-By: Claude Opus 5.5 <[email protected]>
657 lines
24 KiB
Python
657 lines
24 KiB
Python
"""Admin: the Security tab (settings, alerts, the integrity result), the access log and its CSV, who is in now, the
|
|
stamp check, and the directory flags.
|
|
|
|
Every route needs the admin role and a password session, except the stamp check: Admin, Q.C. and Clerical on a password
|
|
session, at /api/stamp-verify (change C9) and at the old /api/admin paths. Every change is audited (`admin.*`).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import io
|
|
import ipaddress
|
|
import json
|
|
import re
|
|
import uuid
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from fastapi import APIRouter, Depends, Query, Request, Response
|
|
from fastapi.encoders import jsonable_encoder
|
|
from psycopg.rows import dict_row
|
|
from pydantic import BaseModel, ConfigDict, Field
|
|
from starlette.concurrency import run_in_threadpool
|
|
from starlette.datastructures import UploadFile
|
|
|
|
from .. import audit
|
|
from ..auth import secsettings
|
|
from ..domain import security_alerts
|
|
from ..pdf import stamps, tracecsv
|
|
from .account import LOG_PREFIXES
|
|
from .deps import (
|
|
Db,
|
|
Principal,
|
|
admin_principal,
|
|
current_principal,
|
|
get_db,
|
|
idle_minutes,
|
|
require_password_session,
|
|
settings_of,
|
|
)
|
|
from .errors import ApiError
|
|
from .jobs import run_out
|
|
from .packets import WRITE
|
|
|
|
router = APIRouter(prefix="/api/admin", tags=["admin"])
|
|
# The stamp check outside /admin (change C9): Admin, Q.C. and Clerical. The /api/admin paths stay as aliases.
|
|
stamp_router = APIRouter(prefix="/api", tags=["stamp"])
|
|
|
|
CSV = "text/csv; charset=utf-8"
|
|
PDF_MAX_BYTES = 25 * 1024 * 1024
|
|
FORM_SLACK = 64 * 1024
|
|
|
|
|
|
# ---------------------------------------------------------------- security settings
|
|
|
|
|
|
class SecPut(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
values: dict[str, Any]
|
|
version: int = Field(ge=0)
|
|
|
|
|
|
def _sec_out(conn: Any, request: Request) -> dict[str, Any]:
|
|
s = settings_of(request)
|
|
stored, version, updated_at, updated_by = secsettings.load_versioned(conn)
|
|
defaults = secsettings.defaults(s)
|
|
return {
|
|
"values": {**defaults, **stored},
|
|
"defaults": defaults,
|
|
"bounds": {k: list(v) for k, v in secsettings.BOUNDS.items()},
|
|
"version": version,
|
|
"updated_at": updated_at,
|
|
"updated_by": str(updated_by) if updated_by else None,
|
|
}
|
|
|
|
|
|
@router.get("/security/settings")
|
|
def security_settings(
|
|
request: Request,
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
return _sec_out(d.conn, request)
|
|
|
|
|
|
@router.put("/security/settings", dependencies=WRITE)
|
|
def security_settings_put(
|
|
body: SecPut,
|
|
request: Request,
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
conn = d.conn
|
|
try:
|
|
secsettings.save(conn, settings_of(request), who=p, values=body.values, version=body.version)
|
|
except secsettings.SecSettingsError as e:
|
|
raise ApiError(422, "bad_settings", e.errors[0], extra={"errors": e.errors}) from None
|
|
except secsettings.VersionConflict:
|
|
raise ApiError(
|
|
409,
|
|
"version_conflict",
|
|
"Someone else changed the security settings. Reload and try again.",
|
|
extra={"current": jsonable_encoder(_sec_out(conn, request))},
|
|
) from None
|
|
return _sec_out(conn, request)
|
|
|
|
|
|
# ---------------------------------------------------------------- alerts and integrity
|
|
|
|
|
|
@router.get("/security/alerts")
|
|
def security_alert_list(
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
hours: int = Query(24, ge=1, le=168),
|
|
) -> list[dict[str, Any]]:
|
|
return security_alerts.alerts(d.conn, hours)
|
|
|
|
|
|
@router.get("/security/integrity")
|
|
def security_integrity(p: Principal = Depends(admin_principal), d: Db = Depends(get_db)) -> dict[str, Any]: # noqa: B008
|
|
with d.conn.cursor(row_factory=dict_row) as cur:
|
|
r = cur.execute(
|
|
"SELECT * FROM job_runs WHERE job = 'integrity' AND finished_at IS NOT NULL ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
findings = (
|
|
cur.execute(
|
|
"SELECT kind, subject, detail, at FROM integrity_findings WHERE run_id = %s ORDER BY id", (r["id"],)
|
|
).fetchall()
|
|
if r
|
|
else []
|
|
)
|
|
run = None
|
|
if r is not None:
|
|
full = run_out(r) or {}
|
|
run = {k: full.get(k) for k in ("id", "started_at", "finished_at", "status", "counts")}
|
|
return {
|
|
"run": run,
|
|
"findings": [
|
|
{"kind": f["kind"], "subject": f["subject"], "detail": f["detail"] or {}, "at": f["at"]} for f in findings
|
|
],
|
|
}
|
|
|
|
|
|
# ---------------------------------------------------------------- access log
|
|
|
|
ACCESS_LIMIT_MAX = 1000
|
|
CSV_ROWS_MAX = 50_000
|
|
FAIL_ACTIONS = ("auth.signin_failed", "auth.lockout", "auth.signin_burst")
|
|
RX_CURSOR = re.compile(r"^(?P<at>[0-9T:.+\-Z ]{10,40}),(?P<src>[01]),(?P<id>[0-9]{1,19})$")
|
|
|
|
ACCESS_SQL = """
|
|
SELECT x.at, x.src, x.id, x.source, x.kind, x.ok, coalesce(x.account, u.sam_account) AS account, x.user_id,
|
|
u.display_name, host(x.ip) AS ip, x.detail
|
|
FROM (
|
|
SELECT a.at, 0 AS src, a.id, 'attempt' AS source, a.kind, a.ok, a.account, a.user_id, a.ip,
|
|
jsonb_strip_nulls(jsonb_build_object('reason', a.reason, 'session_id', a.session_id)) AS detail
|
|
FROM signin_attempts a
|
|
UNION ALL
|
|
SELECT e.at, 1, e.id, 'audit', e.action, NOT (e.action = ANY(%(fail)s)), e.detail->>'account', e.user_id, e.ip,
|
|
coalesce(e.detail, '{{}}'::jsonb)
|
|
FROM audit_events e WHERE e.action LIKE ANY(%(prefixes)s)
|
|
) x LEFT JOIN users u ON u.id = x.user_id
|
|
WHERE true{where}
|
|
ORDER BY x.at DESC, x.src DESC, x.id DESC
|
|
LIMIT %(limit)s
|
|
"""
|
|
|
|
|
|
def _access_where(
|
|
*,
|
|
since: datetime | None,
|
|
until: datetime | None,
|
|
user_id: uuid.UUID | None,
|
|
q: str,
|
|
kind: str,
|
|
ok: bool | None,
|
|
ip: str,
|
|
before: str,
|
|
) -> tuple[str, dict[str, Any]]:
|
|
terms: list[str] = []
|
|
params: dict[str, Any] = {"fail": list(FAIL_ACTIONS), "prefixes": list(LOG_PREFIXES)}
|
|
if since is not None:
|
|
terms.append("x.at >= %(since)s")
|
|
params["since"] = since
|
|
if until is not None:
|
|
terms.append("x.at < %(until)s")
|
|
params["until"] = until
|
|
if user_id is not None:
|
|
terms.append("x.user_id = %(user_id)s")
|
|
params["user_id"] = user_id
|
|
q = q.strip()
|
|
if q:
|
|
pat = "%" + q.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + "%"
|
|
terms.append(
|
|
"(x.account ILIKE %(q)s OR u.sam_account ILIKE %(q)s OR u.display_name ILIKE %(q)s"
|
|
" OR u.emp_no = %(q_exact)s)"
|
|
)
|
|
params["q"], params["q_exact"] = pat, q
|
|
kind = kind.strip()
|
|
if kind:
|
|
terms.append("x.kind = %(kind)s")
|
|
params["kind"] = kind
|
|
if ok is not None:
|
|
terms.append("x.ok = %(ok)s")
|
|
params["ok"] = ok
|
|
ip = ip.strip()
|
|
if ip:
|
|
try:
|
|
net = ipaddress.ip_network(ip, strict=False)
|
|
except ValueError:
|
|
raise ApiError(422, "invalid_request", "ip must be an address or a network like 10.1.2.0/24.") from None
|
|
terms.append("x.ip <<= %(ip)s::inet")
|
|
params["ip"] = str(net)
|
|
before = before.strip()
|
|
if before:
|
|
m = RX_CURSOR.match(before)
|
|
if m is None:
|
|
raise ApiError(422, "invalid_request", "before is the next_before value of a previous page.")
|
|
try:
|
|
at = datetime.fromisoformat(m["at"].replace("Z", "+00:00"))
|
|
except ValueError:
|
|
raise ApiError(422, "invalid_request", "before is the next_before value of a previous page.") from None
|
|
terms.append("(x.at, x.src, x.id) < (%(b_at)s, %(b_src)s, %(b_id)s)")
|
|
params.update(b_at=at, b_src=int(m["src"]), b_id=int(m["id"]))
|
|
return "".join(f" AND {t}" for t in terms), params
|
|
|
|
|
|
def _access_rows(conn: Any, where: str, params: dict[str, Any], limit: int) -> list[dict[str, Any]]:
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
|
rows: list[dict[str, Any]] = cur.execute(ACCESS_SQL.format(where=where), {**params, "limit": limit}).fetchall()
|
|
return rows
|
|
|
|
|
|
def _access_out(r: dict[str, Any]) -> dict[str, Any]:
|
|
return {
|
|
"id": f"{r['source']}:{r['id']}",
|
|
"at": r["at"],
|
|
"source": r["source"],
|
|
"kind": r["kind"],
|
|
"ok": bool(r["ok"]),
|
|
"account": r["account"],
|
|
"user_id": str(r["user_id"]) if r["user_id"] else None,
|
|
"display_name": r["display_name"],
|
|
"ip": r["ip"],
|
|
"detail": r["detail"] or {},
|
|
}
|
|
|
|
|
|
class AccessFilters(BaseModel):
|
|
since: datetime | None = None
|
|
until: datetime | None = None
|
|
user_id: uuid.UUID | None = None
|
|
q: str = Field("", max_length=200)
|
|
kind: str = Field("", max_length=60)
|
|
ok: bool | None = None
|
|
ip: str = Field("", max_length=60)
|
|
|
|
|
|
def access_filters(
|
|
since: datetime | None = Query(None), # noqa: B008
|
|
until: datetime | None = Query(None), # noqa: B008
|
|
user_id: uuid.UUID | None = Query(None), # noqa: B008
|
|
q: str = Query("", max_length=200),
|
|
kind: str = Query("", max_length=60),
|
|
ok: bool | None = Query(None),
|
|
ip: str = Query("", max_length=60),
|
|
) -> AccessFilters:
|
|
return AccessFilters(since=since, until=until, user_id=user_id, q=q, kind=kind, ok=ok, ip=ip)
|
|
|
|
|
|
@router.get("/access-log")
|
|
def access_log(
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
f: AccessFilters = Depends(access_filters), # noqa: B008
|
|
limit: int = Query(100, ge=1, le=ACCESS_LIMIT_MAX),
|
|
before: str = Query("", max_length=80),
|
|
) -> dict[str, Any]:
|
|
where, params = _access_where(**f.model_dump(), before=before)
|
|
rows = _access_rows(d.conn, where, params, limit)
|
|
nxt = None
|
|
if len(rows) == limit:
|
|
last = rows[-1]
|
|
nxt = f"{last['at'].astimezone(UTC).isoformat()},{last['src']},{last['id']}"
|
|
return {"rows": [_access_out(r) for r in rows], "next_before": nxt}
|
|
|
|
|
|
CSV_HEADER = ("at", "source", "kind", "ok", "account", "display_name", "user_id", "ip", "detail")
|
|
|
|
|
|
@router.get("/access-log.csv")
|
|
def access_log_csv(
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
f: AccessFilters = Depends(access_filters), # noqa: B008
|
|
) -> Response:
|
|
conn = d.conn
|
|
where, params = _access_where(**f.model_dump(), before="")
|
|
rows = _access_rows(conn, where, params, CSV_ROWS_MAX)
|
|
lines: list[list[Any]] = [list(CSV_HEADER)]
|
|
for r in rows:
|
|
o = _access_out(r)
|
|
lines.append(
|
|
[
|
|
o["at"].astimezone(UTC).isoformat(),
|
|
o["source"],
|
|
o["kind"],
|
|
"yes" if o["ok"] else "no",
|
|
o["account"],
|
|
o["display_name"],
|
|
o["user_id"],
|
|
o["ip"],
|
|
json.dumps(o["detail"], sort_keys=True, default=str, separators=(",", ":")),
|
|
]
|
|
)
|
|
text = tracecsv.to_csv(lines) + "\r\n"
|
|
filters = {k: (str(v) if v is not None else None) for k, v in f.model_dump().items() if v not in (None, "")}
|
|
with conn.transaction():
|
|
conn.execute(
|
|
"INSERT INTO print_log (by, by_name, session_id, kind, what, pages, ip)"
|
|
" VALUES (%s, %s, %s, 'access-log-csv', %s, NULL, %s)",
|
|
(p.user_id, p.user.get("display_name"), p.session_id, f"Access log CSV ({len(rows)} rows)", p.ip),
|
|
)
|
|
audit.append(
|
|
conn,
|
|
action="admin.access_log_exported",
|
|
user_id=p.user_id,
|
|
session_id=p.session_id,
|
|
role_snapshot=p.role_snapshot,
|
|
subject_type="access_log",
|
|
subject_id="csv",
|
|
detail={"rows": len(rows), "filters": filters, "capped": len(rows) >= CSV_ROWS_MAX},
|
|
ip=p.ip,
|
|
)
|
|
day = datetime.now(UTC).strftime("%Y%m%d")
|
|
return Response(
|
|
content=text.encode("utf-8"),
|
|
media_type=CSV,
|
|
headers={
|
|
"Content-Disposition": f'attachment; filename="access-log-{day}.csv"',
|
|
"X-Content-Type-Options": "nosniff",
|
|
"Cache-Control": "no-store",
|
|
},
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------- who is in now
|
|
|
|
|
|
@router.get("/presence")
|
|
def presence(
|
|
request: Request,
|
|
p: Principal = Depends(admin_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
conn = d.conn
|
|
s = secsettings.overlay(conn, settings_of(request))
|
|
widest = max(s.session_idle_minutes, s.session_desk_minutes)
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
|
rows = cur.execute(
|
|
"SELECT s.id, s.user_id, u.display_name, s.kind::text AS kind, dv.name AS device_name, host(s.ip) AS ip,"
|
|
" s.created_at, s.last_seen_at, s.locked_since, s.roles_snapshot, now() AS db_now,"
|
|
" now() - s.last_seen_at AS idle"
|
|
" FROM sessions s JOIN users u ON u.id = s.user_id LEFT JOIN devices dv ON dv.id = s.device_id"
|
|
" WHERE s.revoked_at IS NULL AND s.expires_at > now()"
|
|
" AND s.last_seen_at > now() - make_interval(mins => %s::int)"
|
|
" ORDER BY lower(u.display_name), s.created_at",
|
|
(widest,),
|
|
).fetchall()
|
|
streams: dict[str, dict[str, Any]] = {}
|
|
hub = getattr(request.app.state, "events", None)
|
|
if hub is not None:
|
|
for screen, users in hub.presence_by_screen().items():
|
|
for u in users:
|
|
streams.setdefault(str(u["user_id"]), {"screen": None if screen == "none" else screen, **u})
|
|
at = rows[0]["db_now"] if rows else datetime.now(UTC)
|
|
out = []
|
|
for r in rows:
|
|
roles = list((r["roles_snapshot"] or {}).get("roles", []))
|
|
idle_min, _ = idle_minutes(s, roles)
|
|
if r["idle"].total_seconds() > idle_min * 60:
|
|
continue
|
|
stream = streams.get(str(r["user_id"]))
|
|
out.append(
|
|
{
|
|
"session_id": str(r["id"]),
|
|
"user_id": str(r["user_id"]),
|
|
"display_name": r["display_name"],
|
|
"kind": r["kind"],
|
|
"device_name": r["device_name"],
|
|
"ip": r["ip"],
|
|
"created_at": r["created_at"],
|
|
"last_seen_at": r["last_seen_at"],
|
|
"locked": r["locked_since"] is not None,
|
|
"screen": stream["screen"] if stream else None,
|
|
"packet_id": stream.get("packet_id") if stream else None,
|
|
}
|
|
)
|
|
return {"at": at, "sessions": out}
|
|
|
|
|
|
# ---------------------------------------------------------------- stamp check
|
|
|
|
RX_SIG_IN_TEXT = re.compile(r"stamp\s+([0-9a-fA-F]{16})\b")
|
|
RX_SIG_ALONE = re.compile(r"^\s*([0-9a-fA-F]{16})\s*$")
|
|
RX_UTC = re.compile(r"(\d{4}-\d{2}-\d{2}) (\d{2}:\d{2}) UTC")
|
|
TEXT_MAX = 4000
|
|
PDF_STAMPS_MAX = 50
|
|
|
|
|
|
class StampText(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
|
|
text: str = Field(max_length=TEXT_MAX)
|
|
|
|
|
|
def _squash(text: str) -> str:
|
|
return "".join(text.split())
|
|
|
|
|
|
def _sign_candidates(conn: Any, r: dict[str, Any]) -> list[tuple[Any, str, str | None]]:
|
|
"""(packet_id, signed kind, seal hash) the print could have been signed with, from its print_log kind."""
|
|
kind, pid = str(r["kind"]), r["packet_id"]
|
|
itp = f"itp-{r['itp_id']}" if r.get("itp_id") is not None else None
|
|
if kind == "report-pdf":
|
|
return [(None, "report", None)]
|
|
if kind == "itp-pdf":
|
|
return [(itp, "itp-pdf", None)]
|
|
if kind == "itp":
|
|
return [(itp, "print:itp", None)]
|
|
if kind == "databook-export":
|
|
seals: list[str | None] = [
|
|
bytes(x[0]).hex()
|
|
for x in conn.execute(
|
|
"SELECT hash FROM packet_seals WHERE packet_id = %s ORDER BY id DESC", (pid,)
|
|
).fetchall()
|
|
]
|
|
return [(pid, "databook", s) for s in [*seals, None]]
|
|
if kind == "traveler":
|
|
return [(pid, "traveler", None), (pid, "print:traveler", None)]
|
|
return [(pid, f"print:{kind}", None)]
|
|
|
|
|
|
def _minutes(r: dict[str, Any]) -> list[str]:
|
|
out: list[str] = []
|
|
m = RX_UTC.search(str(r.get("stamp") or ""))
|
|
if m:
|
|
out.append(f"{m[1]}T{m[2]}Z")
|
|
at = r["at"].astimezone(UTC)
|
|
out.append(stamps.minute_of(at))
|
|
return list(dict.fromkeys(out))
|
|
|
|
|
|
def _print_out(conn: Any, r: dict[str, Any]) -> dict[str, Any]:
|
|
label = None
|
|
if r["packet_id"] is not None:
|
|
row = conn.execute(
|
|
"SELECT pr.number, u.lot, p.drawing FROM packets p JOIN units u ON u.id = p.unit_id"
|
|
" JOIN projects pr ON pr.id = u.project_id WHERE p.id = %s",
|
|
(r["packet_id"],),
|
|
).fetchone()
|
|
if row:
|
|
label = " ".join(x for x in (row[0], f"Lot {row[1]}" if row[1] else "", row[2] or "") if x)
|
|
return {
|
|
"id": r["id"],
|
|
"at": r["at"],
|
|
"by_name": r["by_name"],
|
|
"kind": r["kind"],
|
|
"what": r["what"],
|
|
"pages": r["pages"],
|
|
"packet_id": r["packet_id"],
|
|
"packet_label": label,
|
|
}
|
|
|
|
|
|
def check_stamp(conn: Any, key: bytes | None, sig: str | None, shown: str | None) -> dict[str, Any]:
|
|
"""One stamp: the signature, and the text shown up to and including it (None when only the hex was given)."""
|
|
if not sig:
|
|
return {"found": False, "valid": False, "reason": "No stamp signature in the text.", "print": None}
|
|
sig = sig.lower()
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
|
rows = cur.execute("SELECT * FROM print_log WHERE stamp_sig = %s ORDER BY id", (sig,)).fetchall()
|
|
if not rows:
|
|
return {
|
|
"found": False,
|
|
"valid": False,
|
|
"reason": f"No print has stamp {sig}. The copy is not from Packet Station, or the stamp was changed.",
|
|
"print": None,
|
|
}
|
|
r = rows[0]
|
|
out = {"found": True, "valid": False, "reason": "", "print": _print_out(conn, r)}
|
|
if key is None:
|
|
out["reason"] = "The seal key is not available, so the signature cannot be checked."
|
|
return out
|
|
signed = any(
|
|
stamps.verify_stamp(key, pid, kind, seal, r["by"], minute, sig)
|
|
for pid, kind, seal in _sign_candidates(conn, r)
|
|
for minute in _minutes(r)
|
|
)
|
|
if not signed:
|
|
out["reason"] = "The signature does not match the print log. Treat the copy as not controlled."
|
|
return out
|
|
if shown is not None and r.get("stamp"):
|
|
logged, seen = _squash(str(r["stamp"])), _squash(shown)
|
|
if not (logged.endswith(seen) or seen.endswith(logged)):
|
|
out["reason"] = "The stamp text differs from the print log. The copy was changed."
|
|
return out
|
|
out["valid"] = True
|
|
out["reason"] = "The stamp matches the print log."
|
|
return out
|
|
|
|
|
|
def stamp_principal(p: Principal = Depends(current_principal)) -> Principal: # noqa: B008
|
|
"""Admin, every Q.C. role and every Clerical role (change C9, Gitea issue 32), on a password session as the Admin
|
|
path always needed."""
|
|
if not (p.is_admin or any(r.startswith(("qc", "clerical")) for r in p.roles)):
|
|
raise ApiError(403, "forbidden", "The stamp check is for Admin, Q.C. and Clerical.")
|
|
return require_password_session(p)
|
|
|
|
|
|
def _seal_key(request: Request) -> bytes | None:
|
|
try:
|
|
return settings_of(request).seal_key()
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
def _audit_verify(conn: Any, p: Principal, results: list[dict[str, Any]], source: str) -> None:
|
|
audit.append(
|
|
conn,
|
|
action="admin.stamp_verified",
|
|
user_id=p.user_id,
|
|
session_id=p.session_id,
|
|
role_snapshot=p.role_snapshot,
|
|
subject_type="print",
|
|
subject_id=",".join(str((x["print"] or {}).get("id", "")) for x in results)[:300] or None,
|
|
detail={
|
|
"source": source,
|
|
"results": [
|
|
{"found": x["found"], "valid": x["valid"], "print_id": (x["print"] or {}).get("id")} for x in results
|
|
],
|
|
},
|
|
ip=p.ip,
|
|
)
|
|
|
|
|
|
@router.post("/stamp-verify", dependencies=WRITE)
|
|
@stamp_router.post("/stamp-verify", dependencies=WRITE)
|
|
def stamp_verify(
|
|
body: StampText,
|
|
request: Request,
|
|
p: Principal = Depends(stamp_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
text = body.text
|
|
sig: str | None = None
|
|
shown: str | None = None
|
|
m = RX_SIG_IN_TEXT.search(text)
|
|
if m is not None:
|
|
sig, shown = m[1], text[: m.end()]
|
|
else:
|
|
alone = RX_SIG_ALONE.match(text)
|
|
if alone:
|
|
sig = alone[1]
|
|
res = check_stamp(d.conn, _seal_key(request), sig, shown)
|
|
_audit_verify(d.conn, p, [res], "text")
|
|
return res
|
|
|
|
|
|
def pdf_stamps(data: bytes) -> list[tuple[str, str]]:
|
|
"""(signature, the page text up to and including it) for every stamp in a PDF's text, first page first."""
|
|
from pypdf import PdfReader
|
|
from pypdf.errors import PdfReadError
|
|
|
|
try:
|
|
reader = PdfReader(io.BytesIO(data))
|
|
found: dict[str, str] = {}
|
|
for page in reader.pages:
|
|
text = page.extract_text() or ""
|
|
for m in RX_SIG_IN_TEXT.finditer(text):
|
|
sig = m[1].lower()
|
|
if sig not in found:
|
|
found[sig] = text[: m.end()]
|
|
if len(found) >= PDF_STAMPS_MAX:
|
|
break
|
|
except (PdfReadError, ValueError, KeyError, TypeError) as e:
|
|
raise ApiError(422, "bad_pdf", f"The PDF cannot be read: {type(e).__name__}.") from None
|
|
return list(found.items())
|
|
|
|
|
|
@router.post("/stamp-verify/pdf")
|
|
@stamp_router.post("/stamp-verify/pdf")
|
|
async def stamp_verify_pdf(
|
|
request: Request,
|
|
p: Principal = Depends(stamp_principal), # noqa: B008
|
|
d: Db = Depends(get_db), # noqa: B008
|
|
) -> dict[str, Any]:
|
|
length = request.headers.get("content-length")
|
|
if length is not None and length.isdigit() and int(length) > PDF_MAX_BYTES + FORM_SLACK:
|
|
raise ApiError(413, "too_large", "The PDF is larger than 25 MB.")
|
|
form = await request.form(max_files=1, max_fields=2)
|
|
try:
|
|
file = form.get("file")
|
|
if not isinstance(file, UploadFile):
|
|
raise ApiError(422, "invalid_request", "Send the PDF in the 'file' field.")
|
|
data = await file.read(PDF_MAX_BYTES + 1)
|
|
finally:
|
|
await form.close()
|
|
if len(data) > PDF_MAX_BYTES:
|
|
raise ApiError(413, "too_large", "The PDF is larger than 25 MB.")
|
|
if not data.startswith(b"%PDF-"):
|
|
raise ApiError(422, "not_pdf", "The file is not a PDF.")
|
|
found = await run_in_threadpool(pdf_stamps, data)
|
|
key = _seal_key(request)
|
|
|
|
def check_all() -> list[dict[str, Any]]:
|
|
results = [check_stamp(d.conn, key, sig, shown) for sig, shown in found]
|
|
_audit_verify(d.conn, p, results, "pdf")
|
|
return results
|
|
|
|
return {"stamps": await run_in_threadpool(check_all)}
|
|
|
|
|
|
# ---------------------------------------------------------------- directory flags
|
|
|
|
|
|
@router.get("/directory-flags")
|
|
def directory_flags(p: Principal = Depends(admin_principal), d: Db = Depends(get_db)) -> dict[str, Any]: # noqa: B008
|
|
conn = d.conn
|
|
with conn.cursor(row_factory=dict_row) as cur:
|
|
run = cur.execute(
|
|
"SELECT finished_at FROM job_runs WHERE job = 'ad-reconcile' AND finished_at IS NOT NULL"
|
|
" ORDER BY id DESC LIMIT 1"
|
|
).fetchone()
|
|
rows = cur.execute(
|
|
"SELECT id, sam_account, display_name, emp_no, directory_flags FROM users WHERE active"
|
|
" AND (coalesce((directory_flags->>'emp_missing')::bool, false)"
|
|
" OR coalesce((directory_flags->>'emp_duplicate')::bool, false))"
|
|
" ORDER BY emp_no NULLS FIRST, lower(display_name)"
|
|
).fetchall()
|
|
synced = cur.execute("SELECT max(ad_synced_at) AS at FROM users").fetchone()
|
|
missing: list[dict[str, Any]] = []
|
|
dupes: dict[str, list[dict[str, Any]]] = {}
|
|
for r in rows:
|
|
who = {"user_id": str(r["id"]), "sam_account": r["sam_account"], "display_name": r["display_name"]}
|
|
flags = r["directory_flags"] or {}
|
|
if flags.get("emp_missing"):
|
|
missing.append(who)
|
|
if flags.get("emp_duplicate") and r["emp_no"]:
|
|
dupes.setdefault(str(r["emp_no"]), []).append(who)
|
|
return {
|
|
"checked_at": (run or {}).get("finished_at") or (synced or {}).get("at"),
|
|
"missing": missing,
|
|
"duplicate": [{"emp_no": k, "users": v} for k, v in sorted(dupes.items())],
|
|
}
|