breakpilot-compliance/scripts/obligation_discovery/materialize_capabilities.py

"""#5b — Materialisierung der Capability-Schicht (Modell C, User-Entscheidung 2026-06-26).

Aus `cross_domain_relationships.json` (SHARED_CAPABILITY) + den 6 CRA-P1-Registries:
- `obligations/capabilities.json` — Capability-Knoten: realized_by (n:m) + guidance_basis hochgezogen.
- `obligations/cra_core.json` — 2 CORE-Obligations (Sicherheitsziele): attack_surface_minimization,
  software_integrity_protection (Modell C: KEINE eigene SecurityObjective-Klasse; das Ziel IST eine
  abstrakte CORE-Pflicht).
- patcht DOMAIN-Obligations in ihren Registries: `specializes` (→CORE) + `objective_tags` (Vorwärts-
  Kompat zu Modell B: Tags, keine Klasse).
- markiert `vuln_remediation_patching` als deprecated_alias von `provide_security_updates` (Merge).
- `remote_access_data_export_protection` bleibt BEST_PRACTICE (Notiz: pending Data-Act-Scope).

Deterministisch. Lokal lauffähig (nur json). Danach export_join_keys neu (inkl. cra_core).
"""
from __future__ import annotations

import json

REG_FILES = ["obligations/cra.json", "obligations/cra_authentication.json",
             "obligations/cra_logging.json", "obligations/cra_remote_access.json",
             "obligations/cra_updates.json"]

# Cluster-capability_name -> kanonische capability_id (access_control absichtlich gedroppt: zu schwach)
CONSOLIDATE = {"mfa": "multi_factor_authentication", "session_management": "session_management",
               "tls_encryption": "transport_encryption", "mutual_tls": "transport_encryption",
               "tls_certificate_auth": "transport_encryption", "code_signing": "code_signing",
               "anomaly_detection": "security_monitoring_alerting"}
CAP_META = {
    "multi_factor_authentication": ("Multi-Factor Authentication",
        "Mehrfaktor-Authentisierung als technische Faehigkeit (Besitz/Wissen/Inhaerenz)."),
    "session_management": ("Session Management",
        "Sichere Sitzungsverwaltung: Timeouts, Bindung, Re-Auth, Beendigung."),
    "transport_encryption": ("Transport Encryption",
        "Verschluesselter Transport (TLS, mutual-TLS, Zertifikats-Auth, VPN/Tunnel)."),
    "code_signing": ("Code & Update Signing",
        "Digitale Signatur + Integritaets-/Authentizitaetspruefung von Firmware/Software/Updates."),
    "security_monitoring_alerting": ("Security Monitoring & Alerting",
        "Anomalie-/Bedrohungserkennung und Alarmierung aus Logs/Telemetrie."),
}
CAP_ORDER = ["multi_factor_authentication", "session_management", "transport_encryption",
             "code_signing", "security_monitoring_alerting"]

# DOMAIN-Obligation -> (CORE-Ziel, objective_tags)
SPECIALIZES = {
    "remote_access_attack_surface_min": ("attack_surface_minimization", ["attack_surface"]),
    "component_remote_interface_security": ("attack_surface_minimization", ["attack_surface"]),
    "signed_update_integrity": ("software_integrity_protection", ["integrity"]),
    "firmware_software_authentication": ("software_integrity_protection", ["integrity"]),
}


