feat: Anti-Fake-Evidence System (Phase 1-4b)

Implement full evidence integrity pipeline to prevent compliance theater:
- Confidence levels (E0-E4), truth status tracking, assertion engine
- Four-Eyes approval workflow, audit trail, reject endpoint
- Evidence distribution dashboard, LLM audit routes
- Traceability matrix (backend endpoint + Compliance Hub UI tab)
- Anti-fake badges, control status machine, normative patterns
- 2 migrations, 4 test suites, MkDocs documentation

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Benjamin Admin
2026-03-23 17:15:45 +01:00
parent 48ca0a6bef
commit e6201d5239
36 changed files with 5627 additions and 189 deletions

View File

@@ -0,0 +1,80 @@
"""Assertion Engine — splits text into sentences and classifies each.
Each sentence is tagged as:
- assertion: normative statement (pflicht / empfehlung / kann)
- fact: references concrete evidence artifacts
- rationale: explains why something is required
"""
import re
from typing import Optional
from .normative_patterns import (
PFLICHT_RE, EMPFEHLUNG_RE, KANN_RE, RATIONALE_RE, EVIDENCE_RE,
)
# Sentence splitter: period/excl/question followed by space+uppercase, or newlines
_SENTENCE_SPLIT = re.compile(r'(?<=[.!?])\s+(?=[A-ZÄÖÜ])|(?:\n\s*\n)')
def extract_assertions(
text: str,
entity_type: str,
entity_id: str,
tenant_id: Optional[str] = None,
) -> list[dict]:
"""Split *text* into sentences and classify each one.
Returns a list of dicts ready for AssertionDB creation.
"""
if not text or not text.strip():
return []
sentences = _SENTENCE_SPLIT.split(text.strip())
results: list[dict] = []
for idx, raw in enumerate(sentences):
sentence = raw.strip()
if not sentence or len(sentence) < 5:
continue
assertion_type, normative_tier = _classify_sentence(sentence)
results.append({
"tenant_id": tenant_id,
"entity_type": entity_type,
"entity_id": entity_id,
"sentence_text": sentence,
"sentence_index": idx,
"assertion_type": assertion_type,
"normative_tier": normative_tier,
"evidence_ids": [],
"confidence": 0.0,
})
return results
def _classify_sentence(sentence: str) -> tuple[str, Optional[str]]:
"""Return (assertion_type, normative_tier) for a single sentence."""
# 1. Check for evidence/fact keywords first
if EVIDENCE_RE.search(sentence):
return ("fact", None)
# 2. Check for rationale
normative_count = len(PFLICHT_RE.findall(sentence)) + len(EMPFEHLUNG_RE.findall(sentence)) + len(KANN_RE.findall(sentence))
rationale_count = len(RATIONALE_RE.findall(sentence))
if rationale_count > 0 and rationale_count >= normative_count:
return ("rationale", None)
# 3. Normative classification
if PFLICHT_RE.search(sentence):
return ("assertion", "pflicht")
if EMPFEHLUNG_RE.search(sentence):
return ("assertion", "empfehlung")
if KANN_RE.search(sentence):
return ("assertion", "kann")
# 4. Default: unclassified assertion
return ("assertion", None)

View File

