feat(cra): cyber-meets-safety bridge as real logic (step 2)

Deterministic bridge (cra_safety_bridge.py): a cyber finding's attack capability
(remote_actuation / code_tampering / integrity_loss / auth_bypass, derived from
its CRA category) is matched against what each CE safety function is vulnerable
to. A match re-opens the mitigated hazard, flags the finding safety_impact (which
floors it to P0), and produces the cross-link. Endpoint accepts safety_functions;
frontend passes the project's safety functions and renders the LIVE cross-links
(no more hardcode). Safety functions are demo input now; come from the CE risk
assessment in production.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-06-14 08:59:41 +02:00
parent fb4d7641ab
commit 10c32d7f7c
5 changed files with 176 additions and 9 deletions
@@ -16,6 +16,7 @@ from typing import Optional
from compliance.api.cra_annex_i_data import ANNEX_I_REQUIREMENTS, MEASURES, DEADLINES
from compliance.services.cra_security_crosswalk import security_refs_for
from compliance.services.cra_prioritizer import prioritize, OBJECTIVES
from compliance.services.cra_safety_bridge import build_cross_links
_REQ_INDEX = {r["req_id"]: r for r in ANNEX_I_REQUIREMENTS}
_SEV_ORDER = {"LOW": 1, "MEDIUM": 2, "HIGH": 3, "CRITICAL": 4}
@@ -120,6 +121,7 @@ class CRAAssessment:
coverage_pct: float = 0.0
quick_wins: list = field(default_factory=list) # finding_ids: high impact, low effort
objectives: list = field(default_factory=lambda: list(OBJECTIVES))
cross_links: list = field(default_factory=list) # cyber-meets-safety bridge
deadlines: list = field(default_factory=lambda: list(DEADLINES))
@@ -197,13 +199,21 @@ def map_finding(f: ScannerFinding) -> MappedFinding:
)
def assess_findings(findings: list, weights=None) -> CRAAssessment:
def assess_findings(findings: list, weights=None, safety_functions=None) -> CRAAssessment:
"""Map findings to a deterministic CRA assessment, then prioritise them.
weights: {objective: 'high'|'medium'|'low'} — the customer's priorities for
the discretionary tier (the P0 floor ignores them).
weights: {objective: 'high'|'medium'|'low'} — customer priorities for the
discretionary tier (the P0 floor ignores them).
safety_functions: CE-risk-assessment safety functions for the cyber-meets-
safety bridge; a finding that can defeat one is flagged safety_impact (→ P0).
"""
mapped = prioritize([map_finding(f) for f in findings], weights)
mapped = [map_finding(f) for f in findings]
cross_links = build_cross_links(mapped, safety_functions)
flagged = {fid for cl in cross_links for fid in cl["cyber_finding_ids"]}
for m in mapped:
if m.finding_id in flagged:
m.safety_impact = True
mapped = prioritize(mapped, weights)
by_risk = {"CRITICAL": 0, "HIGH": 0, "MEDIUM": 0, "LOW": 0}
reqs_touched, measure_ids, unmapped = set(), [], []
for m in mapped:
@@ -226,6 +236,7 @@ def assess_findings(findings: list, weights=None) -> CRAAssessment:
unmapped_findings=unmapped,
coverage_pct=round(100.0 * covered / total, 1) if total else 0.0,
quick_wins=[m.finding_id for m in mapped if m.quick_win],
cross_links=cross_links,
)
@@ -236,6 +247,7 @@ def assess_findings_payload(payload: dict) -> dict:
"""
raw = payload.get("findings", []) if isinstance(payload, dict) else []
weights = payload.get("weights") if isinstance(payload, dict) else None
safety_functions = payload.get("safety_functions") if isinstance(payload, dict) else None
findings = [ScannerFinding.from_dict(d) for d in raw]
assessment = assess_findings(findings, weights)
assessment = assess_findings(findings, weights, safety_functions)
return asdict(assessment) # recurses into nested MappedFinding dataclasses