def main() -> None:
    regs = {f: json.load(open(f, encoding="utf-8")) for f in REG_FILES}
    idx = {}
    for f, r in regs.items():
        for o in r.get("obligations", []):
            idx[o["id"]] = (f, o)

    # realized_by aus dem Artefakt
    art = json.load(open("obligations/cross_domain_relationships.json", encoding="utf-8"))
    realized = {cid: set() for cid in CONSOLIDATE.values()}
    for rr in art["raw_results"]:
        if rr["relation"] == "SHARED_CAPABILITY" and rr.get("capability_name") in CONSOLIDATE:
            cid = CONSOLIDATE[rr["capability_name"]]
            realized[cid].update([rr["a"], rr["b"]])

    def guidance_for(ids):
        seen, out = set(), []
        for oid in ids:
            if oid in idx:
                for g in idx[oid][1].get("guidance_basis", []):
                    k = (g.get("source", ""), g.get("anchor", ""))
                    if k not in seen:
                        seen.add(k)
                        out.append(g)
        return out

    caps = []
    for cid in CAP_ORDER:
        obls = sorted(realized[cid])
        name, desc = CAP_META[cid]
        caps.append({"capability_id": cid, "name": name, "description": desc,
                     "type": "technical_capability", "realized_by": obls, "realizes_count": len(obls),
                     "guidance_basis": guidance_for(obls),
                     "domains": sorted({idx[o][1].get("family", "") for o in obls if o in idx}),
                     "provenance": {"source": "cross_domain_relationships.json SHARED_CAPABILITY"}})

    capabilities = {
        "schema_version": "capability_layer_v1", "model": "Modell C (docs-src/development/capability_model_v1.md)",
        "note": "Capability = technische Faehigkeit (regulierungs-agnostisch). realized_by = Obligations, "
                "die sie erfuellt (n:m). guidance_basis hier KANONISCH hochgezogen aus den realisierten "
                "Obligations (die Obligation-Kopien bleiben vorerst als Legacy; Strip = Folge-Cleanup). "
                "Sicherheitsziele sind KEINE Capabilities -> cra_core.json.",
        "dropped": {"access_control": "OVERLAP (credential_confidentiality <-> sbom_confidentiality), nicht materialisiert"},
        "candidate_capabilities_followup": ["automatic_update_delivery", "update_rollback",
            "trusted_update_source", "hash_verification", "secure_boot", "least_functionality",
            "credential_storage"],
        "capabilities": caps}
    json.dump(capabilities, open("obligations/capabilities.json", "w", encoding="utf-8"),
              ensure_ascii=False, indent=1)

    core = [
        {"id": "attack_surface_minimization", "name": "Minimierung der Angriffsflaeche", "family": "core",
         "description": "Das Produkt minimiert seine Angriffsflaeche: unnoetige Funktionen/Ports/Dienste/"
                        "Schnittstellen sind deaktiviert (Least Functionality).",
         "tier": "LEGAL_MINIMUM", "source_role": "LEGAL_BASIS", "applicability": "universal",
         "objective_tags": ["attack_surface"],
         "legal_basis": [{"source": "CRA", "anchor": "Annex I Part I (2)(j)",
                          "citation": "limit attack surfaces, including external interfaces"}],
         "guidance_basis": [{"source": "NIST", "anchor": "CM-7 Least Functionality", "role": "best_practice"}],
         "specialized_by": ["remote_access_attack_surface_min", "component_remote_interface_security"],
         "primary_implementation": "NIST CM-7", "citation_status": "pending_span_anchor",
         "review_status": "core_from_5b"},
        {"id": "software_integrity_protection", "name": "Schutz der Software-/Firmware-Integritaet", "family": "core",
         "description": "Das Produkt schuetzt Integritaet und Authentizitaet von Software/Firmware "
                        "(Manipulationserkennung, Secure Boot, Signaturpruefung, Runtime-Integritaet).",
         "tier": "LEGAL_MINIMUM", "source_role": "LEGAL_BASIS", "applicability": "universal",
         "objective_tags": ["integrity"],
         "legal_basis": [{"source": "CRA", "anchor": "Annex I Part I (2)(f)",
                          "citation": "protect the integrity of stored, transmitted or processed data, software and configuration"}],
         "guidance_basis": [{"source": "NIST", "anchor": "SI-7 Software, Firmware, and Information Integrity", "role": "best_practice"}],
         "specialized_by": ["signed_update_integrity", "firmware_software_authentication"],
         "realized_by_capabilities": ["code_signing"],
         "primary_implementation": "NIST SI-7", "citation_status": "pending_span_anchor",
         "review_status": "core_from_5b"},
    ]
    json.dump({"schema_version": "obligation_registry_v1", "regulation": "CRA", "regulation_code": "CRA",
               "family": "core",
               "theme": "CORE Security Objectives (CRA Annex I als regulierungs-agnostische Sicherheitsziele)",
               "generated_by": "materialize_capabilities.py (#5b, Modell C)",
               "note": "CORE Legal Obligations = Sicherheitsziele (Modell C: KEINE eigene SecurityObjective-Klasse). "
                       "DOMAIN-Obligations specializes-en hierauf. objective_tags = Vorwaerts-Kompat zu Modell B.",
               "citation_status": "pending_span_anchor", "obligations": core, "relationships": []},
              open("obligations/cra_core.json", "w", encoding="utf-8"), ensure_ascii=False, indent=1)

    dirty = set()
    patched = []
    for oid, (coreid, tags) in SPECIALIZES.items():
        if oid in idx:
            f, o = idx[oid]
            o["specializes"] = coreid
            o["objective_tags"] = tags
            dirty.add(f)
            patched.append(oid)
    if "vuln_remediation_patching" in idx:
        f, o = idx["vuln_remediation_patching"]
        o["merged_into"] = "provide_security_updates"
        o["status"] = "deprecated_alias"
        o["merge_note"] = ("SAME_OBLIGATION (Cross-Domain-Review). Kanonisch: provide_security_updates "
                           "((2)(c)/Art.13). ID bleibt als Alias aufloesbar; downstream provide_security_updates nutzen.")
        dirty.add(f)
    if "remote_access_data_export_protection" in idx:
        f, o = idx["remote_access_data_export_protection"]
        o["tier_note"] = ("Bleibt BEST_PRACTICE (NICHT LM) bis Data-Act/Export-Scope sauber ist (User #5b.6). "
                          "Evtl. Capability-or-Procedure statt Obligation.")
        dirty.add(f)
    for f in dirty:
        json.dump(regs[f], open(f, "w", encoding="utf-8"), ensure_ascii=False, indent=1)

    print("capabilities.json:", len(caps), "Capabilities")
    for c in caps:
        print(f"  {c['capability_id']:30s} realizes {c['realizes_count']:2d} | guidance {len(c['guidance_basis'])} | {c['domains']}")
    print("cra_core.json: 2 CORE (attack_surface_minimization, software_integrity_protection)")
    print("specializes gepatcht:", patched)
    print("alias: vuln_remediation_patching -> provide_security_updates")
    print("dirty registries:", sorted(dirty))


if __name__ == "__main__":
    main()