"""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")) # Domain-spezifische Risiken (AI Act Annex III) domain = ctx.get("domain", "") if domain in ("hr", "recruiting") or ctx.get("has_hr_context"): risks.append(("AGG-Verstoss: Diskriminierung bei Bewerberauswahl (§ 1 AGG)", "mittel", "hoch", "hoch")) risks.append(("Beweislastumkehr bei Diskriminierungsklagen (§ 22 AGG)", "mittel", "hoch", "hoch")) risks.append(("Art. 22 DSGVO: Unzulaessige automatisierte Einzelentscheidung", "mittel", "hoch", "hoch")) risks.append(("Proxy-Diskriminierung durch Name/Foto/Alter-Erkennung", "mittel", "hoch", "hoch")) if domain in ("education", "higher_education", "vocational_training"): risks.append(("Chancenungleichheit durch KI-gestuetzte Bewertung", "mittel", "hoch", "hoch")) risks.append(("Benachteiligung Minderjaehriger ohne Lehrkraft-Kontrolle", "niedrig", "gross", "hoch")) risks.append(("Fehlbewertung mit Auswirkung auf Bildungschancen", "mittel", "hoch", "hoch")) if domain in ("healthcare", "medical_devices", "pharma", "elderly_care"): risks.append(("Fehldiagnose durch KI mit gesundheitlichen Folgen", "niedrig", "gross", "hoch")) risks.append(("Falsche Triage-Priorisierung (lebenskritisch)", "niedrig", "gross", "hoch")) risks.append(("Verletzung der Patientenautonomie", "mittel", "hoch", "hoch")) if domain in ("finance", "banking", "insurance", "investment"): risks.append(("Diskriminierendes Kredit-Scoring", "mittel", "hoch", "hoch")) risks.append(("Ungerechtfertigte Verweigerung von Finanzdienstleistungen", "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