feat(cra): MaschinenVO-Gefährdungs-Ableitung + Cyber-Safety-Brücke

3-Tier-MaschinenVO-Verdict (direkt / sicherheitsrelevant / nicht relevant) aus Personengefährdungs-Signal: eine Komponente ist keine Maschine, aber wenn ihre Funktion bei Fehler ODER Manipulation Personen gefaehrden kann (Bewegung, Laser/ Auge, Kraft, Temperatur, elektrisch), ist sie sicherheitsrelevant — Pflicht trifft den Maschinenbauer, Zulieferer liefert Nachweise, und ein Cyber-Angriff kann die Sicherheitsfunktion aushebeln (Cyber-Safety-Bruecke). OWIS-mit-Laser landet so korrekt als 'sicherheitsrelevante Komponente'. Engine + /readiness additiv; Frontend: Gefährdungs-Frage + -Typen, MaschinenVO-Ergebnisblock. Presets aktualisiert (OWIS: Laser+Bewegung, Zwick: Bewegung). 22 Tests gruen. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-16 18:48:52 +02:00
parent 2b5c155f57
commit 62fafaaec5
6 changed files with 187 additions and 2 deletions
@@ -15,7 +15,9 @@ from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel

 from compliance.services.cra_finding_mapper import assess_findings_payload
-from compliance.services.cra_applicability import compute_verdict, maturity as evidence_maturity, MACHINE_INTEGRATOR
+from compliance.services.cra_applicability import (
+    compute_verdict, compute_machinery_verdict, maturity as evidence_maturity, MACHINE_INTEGRATOR,
+)
 from compliance.services.scanner_mcp_client import fetch_findings
 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
@@ -191,6 +193,10 @@ class ReadinessRequest(BaseModel):
    customers_request_cra_evidence: Optional[bool] = False
    provided_evidence: Optional[List[str]] = None     # evidence keys already in place (sbom, vdp, …)
    digital_elements: Optional[List[str]] = None      # detected/declared digital elements
+    # Machinery-Regulation person-safety axis
+    safety_relevant: Optional[bool] = False           # function can endanger persons on fault/manipulation
+    hazard_types: Optional[List[str]] = None          # movement_crush, laser_radiation, force_energy, …
+    is_safety_component: Optional[bool] = False        # marketed as a safety device (Sicherheitsbauteil)


 # CRA Annex I evidence_type -> guideline bucket (Code / Prozess / Dokumentation).
@@ -292,6 +298,10 @@ async def readiness(body: ReadinessRequest):
        "deadlines": list(DEADLINES),
        # Eingangstür verdict layer
        "verdict": verdict,
+        "machinery_verdict": compute_machinery_verdict(
+            body.producer_type or "", bool(body.is_machinery),
+            bool(body.safety_relevant), body.hazard_types, bool(body.is_safety_component),
+        ),
        "maturity": evidence_maturity(body.provided_evidence),
        "digital_elements": body.digital_elements or [],
        "producer_type": body.producer_type or "",
@@ -105,3 +105,65 @@ def maturity(provided_evidence_keys) -> dict:
    total = len(EVIDENCE_ITEMS)
    pct = round(100.0 * len(present) / total) if total else 0
    return {"pct": pct, "present": present, "missing": missing, "total": total}
