Files
breakpilot-core/document-templates/generators/dsfa_template.py
Benjamin Admin fc71117bf2 feat: Document Templates V2 — DSFA, TOM, VVT, AVV, Verpflichtung, Art.13/14
Erweiterte Compliance-Vorlagen fuer den Document Generator:
- DSFA V2: Schwellwertanalyse (9 WP248-Kriterien), SDM-basierte TOM,
  strukturierte Risikobewertung, KI-Modul (AI Act), Art.36-Pruefung
- TOM V2: 7 SDM-Gewaehrleistungsziele, Sektor-Erweiterungen,
  NIS2/ISO27001/AI Act Varianten
- VVT V2: 6 Branchen-Muster (IT/SaaS, Gesundheit, Handel, Handwerk,
  Bildung, Beratung) + allgemeine Art.30-Vorlage
- AVV V2: Vollstaendiger Art.28-Vertrag mit TOM-Anlage
- Verpflichtungserklaerung: Mitarbeiter-Vertraulichkeit
- Art.13/14 Informationspflichten-Muster

Enthalt SQL-Migrations (compliance_legal_templates), Python-Generatoren
und Qdrant-Cleanup-Skript. Feature-Branch fuer spaetere Integration
in breakpilot-compliance.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>
2026-04-10 11:39:39 +02:00

456 lines
20 KiB
Python

