Compare commits

...

11 Commits

Author SHA1 Message Date
Benjamin Admin c5ecfa8f6c feat(bridge): export 7 accepted CRA->OWASP controls for obligation_id proposal
CI / detect-changes (push) Successful in 5s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / build-sha-integrity (push) Successful in 9s
CI / validate-canonical-controls (push) Successful in 5s
CI / loc-budget (push) Successful in 23s
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Has been skipped
CI / test-go (push) Has been skipped
CI / iace-gt-coverage (push) Has been skipped
CI / test-python-backend (push) Has been skipped
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
obligations/controls_for_obligation_mapping.json — the Compliance Execution
Graph's accepted controls (V6 auth / V11 crypto / V16 logging) handed to the
Obligation Registry to propose the SEMANTIC control->obligation_id, replacing
the coarse citation_unit interim join (Befund 1). Registry fills
proposed_obligation_id; we then adopt it into control_mapping.obligation_id.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 11:36:57 +02:00
Benjamin Admin 9e0a9ccef4 Add obligation_id join-key contract (cross-session bridge)
CI / detect-changes (push) Successful in 5s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / build-sha-integrity (push) Successful in 8s
CI / validate-canonical-controls (push) Successful in 7s
CI / loc-budget (push) Successful in 19s
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Has been skipped
CI / test-go (push) Has been skipped
CI / iace-gt-coverage (push) Has been skipped
CI / test-python-backend (push) Has been skipped
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
Macht meine Seite des Cross-Session-Vertrags konkret: obligation_id ist der stabile Join-Key
zwischen Legal Knowledge Graph (citation_spans -> obligation_id) und Compliance Execution Graph
(control_mapping.source_norm -> obligation_id). Export aller 47 obligation_ids (CRA: 11 sbom +
7 vuln + 29 auth) mit citation_units als Interim-Brücke. Disziplin: obligation_id nie neu
vergeben (re-link, Pendant zu span_id/control_uuid).

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 10:29:29 +02:00
Benjamin Admin 7e1c3668bf Merge remote-tracking branch 'origin/main' into feat/obligation-aggregation
CI / detect-changes (push) Successful in 8s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / build-sha-integrity (push) Successful in 5s
CI / validate-canonical-controls (push) Successful in 4s
CI / loc-budget (push) Successful in 14s
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Has been skipped
CI / test-go (push) Has been skipped
CI / iace-gt-coverage (push) Has been skipped
CI / test-python-backend (push) Has been skipped
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
2026-06-25 10:15:25 +02:00
Benjamin Admin ab3cb86b1c feat(ucca): Evidence-Requirement model (step A)
CI / detect-changes (push) Successful in 5s
CI / branch-name (push) Has been skipped
CI / guardrail-integrity (push) Has been skipped
CI / secret-scan (push) Has been skipped
CI / dep-audit (push) Has been skipped
CI / sbom-scan (push) Has been skipped
CI / build-sha-integrity (push) Successful in 7s
CI / validate-canonical-controls (push) Successful in 5s
CI / loc-budget (push) Successful in 20s
CI / go-lint (push) Has been skipped
CI / python-lint (push) Has been skipped
CI / nodejs-lint (push) Has been skipped
CI / nodejs-build (push) Has been skipped
CI / test-go (push) Successful in 1m5s
CI / iace-gt-coverage (push) Successful in 17s
CI / test-python-backend (push) Has been skipped
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
The last edge of the compliance graph: what concrete, fresh evidence proves a
framework control is met (config_export/test_report/sbom/audit_log/pentest/...
from github/ci/scanner/manual_upload, with a freshness requirement).

Seeded for all 7 accepted CRA->OWASP controls (Auth/Crypto/Logging). A graph
test enforces connectivity: every accepted control must carry >=1 required
evidence — no dangling node in Obligation -> Control -> Evidence.