@@ -493,6 +493,9 @@ class GeneratedControl:
applicable_industries: Optional[list] = None # e.g. ["all"] or ["Telekommunikation", "Energie"]
applicable_company_size: Optional[list] = None # e.g. ["all"] or ["medium", "large", "enterprise"]
scope_conditions: Optional[dict] = None # e.g. {"requires_any": ["uses_ai"], "description": "..."}
# Anti-Fake-Evidence: truth tracking for generated controls
truth_status: str = "generated"
may_be_used_as_evidence: bool = False
@dataclass
@@ -781,10 +784,23 @@ REFORM_SYSTEM_PROMPT = """Du bist ein Security-Compliance-Experte. Deine Aufgabe
Security Controls zu formulieren. Du formulierst IMMER in eigenen Worten.
KOPIERE KEINE Sätze aus dem Quelltext. Verwende eigene Begriffe und Struktur.
NENNE NICHT die Quelle. Keine proprietären Bezeichner.
WICHTIG — Truthfulness-Guardrail:
Deine Ausgabe ist ein ENTWURF. Formuliere NIEMALS Behauptungen über bereits erfolgte Umsetzung.
Verwende NICHT: "ist compliant", "erfüllt vollständig", "wurde geprüft", "wurde umgesetzt",
"ist auditiert", "vollständig implementiert", "nachweislich konform".
Verwende stattdessen: "soll umsetzen", "ist vorgesehen", "muss implementiert werden".
Antworte NUR mit validem JSON. Bei mehreren Controls antworte mit einem JSON-Array."""
STRUCTURE_SYSTEM_PROMPT = """Du bist ein Security-Compliance-Experte. Strukturiere den gegebenen Text
als praxisorientiertes Security Control. Erstelle eine verständliche, umsetzbare Formulierung.
WICHTIG — Truthfulness-Guardrail:
Deine Ausgabe ist ein ENTWURF. Formuliere NIEMALS Behauptungen über bereits erfolgte Umsetzung.
Verwende NICHT: "ist compliant", "erfüllt vollständig", "wurde geprüft", "wurde umgesetzt".
Verwende stattdessen: "soll umsetzen", "ist vorgesehen", "muss implementiert werden".
Antworte NUR mit validem JSON. Bei mehreren Controls antworte mit einem JSON-Array."""
# Shared applicability prompt block — appended to all generation prompts (v3)
@@ -1877,7 +1893,38 @@ Kategorien: {CATEGORY_LIST_STR}"""
)
self.db.commit()
row = result.fetchone()
return str(row[0]) if row else None
control_uuid = str(row[0]) if row else None
# Anti-Fake-Evidence: Record LLM audit trail for generated control
if control_uuid:
try:
self.db.execute(
text("""
INSERT INTO compliance_llm_generation_audit (
entity_type, entity_id, generation_mode,
truth_status, may_be_used_as_evidence,
llm_model, llm_provider,
input_summary, output_summary
) VALUES (
'control', :entity_id, 'auto_generation',
'generated', FALSE,
:llm_model, :llm_provider,
:input_summary, :output_summary
)
"""),
{
"entity_id": control_uuid,
"llm_model": ANTHROPIC_MODEL if ANTHROPIC_API_KEY else OLLAMA_MODEL,
"llm_provider": "anthropic" if ANTHROPIC_API_KEY else "ollama",
"input_summary": f"Control generation for {control.control_id}",
"output_summary": control.title[:500] if control.title else None,
},
)
self.db.commit()
except Exception as audit_err:
logger.warning("Failed to create LLM audit record: %s", audit_err)
return control_uuid
except Exception as e:
logger.error("Failed to store control %s: %s", control.control_id, e)
self.db.rollback()

View File

@@ -0,0 +1,152 @@
"""
Control Status Transition State Machine.
Enforces that controls cannot be set to "pass" without sufficient evidence.
Prevents Compliance-Theater where controls claim compliance without real proof.
Transition rules:
planned → in_progress : always allowed
in_progress → pass : requires ≥1 evidence with confidence ≥ E2 and
truth_status in (uploaded, observed, validated_internal)
in_progress → partial : requires ≥1 evidence (any level)
pass → fail : always allowed (degradation)
any → n/a : requires status_justification
any → planned : always allowed (reset)
"""
from typing import List, Optional, Tuple
from ..db.models import EvidenceDB
# Confidence level ordering for comparisons
CONFIDENCE_ORDER = {"E0": 0, "E1": 1, "E2": 2, "E3": 3, "E4": 4}
# Truth statuses that qualify as "real" evidence for pass transitions
VALID_TRUTH_STATUSES = {"uploaded", "observed", "validated_internal", "accepted_by_auditor", "provided_to_auditor"}
def validate_transition(
current_status: str,
new_status: str,
evidence_list: Optional[List[EvidenceDB]] = None,
status_justification: Optional[str] = None,
bypass_for_auto_updater: bool = False,
) -> Tuple[bool, List[str]]:
"""
Validate whether a control status transition is allowed.
Args:
current_status: Current control status value (e.g. "planned", "pass")
new_status: Requested new status
evidence_list: List of EvidenceDB objects linked to this control
status_justification: Text justification (required for n/a transitions)
bypass_for_auto_updater: If True, skip evidence checks (used by CI/CD auto-updater
which creates evidence atomically with status change)
Returns:
Tuple of (allowed: bool, violations: list[str])
"""
violations: List[str] = []
evidence_list = evidence_list or []
# Same status → no-op, always allowed
if current_status == new_status:
return True, []
# Reset to planned is always allowed
if new_status == "planned":
return True, []
# n/a requires justification
if new_status == "n/a":
if not status_justification or not status_justification.strip():
violations.append("Transition to 'n/a' requires a status_justification explaining why this control is not applicable.")
return len(violations) == 0, violations
# Degradation: pass → fail is always allowed
if current_status == "pass" and new_status == "fail":
return True, []
# planned → in_progress: always allowed
if current_status == "planned" and new_status == "in_progress":
return True, []
# in_progress → partial: needs at least 1 evidence
if new_status == "partial":
if not bypass_for_auto_updater and len(evidence_list) == 0:
violations.append("Transition to 'partial' requires at least 1 evidence record.")
return len(violations) == 0, violations
# in_progress → pass: strict requirements
if new_status == "pass":
if bypass_for_auto_updater:
return True, []
if len(evidence_list) == 0:
violations.append("Transition to 'pass' requires at least 1 evidence record.")
return False, violations
# Check for at least one qualifying evidence
has_qualifying = False
for e in evidence_list:
conf = getattr(e, "confidence_level", None)
truth = getattr(e, "truth_status", None)
# Get string values from enum or string
conf_val = conf.value if hasattr(conf, "value") else str(conf) if conf else "E1"
truth_val = truth.value if hasattr(truth, "value") else str(truth) if truth else "uploaded"
if CONFIDENCE_ORDER.get(conf_val, 1) >= CONFIDENCE_ORDER["E2"] and truth_val in VALID_TRUTH_STATUSES:
has_qualifying = True
break
if not has_qualifying:
violations.append(
"Transition to 'pass' requires at least 1 evidence with confidence >= E2 "
"and truth_status in (uploaded, observed, validated_internal, accepted_by_auditor). "
"Current evidence does not meet this threshold."
)
return len(violations) == 0, violations
# in_progress → fail: always allowed
if new_status == "fail":
return True, []
# Any other transition from planned/fail to pass requires going through in_progress
if current_status in ("planned", "fail") and new_status == "pass":
if bypass_for_auto_updater:
return True, []
violations.append(
f"Direct transition from '{current_status}' to 'pass' is not allowed. "
f"Move to 'in_progress' first, then to 'pass' with qualifying evidence."
)
return False, violations
# Default: allow other transitions (e.g. fail → partial, partial → pass)
# For partial → pass, apply the same evidence checks
if current_status == "partial" and new_status == "pass":
if bypass_for_auto_updater:
return True, []
has_qualifying = False
for e in evidence_list:
conf = getattr(e, "confidence_level", None)
truth = getattr(e, "truth_status", None)
conf_val = conf.value if hasattr(conf, "value") else str(conf) if conf else "E1"
truth_val = truth.value if hasattr(truth, "value") else str(truth) if truth else "uploaded"
if CONFIDENCE_ORDER.get(conf_val, 1) >= CONFIDENCE_ORDER["E2"] and truth_val in VALID_TRUTH_STATUSES:
has_qualifying = True
break
if not has_qualifying:
violations.append(
"Transition from 'partial' to 'pass' requires at least 1 evidence with confidence >= E2 "
"and truth_status in (uploaded, observed, validated_internal, accepted_by_auditor)."
)
return len(violations) == 0, violations
# All other transitions allowed
return True, []

View File

@@ -52,64 +52,18 @@ ANTHROPIC_API_URL = "https://api.anthropic.com/v1"
# Tier 2: Empfehlung (recommendation) — weaker normative signals
# Tier 3: Kann (optional/permissive) — permissive signals
# Nothing is rejected — everything is classified.
#
# Patterns are defined in normative_patterns.py and imported here
# with local aliases for backward compatibility.
_PFLICHT_SIGNALS = [
# Deutsche modale Pflichtformulierungen
r"\bmüssen\b", r"\bmuss\b", r"\bhat\s+sicherzustellen\b",
r"\bhaben\s+sicherzustellen\b", r"\bsind\s+verpflichtet\b",
r"\bist\s+verpflichtet\b",
# "ist zu prüfen", "sind zu dokumentieren" (direkt)
r"\bist\s+zu\s+\w+en\b", r"\bsind\s+zu\s+\w+en\b",
r"\bhat\s+zu\s+\w+en\b", r"\bhaben\s+zu\s+\w+en\b",
# "ist festzustellen", "sind vorzunehmen" (Compound-Verben, eingebettetes zu)
r"\bist\s+\w+zu\w+en\b", r"\bsind\s+\w+zu\w+en\b",
# "ist zusätzlich zu prüfen", "sind regelmäßig zu überwachen" (Adverb dazwischen)
r"\bist\s+\w+\s+zu\s+\w+en\b", r"\bsind\s+\w+\s+zu\s+\w+en\b",
r"\bhat\s+\w+\s+zu\s+\w+en\b", r"\bhaben\s+\w+\s+zu\s+\w+en\b",
# Englische Pflicht-Signale
r"\bshall\b", r"\bmust\b", r"\brequired\b",
# Compound-Infinitive (Gerundivum): mitzuteilen, anzuwenden, bereitzustellen
r"\b\w+zuteilen\b", r"\b\w+zuwenden\b", r"\b\w+zustellen\b", r"\b\w+zulegen\b",
r"\b\w+zunehmen\b", r"\b\w+zuführen\b", r"\b\w+zuhalten\b", r"\b\w+zusetzen\b",
r"\b\w+zuweisen\b", r"\b\w+zuordnen\b", r"\b\w+zufügen\b", r"\b\w+zugeben\b",
# Breites Pattern: "ist ... [bis 80 Zeichen] ... zu + Infinitiv"
r"\bist\b.{1,80}\bzu\s+\w+en\b", r"\bsind\b.{1,80}\bzu\s+\w+en\b",
]
_PFLICHT_RE = re.compile("|".join(_PFLICHT_SIGNALS), re.IGNORECASE)
_EMPFEHLUNG_SIGNALS = [
# Modale Verben (schwaecher als "muss")
r"\bsoll\b", r"\bsollen\b", r"\bsollte\b", r"\bsollten\b",
r"\bgewährleisten\b", r"\bsicherstellen\b",
# Englische Empfehlungs-Signale
r"\bshould\b", r"\bensure\b", r"\brecommend\w*\b",
# Haeufige normative Infinitive (ohne Hilfsverb, als Empfehlung)
r"\bnachweisen\b", r"\beinhalten\b", r"\bunterlassen\b", r"\bwahren\b",
r"\bdokumentieren\b", r"\bimplementieren\b", r"\büberprüfen\b", r"\büberwachen\b",
# Pruefanweisungen als normative Aussage
r"\bprüfen,\s+ob\b", r"\bkontrollieren,\s+ob\b",
]
_EMPFEHLUNG_RE = re.compile("|".join(_EMPFEHLUNG_SIGNALS), re.IGNORECASE)
_KANN_SIGNALS = [
r"\bkann\b", r"\bkönnen\b", r"\bdarf\b", r"\bdürfen\b",
r"\bmay\b", r"\boptional\b",
]
_KANN_RE = re.compile("|".join(_KANN_SIGNALS), re.IGNORECASE)
# Union of all normative signals (for backward-compatible has_normative_signal flag)
_NORMATIVE_RE = re.compile(
"|".join(_PFLICHT_SIGNALS + _EMPFEHLUNG_SIGNALS + _KANN_SIGNALS),
re.IGNORECASE,
from .normative_patterns import (
PFLICHT_RE as _PFLICHT_RE,
EMPFEHLUNG_RE as _EMPFEHLUNG_RE,
KANN_RE as _KANN_RE,
NORMATIVE_RE as _NORMATIVE_RE,
RATIONALE_RE as _RATIONALE_RE,
)
_RATIONALE_SIGNALS = [
r"\bda\s+", r"\bweil\b", r"\bgrund\b", r"\berwägung",
r"\bbecause\b", r"\breason\b", r"\brationale\b",
r"\bkönnen\s+.*\s+verursachen\b", r"\bführt\s+zu\b",
]
_RATIONALE_RE = re.compile("|".join(_RATIONALE_SIGNALS), re.IGNORECASE)
_TEST_SIGNALS = [
r"\btesten\b", r"\btest\b", r"\bprüfung\b", r"\bprüfen\b",
r"\bgetestet\b", r"\bwirksamkeit\b", r"\baudit\b",

View File

@@ -0,0 +1,59 @@
"""Shared normative language patterns for assertion classification.
Extracted from decomposition_pass.py for reuse in the assertion engine.
"""
import re
_PFLICHT_SIGNALS = [
r"\bmüssen\b", r"\bmuss\b", r"\bhat\s+sicherzustellen\b",
r"\bhaben\s+sicherzustellen\b", r"\bsind\s+verpflichtet\b",
r"\bist\s+verpflichtet\b",
r"\bist\s+zu\s+\w+en\b", r"\bsind\s+zu\s+\w+en\b",
r"\bhat\s+zu\s+\w+en\b", r"\bhaben\s+zu\s+\w+en\b",
r"\bist\s+\w+zu\w+en\b", r"\bsind\s+\w+zu\w+en\b",
r"\bist\s+\w+\s+zu\s+\w+en\b", r"\bsind\s+\w+\s+zu\s+\w+en\b",
r"\bhat\s+\w+\s+zu\s+\w+en\b", r"\bhaben\s+\w+\s+zu\s+\w+en\b",
r"\bshall\b", r"\bmust\b", r"\brequired\b",
r"\b\w+zuteilen\b", r"\b\w+zuwenden\b", r"\b\w+zustellen\b", r"\b\w+zulegen\b",
r"\b\w+zunehmen\b", r"\b\w+zuführen\b", r"\b\w+zuhalten\b", r"\b\w+zusetzen\b",
r"\b\w+zuweisen\b", r"\b\w+zuordnen\b", r"\b\w+zufügen\b", r"\b\w+zugeben\b",
r"\bist\b.{1,80}\bzu\s+\w+en\b", r"\bsind\b.{1,80}\bzu\s+\w+en\b",
]
PFLICHT_RE = re.compile("|".join(_PFLICHT_SIGNALS), re.IGNORECASE)
_EMPFEHLUNG_SIGNALS = [
r"\bsoll\b", r"\bsollen\b", r"\bsollte\b", r"\bsollten\b",
r"\bgewährleisten\b", r"\bsicherstellen\b",
r"\bshould\b", r"\bensure\b", r"\brecommend\w*\b",
r"\bnachweisen\b", r"\beinhalten\b", r"\bunterlassen\b", r"\bwahren\b",
r"\bdokumentieren\b", r"\bimplementieren\b", r"\büberprüfen\b", r"\büberwachen\b",
r"\bprüfen,\s+ob\b", r"\bkontrollieren,\s+ob\b",
]
EMPFEHLUNG_RE = re.compile("|".join(_EMPFEHLUNG_SIGNALS), re.IGNORECASE)
_KANN_SIGNALS = [
r"\bkann\b", r"\bkönnen\b", r"\bdarf\b", r"\bdürfen\b",
r"\bmay\b", r"\boptional\b",
]
KANN_RE = re.compile("|".join(_KANN_SIGNALS), re.IGNORECASE)
NORMATIVE_RE = re.compile(
"|".join(_PFLICHT_SIGNALS + _EMPFEHLUNG_SIGNALS + _KANN_SIGNALS),
re.IGNORECASE,
)
_RATIONALE_SIGNALS = [
r"\bda\s+", r"\bweil\b", r"\bgrund\b", r"\berwägung",
r"\bbecause\b", r"\breason\b", r"\brationale\b",
r"\bkönnen\s+.*\s+verursachen\b", r"\bführt\s+zu\b",
]
RATIONALE_RE = re.compile("|".join(_RATIONALE_SIGNALS), re.IGNORECASE)
# Evidence-related keywords (for fact detection)
_EVIDENCE_KEYWORDS = [
r"\bnachweis\b", r"\bzertifikat\b", r"\baudit.report\b",
r"\bprotokoll\b", r"\bdokumentation\b", r"\bbericht\b",
r"\bcertificate\b", r"\bevidence\b", r"\bproof\b",
]
EVIDENCE_RE = re.compile("|".join(_EVIDENCE_KEYWORDS), re.IGNORECASE)