+
+
+# --- Machinery Regulation (2023/1230) applicability — person-safety axis ---
+# A bare control component is not "machinery" itself, but if its function can
+# endanger persons (movement, laser, stored energy …) it is safety-relevant: the
+# duty hits the machine builder, the component maker supplies evidence, and a
+# cyber compromise can defeat a safety function (the CRA × MaschinenVO bridge).
+HAZARD_TYPES = [
+    {"key": "movement_crush", "label": "Bewegung / Quetschen"},
+    {"key": "laser_radiation", "label": "Laser / Strahlung (Auge)"},
+    {"key": "force_energy", "label": "Kraft / gespeicherte Energie"},
+    {"key": "temperature", "label": "Temperatur / Hitze"},
+    {"key": "electrical", "label": "Elektrische Gefährdung"},
+]
+_HAZARD_KEYS = {h["key"]: h["label"] for h in HAZARD_TYPES}
+
+MV_DIREKT = "direkt"
+MV_SICHERHEITSRELEVANT = "sicherheitsrelevant"
+MV_NICHT = "nicht_relevant"
+_MV_LABEL = {
+    MV_DIREKT: "Maschinenverordnung direkt betroffen",
+    MV_SICHERHEITSRELEVANT: "Sicherheitsrelevante Komponente (indirekt)",
+    MV_NICHT: "Maschinenverordnung nicht relevant",
+}
+
+
+def compute_machinery_verdict(
+    producer_type: str = "",
+    is_machinery: bool = False,
+    safety_relevant: bool = False,
+    hazard_types=None,
+    is_safety_component: bool = False,
+) -> dict:
+    """3-tier Machinery-Regulation verdict + cyber→safety bridge flag.
+    `safety_relevant`: can the function endanger persons on fault OR manipulation?"""
+    hazards = [{"key": k, "label": _HAZARD_KEYS[k]} for k in (hazard_types or []) if k in _HAZARD_KEYS]
+    direct = bool(is_machinery) or producer_type == MACHINE_INTEGRATOR or bool(is_safety_component)
+    reasons: list = []
+    if direct:
+        tier = MV_DIREKT
+        reasons.append("Maschine/Anlage bzw. Sicherheitsbauteil → MaschinenVO-Pflichten direkt.")
+    elif safety_relevant or hazards:
+        tier = MV_SICHERHEITSRELEVANT
+        reasons.append("Komponente, deren Funktion bei Fehler oder Manipulation Personen gefährden kann.")
+        reasons.append("MaschinenVO-Pflicht trifft den Maschinenbauer; als Zulieferer liefern Sie Sicherheits-/Cyber-Nachweise zu.")
+    else:
+        tier = MV_NICHT
+        reasons.append("Keine Personengefährdung erkennbar.")
+
+    bridge = tier in (MV_DIREKT, MV_SICHERHEITSRELEVANT)
+    if bridge:
+        reasons.append(
+            "Cyber-trifft-Safety: Ein Angriff auf die Steuerung kann eine Sicherheitsfunktion aushebeln "
+            "(Geschwindigkeit/Position/Verriegelung) → Personenschaden."
+        )
+    return {
+        "tier": tier,
+        "label": _MV_LABEL[tier],
+        "hazards": hazards,
+        "cyber_safety_bridge": bridge,
+        "reasons": reasons,
+    }
@@ -1,7 +1,8 @@
 """Neutral CRA applicability verdict (Eingangstür): legal duty vs market pull."""
 from compliance.services.cra_applicability import (
    ZWINGEND, RATSAM, NICHT_BETROFFEN, COMPONENT, MACHINE_INTEGRATOR,
-    compute_verdict, maturity, in_scope, EVIDENCE_ITEMS,
+    MV_DIREKT, MV_SICHERHEITSRELEVANT, MV_NICHT,
+    compute_verdict, compute_machinery_verdict, maturity, in_scope, EVIDENCE_ITEMS,
 )


@@ -60,6 +61,44 @@ class TestVerdict:
        assert v["cra_class"] == "IMPORTANT_II"


+class TestMachineryVerdict:
+    def test_machine_integrator_is_direct(self):
+        v = compute_machinery_verdict(producer_type=MACHINE_INTEGRATOR)
+        assert v["tier"] == MV_DIREKT
+        assert v["cyber_safety_bridge"] is True
+
+    def test_is_machinery_flag_is_direct(self):
+        v = compute_machinery_verdict(producer_type="end_device", is_machinery=True)
+        assert v["tier"] == MV_DIREKT
+
+    def test_component_with_person_hazard_is_safety_relevant(self):
+        # OWIS-with-laser borderline: component, not a machine, but can harm persons
+        v = compute_machinery_verdict(
+            producer_type=COMPONENT, safety_relevant=True, hazard_types=["laser_radiation", "movement_crush"],
+        )
+        assert v["tier"] == MV_SICHERHEITSRELEVANT
+        assert v["cyber_safety_bridge"] is True
+        assert {h["key"] for h in v["hazards"]} == {"laser_radiation", "movement_crush"}
+
+    def test_hazard_types_alone_imply_safety_relevant(self):
+        v = compute_machinery_verdict(producer_type=COMPONENT, hazard_types=["force_energy"])
+        assert v["tier"] == MV_SICHERHEITSRELEVANT
+
+    def test_component_no_hazard_not_relevant(self):
+        v = compute_machinery_verdict(producer_type=COMPONENT)
+        assert v["tier"] == MV_NICHT
+        assert v["cyber_safety_bridge"] is False
+
+    def test_safety_component_is_direct(self):
+        v = compute_machinery_verdict(producer_type=COMPONENT, is_safety_component=True)
+        assert v["tier"] == MV_DIREKT
+
+    def test_unknown_hazard_keys_ignored(self):
+        v = compute_machinery_verdict(producer_type=COMPONENT, hazard_types=["nonsense"])
+        assert v["hazards"] == []
+        assert v["tier"] == MV_NICHT
+
+
 class TestMaturity:
    def test_empty_is_zero(self):
        m = maturity([])