This is what will let the Advisor state "the CRA requirement is fulfilled" from
present evidence, not from the mere existence of a document.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 10:06:09 +02:00
Benjamin Admin 0db0e9a129 feat(ucca): curate CRA Annex I -> OWASP mappings (review B)
7 accepted, 13 rejected (reviewed_by=benjamin, 2026-06-25). The accepted set is
the first audited ground truth of the compliance graph:
  (2c) Zugriff   -> V6.3.1, V6.1.1   (Auth)
  (2d) Crypto    -> V11.2.1, V11.7.1 (corrected from the retriever's wrong V14)
  (2k) Logging   -> V16.3.3, V16.3.4, V16.1.1

Rejected stay as audit trail. (2e) integrity, (2l) updates, (2i) attack surface
rejected with reason "OWASP ASVS not the right target standard, map via NIST/BSI"
— architectural proof for the multi-framework framework_* layer.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 10:01:06 +02:00
Benjamin Admin 53ea388ea0 refactor(ucca): control-mapping model per review feedback
- DROP confidence from the persisted mapping: a curated mapping is a
  professional statement, not an AI guess (retriever score -> rationale only).
- ADD mapping_status (candidate|accepted|rejected|superseded) — the review state.
- ADD audit trail (reviewed_by/review_date/review_reason); accepted/rejected
  fail-closed without it.
- EXTEND mapping_type: + implements, + contradicts.
- Advisor truth = mapping_status=accepted (acceptedOnly filter).
- migrate the 18 CRA->OWASP rows to mapping_status=candidate.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:50:37 +02:00
Benjamin Admin e5cce9caff Extend advisor proof with procedure→evidence chain
Vollständige Begründungskette aus der Registry: Rechtsgrundlage → Obligation → Procedure
→ Controls → Evidence → Antwort. Join cra.json × cra_procedures.json, deterministisch, kein LLM.

SBOM-Beweis: 7 Pflichten je mit CRA-Rechtsgrundlage + Procedure (wie umgesetzt) + Controls
(Prüfung) + aggregierte Required Evidence; 4 Best-Practice (Guidance OWASP/NIST/ENISA);
Beziehung sbom_*→supports→vuln_identification; citation 7/7 pending_span_anchor.

Der Unterschied zu RAG sichtbar: RAG beantwortet — BreakPilot begründet UND operationalisiert.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:44:27 +02:00
Benjamin Admin 2f3c98fbe0 feat(ucca): first CRA Annex I -> OWASP retriever candidates (step 3)
18 retriever_candidate mappings generated via the sdk-dev control-intent
retriever. All marked retriever_candidate (NOT curated truth) — the review
step turns the good ones into human_curated.

Empirical validation of the A-decision: the retriever proposes, but produces
wrong candidates (e.g. encryption -> V14 Config instead of V11 Crypto;
V14.2.4 over-appears) that only human review catches. Review notes inline.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:36:53 +02:00
Benjamin Admin d987e4fde6 feat(ucca): persisted Control-Mapping data model (Obligation -> framework control)
Versioned JSONL store + Go model for Regulation->Control mappings, per the
A-decision: the retriever only PROPOSES candidates; the curated mapping is the
audited truth the Advisor uses at runtime, never re-invented per query.

- ControlMapping struct (source_norm/source_role/target_framework/target_control/
  mapping_type/confidence/provenance/rationale/version)
- enum validation (rule layer), fail-closed loader, forward+reverse index,
  curated-only filter (IsCurated)
- seed: 2 retriever_candidate rows CRA Annex I -> OWASP ASVS (not yet curated)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:32:15 +02:00
Benjamin Admin 67dba5f641 Add CRA procedure model (SBOM + Vuln)
Schließt die Lücke Obligation→Procedure→Control→Evidence (Schritt 3, Compliance-OS-Ebene).
Procedure = Umsetzungs-/Nachweisebene EINER Obligation, KEINE neue Pflicht (LEGAL_MINIMUM
bleibt an der Obligation; Procedure beschreibt Umsetzung; Evidence belegt sie).

- 11 Procedures (5 SBOM + 6 Vuln), 2 Worked Examples; source_role=procedural_requirement
  (Konvergenz mit der Legal-Knowledge-Engine der anderen Session)
- fulfills_obligations[] referenziert die cra.json-Obligations (alle gültig, volle Abdeckung)
- steps/controls/evidence je Procedure; KEINE tier/legal_basis-Felder (kein Pflicht-Duplikat)
- citation_spans: [] / pending_span_anchor (Join folgt mit dem zitierfähigen Re-Ingest)

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:28:40 +02:00
Benjamin Admin db2fd9d8e9 Add obligation advisor proof (P3)
Demonstriert den Produktnutzen der Registry: obligation-basierte Antwort statt RAG-Text.
Frage → Pflicht (LEGAL_MINIMUM + Rechtsgrundlage + Applicability) ⊥ Best Practice
(guidance_basis) ⊥ Nachweise (evidence_facets + member controls) + Beziehungen, deterministisch
aus obligations/cra.json (kein LLM, zitierfähig).

Beleg (SBOM, Maschinenbauer): JA — 7 CRA-Mindestpflichten + 4 Best-Practice (OWASP/NIST/ENISA);
sbom_* supports vuln_identification_inventory.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
2026-06-25 09:06:34 +02:00
11 changed files with 1340 additions and 0 deletions
@@ -0,0 +1,24 @@
// Control-Mapping: CRA Annex I -> OWASP ASVS 5.0. Eine Zeile = ein Mapping (Schema: ControlMapping).
// Reviewt 2026-06-25 (benjamin): 7 accepted, 13 rejected. accepted = Audit-Wahrheit (Advisor nutzt acceptedOnly).
// rejected bleiben als Audit-Spur ("warum verworfen"). KEIN confidence — kuratiert = fachliche Feststellung.
// Architekturbeweis: CRA -> OWASP fuer AppSec/Auth/Crypto/Logging; Ops/Update/Attack-Surface/Integritaet -> NIST/BSI.
{"source_norm": "CRA Annex I Part I (2)(c) — Schutz vor unbefugtem Zugriff", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V6.3.1", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V6 = Authentication.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V6 = Authentication, sauberer Treffer fuer Zugriffsschutz/Authentisierung.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(c) — Schutz vor unbefugtem Zugriff", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V6.1.1", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V6 = Authentication.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V6 = Authentication, sauberer Treffer fuer Zugriffsschutz/Authentisierung.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V11.2.1", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V11 = Cryptography.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "Korrektur von V14: V11 = Cryptography, richtiger Bereich fuer Verschluesselung.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V11.7.1", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V11.7 = Key Management.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "Korrektur von V14: V11.7 = Key Management fuer Verschluesselung/Schluesselverwaltung.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V16.3.3", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V16 = Security Logging.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V16 = Logging, sauberer Treffer fuer sicherheitsrelevante Ereignisse.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V16.3.4", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V16 = Security Logging.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V16 = Logging, sauberer Treffer fuer sicherheitsrelevante Ereignisse.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V16.1.1", "mapping_type": "supports", "mapping_status": "accepted", "provenance": "human_curated", "rationale": "V16 = Security Logging.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V16 = Logging, sauberer Treffer fuer sicherheitsrelevante Ereignisse.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(c) — Schutz vor unbefugtem Zugriff", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V14 = Config, kein Auth — verworfen.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V14 = Config, Crypto gehoert zu V11 — verworfen.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.3.2", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V14 = Config, Crypto gehoert zu V11 — verworfen.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.2.3", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "V14 = Config, Crypto gehoert zu V11 — verworfen.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(e) — Integritaet", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(e) — Integritaet", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V1.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(e) — Integritaet", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V6.1.1", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(l) — Sichere Updates", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V14.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(l) — Sichere Updates", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V2.4.1", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(l) — Sichere Updates", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V6.1.1", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(i) — Angriffsflaeche minimieren", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V6.1.1", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(i) — Angriffsflaeche minimieren", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V15.3.3", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
{"source_norm": "CRA Annex I Part I (2)(i) — Angriffsflaeche minimieren", "source_role": "operational_requirement", "target_framework": "OWASP ASVS", "target_control": "V8.2.4", "mapping_type": "related", "mapping_status": "rejected", "provenance": "human_curated", "rationale": "Retriever-Kandidat.", "reviewed_by": "benjamin", "review_date": "2026-06-25", "review_reason": "OWASP ASVS ist hier nicht der passende Zielstandard. Mapping ueber NIST/BSI erforderlich.", "version": "2026-06-25"}
@@ -0,0 +1,16 @@
// Evidence-Requirements je OWASP-ASVS-Control (Schema: EvidenceRequirement). Eine Zeile = eine geforderte Evidenz.
// Autoriert/kuratiert (nicht Retriever). Der Advisor kann eine CRA-Anforderung erst dann als erfuellt melden,
// wenn die required Evidenzen der gemappten, accepted Controls vorliegen + frisch genug sind.
// Stand 2026-06-25, Basis: die 7 accepted CRA->OWASP-Mappings (Auth V6, Crypto V11, Logging V16).
{"framework": "OWASP ASVS", "control": "V6.3.1", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "IAM-/Zugriffskonfiguration als Nachweis der Authentisierungs-Anforderung.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V6.3.1", "evidence_type": "test_report", "evidence_source": "ci", "freshness_requirement": "per_release", "required": true, "rationale": "Automatisierter Zugriffstest (CI) belegt funktionierende Zugriffskontrolle.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V6.3.1", "evidence_type": "pentest", "evidence_source": "manual_upload", "freshness_requirement": "annually", "required": false, "rationale": "Jaehrlicher PenTest der Authentisierung — vertieft, aber nicht Pflicht je Release.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V6.1.1", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "Rollenmodell/Auth-Architektur als Nachweis.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V11.2.1", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "Krypto-Konfiguration (zugelassene Algorithmen) als Nachweis der Verschluesselung.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V11.2.1", "evidence_type": "sbom", "evidence_source": "ci", "freshness_requirement": "per_release", "required": true, "rationale": "SBOM weist die eingesetzten Krypto-Bibliotheken/-Versionen nach.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V11.7.1", "evidence_type": "policy", "evidence_source": "manual_upload", "freshness_requirement": "annually", "required": true, "rationale": "Key-Management-Policy (Rotation, Aufbewahrung) als organisatorischer Nachweis.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V11.7.1", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "Konfiguration der Schluesselverwaltung als technischer Nachweis.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V16.3.3", "evidence_type": "audit_log", "evidence_source": "ci", "freshness_requirement": "continuous", "required": true, "rationale": "Security-Audit-Logs belegen, dass sicherheitsrelevante Ereignisse protokolliert werden.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V16.3.3", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "Logging-Konfiguration als Nachweis der erfassten Ereignisarten.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V16.3.4", "evidence_type": "audit_log", "evidence_source": "ci", "freshness_requirement": "continuous", "required": true, "rationale": "Security-Audit-Logs.", "version": "2026-06-25"}
{"framework": "OWASP ASVS", "control": "V16.1.1", "evidence_type": "config_export", "evidence_source": "github", "freshness_requirement": "per_release", "required": true, "rationale": "Logging-Architektur-Konfiguration als Nachweis.", "version": "2026-06-25"}
@@ -0,0 +1,151 @@
package ucca
import (
"bufio"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
)
// ControlMapping is one persisted, versioned, REVIEWABLE link from a legal
// obligation/requirement to a concrete framework control — a node in the curated
// compliance graph (Regulation -> Obligation -> Control -> Evidence). The retriever only
// PROPOSES candidates (mapping_status=candidate); a human/rule decision turns the good ones
// into mapping_status=accepted, which is the audited truth the Advisor uses at runtime.
//
// There is intentionally NO probabilistic "confidence" field: once curated, a mapping is a
// professional statement, not an AI guess. The retriever's score lives only in the rationale
// of a candidate, never as structured truth.
type ControlMapping struct {
SourceNorm string `json:"source_norm"` // e.g. "CRA Annex I Part I (2)(c)"
SourceRole string `json:"source_role"` // source_role of the norm (operational_requirement, ...)
TargetFramework string `json:"target_framework"` // e.g. "OWASP ASVS"
TargetControl string `json:"target_control"` // e.g. "V6.3.1"
MappingType string `json:"mapping_type"` // supports | partially_supports | implements | related | contradicts
MappingStatus string `json:"mapping_status"` // candidate | accepted | rejected | superseded
Provenance string `json:"provenance"` // retriever_candidate | human_curated | rule_based
Rationale string `json:"rationale"`
ReviewedBy string `json:"reviewed_by,omitempty"` // who decided (human or rule id)
ReviewDate string `json:"review_date,omitempty"` // YYYY-MM-DD
ReviewReason string `json:"review_reason,omitempty"`
Version string `json:"version"`
}
// Allowed enum values — the deterministic "rule" layer that keeps the curated store clean.
var (
mappingTypeValues = map[string]bool{"supports": true, "partially_supports": true, "implements": true, "related": true, "contradicts": true}
mappingStatusValues = map[string]bool{"candidate": true, "accepted": true, "rejected": true, "superseded": true}
provenanceValues = map[string]bool{"retriever_candidate": true, "human_curated": true, "rule_based": true}
)
// Validate checks required fields + enum membership, and enforces the audit trail: any
// human/rule DECISION (accepted/rejected) must carry who/when/why. Fail-closed at load.
func (m ControlMapping) Validate() error {
switch {
case m.SourceNorm == "":
return fmt.Errorf("control mapping: source_norm required")
case m.TargetFramework == "":
return fmt.Errorf("control mapping: target_framework required")
case m.TargetControl == "":
return fmt.Errorf("control mapping: target_control required")
case !mappingTypeValues[m.MappingType]:
return fmt.Errorf("control mapping: invalid mapping_type %q", m.MappingType)
case !mappingStatusValues[m.MappingStatus]:
return fmt.Errorf("control mapping: invalid mapping_status %q", m.MappingStatus)
case !provenanceValues[m.Provenance]:
return fmt.Errorf("control mapping: invalid provenance %q", m.Provenance)
}
if m.MappingStatus == "accepted" || m.MappingStatus == "rejected" {
if m.ReviewedBy == "" || m.ReviewDate == "" || m.ReviewReason == "" {
return fmt.Errorf("control mapping %s->%s: status %q requires reviewed_by + review_date + review_reason (audit trail)",
m.SourceNorm, m.TargetControl, m.MappingStatus)
}
}
return nil
}
// IsAccepted reports whether this mapping is the active audited truth.
func (m ControlMapping) IsAccepted() bool { return m.MappingStatus == "accepted" }
// ControlMappingSet is the loaded, indexed mapping store (forward + reverse lookup).
type ControlMappingSet struct {
All []ControlMapping
bySourceNorm map[string][]ControlMapping
byControl map[string][]ControlMapping
}
func controlKey(framework, control string) string { return framework + ":" + control }
// ControlsFor returns the controls mapped to a source norm. acceptedOnly restricts to the
// audited truth (what the Advisor may treat as fact).
func (s *ControlMappingSet) ControlsFor(sourceNorm string, acceptedOnly bool) []ControlMapping {
return filterAccepted(s.bySourceNorm[sourceNorm], acceptedOnly)
}
// ObligationsFor returns the norms mapped to a framework control (reverse lookup).
func (s *ControlMappingSet) ObligationsFor(framework, control string, acceptedOnly bool) []ControlMapping {
return filterAccepted(s.byControl[controlKey(framework, control)], acceptedOnly)
}
func filterAccepted(in []ControlMapping, acceptedOnly bool) []ControlMapping {
if !acceptedOnly {
return in
}
out := make([]ControlMapping, 0, len(in))
for _, m := range in {
if m.IsAccepted() {
out = append(out, m)
}
}
return out
}
// LoadControlMappings reads every *.jsonl file under dir (one mapping per line; blank and
// //-prefixed lines ignored), validates each row, and builds the index. An invalid row
// aborts the whole load — fail-closed, because this is the audit truth, not best-effort.
func LoadControlMappings(dir string) (*ControlMappingSet, error) {
files, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
if err != nil {
return nil, err
}
set := &ControlMappingSet{
bySourceNorm: map[string][]ControlMapping{},
byControl: map[string][]ControlMapping{},
}
for _, f := range files {
fh, err := os.Open(f)
if err != nil {
return nil, err
}
sc := bufio.NewScanner(fh)
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
line := 0
for sc.Scan() {
line++
raw := strings.TrimSpace(sc.Text())
if raw == "" || strings.HasPrefix(raw, "//") {
continue
}
var m ControlMapping
if err := json.Unmarshal([]byte(raw), &m); err != nil {
fh.Close()
return nil, fmt.Errorf("%s:%d: %w", f, line, err)
}
if err := m.Validate(); err != nil {
fh.Close()
return nil, fmt.Errorf("%s:%d: %w", f, line, err)
}
set.All = append(set.All, m)
set.bySourceNorm[m.SourceNorm] = append(set.bySourceNorm[m.SourceNorm], m)
k := controlKey(m.TargetFramework, m.TargetControl)
set.byControl[k] = append(set.byControl[k], m)
}
fh.Close()
if err := sc.Err(); err != nil {
return nil, err
}
}
return set, nil
}
@@ -0,0 +1,85 @@
package ucca
import (
"os"
"path/filepath"
"testing"
)
func TestControlMapping_Validate(t *testing.T) {
candidate := ControlMapping{SourceNorm: "CRA Annex I", TargetFramework: "OWASP ASVS", TargetControl: "V6.3.1", MappingType: "supports", MappingStatus: "candidate", Provenance: "retriever_candidate"}
if err := candidate.Validate(); err != nil {
t.Fatalf("valid candidate rejected: %v", err)
}
accepted := ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "implements", MappingStatus: "accepted", Provenance: "human_curated", ReviewedBy: "benjamin", ReviewDate: "2026-06-25", ReviewReason: "passt"}
if err := accepted.Validate(); err != nil {
t.Fatalf("valid accepted rejected: %v", err)
}
bad := []struct {
name string
m ControlMapping
}{
{"no source_norm", ControlMapping{TargetFramework: "X", TargetControl: "Y", MappingType: "supports", MappingStatus: "candidate", Provenance: "retriever_candidate"}},
{"bad mapping_type", ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "nope", MappingStatus: "candidate", Provenance: "retriever_candidate"}},
{"bad mapping_status", ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "supports", MappingStatus: "maybe", Provenance: "retriever_candidate"}},
{"bad provenance", ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "supports", MappingStatus: "candidate", Provenance: "guessed"}},
{"accepted without audit trail", ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "supports", MappingStatus: "accepted", Provenance: "human_curated"}},
{"rejected without reason", ControlMapping{SourceNorm: "A", TargetFramework: "X", TargetControl: "Y", MappingType: "supports", MappingStatus: "rejected", Provenance: "human_curated", ReviewedBy: "b", ReviewDate: "2026-06-25"}},
}
for _, tt := range bad {
if err := tt.m.Validate(); err == nil {
t.Errorf("%s: expected rejection", tt.name)
}
}
}
func TestLoadControlMappings(t *testing.T) {
dir := t.TempDir()
content := `// header comment, ignored
{"source_norm":"CRA Annex I","source_role":"operational_requirement","target_framework":"OWASP ASVS","target_control":"V6.3.1","mapping_type":"supports","mapping_status":"accepted","provenance":"human_curated","reviewed_by":"benjamin","review_date":"2026-06-25","review_reason":"V6=Auth passt","rationale":"r","version":"2026-06-25"}
{"source_norm":"CRA Annex I","source_role":"operational_requirement","target_framework":"OWASP ASVS","target_control":"V14.2.4","mapping_type":"related","mapping_status":"candidate","provenance":"retriever_candidate","rationale":"r","version":"2026-06-25"}
`
if err := os.WriteFile(filepath.Join(dir, "m.jsonl"), []byte(content), 0o644); err != nil {
t.Fatal(err)
}
set, err := LoadControlMappings(dir)
if err != nil {
t.Fatalf("load: %v", err)
}
if len(set.All) != 2 {
t.Fatalf("want 2 mappings, got %d", len(set.All))
}
if got := set.ControlsFor("CRA Annex I", false); len(got) != 2 {
t.Errorf("ControlsFor(all): want 2, got %d", len(got))
}
if got := set.ControlsFor("CRA Annex I", true); len(got) != 1 {
t.Errorf("ControlsFor(acceptedOnly): want 1 (only accepted), got %d", len(got))
}
if got := set.ObligationsFor("OWASP ASVS", "V6.3.1", true); len(got) != 1 {
t.Errorf("ObligationsFor accepted reverse lookup: want 1, got %d", len(got))
}
}
func TestLoadControlMappings_RejectsInvalid(t *testing.T) {
dir := t.TempDir()
// accepted without the who/when/why audit trail must fail-closed.
if err := os.WriteFile(filepath.Join(dir, "bad.jsonl"), []byte(`{"source_norm":"A","target_framework":"X","target_control":"Y","mapping_type":"supports","mapping_status":"accepted","provenance":"human_curated","rationale":"r","version":"v"}`), 0o644); err != nil {
t.Fatal(err)
}
if _, err := LoadControlMappings(dir); err == nil {
t.Error("accepted mapping without audit trail must fail the load (fail-closed)")
}
}
func TestControlMappings_SeedFileValid(t *testing.T) {
// The committed seed store must always load + validate.
set, err := LoadControlMappings("../../data/control_mappings")
if err != nil {
t.Fatalf("seed control_mappings failed to load: %v", err)
}
if len(set.All) == 0 {
t.Fatal("seed control_mappings is empty")
}
}
@@ -0,0 +1,117 @@
package ucca
import (
"bufio"
"encoding/json"
"fmt"
"os"
"path/filepath"
"strings"
)
// EvidenceRequirement is the last edge of the compliance graph: it says WHAT concrete
// evidence proves a framework control is met, and how fresh that evidence must be. This is
// what lets the Advisor eventually state "the CRA requirement is fulfilled" — not because a
// document exists, but because the required, current evidence is present. Authored/curated,
// not retriever-generated.
type EvidenceRequirement struct {
Framework string `json:"framework"` // e.g. "OWASP ASVS"
Control string `json:"control"` // e.g. "V6.3.1"
EvidenceType string `json:"evidence_type"` // sbom|test_report|config_export|repo_scan|policy|ticket|audit_log|pentest
EvidenceSource string `json:"evidence_source"` // github|ci|scanner|manual_upload
FreshnessRequirement string `json:"freshness_requirement"` // per_release|quarterly|annually|continuous
Required bool `json:"required"`
Rationale string `json:"rationale"`
Version string `json:"version"`
}
// Allowed enum values — the rule layer that keeps the evidence catalog clean.
var (
evidenceTypeValues = map[string]bool{"sbom": true, "test_report": true, "config_export": true, "repo_scan": true, "policy": true, "ticket": true, "audit_log": true, "pentest": true}
evidenceSourceValues = map[string]bool{"github": true, "ci": true, "scanner": true, "manual_upload": true}
freshnessValues = map[string]bool{"per_release": true, "quarterly": true, "annually": true, "continuous": true}
)
// Validate checks required fields + enum membership. Fail-closed at load.
func (e EvidenceRequirement) Validate() error {
switch {
case e.Framework == "":
return fmt.Errorf("evidence requirement: framework required")
case e.Control == "":
return fmt.Errorf("evidence requirement: control required")
case !evidenceTypeValues[e.EvidenceType]:
return fmt.Errorf("evidence requirement: invalid evidence_type %q", e.EvidenceType)
case !evidenceSourceValues[e.EvidenceSource]:
return fmt.Errorf("evidence requirement: invalid evidence_source %q", e.EvidenceSource)
case !freshnessValues[e.FreshnessRequirement]:
return fmt.Errorf("evidence requirement: invalid freshness_requirement %q", e.FreshnessRequirement)
}
return nil
}
// EvidenceRequirementSet is the loaded, indexed evidence catalog.
type EvidenceRequirementSet struct {
All []EvidenceRequirement
byControl map[string][]EvidenceRequirement
}
// For returns all evidence requirements declared for a framework control.
func (s *EvidenceRequirementSet) For(framework, control string) []EvidenceRequirement {
return s.byControl[controlKey(framework, control)]
}
// RequiredFor returns only the required evidence for a control — the minimum that must be
// present before the control may be treated as met.
func (s *EvidenceRequirementSet) RequiredFor(framework, control string) []EvidenceRequirement {
out := make([]EvidenceRequirement, 0)
for _, e := range s.byControl[controlKey(framework, control)] {
if e.Required {
out = append(out, e)
}
}
return out
}
// LoadEvidenceRequirements reads every *.jsonl file under dir (one requirement per line;
// blank and //-prefixed lines ignored), validates each, and builds the per-control index.
// An invalid row aborts the load — fail-closed.
func LoadEvidenceRequirements(dir string) (*EvidenceRequirementSet, error) {
files, err := filepath.Glob(filepath.Join(dir, "*.jsonl"))
if err != nil {
return nil, err
}
set := &EvidenceRequirementSet{byControl: map[string][]EvidenceRequirement{}}
for _, f := range files {
fh, err := os.Open(f)
if err != nil {
return nil, err
}
sc := bufio.NewScanner(fh)
sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
line := 0
for sc.Scan() {
line++
raw := strings.TrimSpace(sc.Text())
if raw == "" || strings.HasPrefix(raw, "//") {
continue
}
var e EvidenceRequirement
if err := json.Unmarshal([]byte(raw), &e); err != nil {
fh.Close()
return nil, fmt.Errorf("%s:%d: %w", f, line, err)
}
if err := e.Validate(); err != nil {
fh.Close()
return nil, fmt.Errorf("%s:%d: %w", f, line, err)
}
set.All = append(set.All, e)
k := controlKey(e.Framework, e.Control)
set.byControl[k] = append(set.byControl[k], e)
}
fh.Close()
if err := sc.Err(); err != nil {
return nil, err
}
}
return set, nil
}
@@ -0,0 +1,84 @@
package ucca
import (
"os"
"path/filepath"
"testing"
)
func TestEvidenceRequirement_Validate(t *testing.T) {
valid := EvidenceRequirement{Framework: "OWASP ASVS", Control: "V6.3.1", EvidenceType: "config_export", EvidenceSource: "github", FreshnessRequirement: "per_release", Required: true}
if err := valid.Validate(); err != nil {
t.Fatalf("valid rejected: %v", err)
}
bad := []struct {
name string
e EvidenceRequirement
}{
{"no control", EvidenceRequirement{Framework: "X", EvidenceType: "sbom", EvidenceSource: "ci", FreshnessRequirement: "per_release"}},
{"bad evidence_type", EvidenceRequirement{Framework: "X", Control: "Y", EvidenceType: "screenshot", EvidenceSource: "ci", FreshnessRequirement: "per_release"}},
{"bad evidence_source", EvidenceRequirement{Framework: "X", Control: "Y", EvidenceType: "sbom", EvidenceSource: "email", FreshnessRequirement: "per_release"}},
{"bad freshness", EvidenceRequirement{Framework: "X", Control: "Y", EvidenceType: "sbom", EvidenceSource: "ci", FreshnessRequirement: "weekly"}},
}
for _, tt := range bad {
if err := tt.e.Validate(); err == nil {
t.Errorf("%s: expected rejection", tt.name)
}
}
}
func TestLoadEvidenceRequirements(t *testing.T) {
dir := t.TempDir()
content := `// header
{"framework":"OWASP ASVS","control":"V6.3.1","evidence_type":"config_export","evidence_source":"github","freshness_requirement":"per_release","required":true,"version":"2026-06-25"}
{"framework":"OWASP ASVS","control":"V6.3.1","evidence_type":"pentest","evidence_source":"manual_upload","freshness_requirement":"annually","required":false,"version":"2026-06-25"}
`
if err := os.WriteFile(filepath.Join(dir, "e.jsonl"), []byte(content), 0o644); err != nil {
t.Fatal(err)
}
set, err := LoadEvidenceRequirements(dir)
if err != nil {
t.Fatalf("load: %v", err)
}
if len(set.All) != 2 {
t.Fatalf("want 2, got %d", len(set.All))
}
if got := set.For("OWASP ASVS", "V6.3.1"); len(got) != 2 {
t.Errorf("For: want 2, got %d", len(got))
}
if got := set.RequiredFor("OWASP ASVS", "V6.3.1"); len(got) != 1 {
t.Errorf("RequiredFor: want 1 (pentest is optional), got %d", len(got))
}
}
func TestEvidenceRequirements_SeedFileValid(t *testing.T) {
set, err := LoadEvidenceRequirements("../../data/evidence_requirements")
if err != nil {
t.Fatalf("seed evidence_requirements failed to load: %v", err)
}
if len(set.All) == 0 {
t.Fatal("seed evidence_requirements is empty")
}
}
// TestGraph_AcceptedControlsHaveEvidence wires the two layers: every control an accepted
// CRA->OWASP mapping points to must have >=1 required evidence — the Obligation -> Control ->
// Evidence chain must be connected, no dangling control nodes.
func TestGraph_AcceptedControlsHaveEvidence(t *testing.T) {
maps, err := LoadControlMappings("../../data/control_mappings")
if err != nil {
t.Fatal(err)
}
ev, err := LoadEvidenceRequirements("../../data/evidence_requirements")
if err != nil {
t.Fatal(err)
}
for _, m := range maps.All {
if !m.IsAccepted() {
continue
}
if len(ev.RequiredFor(m.TargetFramework, m.TargetControl)) == 0 {
t.Errorf("accepted control %s %s has no required evidence (dangling graph node)", m.TargetFramework, m.TargetControl)
}
}
}
@@ -0,0 +1,71 @@
{
"schema_version": "controls_for_obligation_mapping_v1",
"purpose": "Accepted CRA->OWASP controls (Compliance Execution Graph) for the Obligation Registry to propose the SEMANTIC control->obligation_id, replacing the coarse citation_unit interim join. Fill proposed_obligation_id per control, then we adopt it into control_mapping.obligation_id.",
"source": "ai-compliance-sdk control_mappings, mapping_status=accepted, reviewed_by=benjamin 2026-06-25",
"count": 7,
"controls": [
{
"framework": "OWASP ASVS",
"control": "V6.3.1",
"source_norm": "CRA Annex I Part I (2)(c) — Schutz vor unbefugtem Zugriff",
"citation_unit": "Annex I (2)(c)",
"family": "auth",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V6.1.1",
"source_norm": "CRA Annex I Part I (2)(c) — Schutz vor unbefugtem Zugriff",
"citation_unit": "Annex I (2)(c)",
"family": "auth",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V11.2.1",
"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung",
"citation_unit": "Annex I (2)(d)",
"family": "crypto",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V11.7.1",
"source_norm": "CRA Annex I Part I (2)(d) — Vertraulichkeit / Verschluesselung",
"citation_unit": "Annex I (2)(d)",
"family": "crypto",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V16.3.3",
"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging",
"citation_unit": "Annex I (2)(k)",
"family": "logging",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V16.3.4",
"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging",
"citation_unit": "Annex I (2)(k)",
"family": "logging",
"mapping_type": "supports",
"proposed_obligation_id": ""
},
{
"framework": "OWASP ASVS",
"control": "V16.1.1",
"source_norm": "CRA Annex I Part I (2)(k) — Sicherheitsrelevante Ereignisse / Logging",
"citation_unit": "Annex I (2)(k)",
"family": "logging",
"mapping_type": "supports",
"proposed_obligation_id": ""
}
]
}
+227
View File
@@ -0,0 +1,227 @@
{
"schema_version": "obligation_procedures_v1",
"regulation": "CRA",
"layer": "Regulation -> Legal Obligation -> Procedure -> Control -> Evidence",
"note": "Procedure ist KEINE neue Compliance-Pflicht. LEGAL_MINIMUM liegt an der Obligation; die Procedure beschreibt, WIE sie umgesetzt wird; Evidence belegt die Umsetzung. source_role=procedural_requirement (Konvergenz mit der Legal-Knowledge-Engine der anderen Session).",
"citation_status": "pending_span_anchor",
"scope": "worked examples: SBOM + Vulnerability Handling",
"procedures": [
{
"procedure_id": "sbom_generation_process",
"name": "SBOM-Erstellungsprozess",
"description": "Erzeugen einer vollstaendigen, maschinenlesbaren Software Bill of Materials fuer ein Produkt mit digitalen Elementen.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["sbom_creation", "sbom_dependency_coverage", "sbom_format_standard", "sbom_tooling_automation"],
"steps": [
"Komponenten und (direkte + transitive) Abhaengigkeiten inventarisieren",
"SBOM automatisiert in der Build-/Toolchain generieren",
"Komponenten, Versionen, Lizenzen und Lieferanten erfassen",
"in anerkanntem maschinenlesbarem Format (CycloneDX/SPDX) ausgeben",
"Format- und Schemavalidierung durchfuehren"
],
"controls": [
"SBOM-Datei vorhanden",
"Format ist maschinenlesbar und standardkonform (CycloneDX/SPDX)",
"direkte und transitive Abhaengigkeiten enthalten"
],
"evidence": ["sbom.cyclonedx.json", "Format-Validierungs-Log", "Build-/Toolchain-Konfiguration"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "sbom_update_process",
"name": "SBOM-Aktualisierungsprozess",
"description": "Halten der SBOM aktuell ueber den Produktlebenszyklus bei Komponenten-, Versions- und Patch-Aenderungen.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["sbom_maintenance_update"],
"steps": [
"Komponentenaenderung erkennen (Dependency-/Patch-/Versionsaenderung)",
"SBOM neu generieren",
"Lieferanten-SBOMs aktualisieren",
"neue SBOM-Version speichern",
"SBOM in Release-Artefakte uebernehmen"
],
"controls": [
"CI prueft SBOM vorhanden",
"SBOM-Version passt zum Release",
"Supplier-Komponenten enthalten"
],
"evidence": ["sbom.json", "CI-Log", "Release-Artefakt", "Supplier-SBOM"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "sbom_supplier_integration_process",
"name": "Lieferanten-SBOM-Integration",
"description": "Beschaffen und Einarbeiten von Lieferanten-/Drittkomponenten-SBOMs in die Produkt-SBOM.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["sbom_supply_chain_contracts", "sbom_dependency_coverage"],
"steps": [
"SBOM-Anforderung in Lieferantenvertraege aufnehmen",
"Lieferanten-SBOMs einsammeln",
"in die Produkt-SBOM mergen",
"Drittkomponenten und deren Abhaengigkeiten nachverfolgen"
],
"controls": [
"vertragliche SBOM-Klausel vorhanden",
"Lieferanten-SBOMs eingegangen",
"Drittkomponenten in der SBOM gelistet"
],
"evidence": ["Lieferantenvertrag-Klausel", "eingegangene Supplier-SBOMs", "gemergte SBOM"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "sbom_provision_process",
"name": "SBOM-Bereitstellungsprozess",
"description": "Zugaenglichmachen der SBOM fuer berechtigte Parteien (Nutzer, Behoerde) unter Wahrung der Vertraulichkeit.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["sbom_access_provision", "sbom_authority_provision", "sbom_confidentiality"],
"steps": [
"Zugangskanal definieren (Portal/API/dokumentierter Pfad)",
"Nutzer ueber den Zugangsweg informieren",
"auf begruendetes Verlangen der Marktueberwachungsbehoerde vertraulich bereitstellen",
"Zugriffskontrolle und Vertraulichkeitsmassnahmen anwenden"
],
"controls": [
"Zugangspfad dokumentiert",
"Zugriffskontrolle/Vertraulichkeit umgesetzt",
"Behoerden-Bereitstellungsprozess definiert"
],
"evidence": ["Zugangskanal-Dokumentation", "Behoerden-Anfrage-Log", "Zugriffskontroll-Konfiguration"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "sbom_conformity_documentation_process",
"name": "SBOM in technischer Dokumentation/Konformitaet",
"description": "Aufnehmen der SBOM in die technische Dokumentation und Verifizieren der Vollstaendigkeit fuer die Konformitaetsbewertung.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["sbom_technical_documentation", "sbom_completeness_verification"],
"steps": [
"SBOM in die technische Dokumentation aufnehmen",
"Vollstaendigkeit gegen die real eingesetzte Softwarekomposition pruefen",
"der Konformitaetsbewertung beilegen (ggf. EUCC)"
],
"controls": [
"SBOM Teil der technischen Dokumentation",
"Vollstaendigkeit verifiziert",
"Konformitaetsnachweis vorhanden"
],
"evidence": ["technische Dokumentation", "Vollstaendigkeits-Pruefbericht", "Konformitaetsnachweis"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_handling_process_setup",
"name": "Schwachstellenbehandlungsprozess einrichten",
"description": "Dokumentierten Prozess und Meldekanal (CVD) fuer die Schwachstellenbehandlung etablieren.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["vuln_handling_process"],
"steps": [
"dokumentierten Schwachstellenbehandlungsprozess definieren",
"Coordinated-Vulnerability-Disclosure-Richtlinie und Meldekanal veroeffentlichen",
"eingehende Meldungen triagieren"
],
"controls": [
"Behandlungsprozess dokumentiert",
"Meldekanal/Kontaktstelle auffindbar (z.B. security.txt)",
"Triage-Verfahren vorhanden"
],
"evidence": ["Prozessdokument", "security.txt / Kontaktstelle", "Triage-Log"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_identification_process",
"name": "Schwachstellen-Identifikation",
"description": "Bekannte Schwachstellen in eingesetzten Komponenten erkennen und inventarisieren.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["vuln_identification_inventory"],
"steps": [
"Advisories/CVE-Feeds beobachten",
"gegen die SBOM-Komponenten abgleichen",
"Schwachstellen-Inventar pflegen"
],
"controls": [
"Advisory-/CVE-Monitoring aktiv",
"SBOM-zu-CVE-Abgleich durchgefuehrt",
"Schwachstellen-Inventar gepflegt"
],
"evidence": ["CVE-Abgleich-Report", "Schwachstellen-Register"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_assessment_process",
"name": "Schwachstellen-Bewertung/Priorisierung",
"description": "Identifizierte Schwachstellen nach Schweregrad, Ausnutzbarkeit und Exposition bewerten und priorisieren.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["vuln_assessment_prioritization"],
"steps": [
"Schweregrad bewerten (z.B. CVSS)",
"Ausnutzbarkeit/Exposition einschaetzen",
"risikobasiert priorisieren"
],
"controls": [
"Schweregrad standardisiert bewertet",
"risikobasierte Priorisierung vorhanden"
],
"evidence": ["Bewertungsdatensatz (CVSS)", "Prioritaetenliste"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_remediation_process",
"name": "Schwachstellen-Behebung",
"description": "Bekannte Schwachstellen fristgerecht durch Patches/Gegenmassnahmen beheben und Sicherheitsupdates bereitstellen.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["vuln_remediation_patching"],
"steps": [
"Fix/Gegenmassnahme entwickeln",
"testen",
"Sicherheitsupdate kostenfrei und zeitnah bereitstellen",
"bis zum Abschluss nachverfolgen"
],
"controls": [
"zeitnahe Behebung",
"Sicherheitsupdate bereitgestellt",
"Follow-up bis Closure"
],
"evidence": ["Patch/Release", "Behebungs-Zeitleiste", "Follow-up-Log"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_disclosure_process",
"name": "Offenlegung + Nutzerinformation",
"description": "Koordinierte Offenlegung behobener Schwachstellen und Information der Nutzer ueber Schutzmassnahmen.",
"source_role": "procedural_requirement",
"fulfills_obligations": ["coordinated_vulnerability_disclosure", "vuln_info_dissemination_users"],
"steps": [
"Offenlegungszeitpunkt koordinieren",
"Security Advisory / CVE-Eintrag veroeffentlichen",
"Nutzer ueber behobene Schwachstelle und Schutzmassnahmen informieren"
],
"controls": [
"Advisory veroeffentlicht",
"Nutzer informiert"
],
"evidence": ["Security Advisory", "CVE-Eintrag", "Nutzer-Benachrichtigung"],
"citation_spans": [], "citation_status": "pending_span_anchor"
},
{
"procedure_id": "vuln_authority_reporting_process",
"name": "Behoerdenmeldung aktiv ausgenutzter Schwachstellen",
"description": "Aktiv ausgenutzte Schwachstellen fristgerecht an CSIRT/ENISA melden (CRA Art. 14-Kaskade).",
"source_role": "procedural_requirement",
"fulfills_obligations": ["exploited_vuln_reporting_authorities"],
"applicability_note": "bedingt: nur bei aktiv ausgenutzter Schwachstelle",
"steps": [
"aktive Ausnutzung erkennen",
"Fruehwarnung an CSIRT/ENISA (24h)",
"vollstaendige Meldung (72h)",
"Abschlussbericht (14 Tage)"
],
"controls": [
"24h-Fruehwarnung erfolgt",
"72h-Meldung erfolgt",
"14d-Abschlussbericht erfolgt"
],
"evidence": ["CSIRT/ENISA-Meldungsbelege", "Zeitstempel der Kaskade"],
"citation_spans": [], "citation_status": "pending_span_anchor"
}
]
}
+423
View File
@@ -0,0 +1,423 @@
{
"schema_version": "obligation_join_keys_v1",
"contract": "obligation_id ist der stabile Join-Key. Legal Knowledge Graph haengt citation_spans an obligation_id; Compliance Execution Graph mappt control_mapping.source_norm -> obligation_id. Interim-Bruecke = citation_units. obligation_id NIE neu vergeben (re-link).",
"count": 47,
"obligation_ids": [
{
"obligation_id": "sbom_creation",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_dependency_coverage",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Art. 3(36) i.V.m. Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_format_standard",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_maintenance_update",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_completeness_verification",
"regulation": "CRA",
"family": "sbom",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "sbom_tooling_automation",
"regulation": "CRA",
"family": "sbom",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "IMPLEMENTATION"
},
{
"obligation_id": "sbom_access_provision",
"regulation": "CRA",
"family": "sbom",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "sbom_authority_provision",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Art. 31 / Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_confidentiality",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Art. 31(4)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "sbom_supply_chain_contracts",
"regulation": "CRA",
"family": "sbom",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "sbom_technical_documentation",
"regulation": "CRA",
"family": "sbom",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Art. 31 i.V.m. Annex VII"
],
"source_role": "EVIDENCE"
},
{
"obligation_id": "vuln_identification_inventory",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "vuln_assessment_prioritization",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (1)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "vuln_remediation_patching",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (2) & (8)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "vuln_handling_process",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Article 13(8) & Annex VII"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "coordinated_vulnerability_disclosure",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (5)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "exploited_vuln_reporting_authorities",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Article 14 & Article 16"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "vuln_info_dissemination_users",
"regulation": "CRA",
"family": "vuln",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I Part II (4) & (6)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "user_authentication_required",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(d)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "authentication_policy_documented",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "auth_exceptions_documented",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "mfa_required",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "step_up_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "privileged_op_reauth",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "strong_crypto_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(e)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "credential_lifecycle_management",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "credential_confidentiality_protection",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(e)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "password_policy",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "no_default_credentials",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(a)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "account_lockout_failed_attempts",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "server_side_validation",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "session_binding_management",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "reauth_after_inactivity",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "token_validation_lifecycle",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "mutual_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "revocation_check",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "encrypted_auth_channel",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(e)"
],
"source_role": "LEGAL_BASIS"
},
{
"obligation_id": "tls_certificate_auth",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "service_to_service_auth",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "auth_key_management",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "biometric_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "federated_auth_assertions",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "separate_authn_authz",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "remote_access_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "supplier_access_auth",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "personal_admin_accounts",
"regulation": "CRA",
"family": "authentication",
"tier": "BEST_PRACTICE",
"citation_units": [],
"source_role": "GUIDANCE"
},
{
"obligation_id": "firmware_software_authentication",
"regulation": "CRA",
"family": "authentication",
"tier": "LEGAL_MINIMUM",
"citation_units": [
"Annex I (2)(c)"
],
"source_role": "LEGAL_BASIS"
}
]
}
@@ -0,0 +1,90 @@
"""P3 — Compliance-Advisor-Proof: obligation-basierte Antwort als vollstaendige
BEGRUENDUNGSKETTE aus der Registry (NICHT RAG-Text, KEIN LLM):
Rechtsgrundlage -> Obligation -> Procedure -> Controls -> Evidence -> Antwort.
Deterministisch + zitierfaehig. Der Unterschied zu RAG: RAG beantwortet — BreakPilot
begruendet UND operationalisiert.
python3 scripts/obligation_discovery/advisor_proof.py --registry obligations/cra.json \
--procedures obligations/cra_procedures.json --topic sbom --has-digital-elements
"""
from __future__ import annotations
import argparse
import json
def applies(obl: dict, has_digital: bool) -> tuple[bool, str]:
a = obl.get("applicability", "universal")
if a == "universal":
return True, ""
if a.startswith("domain:products_with_digital_elements"):
return has_digital, "nur fuer Produkte mit digitalen Elementen (CRA Art. 3)"
if a.startswith("domain:"):
return True, a.split(":", 1)[1]
if a.startswith("conditional:"):
return True, f"bedingt: {a.split(':',1)[1]}"
return True, ""
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("--registry", required=True)
ap.add_argument("--procedures", required=True)
ap.add_argument("--topic", default="sbom")
ap.add_argument("--has-digital-elements", action="store_true")
ap.add_argument("--question", default="Muss ich als Maschinenbauer eine SBOM bereitstellen?")
a = ap.parse_args()
reg = json.load(open(a.registry, encoding="utf-8"))
procs = json.load(open(a.procedures, encoding="utf-8"))["procedures"]
obls = [o for o in reg["obligations"]
if a.topic in o.get("family", "") or a.topic in o["id"]]
ids = {o["id"] for o in obls}
by_obl: dict[str, list] = {}
for p in procs:
for oid in p.get("fulfills_obligations", []):
by_obl.setdefault(oid, []).append(p)
pflicht = [o for o in obls if o["tier"] == "LEGAL_MINIMUM" and applies(o, a.has_digital_elements)[0]]
best = [o for o in obls if o["tier"] != "LEGAL_MINIMUM"]
print(f"FRAGE: {a.question}")
print(f"\nANTWORT: {'JA' if pflicht and a.has_digital_elements else 'NUR WENN CRA-anwendbar'}"
f"sofern das Produkt unter den CRA faellt (product with digital elements, Art. 3).")
print("\n══ BEGRUENDUNGSKETTE (Recht → Obligation → Procedure → Controls → Evidence) ══")
req_evidence: list[str] = []
for o in pflicht:
lb = "; ".join(f"{b.get('source','')} {b.get('anchor','')}".strip() for b in o.get("legal_basis", []))
print(f"\n● PFLICHT: {o['id']}{o.get('description','')[:80]}")
print(f" Rechtsgrundlage: {lb or ''}")
ps = by_obl.get(o["id"], [])
for p in ps:
print(f" Procedure (wie umgesetzt): {p['procedure_id']} — Schritte: {len(p.get('steps',[]))}")
print(f" Controls (Pruefung): {' · '.join(p.get('controls', []))[:96]}")
print(f" Nachweis: {' · '.join(p.get('evidence', []))}")
req_evidence += p.get("evidence", [])
if not ps:
print(" Procedure: (noch keine modelliert)")
print("\n── REQUIRED EVIDENCE (aggregiert, womit wird es nachgewiesen) ──")
print(" " + " · ".join(dict.fromkeys(req_evidence)) if req_evidence else "")
print("\n── BEST PRACTICE (anerkannte Umsetzung, KEINE CRA-Wortlautpflicht) ──")
for o in best:
gb = "; ".join(b.get("source", "") for b in o.get("guidance_basis", []))
print(f"{o['id']}{o.get('description','')[:64]} | Guidance: {gb or ''}")
print("\n── BEZIEHUNG (warum es zaehlt) ──")
for r in reg.get("relationships", []):
if r.get("from") in ids and r.get("to") not in ids:
print(f"{r['from']} --{r['type']}--> {r['to']}: {r.get('note','')[:64]}")
pend = sum(1 for o in pflicht if o.get("citation_status") == "pending_span_anchor")
print(f"\n── CITATION ──\n {pend}/{len(pflicht)} Pflichten: pending_span_anchor "
f"(Textstellen-Anker folgen mit dem zitierfaehigen Re-Ingest)")
print("\n(RAG beantwortet — BreakPilot begruendet UND operationalisiert.)")
if __name__ == "__main__":
main()
@@ -0,0 +1,52 @@
"""Exportiert den OBLIGATION_ID-Join-Key-Vertrag aus den Registry-Artefakten.
Die obligation_id ist der stabile Brueckenschluessel zwischen Legal Knowledge Graph
(citation_spans haengen an obligation_id) und Compliance Execution Graph
(control_mapping.source_norm -> obligation_id). citation_units = die legal_basis-Anker,
ueber die beide Seiten heute (vor obligation_id-Adoption) bruecken koennen.
DISZIPLIN: obligation_id wird RE-GELINKT, NIE neu vergeben (Pendant zu span_id/control_uuid).
python3 scripts/obligation_discovery/export_join_keys.py obligations/cra.json obligations/cra_authentication.json
"""
from __future__ import annotations
import argparse
import json
def main() -> None:
ap = argparse.ArgumentParser()
ap.add_argument("registries", nargs="+")
ap.add_argument("--out", default="obligations/obligation_join_keys.json")
a = ap.parse_args()
keys = []
for path in a.registries:
reg = json.load(open(path, encoding="utf-8"))
for o in reg.get("obligations", []):
citation_units = [b.get("anchor", "") for b in o.get("legal_basis", []) if b.get("anchor")]
keys.append({
"obligation_id": o["id"],
"regulation": reg.get("regulation", ""),
"family": o.get("family", ""),
"tier": o.get("tier", ""),
"citation_units": citation_units,
"source_role": o.get("source_role", ""),
})
out = {
"schema_version": "obligation_join_keys_v1",
"contract": "obligation_id ist der stabile Join-Key. Legal Knowledge Graph haengt "
"citation_spans an obligation_id; Compliance Execution Graph mappt "
"control_mapping.source_norm -> obligation_id. Interim-Bruecke = citation_units. "
"obligation_id NIE neu vergeben (re-link).",
"count": len(keys),
"obligation_ids": keys,
}
json.dump(out, open(a.out, "w", encoding="utf-8"), ensure_ascii=False, indent=1)
from collections import Counter
print(f"exportiert: {a.out} ({len(keys)} obligation_ids)")
print("Regulierungen:", dict(Counter(k["regulation"] for k in keys)))
print("Familien:", dict(Counter(k["family"] for k in keys)))
if __name__ == "__main__":
main()