PacketStation/server/packetstation/api/admin_security.py

"""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())],
    }