"""Auditor services that don't fit the upstream-sync shape. * :class:`PrivilegedReviewService` — quarterly cycle CRUD + holder snapshot + per-holder decision flow * :class:`RiskScoreService` — recompute per-person risk scores from every upstream auditor signal * :class:`EvidencePackService` — async PDF + ZIP package builder * :class:`DashboardService` — KPI aggregations for the auditor landing page Plus list/query helpers used by the router for paginated endpoints (MFA list, public exposure list, secret rotation list, signin events, PAT audit, cert expiries, risk scores, evidence packs). All write operations write an audit log row; all RBAC enforcement is done at the router boundary via ``require_role(...)``. Inside the service layer we trust the caller — the layer's contract is "valid input → valid state transition + audit row". """ from __future__ import annotations import asyncio import csv import hashlib import io import os import uuid import zipfile from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from app.audit.service import AuditService from app.core.exceptions import AppException, NotFoundError from app.core.exporting import to_cell from app.core.safe_logging import get_safe_logger from app.domains.auditor.constants import ( ALL_EVIDENCE_MODULES, ANOMALY_LEVEL_CRITICAL, ANOMALY_LEVEL_HIGH, CERT_ALERT_30D, CERT_ALERT_60D, EVIDENCE_MODULE_ACCESS_EVIDENCE, EVIDENCE_MODULE_CERT_EXPIRY, EVIDENCE_MODULE_MFA, EVIDENCE_MODULE_PAT_AUDIT, EVIDENCE_MODULE_PRIVILEGED_REVIEW, EVIDENCE_MODULE_PUBLIC_EXPOSURE, EVIDENCE_MODULE_RISK_SCORE, EVIDENCE_MODULE_SECRET_ROTATION, EVIDENCE_MODULE_SIGNIN_ANOMALY, EVIDENCE_PACK_DEFAULT_ROOT, EVIDENCE_PACK_ROOT_ENV, EVIDENCE_RETENTION_DAYS, EVIDENCE_STATUS_COMPLETED, EVIDENCE_STATUS_FAILED, EVIDENCE_STATUS_GENERATING, EVIDENCE_STATUS_PENDING, EXPOSURE_SEVERITY_CRITICAL, EXPOSURE_STATUS_ACCEPTED_RISK, EXPOSURE_STATUS_OPEN, EXPOSURE_STATUS_RESOLVED, PAT_RISK_BROAD_SCOPE_ADMIN, PAT_RISK_DORMANT, PAT_RISK_OWNER_LEFT, PRIV_CYCLE_IN_PROGRESS, RISK_FACTOR_WEIGHTS, RISK_LEVEL_CRITICAL, RISK_LEVEL_HIGH, SECRET_STATUS_OVERDUE, risk_level_for_score, ) from app.domains.auditor.evidence_query.contracts import persisted_utc from app.domains.auditor.models import ( AccessCertification, AuditorCertExpiry, AuditorEvidencePack, AuditorMfaStatus, AuditorPatAudit, AuditorPublicExposureFinding, AuditorRiskScore, AuditorSecretRotationRecord, AuditorSigninEvent, ) from app.identity.person_models import IdentityPerson logger = get_safe_logger(__name__) def _utc_now() -> datetime: return datetime.now(UTC) # PrivilegedReviewService + AccessReviewEscalationService were folded # into the unified access-certification engine on 2026-06-04 (GRC PR3): # privileged-role review is now campaign_type='privileged_role' in # app/domains/auditor/access_certification_service.py. The old # auditor_privileged_* tables were dropped (migration 252). # ═════════════════════════════════════════════════════════════════════ # Risk score # ═════════════════════════════════════════════════════════════════════ class RiskScoreService: """Recompute per-active-person risk scores from every upstream auditor signal. Idempotent — overwrites the existing row per person. Score caps at 100. Level mapping in ``constants.risk_level_for_score``. """ @staticmethod async def recompute_all( session: AsyncSession, *, actor_user_id: uuid.UUID | None = None, trace_id: uuid.UUID | None = None, ) -> dict[str, Any]: now = _utc_now() persons = (await session.execute( select(IdentityPerson).where( # Includes left employees so we surface "logged in # after offboarding" cases. The risk score on a left # person is informational only — they cannot log in # if leaver_blocked & force_logout did their job. # # Tombstones are a different story: a soft-deleted row # keeps employment_status='active' here, so what used to # be a tautological ``id IS NOT NULL`` predicate scored a # purged duplicate every hour, and that ghost row counted # towards risk_critical_count → the endpoint_risk_scoring # control's pass/fail. IdentityPerson.deleted_at.is_(None), ) )).scalars().all() # Pre-fetch upstream signals once mfa_rows = (await session.execute(select(AuditorMfaStatus))).scalars().all() mfa_by_person: dict[uuid.UUID, list[AuditorMfaStatus]] = {} for r in mfa_rows: mfa_by_person.setdefault(r.person_id, []).append(r) signin_window = now - timedelta(days=30) signins = (await session.execute( select(AuditorSigninEvent).where( AuditorSigninEvent.event_at >= signin_window, ) )).scalars().all() signin_by_person: dict[uuid.UUID, list[AuditorSigninEvent]] = {} for r in signins: if r.person_id is not None: signin_by_person.setdefault(r.person_id, []).append(r) secrets_overdue = (await session.execute( select(AuditorSecretRotationRecord).where( AuditorSecretRotationRecord.status == SECRET_STATUS_OVERDUE, ) )).scalars().all() secret_by_owner: dict[uuid.UUID, int] = {} for r in secrets_overdue: if r.owner_person_id is not None: secret_by_owner[r.owner_person_id] = ( secret_by_owner.get(r.owner_person_id, 0) + 1 ) pat_rows = (await session.execute( select(AuditorPatAudit).where( AuditorPatAudit.person_id.is_not(None), ) )).scalars().all() pat_by_owner: dict[uuid.UUID, list[AuditorPatAudit]] = {} for r in pat_rows: pat_by_owner.setdefault(r.person_id, []).append(r) certs_expired = (await session.execute( select(AuditorCertExpiry).where( AuditorCertExpiry.alert_level == "expired", AuditorCertExpiry.owner_person_id.is_not(None), ) )).scalars().all() cert_by_owner: dict[uuid.UUID, int] = {} for r in certs_expired: cert_by_owner[r.owner_person_id] = ( cert_by_owner.get(r.owner_person_id, 0) + 1 ) recomputed = 0 high_count = 0 critical_count = 0 # Pre-fetch every existing score row once (#2188-1). The loop # below runs for the whole company every hour; a per-person # SELECT inside it is N round-trips for one indexed scan. existing_by_person = { row.person_id: row for row in (await session.execute( select(AuditorRiskScore) )).scalars().all() } for p in persons: score, factors, breakdown = RiskScoreService._score_person( p, mfa_records=mfa_by_person.get(p.id, []), signins=signin_by_person.get(p.id, []), overdue_secret_count=secret_by_owner.get(p.id, 0), pats=pat_by_owner.get(p.id, []), expired_cert_count=cert_by_owner.get(p.id, 0), ) level = risk_level_for_score(score) existing = existing_by_person.get(p.id) if existing is None: session.add(AuditorRiskScore( id=uuid.uuid4(), person_id=p.id, score=score, level=level, factors=factors, breakdown=breakdown, computed_at=now, )) else: existing.score = score existing.level = level existing.factors = factors existing.breakdown = breakdown existing.computed_at = now recomputed += 1 if level == RISK_LEVEL_CRITICAL: critical_count += 1 elif level == RISK_LEVEL_HIGH: high_count += 1 await session.flush() await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user" if actor_user_id else "system", actor_user_id=actor_user_id, action="auditor.risk_score.recompute", resource_type="auditor_risk_scores", status="success", detail={ "person_count": recomputed, "high_count": high_count, "critical_count": critical_count, }, ) return { "person_count": recomputed, "high_count": high_count, "critical_count": critical_count, "computed_at": now, } @staticmethod def _score_person( person: IdentityPerson, *, mfa_records: list[AuditorMfaStatus], signins: list[AuditorSigninEvent], overdue_secret_count: int, pats: list[AuditorPatAudit], expired_cert_count: int, ) -> tuple[int, dict[str, int], dict[str, Any]]: factors: dict[str, int] = {} breakdown: dict[str, Any] = {} def _add(name: str, hits: int = 1) -> None: weight, cap = RISK_FACTOR_WEIGHTS[name] current = factors.get(name, 0) factors[name] = min(cap, current + weight * hits) # MFA: per-provider missing penalty missing_providers = [ r.provider for r in mfa_records # Only observed non-enrolment is a risk factor; an unreadable # provider policy penalised 528 of 570 identities for a fact # Helix never established. if not r.is_enrolled and r.enrollment_observed ] if missing_providers: _add("mfa_missing_provider", hits=len(missing_providers)) breakdown["mfa_missing_providers"] = missing_providers # Signin anomalies high_anom = [ s for s in signins if s.anomaly_level in (ANOMALY_LEVEL_HIGH, ANOMALY_LEVEL_CRITICAL) ] if high_anom: _add("anomalous_signin_event", hits=len(high_anom)) breakdown["high_anomaly_event_ids"] = [ str(s.id) for s in high_anom[:10] ] # A signin is "post-offboard" only when (a) the person has left # and (b) the signin event happened on or after their termination # date. The previous filter omitted (b) and labelled every # historical signin of a left employee as post-offboard, which # both inflated the breakdown and hid the real signal. if person.employment_status == "left" and person.termination_date: term_date = person.termination_date post_offboard = [ s for s in signins if s.event_at is not None and s.event_at.date() >= term_date ] if post_offboard: # Each post-termination signin is a discrete violation; # scale the score by count (capped via the factor's cap). _add("post_offboard_signin", hits=len(post_offboard)) breakdown["post_offboard_signin_ids"] = [ str(s.id) for s in post_offboard[:10] ] breakdown["post_offboard_signin_count"] = len(post_offboard) # Secrets if overdue_secret_count: _add("overdue_secret", hits=overdue_secret_count) breakdown["overdue_secret_count"] = overdue_secret_count # PATs dormant = [ p for p in pats if p.risk_flags and PAT_RISK_DORMANT in p.risk_flags ] if dormant: _add("dormant_pat", hits=len(dormant)) breakdown["dormant_pat_ids"] = [str(p.id) for p in dormant[:10]] over_scoped = [ p for p in pats if p.risk_flags and PAT_RISK_BROAD_SCOPE_ADMIN in p.risk_flags ] if over_scoped: _add("over_scoped_pat", hits=len(over_scoped)) breakdown["over_scoped_pat_ids"] = [ str(p.id) for p in over_scoped[:10] ] owner_left = [ p for p in pats if p.risk_flags and PAT_RISK_OWNER_LEFT in p.risk_flags ] if owner_left: _add("owner_left_pat", hits=1) breakdown["owner_left_pat_ids"] = [ str(p.id) for p in owner_left[:10] ] # Certs if expired_cert_count: _add("expired_cert_owned", hits=expired_cert_count) breakdown["expired_cert_count"] = expired_cert_count score = min(100, sum(factors.values())) return score, factors, breakdown # ═════════════════════════════════════════════════════════════════════ # Evidence pack # ═════════════════════════════════════════════════════════════════════ def _csv_row(values: list[Any]) -> str: """One RFC-4180 CSV row (#2188-6). The evidence CSVs were assembled with ``",".join(...)``, so any field containing a comma, a quote or a newline broke the row's column alignment — silently, in a compliance artefact that goes to an auditor. Display names, project names and resource labels all contain commas in practice. ``csv.writer`` quotes and escapes them. Every cell also goes through the shared ``to_cell``, which is this repo's single source of truth for the neutralised formula-prefix set (``= + - @`` plus the ``\\t`` / ``\\r`` DDE vectors — see OWASP "CSV injection"). Quoting alone does NOT stop it: Excel evaluates a leading ``=`` inside a quoted field just the same. These CSVs carry ``subject_label``, i.e. ``IdentityPerson.display_name``, which arrives from HRIS — an upstream this platform does not control — and the artefact is a ZIP handed to an external auditor, who opens it in Excel. Five other exporters in this codebase already do this; the evidence pack was the one that never adopted it. """ buf = io.StringIO() csv.writer(buf, lineterminator="").writerow( ["" if v is None else str(to_cell(v)) for v in values] ) return buf.getvalue() def _evidence_root() -> str: root = os.environ.get(EVIDENCE_PACK_ROOT_ENV) or EVIDENCE_PACK_DEFAULT_ROOT os.makedirs(root, exist_ok=True) return root class EvidencePackService: """Async PDF + ZIP package builder. Caller (router) creates an ``AuditorEvidencePack`` row in ``pending`` state and ENQUEUES generation. The worker (scheduled or on-demand) picks up pending rows and runs ``generate(row_id)`` which marks generating → completed/failed. """ @staticmethod async def request_build( session: AsyncSession, *, title: str, period_start: datetime, period_end: datetime, modules: list[str], actor_user_id: uuid.UUID, trace_id: uuid.UUID | None = None, ) -> AuditorEvidencePack: if period_end <= period_start: raise AppException( code="INVALID_PERIOD", message="period_end must be after period_start", status_code=422, ) invalid = [m for m in modules if m not in ALL_EVIDENCE_MODULES] if invalid: raise AppException( code="INVALID_MODULES", message=f"unknown evidence modules: {invalid}", status_code=422, ) retention = _utc_now() + timedelta(days=EVIDENCE_RETENTION_DAYS) pack = AuditorEvidencePack( id=uuid.uuid4(), requested_by_person_id=actor_user_id, title=title, period_start=period_start, period_end=period_end, modules=list(modules), status=EVIDENCE_STATUS_PENDING, retention_until=retention, ) session.add(pack) await session.flush() await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user", actor_user_id=actor_user_id, action="auditor.evidence_pack.request", resource_type="auditor_evidence_packs", resource_id=str(pack.id), status="success", detail={"title": title, "modules": modules}, ) return pack @staticmethod async def generate( session: AsyncSession, *, pack_id: uuid.UUID, ) -> AuditorEvidencePack: """Generate a pending pack synchronously. Caller is the worker loop or an admin "build now" endpoint.""" pack = await session.get(AuditorEvidencePack, pack_id) if pack is None: raise NotFoundError("AuditorEvidencePack", str(pack_id)) if pack.status != EVIDENCE_STATUS_PENDING: return pack pack.status = EVIDENCE_STATUS_GENERATING pack.started_at = _utc_now() await session.flush() try: file_path, file_size, sha256 = await EvidencePackService._build_pack( session, pack=pack, ) pack.file_path = file_path pack.file_size = file_size pack.checksum_sha256 = sha256 pack.status = EVIDENCE_STATUS_COMPLETED pack.completed_at = _utc_now() except Exception as exc: # noqa: BLE001 logger.exception("evidence pack %s build failed", pack_id) pack.status = EVIDENCE_STATUS_FAILED pack.error_message = str(exc)[:1000] pack.completed_at = _utc_now() await session.flush() await AuditService(session).log( trace_id=uuid.uuid4(), actor_type="system", action=f"auditor.evidence_pack.{pack.status}", resource_type="auditor_evidence_packs", resource_id=str(pack.id), status=( "success" if pack.status == EVIDENCE_STATUS_COMPLETED else "failure" ), detail={ "title": pack.title, "file_size": pack.file_size, "error": pack.error_message, }, ) return pack @staticmethod async def _build_pack( session: AsyncSession, *, pack: AuditorEvidencePack, ) -> tuple[str, int, str]: """Generate the PDF + ZIP on disk; return ``(path, size, sha256)``. Per-module sections — each module renders both a PDF chapter AND a CSV export inside the ZIP. The PDF is the human-readable narrative; the CSV is for offline tooling. """ from reportlab.lib import colors from reportlab.lib.pagesizes import A4 from reportlab.lib.styles import getSampleStyleSheet from reportlab.platypus import ( PageBreak, Paragraph, SimpleDocTemplate, Spacer, Table, TableStyle, ) root = _evidence_root() out_dir = os.path.join(root, str(pack.id)) os.makedirs(out_dir, exist_ok=True) pdf_path = os.path.join(out_dir, "evidence.pdf") zip_path = os.path.join(out_dir, "evidence_pack.zip") styles = getSampleStyleSheet() story: list[Any] = [] csv_files: list[tuple[str, str]] = [] # (filename, csv_text) story.append(Paragraph(pack.title, styles["Title"])) story.append(Spacer(1, 12)) story.append(Paragraph( f"Period: {pack.period_start.isoformat()} → " f"{pack.period_end.isoformat()}", styles["Normal"], )) story.append(Paragraph( f"Modules included: {', '.join(pack.modules)}", styles["Normal"], )) story.append(Paragraph( f"Generated at: {datetime.now(UTC).isoformat()}", styles["Normal"], )) story.append(PageBreak()) for module_name in pack.modules: section = await EvidencePackService._render_module( session, module_name=module_name, pack=pack, ) story.append(Paragraph(section["title"], styles["Heading1"])) story.append(Spacer(1, 8)) story.append(Paragraph(section["summary"], styles["Normal"])) story.append(Spacer(1, 8)) data = section["table"] if data: tbl = Table(data, repeatRows=1) tbl.setStyle(TableStyle([ ("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey), ("GRID", (0, 0), (-1, -1), 0.25, colors.grey), ("FONTSIZE", (0, 0), (-1, -1), 8), ("VALIGN", (0, 0), (-1, -1), "TOP"), ])) story.append(tbl) story.append(PageBreak()) csv_files.append(( f"{module_name}.csv", section["csv"], )) # reportlab rendering and the zip/sha pass are CPU-bound and # SYNCHRONOUS — on a big pack they block the event loop for # seconds, and the "build now" endpoint reaches this straight # from an API worker, so every other request on that process # stalls with it (#2188-3). Push the whole file-producing tail # onto a thread; nothing in it touches the DB session. def _render_and_package() -> tuple[str, int, str]: doc = SimpleDocTemplate(pdf_path, pagesize=A4) doc.build(story) with zipfile.ZipFile( zip_path, "w", zipfile.ZIP_DEFLATED, ) as zf: zf.write(pdf_path, "evidence.pdf") for name, text in csv_files: zf.writestr(name, text) zf.writestr("manifest.json", _build_manifest(pack)) size_bytes = os.path.getsize(zip_path) with open(zip_path, "rb") as f: digest = hashlib.sha256(f.read()).hexdigest() return zip_path, size_bytes, digest return await asyncio.to_thread(_render_and_package) @staticmethod async def _render_module( session: AsyncSession, *, module_name: str, pack: AuditorEvidencePack, ) -> dict[str, Any]: """Return dict with title / summary / table / csv for a module.""" if module_name == EVIDENCE_MODULE_MFA: rows = (await session.execute( select(AuditorMfaStatus).order_by( AuditorMfaStatus.is_enrolled, AuditorMfaStatus.provider, ) )).scalars().all() table = [["person_id", "provider", "enrolled", "enforced", "last_synced_at"]] csv_lines = ["person_id,provider,enrolled,enforced,last_synced_at"] for r in rows: table.append([ str(r.person_id), r.provider, "yes" if r.is_enrolled else "no", "yes" if r.is_enforced else "no", r.last_synced_at.isoformat() if r.last_synced_at else "", ]) csv_lines.append(_csv_row([ str(r.person_id), r.provider, "1" if r.is_enrolled else "0", "1" if r.is_enforced else "0", r.last_synced_at.isoformat() if r.last_synced_at else "", ])) return { "title": "MFA Compliance", "summary": ( f"{len(rows)} (person, provider) MFA rows captured. " f"Unenrolled rows are listed first." ), "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_PRIVILEGED_REVIEW: cycles = (await session.execute( select(AccessCertification).where( AccessCertification.campaign_type == "privileged_role", ).order_by(AccessCertification.started_at.desc()) )).scalars().all() table = [["cycle_label", "status", "started_at", "due_at", "decided", "revoked"]] csv_lines = ["cycle_label,status,started_at,due_at," "snapshot,decided,revoked"] for c in cycles: table.append([ c.label, c.status, c.started_at.isoformat(), c.due_at.isoformat(), str(c.decided_count), str(c.revoked_count), ]) csv_lines.append(_csv_row([ c.label, c.status, c.started_at.isoformat(), c.due_at.isoformat(), str(c.snapshot_item_count), str(c.decided_count), str(c.revoked_count), ])) return { "title": "Privileged Role Quarterly Review", "summary": f"{len(cycles)} review cycles on record.", "table": table, "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_PUBLIC_EXPOSURE: rows = (await session.execute( select(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.last_seen_at >= pack.period_start, AuditorPublicExposureFinding.last_seen_at <= pack.period_end, ).order_by(AuditorPublicExposureFinding.severity.desc()) )).scalars().all() table = [["source", "resource_id", "exposure_type", "severity", "status", "last_seen_at"]] csv_lines = ["source,resource_id,exposure_type,severity," "status,first_seen,last_seen,resolved_at"] for r in rows: table.append([ r.source, r.resource_id, r.exposure_type, r.severity, r.status, r.last_seen_at.isoformat(), ]) csv_lines.append(_csv_row([ r.source, r.resource_id, r.exposure_type, r.severity, r.status, r.first_seen_at.isoformat(), r.last_seen_at.isoformat(), r.resolved_at.isoformat() if r.resolved_at else "", ])) return { "title": "Public Exposure Findings", "summary": ( f"{len(rows)} findings within the period." ), "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_SECRET_ROTATION: rows = (await session.execute( select(AuditorSecretRotationRecord).order_by( AuditorSecretRotationRecord.status.desc(), AuditorSecretRotationRecord.next_rotation_due_at, ) )).scalars().all() table = [["secret_kind", "source", "resource_id", "owner_email", "status", "next_due"]] csv_lines = ["kind,source,resource,owner_email,last_rotated," "next_due,status,days_until"] for r in rows: table.append([ r.secret_kind, r.source, r.resource_id, r.owner_email or "", r.status, r.next_rotation_due_at.isoformat() if r.next_rotation_due_at else "", ]) csv_lines.append(_csv_row([ r.secret_kind, r.source, r.resource_id, r.owner_email or "", r.last_rotated_at.isoformat() if r.last_rotated_at else "", r.next_rotation_due_at.isoformat() if r.next_rotation_due_at else "", r.status, # The int, NOT ``str(...)``. Pre-stringifying makes # ``to_cell`` see "-12" and neutralise the leading # minus, so every OVERDUE secret — the only ones with # a negative value, and the rows an auditor sorts on # — arrived as apostrophe-prefixed text. ``to_cell`` # passes ints through untouched. r.days_until_due if r.days_until_due is not None else "", ])) return { "title": "Secret Rotation Ledger", "summary": f"{len(rows)} secrets tracked.", "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_SIGNIN_ANOMALY: rows = (await session.execute( select(AuditorSigninEvent).where( AuditorSigninEvent.event_at >= pack.period_start, AuditorSigninEvent.event_at <= pack.period_end, AuditorSigninEvent.anomaly_level.in_( [ANOMALY_LEVEL_HIGH, ANOMALY_LEVEL_CRITICAL] ), ).order_by(AuditorSigninEvent.event_at.desc()) )).scalars().all() table = [["event_at", "provider", "person_id", "status", "anomaly_level", "score"]] csv_lines = ["event_at,provider,person_id,email,status," "level,score,reasons"] for r in rows: table.append([ r.event_at.isoformat(), r.provider, str(r.person_id) if r.person_id else "", r.status, r.anomaly_level, str(r.anomaly_score), ]) csv_lines.append(_csv_row([ r.event_at.isoformat(), r.provider, str(r.person_id) if r.person_id else "", r.provider_email or "", r.status, r.anomaly_level, str(r.anomaly_score), "|".join(r.anomaly_reasons or []), ])) return { "title": "High-Anomaly Signin Events", "summary": f"{len(rows)} high/critical anomaly events.", "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_PAT_AUDIT: rows = (await session.execute( select(AuditorPatAudit).order_by( AuditorPatAudit.kind, AuditorPatAudit.last_used_at, ) )).scalars().all() table = [["kind", "source", "owner_email", "scopes", "last_used", "expires", "status"]] csv_lines = ["kind,source,token_id,owner_email,scopes," "last_used,expires,status,risk_flags"] for r in rows: table.append([ r.kind, r.source, r.owner_email or "", ",".join(r.scopes or [])[:60], r.status, r.expires_at.isoformat() if r.expires_at else "", r.status, ]) csv_lines.append(_csv_row([ r.kind, r.source, r.token_external_id, r.owner_email or "", "|".join(r.scopes or []), r.last_used_at.isoformat() if r.last_used_at else "", r.expires_at.isoformat() if r.expires_at else "", r.status, "|".join(r.risk_flags or []), ])) return { "title": "PAT / MCP Token Audit", "summary": f"{len(rows)} tokens tracked.", "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_CERT_EXPIRY: rows = (await session.execute( select(AuditorCertExpiry).order_by( AuditorCertExpiry.expires_at, ) )).scalars().all() table = [["asset_kind", "name", "issuer", "expires_at", "alert_level", "status"]] csv_lines = ["kind,source,external_id,name,issuer," "expires_at,alert_level,status,owner_email"] for r in rows: table.append([ r.asset_kind, r.name, r.issuer or "", r.expires_at.isoformat(), r.alert_level, r.status, ]) csv_lines.append(_csv_row([ r.asset_kind, r.source, r.external_id, r.name, r.issuer or "", r.expires_at.isoformat(), r.alert_level, r.status, r.owner_email or "", ])) return { "title": "Certificate / ICP Expiry", "summary": f"{len(rows)} certs / domains / ICP filings.", "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_RISK_SCORE: rows = (await session.execute( select(AuditorRiskScore).order_by( AuditorRiskScore.score.desc() ) )).scalars().all() table = [["person_id", "score", "level", "computed_at", "factor_count"]] csv_lines = ["person_id,score,level,computed_at,factors"] for r in rows: table.append([ str(r.person_id), str(r.score), r.level, r.computed_at.isoformat(), str(len(r.factors or {})), ]) csv_lines.append(_csv_row([ str(r.person_id), str(r.score), r.level, r.computed_at.isoformat(), "|".join(f"{k}={v}" for k, v in (r.factors or {}).items()), ])) return { "title": "Per-Person Risk Score", "summary": f"{len(rows)} scored persons.", "table": table[:200], "csv": "\n".join(csv_lines), } if module_name == EVIDENCE_MODULE_ACCESS_EVIDENCE: return await EvidencePackService._render_access_evidence( session, pack=pack, ) # Should never reach here — modules validated at request time return { "title": module_name, "summary": "(unknown module)", "table": [], "csv": "", } @staticmethod async def _render_access_evidence( session: AsyncSession, *, pack: AuditorEvidencePack, ) -> dict[str, Any]: """Render the frozen evidence bundles covering the pack's period. Deliberately renders **stored** bundles rather than re-running the queries: a pack that re-derived its numbers at build time would disagree with the bundle an auditor already reviewed, and the hash printed alongside would no longer verify anything. Each row carries the citation (source table + pk, or audit-log id) so a reviewer working from the CSV can open the originating record rather than trust the summary. """ from app.domains.auditor.models import ( AuditorEvidenceBundle, AuditorEvidenceBundleRow, ) # Selected on the period the evidence COVERS, not on when the # bundle was written — but only for the SCHEDULED, company-wide # producer, which is the one those two columns exist for. The # monthly pre-compile runs at 03:20 on the 2nd, so May's # evidence is created in June; filtering everything on # ``created_at`` excluded every pre-compiled bundle for the # pack's own month and pulled in the previous month's, leaving a # May pack rendering "0 frozen evidence bundle(s)". # # Manual bundles stay on ``created_at``. A plain interval # overlap put no upper bound on the bundle's own window, so one # auditor running ``person_footprint`` over 2020-to-now — a # named individual's complete access history, display names in # the clear — would then appear in EVERY monthly pack from 2020 # onward. "When someone went looking" is the right axis for a # question someone asked by hand; "what period it covers" is the # right axis for routine collection. # Strict overlap on the period the evidence COVERS. The pack's # own range is expected to be whole days — the picker sends # start-of-day / end-of-day — so a scheduled bundle, whose # window IS one calendar month, lands in exactly the packs whose # range touches that month. # # An earlier version padded both bounds by a day to rescue a # UTC+8 picker that serialised "May" to 04-30T16:00Z.. # 05-30T16:00Z. That padding pulled in BOTH neighbours: April's # period_end and June's period_start each sat exactly on a padded # edge, so a May pack shipped three months of evidence, two of # them wrong. The picker is the thing that was broken; it now # sends whole days, and this predicate stays honest. # The offboarding archive rides the coverage axis too. Its # window is a single point — the leaver's last day — so it # carries none of the wide-window risk a hand-run footprint # does, and putting it on ``created_at`` filed a 29 May leaver # whose offboarding completed on 3 June into the JUNE pack. _COVERED = ("scheduled", "offboarding") scheduled_for_period = and_( AuditorEvidenceBundle.generated_via.in_(_COVERED), AuditorEvidenceBundle.period_start.is_not(None), AuditorEvidenceBundle.period_start <= pack.period_end, or_( AuditorEvidenceBundle.period_end.is_(None), AuditorEvidenceBundle.period_end >= pack.period_start, ), ) # A scheduled bundle with no period at all cannot be placed on # the coverage axis, so it falls back to when it was produced # rather than through both branches into nothing. undated_scheduled = and_( AuditorEvidenceBundle.generated_via.in_(_COVERED), AuditorEvidenceBundle.period_start.is_(None), AuditorEvidenceBundle.created_at >= pack.period_start, AuditorEvidenceBundle.created_at <= pack.period_end, ) # Manual and offboarding bundles stay on ``created_at``. A plain # overlap put no upper bound on the BUNDLE's window, so one # auditor running person_footprint over 2020-to-now — a named # individual's complete access history, names in the clear — # would appear in every monthly pack since 2020. asked_in_period = and_( AuditorEvidenceBundle.generated_via.not_in(_COVERED), AuditorEvidenceBundle.created_at >= pack.period_start, AuditorEvidenceBundle.created_at <= pack.period_end, ) bundles = list((await session.execute( select(AuditorEvidenceBundle) .where(or_( scheduled_for_period, undated_scheduled, asked_in_period, )) .order_by(AuditorEvidenceBundle.created_at) )).scalars().all()) table: list[list[str]] = [[ "bundle_id", "query", "version", "rows", "content_hash", "generated_via", "created_at", ]] csv_lines = [_csv_row([ "bundle_id", "query_key", "query_version", "seq", "subject_type", "subject_label", "source_table", "source_pk", "audit_log_id", "occurred_at", ])] total_rows = 0 for bundle in bundles: table.append([ str(bundle.id)[:8], bundle.query_key, bundle.query_version, str(bundle.row_count), bundle.content_hash[:16], bundle.generated_via, bundle.created_at.isoformat(), ]) rows = list((await session.execute( select(AuditorEvidenceBundleRow) .where(AuditorEvidenceBundleRow.bundle_id == bundle.id) .order_by(AuditorEvidenceBundleRow.seq) )).scalars().all()) total_rows += len(rows) for row in rows: # ``_csv_row`` and not ``",".join(...)``: subject labels # carry display names, and a comma in one silently # shifts every later column in a compliance artefact. csv_lines.append(_csv_row([ str(bundle.id), bundle.query_key, bundle.query_version, row.seq, row.subject_type, row.subject_label or "", row.source_table or "", row.source_pk or "", str(row.audit_log_id) if row.audit_log_id else "", # UTC-explicit: the column stores no offset, and an # external auditor re-hashing this CSV must get the # digest that was frozen with the bundle. persisted_utc(row.occurred_at).isoformat() if row.occurred_at else "", ])) gap_count = sum(len(b.data_gaps or []) for b in bundles) summary = ( f"{len(bundles)} frozen evidence bundle(s), {total_rows} cited " f"row(s)." ) if gap_count: # Surfaced in the PDF narrative, not just the CSV: a reader # skimming the chapter must see that some of it is declared # incomplete. summary += f" {gap_count} declared data gap(s) — see bundle detail." return { "title": "Access Evidence (frozen bundles)", "summary": summary, "table": table[:200], "csv": "\n".join(csv_lines), } def _build_manifest(pack: AuditorEvidencePack) -> str: import json return json.dumps({ "id": str(pack.id), "title": pack.title, "period_start": pack.period_start.isoformat(), "period_end": pack.period_end.isoformat(), "modules": pack.modules, "requested_by_person_id": str(pack.requested_by_person_id), "generated_at": ( pack.completed_at.isoformat() if pack.completed_at else _utc_now().isoformat() ), }, indent=2) # ═════════════════════════════════════════════════════════════════════ # Dashboard summary # ═════════════════════════════════════════════════════════════════════ class DashboardService: @staticmethod async def summary(session: AsyncSession) -> dict[str, Any]: now = _utc_now() # MFA coverage population — scope to ACTIVE, non-deleted employees, # matching the per-person worklist (`reconcile_mfa_findings`). The # MFA sync upserts a row for every person and never prunes leavers, # and a departed employee's suspended Google account reports # `is_enrolled=False`. Counting those dragged `mfa_coverage` below # the 95% SOC2/ISO threshold and could FAIL the control on people # who are out of scope — the dashboard count and the worklist must # agree on the population. # Third population rule, same reason as the two above: a row whose # provider never reported enrollment (Atlassian ``policy_absent``) # is not evidence of a missing enrolment. Counting those made the # ``mfa_coverage`` ISMS control read ``0/412 enrolled (0.0%)`` and # FAIL on people who were in fact covered by Google 2SV. _mfa_active = ( select(AuditorMfaStatus.id) .join(IdentityPerson, AuditorMfaStatus.person_id == IdentityPerson.id) .where( IdentityPerson.employment_status == "active", IdentityPerson.deleted_at.is_(None), AuditorMfaStatus.enrollment_observed.is_(True), ) ) mfa_total = int((await session.execute( select(func.count()).select_from(_mfa_active.subquery()) )).scalar_one()) mfa_unenrolled = int((await session.execute( select(func.count()).select_from( _mfa_active.where(AuditorMfaStatus.is_enrolled.is_(False)).subquery() ) )).scalar_one()) # Privileged review — now a privileged_role access-certification # campaign. Returned as a dict mapping the legacy # PrivilegedReviewCycleSummary field names so the dashboard schema # is unchanged. _priv = (await session.execute( select(AccessCertification).where( AccessCertification.campaign_type == "privileged_role", AccessCertification.status == PRIV_CYCLE_IN_PROGRESS, ).order_by( AccessCertification.started_at.desc() ).limit(1) )).scalar_one_or_none() cycle = None if _priv is None else { "id": _priv.id, "cycle_label": _priv.label, "status": _priv.status, "started_at": _priv.started_at, "due_at": _priv.due_at, "completed_at": _priv.completed_at, "snapshot_holder_count": _priv.snapshot_item_count, "decided_count": _priv.decided_count, "revoked_count": _priv.revoked_count, "notes": _priv.notes, } # Public exposure pubex_open = int((await session.execute( select(func.count()).select_from(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.status == EXPOSURE_STATUS_OPEN, ) )).scalar_one()) pubex_critical = int((await session.execute( select(func.count()).select_from(AuditorPublicExposureFinding).where( AuditorPublicExposureFinding.status == EXPOSURE_STATUS_OPEN, AuditorPublicExposureFinding.severity == EXPOSURE_SEVERITY_CRITICAL, ) )).scalar_one()) # Secret rotation secrets_overdue = int((await session.execute( select(func.count()).select_from(AuditorSecretRotationRecord).where( AuditorSecretRotationRecord.status == SECRET_STATUS_OVERDUE, ) )).scalar_one()) # Signin twenty_four_h_ago = now - timedelta(hours=24) signin_high = int((await session.execute( select(func.count()).select_from(AuditorSigninEvent).where( AuditorSigninEvent.event_at >= twenty_four_h_ago, AuditorSigninEvent.anomaly_level.in_( [ANOMALY_LEVEL_HIGH, ANOMALY_LEVEL_CRITICAL] ), ) )).scalar_one()) # PAT pat_risky = int((await session.execute( select(func.count()).select_from(AuditorPatAudit).where( AuditorPatAudit.risk_flags.is_not(None), ) )).scalar_one()) # Cert cert_expiring = int((await session.execute( select(func.count()).select_from(AuditorCertExpiry).where( AuditorCertExpiry.alert_level.in_( [CERT_ALERT_30D, CERT_ALERT_60D] ), ) )).scalar_one()) # Risk risk_critical = int((await session.execute( select(func.count()).select_from(AuditorRiskScore).where( AuditorRiskScore.level == RISK_LEVEL_CRITICAL, ) )).scalar_one()) risk_high = int((await session.execute( select(func.count()).select_from(AuditorRiskScore).where( AuditorRiskScore.level == RISK_LEVEL_HIGH, ) )).scalar_one()) # Last evidence pack last_ep = (await session.execute( select(AuditorEvidencePack).order_by( AuditorEvidencePack.created_at.desc() ).limit(1) )).scalar_one_or_none() return { "mfa_total_count": mfa_total, "mfa_unenrolled_count": mfa_unenrolled, "privileged_open_cycle": cycle, "public_exposure_open_count": pubex_open, "public_exposure_critical_count": pubex_critical, "secret_overdue_count": secrets_overdue, "signin_high_anomaly_24h": signin_high, "pat_risky_count": pat_risky, "cert_expiring_30d": cert_expiring, "risk_critical_count": risk_critical, "risk_high_count": risk_high, "last_evidence_pack": last_ep, } # ═════════════════════════════════════════════════════════════════════ # Resolve helpers (router glue) # ═════════════════════════════════════════════════════════════════════ class ExposureService: @staticmethod async def resolve( session: AsyncSession, *, finding_id: uuid.UUID, resolution: str, note: str, actor_user_id: uuid.UUID, trace_id: uuid.UUID | None = None, ) -> AuditorPublicExposureFinding: if resolution not in ( EXPOSURE_STATUS_RESOLVED, EXPOSURE_STATUS_ACCEPTED_RISK, ): raise AppException( code="INVALID_RESOLUTION", message="resolution must be 'resolved' or 'accepted_risk'", status_code=422, ) f = await session.get(AuditorPublicExposureFinding, finding_id) if f is None: raise NotFoundError( "AuditorPublicExposureFinding", str(finding_id), ) f.status = resolution f.resolved_at = _utc_now() f.resolved_by_person_id = actor_user_id f.resolution_note = note from app.domains.auditor import findings_service as _findings await session.flush() await _findings.mirror_exposure_row(session, f) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user", actor_user_id=actor_user_id, action=f"auditor.public_exposure.{resolution}", resource_type="auditor_public_exposure_findings", resource_id=str(f.id), status="success", detail={ "source": f.source, "resource_id": f.resource_id, "note": note, }, ) return f class SigninAcknowledgeService: @staticmethod async def acknowledge( session: AsyncSession, *, event_id: uuid.UUID, note: str, actor_user_id: uuid.UUID, trace_id: uuid.UUID | None = None, ) -> AuditorSigninEvent: ev = await session.get(AuditorSigninEvent, event_id) if ev is None: raise NotFoundError("AuditorSigninEvent", str(event_id)) ev.acknowledged_at = _utc_now() ev.acknowledged_by_person_id = actor_user_id ev.acknowledged_note = note from app.domains.auditor import findings_service as _findings await session.flush() await _findings.mirror_signin_row(session, ev) await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user", actor_user_id=actor_user_id, action="auditor.signin.acknowledge", resource_type="auditor_signin_events", resource_id=str(ev.id), status="success", detail={ "provider": ev.provider, "anomaly_score": ev.anomaly_score, "anomaly_level": ev.anomaly_level, "note": note, }, ) return ev class CertManualService: @staticmethod async def upsert_manual( session: AsyncSession, *, asset_kind: str, external_id: str, name: str, subject: str | None, issuer: str | None, issued_at: datetime | None, expires_at: datetime, owner_person_id: uuid.UUID | None, owner_email: str | None, environment: str | None, notes: str | None, actor_user_id: uuid.UUID, trace_id: uuid.UUID | None = None, ) -> AuditorCertExpiry: from app.domains.auditor.constants import ( CERT_SOURCE_MANUAL, CERT_STATUS_ACTIVE, CERT_STATUS_EXPIRED, cert_alert_level_for_days, ) days_until = int( (expires_at - _utc_now()).total_seconds() // 86400 ) alert = cert_alert_level_for_days(days_until) status = CERT_STATUS_EXPIRED if days_until < 0 else CERT_STATUS_ACTIVE existing = (await session.execute( select(AuditorCertExpiry).where( AuditorCertExpiry.source == CERT_SOURCE_MANUAL, AuditorCertExpiry.asset_kind == asset_kind, AuditorCertExpiry.external_id == external_id, ) )).scalar_one_or_none() if existing is None: row = AuditorCertExpiry( id=uuid.uuid4(), asset_kind=asset_kind, source=CERT_SOURCE_MANUAL, external_id=external_id, name=name, subject=subject, issuer=issuer, issued_at=issued_at, expires_at=expires_at, days_until_expiry=days_until, alert_level=alert, status=status, owner_person_id=owner_person_id, owner_email=owner_email, environment=environment, notes=notes, last_synced_at=_utc_now(), ) session.add(row) await session.flush() else: row = existing row.name = name row.subject = subject row.issuer = issuer row.issued_at = issued_at row.expires_at = expires_at row.days_until_expiry = days_until row.alert_level = alert row.status = status row.owner_person_id = owner_person_id row.owner_email = owner_email row.environment = environment row.notes = notes row.last_synced_at = _utc_now() await AuditService(session).log( trace_id=trace_id or uuid.uuid4(), actor_type="user", actor_user_id=actor_user_id, action="auditor.cert_expiry.manual_upsert", resource_type="auditor_cert_expiries", resource_id=str(row.id), status="success", detail={ "asset_kind": asset_kind, "external_id": external_id, "name": name, "expires_at": expires_at.isoformat(), }, ) return row