"""GRC spine service — catalog reconcile, control evaluation, posture. This is the engine behind the redesigned 安全合规 menu. It does three things: 1. ``sync_compliance_catalog`` — idempotently upserts frameworks / requirements / controls / crosswalk from :mod:`compliance_catalog` (the single source of truth). Run by the recompute cron and on demand, so editing the catalog propagates without a migration. 2. ``recompute_control_status`` — evaluates every enabled control from its backing signal (detective-job freshness + the metric the job produces) and upserts a ``compliance_control_status`` row. This is continuous control monitoring (CCM). 3. Read queries — posture summary, per-framework readiness, control catalog — projections over controls × statuses × the crosswalk. No duplicated state: a framework's readiness is derived, never stored. Datetime discipline: the codebase round-trips ``DateTime(timezone=True)`` columns as naive on fresh load, so this module computes in **naive UTC** throughout to avoid the aware/naive subtraction class of 500s. """ from __future__ import annotations import uuid from datetime import UTC, datetime, timedelta from typing import Any from sqlalchemy import select from sqlalchemy.dialects.mysql import insert as mysql_insert from sqlalchemy.ext.asyncio import AsyncSession from app.domains.auditor import compliance_catalog as cat from app.domains.auditor.models import ( AccessCertification, ComplianceControl, ComplianceControlRequirement, ComplianceControlStatus, ComplianceFramework, ComplianceRequirement, ) from app.domains.auditor.services import DashboardService from app.scheduled_jobs.models import ScheduledJob def _naive_now() -> datetime: return datetime.now(UTC).replace(tzinfo=None) def _to_naive(dt: datetime | None) -> datetime | None: if dt is None: return None if dt.tzinfo is not None: return dt.astimezone(UTC).replace(tzinfo=None) return dt # How old the feeding signal may be before a control is "stale". Keyed # by control key; chosen ~2× the job's natural cadence so a single missed # tick doesn't flip the control. _STALE_AFTER_HOURS: dict[str, int] = { "mfa_coverage": 48, "privileged_access_review": 24 * 100, # quarterly "pat_audit": 48, "public_exposure_scan": 48, "endpoint_risk_scoring": 4, "secret_rotation": 48, "cert_expiry_monitoring": 48, "signin_anomaly_monitoring": 4, } # ════════════════════════════════════════════════════════════════════ # Catalog reconcile # ════════════════════════════════════════════════════════════════════ async def sync_compliance_catalog(db: AsyncSession) -> dict[str, int]: """Idempotently upsert the GRC catalog from :mod:`compliance_catalog`. Safe to call repeatedly. Resolves all FKs by natural key in Python. Never deletes — a framework/control removed from the catalog stays in the DB (manual cleanup), so we never orphan an auditor's notes. """ counts = {"frameworks": 0, "requirements": 0, "controls": 0, "mappings": 0} # ── Frameworks ────────────────────────────────────────────────── existing_fw = { f.key: f for f in (await db.execute(select(ComplianceFramework))).scalars().all() } fw_by_key: dict[str, ComplianceFramework] = dict(existing_fw) for spec in cat.FRAMEWORKS: row = existing_fw.get(spec["key"]) if row is None: row = ComplianceFramework( key=spec["key"], name=spec["name"], version=spec.get("version"), description=spec.get("description"), sort_order=spec.get("sort_order", 0), ) db.add(row) fw_by_key[spec["key"]] = row counts["frameworks"] += 1 else: row.name = spec["name"] row.version = spec.get("version") row.description = spec.get("description") row.sort_order = spec.get("sort_order", 0) await db.flush() # ── Requirements ──────────────────────────────────────────────── existing_reqs = ( await db.execute(select(ComplianceRequirement)) ).scalars().all() req_by_fk_ref: dict[tuple[Any, str], ComplianceRequirement] = { (r.framework_id, r.ref): r for r in existing_reqs } for fw_key, reqs in cat.REQUIREMENTS.items(): fw = fw_by_key.get(fw_key) if fw is None: continue for spec in reqs: key = (fw.id, spec["ref"]) row = req_by_fk_ref.get(key) if row is None: row = ComplianceRequirement( framework_id=fw.id, ref=spec["ref"], title=spec["title"], description=spec.get("description"), category=spec.get("category"), sort_order=spec.get("sort_order", 0), ) db.add(row) req_by_fk_ref[key] = row counts["requirements"] += 1 else: row.title = spec["title"] row.description = spec.get("description") row.category = spec.get("category") row.sort_order = spec.get("sort_order", 0) await db.flush() # ── Controls ──────────────────────────────────────────────────── existing_ctrls = { c.key: c for c in (await db.execute(select(ComplianceControl))).scalars().all() } ctrl_by_key: dict[str, ComplianceControl] = dict(existing_ctrls) for spec in cat.CONTROLS: row = existing_ctrls.get(spec["key"]) if row is None: row = ComplianceControl( key=spec["key"], name=spec["name"], category=spec["category"], description=spec.get("description"), signal_type=spec.get("signal_type", "scheduled_job"), signal_ref=spec.get("signal_ref"), config=spec.get("config") or {}, sort_order=spec.get("sort_order", 0), ) db.add(row) ctrl_by_key[spec["key"]] = row counts["controls"] += 1 else: row.name = spec["name"] row.category = spec["category"] row.description = spec.get("description") row.signal_type = spec.get("signal_type", "scheduled_job") row.signal_ref = spec.get("signal_ref") # The catalog is the single source of truth for tuning config # (see module docstring) — refresh it from the spec on every # reconcile so a catalog edit propagates without a migration. row.config = spec.get("config") or {} row.sort_order = spec.get("sort_order", 0) await db.flush() # ── Crosswalk mappings ────────────────────────────────────────── existing_maps = { (m.control_id, m.requirement_id) for m in ( await db.execute(select(ComplianceControlRequirement)) ).scalars().all() } for ctrl_key, fw_refs in cat.MAPPINGS.items(): ctrl = ctrl_by_key.get(ctrl_key) if ctrl is None: continue for fw_key, refs in fw_refs.items(): fw = fw_by_key.get(fw_key) if fw is None: continue for ref in refs: req = req_by_fk_ref.get((fw.id, ref)) if req is None: continue if (ctrl.id, req.id) in existing_maps: continue db.add(ComplianceControlRequirement( control_id=ctrl.id, requirement_id=req.id, )) existing_maps.add((ctrl.id, req.id)) counts["mappings"] += 1 await db.flush() return counts # ════════════════════════════════════════════════════════════════════ # Control evaluation (continuous control monitoring) # ════════════════════════════════════════════════════════════════════ async def _evaluate_controls(db: AsyncSession) -> dict[str, dict[str, Any]]: """Compute {control_key: {status, score, summary, detail, source_fresh_at}}. Reuses :meth:`DashboardService.summary` so posture numbers stay identical to the existing per-control dashboard, then layers a freshness gate from each control's feeding scheduled job. """ now = _naive_now() summary = await DashboardService.summary(db) # Binary controls derive pass/fail from the unified findings worklist # (open + acknowledged), so posture honors operator decisions: # accepting-risk or resolving a finding clears the control. from app.domains.auditor import findings_service open_by_control = await findings_service.count_open_by_control(db) # Job freshness for every control's signal_ref. handler_keys = [ c["signal_ref"] for c in cat.CONTROLS if c.get("signal_ref") ] job_rows = (await db.execute( select(ScheduledJob).where(ScheduledJob.handler_key.in_(handler_keys)) )).scalars().all() job_by_handler = {j.handler_key: j for j in job_rows} def freshness(control_key: str, handler_key: str | None): """Return (gate_status_or_None, source_fresh_at). gate_status: 'not_configured' / 'stale' if the signal can't be trusted, else None (proceed to metric evaluation). """ if not handler_key: return None, None job = job_by_handler.get(handler_key) if job is None or not job.enabled: return cat.STATUS_NOT_CONFIGURED, None last_run = _to_naive(job.last_run_at) if last_run is None: return cat.STATUS_STALE, None max_age = _STALE_AFTER_HOURS.get(control_key, 48) if now - last_run > timedelta(hours=max_age): return cat.STATUS_STALE, last_run return None, last_run out: dict[str, dict[str, Any]] = {} # Latest privileged-role access-certification campaign (the only # control needing an extra query beyond the dashboard summary). latest_cycle = (await db.execute( select(AccessCertification).where( AccessCertification.campaign_type == "privileged_role", ).order_by(AccessCertification.started_at.desc()).limit(1) )).scalar_one_or_none() def result(status, *, score=None, summary_text=None, detail=None, fresh_at=None): return { "status": status, "score": score, "summary": summary_text, "detail": detail or {}, "source_fresh_at": fresh_at, } for spec in cat.CONTROLS: key = spec["key"] gate, fresh_at = freshness(key, spec.get("signal_ref")) if gate is not None: out[key] = result(gate, fresh_at=fresh_at) continue if key == "mfa_coverage": total = summary["mfa_total_count"] unenrolled = summary["mfa_unenrolled_count"] if total == 0: out[key] = result(cat.STATUS_NOT_CONFIGURED, fresh_at=fresh_at) continue enrolled = total - unenrolled pct = round(enrolled / total * 100, 1) threshold = (spec.get("config") or {}).get("min_coverage_pct", 95) status = cat.STATUS_PASS if pct >= threshold else cat.STATUS_FAIL out[key] = result( status, score=int(pct), summary_text=f"{enrolled}/{total} enrolled ({pct}%)", detail={"enrolled": enrolled, "total": total, "threshold_pct": threshold}, fresh_at=fresh_at, ) elif key == "privileged_access_review": if latest_cycle is None: out[key] = result( cat.STATUS_NOT_CONFIGURED, summary_text="no review cycle opened yet", fresh_at=fresh_at, ) continue due = _to_naive(latest_cycle.due_at) if latest_cycle.status == "completed": out[key] = result( cat.STATUS_PASS, summary_text=f"cycle {latest_cycle.label} completed", detail={"cycle_label": latest_cycle.label}, fresh_at=fresh_at, ) elif due is not None and now > due: out[key] = result( cat.STATUS_FAIL, summary_text=( f"cycle {latest_cycle.label} overdue" ), detail={"cycle_label": latest_cycle.label, "due_at": due.isoformat()}, fresh_at=fresh_at, ) else: out[key] = result( cat.STATUS_PASS, summary_text=( f"cycle {latest_cycle.label} in progress" ), detail={"cycle_label": latest_cycle.label}, fresh_at=fresh_at, ) elif key == "pat_audit": risky = open_by_control.get(key, 0) out[key] = _binary_clean(risky, "risky token", fresh_at, result) elif key == "public_exposure_scan": open_ct = open_by_control.get(key, 0) status = cat.STATUS_PASS if open_ct == 0 else cat.STATUS_FAIL out[key] = result( status, summary_text=( "no open exposure findings" if open_ct == 0 else f"{open_ct} open finding(s)" ), detail={"open": open_ct}, fresh_at=fresh_at, ) elif key == "endpoint_risk_scoring": crit = summary["risk_critical_count"] high = summary["risk_high_count"] status = cat.STATUS_PASS if crit == 0 else cat.STATUS_FAIL out[key] = result( status, summary_text=( "no critical-risk identities" if crit == 0 else f"{crit} critical-risk identit(ies)" ), detail={"critical": crit, "high": high}, fresh_at=fresh_at, ) elif key == "secret_rotation": overdue = open_by_control.get(key, 0) out[key] = _binary_clean( overdue, "overdue secret", fresh_at, result) elif key == "cert_expiry_monitoring": expiring = open_by_control.get(key, 0) out[key] = _binary_clean( expiring, "certificate expiring soon", fresh_at, result) elif key == "signin_anomaly_monitoring": high = open_by_control.get(key, 0) out[key] = _binary_clean( high, "high-anomaly sign-in", fresh_at, result) elif key == "access_certification": from app.domains.auditor import ( access_certification_service as acs, ) # This control attests *general user-access* review (manager # re-attests software-seat entitlements). It must look only at # ``software_seat`` campaigns — the ``privileged_role`` campaign # auto-opens every quarter and would otherwise mask the absence # of a user-access review, giving a false PASS. The # ``privileged_access_review`` control tracks privileged_role # separately. latest = await acs.latest_campaign_status( db, campaign_type="software_seat") if latest is None: out[key] = result( cat.STATUS_NOT_CONFIGURED, summary_text="no access-review campaign run yet", ) elif latest["status"] == acs.STATUS_COMPLETED: stale_days = (spec.get("config") or {}).get( "stale_after_days", 100) completed = latest["completed_at"] if completed is not None and ( now - completed > timedelta(days=stale_days) ): out[key] = result( cat.STATUS_STALE, summary_text=( f"last campaign '{latest['label']}' completed " f">{stale_days}d ago" ), ) else: out[key] = result( cat.STATUS_PASS, summary_text=f"'{latest['label']}' completed", ) else: due = latest["due_at"] if due is not None and now > due: out[key] = result( cat.STATUS_FAIL, summary_text=f"'{latest['label']}' overdue", ) else: out[key] = result( cat.STATUS_PASS, summary_text=f"'{latest['label']}' in progress", ) elif key == "access_self_attestation": from app.domains.auditor import ( access_certification_service as acs, ) # Holder-driven half-yearly self-attestation. Filtered to # ``self_attestation`` campaigns so the manager-led software_seat # review (above) and the quarterly privileged_role review never # mask a missing all-employee self-attestation. latest = await acs.latest_campaign_status( db, campaign_type=acs.CAMPAIGN_TYPE_SELF_ATTESTATION) if latest is None: out[key] = result( cat.STATUS_NOT_CONFIGURED, summary_text="no self-attestation campaign run yet", ) elif latest["status"] == acs.STATUS_COMPLETED: stale_days = (spec.get("config") or {}).get( "stale_after_days", 200) completed = latest["completed_at"] if completed is not None and ( now - completed > timedelta(days=stale_days) ): out[key] = result( cat.STATUS_STALE, summary_text=( f"last campaign '{latest['label']}' completed " f">{stale_days}d ago" ), ) else: out[key] = result( cat.STATUS_PASS, summary_text=f"'{latest['label']}' completed", ) else: due = latest["due_at"] if due is not None and now > due: out[key] = result( cat.STATUS_FAIL, summary_text=f"'{latest['label']}' overdue", ) else: out[key] = result( cat.STATUS_PASS, summary_text=f"'{latest['label']}' in progress", ) elif key == "security_awareness_training": # Cross-domain read by IMPORT of a pure evaluator, matching # the ``periodic_self_check`` precedent below — the auditor # domain owns the control catalog, and each domain owns the # function that turns its own state into a control verdict. from app.domains.security_awareness.compliance import ( evaluate_awareness_control, ) ev = await evaluate_awareness_control(db) out[key] = result( ev["status"], score=ev["score"], summary_text=ev["summary"], detail=ev["detail"], fresh_at=fresh_at, ) elif key == "periodic_self_check": from app.domains.auditor.self_check_service import ( evaluate_self_check_control, ) ev = await evaluate_self_check_control(db) out[key] = result( ev["status"], score=ev["score"], summary_text=ev["summary"], detail=ev["detail"], fresh_at=fresh_at, ) else: out[key] = result(cat.STATUS_UNKNOWN, fresh_at=fresh_at) return out def _binary_clean(count, noun, fresh_at, result): """Helper: clean (count==0) → pass, else fail with a count summary.""" if count == 0: return result( cat.STATUS_PASS, summary_text=f"no {noun}s", fresh_at=fresh_at) return result( cat.STATUS_FAIL, summary_text=f"{count} {noun}(s)", detail={"count": count}, fresh_at=fresh_at) async def recompute_control_status(db: AsyncSession) -> dict[str, Any]: """Reconcile catalog, evaluate every control, upsert status rows. The entrypoint for the ``compliance.control_status_recompute`` cron and the manual "recompute" button. """ await sync_compliance_catalog(db) evaluated = await _evaluate_controls(db) now = _naive_now() controls = (await db.execute( select(ComplianceControl).where(ComplianceControl.enabled.is_(True)) )).scalars().all() by_status: dict[str, int] = {} values: list[dict[str, Any]] = [] for ctrl in controls: ev = evaluated.get(ctrl.key) if ev is None: ev = {"status": cat.STATUS_UNKNOWN, "score": None, "summary": None, "detail": {}, "source_fresh_at": None} by_status[ev["status"]] = by_status.get(ev["status"], 0) + 1 values.append({ "id": uuid.uuid4(), "control_id": ctrl.id, "status": ev["status"], "score": ev["score"], "summary": ev["summary"], "detail": ev["detail"], "evaluated_at": now, "source_fresh_at": ev["source_fresh_at"], }) # Upsert on the UNIQUE(control_id) key. ON DUPLICATE KEY UPDATE keeps # this idempotent under concurrency — the manual "recompute" button and # the hourly cron run in separate sessions, and a read-then-insert would # race into a duplicate-key 500 the first time a freshly-added control is # seeded by both at once. if values: stmt = mysql_insert(ComplianceControlStatus).values(values) stmt = stmt.on_duplicate_key_update( status=stmt.inserted.status, score=stmt.inserted.score, summary=stmt.inserted.summary, detail=stmt.inserted.detail, evaluated_at=stmt.inserted.evaluated_at, source_fresh_at=stmt.inserted.source_fresh_at, updated_at=now, ) await db.execute(stmt) await db.commit() return { "control_count": len(controls), "by_status": by_status, "computed_at": now.isoformat(), } # ════════════════════════════════════════════════════════════════════ # Read queries (projections) # ════════════════════════════════════════════════════════════════════ async def _load_graph(db: AsyncSession): """Load frameworks, requirements, controls, mappings, statuses once.""" # Frameworks and controls are filtered to ``enabled=True`` to match # ``recompute_control_status`` (which only ever writes a status row # for enabled controls). Without this, a disabled control surfaced in # the control rollup as a phantom STATUS_UNKNOWN entry. (Coverage math # below still keys off ``mappings``/``statuses`` — intentionally left # untouched so disabling a control doesn't silently rewrite a # requirement's coverage; that's a product decision, not a bug fix.) frameworks = (await db.execute( select(ComplianceFramework) .where(ComplianceFramework.enabled.is_(True)) .order_by(ComplianceFramework.sort_order) )).scalars().all() requirements = (await db.execute( select(ComplianceRequirement).order_by( ComplianceRequirement.sort_order ) )).scalars().all() controls = (await db.execute( select(ComplianceControl) .where(ComplianceControl.enabled.is_(True)) .order_by(ComplianceControl.sort_order) )).scalars().all() mappings = (await db.execute( select(ComplianceControlRequirement) )).scalars().all() statuses = { s.control_id: s for s in ( await db.execute(select(ComplianceControlStatus)) ).scalars().all() } return frameworks, requirements, controls, mappings, statuses def _coverage_for(control_ids: list[Any], statuses: dict) -> str: """Coverage state of a requirement given its mapped controls.""" if not control_ids: return cat.COVERAGE_NONE pass_n = 0 for cid in control_ids: st = statuses.get(cid) if st is not None and st.status == cat.STATUS_PASS: pass_n += 1 if pass_n == len(control_ids): return cat.COVERAGE_PASS if pass_n == 0: return cat.COVERAGE_FAIL return cat.COVERAGE_PARTIAL async def get_posture(db: AsyncSession) -> dict[str, Any]: """Headline GRC posture: per-framework readiness + control rollup.""" frameworks, requirements, controls, mappings, statuses = ( await _load_graph(db) ) reqs_by_fw: dict[Any, list[ComplianceRequirement]] = {} for r in requirements: reqs_by_fw.setdefault(r.framework_id, []).append(r) ctrls_by_req: dict[Any, list[Any]] = {} for m in mappings: ctrls_by_req.setdefault(m.requirement_id, []).append(m.control_id) framework_summaries = [] for fw in frameworks: reqs = reqs_by_fw.get(fw.id, []) cov_counts = {cat.COVERAGE_PASS: 0, cat.COVERAGE_PARTIAL: 0, cat.COVERAGE_FAIL: 0, cat.COVERAGE_NONE: 0} for r in reqs: cov_counts[_coverage_for( ctrls_by_req.get(r.id, []), statuses)] += 1 total = len(reqs) readiness = ( round(cov_counts[cat.COVERAGE_PASS] / total * 100) if total else 0 ) framework_summaries.append({ "key": fw.key, "name": fw.name, "version": fw.version, "requirement_count": total, "readiness_pct": readiness, "coverage": cov_counts, }) # Control rollup by status. by_status: dict[str, int] = {} for c in controls: st = statuses.get(c.id) s = st.status if st else cat.STATUS_UNKNOWN by_status[s] = by_status.get(s, 0) + 1 return { "frameworks": framework_summaries, "controls": { "total": len(controls), "by_status": by_status, }, } async def list_frameworks(db: AsyncSession) -> list[dict[str, Any]]: posture = await get_posture(db) return posture["frameworks"] def _control_view(c: ComplianceControl, statuses: dict) -> dict[str, Any]: st = statuses.get(c.id) return { "key": c.key, "name": c.name, "category": c.category, "description": c.description, "signal_type": c.signal_type, "signal_ref": c.signal_ref, "enabled": bool(c.enabled), "status": st.status if st else cat.STATUS_UNKNOWN, "score": st.score if st else None, "status_summary": st.summary if st else None, "evaluated_at": ( _to_naive(st.evaluated_at).isoformat() if st and st.evaluated_at else None ), "source_fresh_at": ( _to_naive(st.source_fresh_at).isoformat() if st and st.source_fresh_at else None ), } async def get_framework( db: AsyncSession, framework_key: str, ) -> dict[str, Any] | None: frameworks, requirements, controls, mappings, statuses = ( await _load_graph(db) ) fw = next((f for f in frameworks if f.key == framework_key), None) if fw is None: return None ctrl_by_id = {c.id: c for c in controls} ctrls_by_req: dict[Any, list[Any]] = {} for m in mappings: ctrls_by_req.setdefault(m.requirement_id, []).append(m.control_id) reqs = [r for r in requirements if r.framework_id == fw.id] req_views = [] cov_counts = {cat.COVERAGE_PASS: 0, cat.COVERAGE_PARTIAL: 0, cat.COVERAGE_FAIL: 0, cat.COVERAGE_NONE: 0} for r in reqs: cids = ctrls_by_req.get(r.id, []) coverage = _coverage_for(cids, statuses) cov_counts[coverage] += 1 req_views.append({ "ref": r.ref, "title": r.title, "description": r.description, "category": r.category, "coverage": coverage, "controls": [ _control_view(ctrl_by_id[cid], statuses) for cid in cids if cid in ctrl_by_id ], }) total = len(reqs) return { "key": fw.key, "name": fw.name, "version": fw.version, "description": fw.description, "requirement_count": total, "readiness_pct": ( round(cov_counts[cat.COVERAGE_PASS] / total * 100) if total else 0 ), "coverage": cov_counts, "requirements": req_views, } async def list_controls(db: AsyncSession) -> list[dict[str, Any]]: frameworks, requirements, controls, mappings, statuses = ( await _load_graph(db) ) req_by_id = {r.id: r for r in requirements} fw_by_id = {f.id: f for f in frameworks} # control_id -> list of "soc2:CC6.1" refs_by_ctrl: dict[Any, list[str]] = {} for m in mappings: req = req_by_id.get(m.requirement_id) if req is None: continue fw = fw_by_id.get(req.framework_id) if fw is None: continue refs_by_ctrl.setdefault(m.control_id, []).append( f"{fw.key}:{req.ref}") out = [] for c in controls: view = _control_view(c, statuses) view["mapped_refs"] = sorted(refs_by_ctrl.get(c.id, [])) view["mapped_count"] = len(refs_by_ctrl.get(c.id, [])) out.append(view) return out async def get_control( db: AsyncSession, control_key: str, ) -> dict[str, Any] | None: frameworks, requirements, controls, mappings, statuses = ( await _load_graph(db) ) c = next((x for x in controls if x.key == control_key), None) if c is None: return None req_by_id = {r.id: r for r in requirements} fw_by_id = {f.id: f for f in frameworks} mapped = [] for m in mappings: if m.control_id != c.id: continue req = req_by_id.get(m.requirement_id) if req is None: continue fw = fw_by_id.get(req.framework_id) mapped.append({ "framework_key": fw.key if fw else None, "framework_name": fw.name if fw else None, "ref": req.ref, "title": req.title, }) view = _control_view(c, statuses) st = statuses.get(c.id) view["detail"] = st.detail if st else None view["config"] = c.config or {} view["requirements"] = sorted( mapped, key=lambda x: (x["framework_key"] or "", x["ref"])) return view