feat(cra): attach network_security regulatory breadth (shared Controls-API)

Semantic breadth (2): each finding's CRA-AI is mapped to a network_security
sub_topic and enriched with atom-grain, framework-traceable obligations from the
shared Controls-API (compliance.atom_classification) — at the endpoint/view layer
(SessionLocal), NOT in the pure mapper. CRA-AI anchor + curated measure +
NIST/OWASP crosswalk stay the lead; this is breadth + source evidence. Only
network_security is queried (atom-grain), scoped by sub_topic + limit. Frontend
renders it under the collapsible best-practice depth (control_id · title · source).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

backend-compliance/compliance/api/cra_assess_routes.py

@@ -16,6 +16,8 @@ from pydantic import BaseModel
 
 from compliance.services.cra_finding_mapper import assess_findings_payload
 from compliance.services.cra_snapshot_store import save_snapshot, list_snapshots, get_snapshot
+from compliance.services.cra_use_case_controls import enrich_findings_with_breadth
+from database import SessionLocal
 from .tenant_utils import get_tenant_id
 
 router = APIRouter(prefix="/v1/cra", tags=["cra"])
@@ -59,15 +61,29 @@ def _payload(body: AssessRequest) -> dict:
     }
 
 
+def _assess_enriched(body: AssessRequest) -> dict:
+    """Assessment + the network_security regulatory breadth (atom-grain).
+
+    Breadth is attached at this view layer (db here), never in the pure mapper.
+    """
+    result = assess_findings_payload(_payload(body))
+    db = SessionLocal()
+    try:
+        enrich_findings_with_breadth(result.get("mapped", []), db)
+    finally:
+        db.close()
+    return result
+
+
 @router.post("/assess")
 async def assess(body: AssessRequest):
-    return assess_findings_payload(_payload(body))
+    return _assess_enriched(body)
 
 
 @router.post("/projects/{project_id}/assess-snapshot")
 async def assess_snapshot(project_id: str, body: AssessRequest, tenant_id: str = Depends(get_tenant_id)):
     """Run the assessment and persist it as a versioned snapshot (running system)."""
-    assessment = assess_findings_payload(_payload(body))
+    assessment = _assess_enriched(body)
     snap = save_snapshot(project_id, tenant_id, assessment)
     return {"snapshot": snap, "assessment": assessment}
 

backend-compliance/compliance/services/cra_use_case_controls.py

@@ -0,0 +1,69 @@
+"""Attach the atom-grain network_security regulatory breadth to CRA findings.
+
+This is the "semantic breadth (2)" from the handoff: the shared Controls-API
+(compliance.atom_classification, use_case=network_security, ~11k precise,
+framework-traceable obligations). It runs at the ENDPOINT/VIEW layer — NOT in
+the pure cra_finding_mapper, which stays deterministic. The CRA Annex I anchor +
+the curated measure + the NIST/OWASP golden-set crosswalk remain the lead; this
+is breadth + source evidence, not a replacement.
+
+Only network_security is atom-grain — we query only that, always scoped by
+sub_topic + limit (per the caveats).
+"""
+from compliance.api.cra_annex_i_data import ANNEX_I_REQUIREMENTS
+from compliance.services.use_case_controls import UseCaseControlsService
+
+# CRA-AI requirement -> network_security sub_topic (via the NIST families per
+# CRA-AI). Exact sub_topic keys verified against the live atom_classification.
+_REQ_TO_SUBTOPIC = {
+    "CRA-AI-1": "secure_development", "CRA-AI-2": "network_segmentation",
+    "CRA-AI-3": "network_segmentation", "CRA-AI-4": "access_control",
+    "CRA-AI-5": "secure_development", "CRA-AI-6": "secure_development",
+    "CRA-AI-7": "authentication", "CRA-AI-8": "authentication", "CRA-AI-9": "authentication",
+    "CRA-AI-10": "access_control", "CRA-AI-11": "authentication", "CRA-AI-12": "access_control",
+    "CRA-AI-13": "cryptography", "CRA-AI-14": "cryptography", "CRA-AI-15": "cryptography",
+    "CRA-AI-16": "cryptography", "CRA-AI-17": "data_protection",
+    "CRA-AI-18": "secure_development", "CRA-AI-19": "secure_development", "CRA-AI-20": "secure_development",
+    "CRA-AI-21": "supply_chain_security", "CRA-AI-22": "vulnerability_management",
+    "CRA-AI-23": "supply_chain_security",
+    "CRA-AI-24": "logging_monitoring", "CRA-AI-25": "logging_monitoring",
+    "CRA-AI-26": "logging_monitoring", "CRA-AI-27": "logging_monitoring",
+    "CRA-AI-28": "vulnerability_management", "CRA-AI-29": "vulnerability_management",
+    "CRA-AI-30": "vulnerability_management", "CRA-AI-31": "vulnerability_management",
+    "CRA-AI-32": "vulnerability_management", "CRA-AI-33": "vulnerability_management",
+    "CRA-AI-34": "vulnerability_management",
+    "CRA-AI-35": "incident_response", "CRA-AI-36": "incident_response",
+    "CRA-AI-37": "incident_response", "CRA-AI-38": "incident_response",
+    "CRA-AI-39": "vulnerability_management", "CRA-AI-40": "incident_response",
+}
+
+
+def subtopic_for(req_id: str):
+    return _REQ_TO_SUBTOPIC.get(req_id)
+
+
+def enrich_findings_with_breadth(mapped: list, db, limit: int = 5) -> None:
+    """Attach `sub_topic` + `regulatory_breadth` (atom controls) to each finding.
+
+    Queries network_security once per distinct sub_topic (cached). Best-effort:
+    on any error a finding just gets an empty breadth — never breaks the assessment.
+    """
+    svc = UseCaseControlsService(db)
+    cache: dict = {}
+    for m in mapped:
+        st = _REQ_TO_SUBTOPIC.get(m.get("primary_requirement"))
+        m["sub_topic"] = st
+        if not st:
+            m["regulatory_breadth"] = []
+            continue
+        if st not in cache:
+            try:
+                res = svc.controls_for_use_case("network_security", sub_topic=st, limit=limit)
+                cache[st] = [
+                    {"control_id": c.get("control_id"), "title": c.get("title"),
+                     "source_regulation": c.get("source_regulation"), "severity": c.get("severity")}
+                    for c in res.get("controls", [])
+                ]
+            except Exception:
+                cache[st] = []
+        m["regulatory_breadth"] = cache[st]