"""DSFA template generator V2 — creates DSFA skeleton from company profile.
Enhanced with:
- Schwellwertanalyse (9 WP248 criteria)
- Bundesland-specific Muss-Listen references
- SDM-based TOM structure (7 Gewaehrleistungsziele)
- Structured risk assessment (ISO 29134 methodology)
- AI Act module (Section 8)
- Art. 36 consultation assessment
"""
from typing import Optional
# -- WP248 Kriterien --------------------------------------------------------
WP248_CRITERIA = [
{"id": "K1", "label": "Bewertung oder Scoring (einschl. Profiling und Prognose)",
"ctx_keys": ["has_profiling", "has_scoring"]},
{"id": "K2", "label": "Automatisierte Entscheidungsfindung mit Rechtswirkung",
"ctx_keys": ["has_automated_decisions"]},
{"id": "K3", "label": "Systematische Ueberwachung von Personen",
"ctx_keys": ["has_surveillance", "has_employee_monitoring", "has_video_surveillance"]},
{"id": "K4", "label": "Verarbeitung sensibler Daten (Art. 9/10 DS-GVO)",
"ctx_keys": ["processes_health_data", "processes_biometric_data", "processes_criminal_data"]},
{"id": "K5", "label": "Datenverarbeitung in grossem Umfang",
"ctx_keys": ["large_scale_processing"]},
{"id": "K6", "label": "Verknuepfung oder Zusammenfuehrung von Datenbestaenden",
"ctx_keys": ["data_matching", "data_combining"]},
{"id": "K7", "label": "Daten zu schutzbeduerftigen Betroffenen",
"ctx_keys": ["processes_minors_data", "processes_employee_data", "processes_patient_data"]},
{"id": "K8", "label": "Innovative Nutzung neuer technologischer Loesungen",
"ctx_keys": ["uses_ai", "uses_biometrics", "uses_iot"]},
{"id": "K9", "label": "Verarbeitung hindert Betroffene an Rechtsausuebung",
"ctx_keys": ["blocks_service_access", "blocks_contract"]},
]
# -- Bundesland -> Aufsichtsbehoerde Mapping --------------------------------
BUNDESLAND_AUFSICHT = {
"baden-wuerttemberg": ("LfDI Baden-Wuerttemberg", "DSK Muss-Liste + BW-spezifische Liste (Art. 35 Abs. 4)"),
"bayern": ("BayLDA (nicht-oeffentlicher Bereich)", "BayLDA Muss-Liste (17.10.2018) + Fallbeispiel ISO 29134"),
"berlin": ("BlnBDI", "BlnBDI Muss-Liste nicht-oeffentlich / oeffentlich"),
"brandenburg": ("LDA Brandenburg", "LDA BB Muss-Liste allgemein / oeffentlich"),
"bremen": ("LfDI Bremen", "LfDI HB Muss-Liste"),
"hamburg": ("HmbBfDI", "HmbBfDI Muss-Liste nicht-oeffentlich / oeffentlich"),
"hessen": ("HBDI", "DSK Muss-Liste (HBDI uebernimmt DSK-Liste)"),
"mecklenburg-vorpommern": ("LfDI M-V", "LfDI M-V Muss-Liste"),
"niedersachsen": ("LfD Niedersachsen", "LfD NI Muss-Liste + Pruefschema"),
"nordrhein-westfalen": ("LDI NRW", "LDI NRW Muss-Liste nicht-oeffentlich / oeffentlich"),
"rheinland-pfalz": ("LfDI RLP", "LfDI RLP Muss-Liste allgemein / oeffentlich"),
"saarland": ("UDS Saarland", "DSK Muss-Liste (UDS uebernimmt DSK-Liste)"),
"sachsen": ("SDB Sachsen", "SDB Sachsen Muss-Liste"),
"sachsen-anhalt": ("LfD Sachsen-Anhalt", "LfD SA Muss-Liste allgemein / oeffentlich"),
"schleswig-holstein": ("ULD Schleswig-Holstein", "ULD Muss-Liste + Planspiel-DSFA"),
"thueringen": ("TLfDI", "TLfDI Muss-Liste (04.07.2018)"),
"bund": ("BfDI", "BfDI Muss-Liste / DSFA-Hinweise"),
}
# -- SDM Gewaehrleistungsziele -----------------------------------------------
SDM_GOALS = [
{
"id": "verfuegbarkeit",
"label": "Verfuegbarkeit",
"description": "Personenbezogene Daten stehen zeitgerecht zur Verfuegung und koennen ordnungsgemaess verarbeitet werden.",
"default_measures": [
"Redundante Datenhaltung und regelmaessige Backups",
"Disaster-Recovery-Plan mit definierten RTO/RPO-Werten",
"USV und Notstromversorgung fuer kritische Systeme",
],
},
{
"id": "integritaet",
"label": "Integritaet",
"description": "Personenbezogene Daten bleiben waehrend der Verarbeitung unversehrt, vollstaendig und aktuell.",
"default_measures": [
"Pruefsummen und digitale Signaturen fuer Datenuebertragungen",
"Eingabevalidierung und Plausibilitaetspruefungen",
"Versionierung und Change-Management-Verfahren",
],
},
{
"id": "vertraulichkeit",
"label": "Vertraulichkeit",
"description": "Nur befugte Personen koennen personenbezogene Daten zur Kenntnis nehmen.",
"default_measures": [
"Verschluesselung: TLS 1.3 im Transit, AES-256 at Rest",
"Rollenbasiertes Zugriffskonzept (RBAC) mit Least-Privilege-Prinzip",
"Multi-Faktor-Authentifizierung fuer administrative Zugaenge",
],
},
{
"id": "nichtverkettung",
"label": "Nichtverkettung",
"description": "Personenbezogene Daten werden nur fuer den Zweck verarbeitet, zu dem sie erhoben wurden.",
"default_measures": [
"Technische Zweckbindung durch Mandantentrennung",
"Pseudonymisierung wo fachlich moeglich",
"Getrennte Datenbanken / Schemata je Verarbeitungszweck",
],
},
{
"id": "transparenz",
"label": "Transparenz",
"description": "Betroffene, der Verantwortliche und die Aufsichtsbehoerde koennen die Verarbeitung nachvollziehen.",
"default_measures": [
"Vollstaendiges Audit-Log aller Datenzugriffe und -aenderungen",
"Verzeichnis der Verarbeitungstaetigkeiten (Art. 30 DS-GVO)",
"Informationspflichten gemaess Art. 13/14 DS-GVO umgesetzt",
],
},
{
"id": "intervenierbarkeit",
"label": "Intervenierbarkeit",
"description": "Betroffenenrechte (Auskunft, Berichtigung, Loeschung, Widerspruch) koennen wirksam ausgeuebt werden.",
"default_measures": [
"Self-Service-Portal oder dokumentierter Prozess fuer Betroffenenanfragen",
"Technische Loeschfaehigkeit mit Nachweis (Loeschprotokoll)",
"Datenexport in maschinenlesbarem Format (Art. 20 DS-GVO)",
],
},
{
"id": "datenminimierung",
"label": "Datenminimierung",
"description": "Die Verarbeitung beschraenkt sich auf das erforderliche Mass.",
"default_measures": [
"Regelmaessige Pruefung der Erforderlichkeit erhobener Datenfelder",
"Automatisierte Loeschung nach Ablauf der Aufbewahrungsfrist",
"Anonymisierung / Aggregation fuer statistische Zwecke",
],
},
]
def generate_dsfa_draft(ctx: dict) -> dict:
"""Generate a DSFA draft document from template context.
Args:
ctx: Flat dict from company-profile/template-context endpoint.
Returns:
Dict with DSFA fields ready for creation via POST /dsfa.
"""
company = ctx.get("company_name", "Unbekannt")
dpo = ctx.get("dpo_name", "")
dpo_email = ctx.get("dpo_email", "")
federal_state = ctx.get("federal_state", "").lower().replace(" ", "-")
# --- Section 0: Schwellwertanalyse ---
schwellwert = _generate_schwellwertanalyse(ctx)
# --- Section 1: Verarbeitungsbeschreibung ---
section_1 = _generate_section_1(ctx, company, dpo, dpo_email)
# --- Section 2: Notwendigkeit ---
section_2 = _generate_section_2(ctx)
# --- Section 3: Risikobewertung ---
section_3 = _generate_risk_assessment(ctx)
# --- Section 4: Stakeholder-Konsultation ---
section_4 = _generate_section_4(ctx)
# --- Section 5: TOM nach SDM ---
section_5 = _generate_sdm_tom_section(ctx)
# --- Section 6: DSB-Stellungnahme ---
section_6 = _generate_section_6(ctx, dpo)
# --- Section 7: Ergebnis ---
section_7 = _generate_section_7(ctx)
# --- Section 8: KI-Modul ---
ai_systems = ctx.get("ai_systems", [])
involves_ai = len(ai_systems) > 0
section_8 = _generate_ai_module(ctx) if involves_ai else None
sections = {
"section_0": {"title": "Schwellwertanalyse", "content": schwellwert["content"]},
"section_1": {"title": "Allgemeine Informationen und Verarbeitungsbeschreibung", "content": section_1},
"section_2": {"title": "Notwendigkeit und Verhaeltnismaessigkeit", "content": section_2},
"section_3": {"title": "Risikobewertung", "content": section_3},
"section_4": {"title": "Konsultation der Betroffenen", "content": section_4},
"section_5": {"title": "Technische und organisatorische Massnahmen (SDM)", "content": section_5},
"section_6": {"title": "Stellungnahme des DSB", "content": section_6},
"section_7": {"title": "Ergebnis und Ueberprufungsplan", "content": section_7},
}
if section_8:
sections["section_8"] = {"title": "KI-spezifisches Modul (EU AI Act)", "content": section_8}
# Assess Art. 36 consultation requirement
art36_required = _assess_art36_consultation(ctx, schwellwert)
return {
"title": f"DSFA — {company}",
"description": f"Datenschutz-Folgenabschaetzung fuer {company}",
"status": "draft",
"risk_level": "high" if involves_ai or schwellwert["criteria_met"] >= 3 else "medium",
"involves_ai": involves_ai,
"dpo_name": dpo,
"federal_state": ctx.get("federal_state", ""),
"sections": sections,
"wp248_criteria_met": schwellwert["criteria_details"],
"art35_abs3_triggered": schwellwert["art35_abs3"],
"threshold_analysis": {
"criteria_met_count": schwellwert["criteria_met"],
"dsfa_required": schwellwert["dsfa_required"],
"muss_liste_ref": schwellwert.get("muss_liste_ref", ""),
},
"consultation_requirement": {
"art36_required": art36_required,
"reason": "Restrisiko bleibt nach Massnahmen hoch" if art36_required else "Restrisiko akzeptabel",
},
"processing_systems": [s.get("name", "") for s in ctx.get("processing_systems", [])],
"ai_systems_summary": [
{"name": s.get("name"), "risk": s.get("risk_category", "unknown")}
for s in ai_systems
],
}
# -- Internal helpers --------------------------------------------------------
def _generate_schwellwertanalyse(ctx: dict) -> dict:
"""Evaluate 9 WP248 criteria against company profile."""
criteria_details = []
criteria_met = 0
for criterion in WP248_CRITERIA:
met = any(ctx.get(key) for key in criterion["ctx_keys"])
criteria_details.append({
"id": criterion["id"],
"label": criterion["label"],
"met": met,
})
if met:
criteria_met += 1
# Art. 35 Abs. 3 specific triggers
art35_abs3 = []
if ctx.get("has_profiling") and ctx.get("has_automated_decisions"):
art35_abs3.append("Art. 35 Abs. 3 lit. a: Profiling mit Rechtswirkung")
if any(ctx.get(k) for k in ["processes_health_data", "processes_biometric_data", "processes_criminal_data"]):
if ctx.get("large_scale_processing"):
art35_abs3.append("Art. 35 Abs. 3 lit. b: Umfangreiche Verarbeitung besonderer Kategorien")
if ctx.get("has_surveillance"):
art35_abs3.append("Art. 35 Abs. 3 lit. c: Systematische Ueberwachung oeffentlicher Bereiche")
dsfa_required = criteria_met >= 2 or len(art35_abs3) > 0
# Bundesland reference
federal_state = ctx.get("federal_state", "").lower().replace(" ", "-")
aufsicht_info = BUNDESLAND_AUFSICHT.get(federal_state, ("Nicht zugeordnet", "DSK Muss-Liste (allgemein)"))
met_labels = [c["label"] for c in criteria_details if c["met"]]
content_lines = [
f"**Anzahl erfuellter WP248-Kriterien:** {criteria_met} von 9\n",
f"**Erfuellte Kriterien:** {', '.join(met_labels) if met_labels else 'Keine'}\n",
]
if art35_abs3:
content_lines.append(f"**Art. 35 Abs. 3 DS-GVO direkt ausgeloest:** {'; '.join(art35_abs3)}\n")
content_lines.append(
f"\n**Ergebnis:** DSFA ist {'**erforderlich**' if dsfa_required else '**nicht erforderlich**'}."
)
if dsfa_required and criteria_met < 2:
content_lines.append(" (Ausgeloest durch Art. 35 Abs. 3 DS-GVO)")
return {
"content": "\n".join(content_lines),
"criteria_met": criteria_met,
"criteria_details": criteria_details,
"art35_abs3": art35_abs3,
"dsfa_required": dsfa_required,
"muss_liste_ref": aufsicht_info[1],
}
def _generate_section_1(ctx: dict, company: str, dpo: str, dpo_email: str) -> str:
federal_state = ctx.get("federal_state", "")
aufsicht = BUNDESLAND_AUFSICHT.get(
federal_state.lower().replace(" ", "-"), ("Nicht zugeordnet",)
)[0]
lines = [
f"**Verantwortlicher:** {company}",
f"**Datenschutzbeauftragter:** {dpo}" + (f" ({dpo_email})" if dpo_email else ""),
f"**Zustaendige Aufsichtsbehoerde:** {aufsicht}",
]
systems = ctx.get("processing_systems", [])
if systems:
lines.append("\n**Eingesetzte Verarbeitungssysteme:**")
for s in systems:
hosting = s.get("hosting", "")
lines.append(f"- {s.get('name', 'N/A')}" + (f" ({hosting})" if hosting else ""))
return "\n".join(lines)
def _generate_section_2(ctx: dict) -> str:
lines = [
"### Notwendigkeit\n",
"Die Verarbeitung ist zur Erreichung des beschriebenen Zwecks erforderlich. ",
"Alternative, weniger eingriffsintensive Massnahmen wurden geprueft.\n",
"### Datenminimierung\n",
"Die verarbeiteten Datenkategorien beschraenken sich auf das fuer den ",
"Verarbeitungszweck erforderliche Minimum (Art. 5 Abs. 1 lit. c DS-GVO).\n",
]
return "".join(lines)
def _generate_risk_assessment(ctx: dict) -> str:
lines = ["## Risikoanalyse\n"]
# Standard risks
risks = [
("Unbefugter Zugriff auf personenbezogene Daten", "mittel", "hoch", "hoch"),
("Datenverlust durch technischen Ausfall", "niedrig", "hoch", "mittel"),
("Fehlerhafte Verarbeitung / Datenqualitaet", "niedrig", "mittel", "niedrig"),
("Zweckentfremdung erhobener Daten", "niedrig", "hoch", "mittel"),
]
if ctx.get("has_ai_systems") or ctx.get("uses_ai"):
risks.append(("Diskriminierung durch algorithmische Entscheidungen", "mittel", "hoch", "hoch"))
risks.append(("Mangelnde Erklaerbarkeit von KI-Entscheidungen", "mittel", "mittel", "mittel"))
if ctx.get("processes_health_data"):
risks.append(("Offenlegung von Gesundheitsdaten", "niedrig", "gross", "hoch"))
if any(ctx.get(k) for k in ["third_country_transfer", "processes_in_third_country"]):
risks.append(("Zugriff durch Behoerden in Drittlaendern", "mittel", "hoch", "hoch"))
lines.append("| Risiko | Eintrittswahrscheinlichkeit | Schwere | Gesamt |")
lines.append("|--------|----------------------------|---------|--------|")
for risk_name, likelihood, severity, overall in risks:
lines.append(f"| {risk_name} | {likelihood} | {severity} | **{overall}** |")
lines.append("")
high_risks = sum(1 for _, _, _, o in risks if o == "hoch")
if high_risks > 0:
lines.append(f"\n**{high_risks} Risiken mit Stufe 'hoch' identifiziert.** "
"Massnahmen gemaess Abschnitt 5 reduzieren das Restrisiko.")
return "\n".join(lines)
def _generate_section_4(ctx: dict) -> str:
lines = []
if ctx.get("has_works_council"):
lines.append("Der Betriebsrat wurde informiert und angehoert.")
lines.append(
"Eine Konsultation der Betroffenen gemaess Art. 35 Abs. 9 DS-GVO "
"wird empfohlen, soweit verhaeltnismaessig und praktikabel."
)
return "\n".join(lines)
def _generate_sdm_tom_section(ctx: dict) -> str:
"""Generate TOM section structured by 7 SDM Gewaehrleistungsziele."""
lines = []
for goal in SDM_GOALS:
lines.append(f"**{goal['label']}** — {goal['description']}\n")
lines.append("| Massnahme | Typ | Status |")
lines.append("|-----------|-----|--------|")
for measure in goal["default_measures"]:
mtype = "technisch" if any(
kw in measure.lower()
for kw in ["verschluesselung", "backup", "redundanz", "tls", "aes", "rbac", "mfa",
"pruefsumm", "validierung", "loeschfaehigkeit", "export", "automatisiert"]
) else "organisatorisch"
lines.append(f"| {measure} | {mtype} | geplant |")
lines.append("")
return "\n".join(lines)
def _generate_section_6(ctx: dict, dpo: str) -> str:
if dpo:
return (
f"Der Datenschutzbeauftragte ({dpo}) wurde konsultiert. "
"Die Stellungnahme liegt bei bzw. wird nachgereicht."
)
return (
"Ein Datenschutzbeauftragter wurde noch nicht benannt. "
"Sofern eine Benennungspflicht besteht (Art. 37 DS-GVO), "
"ist dies vor Abschluss der DSFA nachzuholen."
)
def _generate_section_7(ctx: dict) -> str:
review_months = ctx.get("review_cycle_months", 12)
lines = [
"### Ergebnis\n",
"Die DSFA wurde gemaess Art. 35 DS-GVO durchgefuehrt. Die identifizierten Risiken ",
"wurden bewertet und durch geeignete Massnahmen auf ein akzeptables Niveau reduziert.\n",
"### Ueberprufungsplan\n",
f"- **Regelmaessige Ueberprufung:** alle {review_months} Monate\n",
"- **Trigger fuer ausserplanmaessige Ueberprufung:**\n",
" - Wesentliche Aenderung der Verarbeitungstaetigkeit\n",
" - Neue oder geaenderte Rechtsgrundlage\n",
" - Sicherheitsvorfall mit Bezug zur Verarbeitung\n",
" - Aenderung der eingesetzten Technologie oder Auftragsverarbeiter\n",
" - Neue Erkenntnisse zu Risiken oder Bedrohungen\n",
]
return "".join(lines)
def _generate_ai_module(ctx: dict) -> str:
"""Generate Section 8 for AI systems (EU AI Act)."""
lines = ["### Eingesetzte KI-Systeme\n"]
ai_systems = ctx.get("ai_systems", [])
if ai_systems:
lines.append("| System | Zweck | Risikokategorie | Human Oversight |")
lines.append("|--------|-------|-----------------|-----------------|")
for s in ai_systems:
risk = s.get("risk_category", "unbekannt")
oversight = "Ja" if s.get("has_human_oversight") else "Nein"
lines.append(f"| {s.get('name', 'N/A')} | {s.get('purpose', 'N/A')} | {risk} | {oversight} |")
lines.append("")
if ctx.get("subject_to_ai_act"):
lines.append(
"**Hinweis:** Das Unternehmen unterliegt dem EU AI Act (Verordnung (EU) 2024/1689). "
"Fuer Hochrisiko-KI-Systeme ist eine grundrechtliche Folgenabschaetzung "
"gemaess Art. 27 KI-VO durchzufuehren.\n"
)
high_risk = [s for s in ai_systems if s.get("risk_category") in ("high", "hoch")]
if high_risk:
lines.append("### Hochrisiko-KI-Systeme — Zusatzanforderungen\n")
lines.append("Fuer die folgenden Systeme gelten die Anforderungen aus Kapitel III KI-VO:\n")
for s in high_risk:
lines.append(f"- **{s.get('name', 'N/A')}**: Risikomanagement (Art. 9), "
f"Daten-Governance (Art. 10), Transparenz (Art. 13), "
f"Human Oversight (Art. 14)\n")
return "\n".join(lines)
def _assess_art36_consultation(ctx: dict, schwellwert: dict) -> bool:
"""Determine if Art. 36 DSGVO consultation with supervisory authority is required.
Art. 36 requires prior consultation when the DSFA indicates that the processing
would result in a HIGH residual risk despite mitigation measures.
"""
if schwellwert["criteria_met"] >= 4:
return True
if len(schwellwert.get("art35_abs3", [])) >= 2:
return True
ai_systems = ctx.get("ai_systems", [])
high_risk_ai = [s for s in ai_systems if s.get("risk_category") in ("high", "hoch", "unacceptable")]
if len(high_risk_ai) >= 2:
return True
return False