Files
breakpilot-compliance/backend-compliance/compliance/api/agent_doc_check_extras.py
T
Benjamin Admin 6c223c7c9b
CI / detect-changes (push) Successful in 10s
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 / validate-canonical-controls (push) Successful in 14s
CI / loc-budget (push) Failing after 15s
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) Successful in 2m43s
CI / test-go (push) Has been skipped
CI / iace-gt-coverage (push) Has been skipped
CI / test-python-backend (push) Successful in 37s
CI / test-python-document-crawler (push) Has been skipped
CI / test-python-dsms-gateway (push) Has been skipped
feat(compliance-check): exec-summary + voll-audit + TDM-respect + cookie-KB-extended + saving-scan-funnel
P1 — Exec-Summary oben im Email-Report (4 KPIs + 2 CTAs, dunkler Gradient)
P3 — no_direct_sales-Flag fuer OEM-Konfigurator-Sites; AGB/Widerruf/AGB als
     "NICHT ANWENDBAR" (grau) statt "NICHT GEFUNDEN" (rot)
P5 — Voll-Audit Unification: alle Findings (MC + Pflichtangaben + Vendor +
     Redundanz) in /data/compliance_audits.db.unified_findings; neuer
     /api/compliance/agent/findings/<id> Endpoint + FindingsTab im Audit-UI
     mit Filter + CSV-Export
P7 — Crawl-Hardening: TDM-Reservation-Check (robots.txt / ai.txt / Header /
     Meta) vor jedem Run mit 24h-Cache; HeadlessChrome-UA (Firma noch nicht
     gegruendet — Switch via BREAKPILOT_BRANDED_UA env); per-Domain
     Rate-Limit 1 req/s + max 2 concurrent
P2 — Cookie-Knowledge-DB additiv erweitert (35 -> 74 Cookies): Adobe, Meta,
     Microsoft, LinkedIn, TikTok, HubSpot, Marketo, Salesforce, Hotjar,
     FullStory, Mouseflow, Intercom, Drift, Zendesk, Cloudflare, Stripe,
     OneTrust/Cookiebot/Usercentrics, Matomo, Pinterest, Snapchat, X/Twitter,
     YouTube, Vimeo, Klaviyo, Mailchimp, Mixpanel, Segment, Amplitude,
     Optimizely, Datadog; Wire-in in cookie_function_classifier liefert
     compliance_risk-Label (kritisch/hoch/mittel/gering) pro Vendor
A  — k-Anonymitaets-Helper (benchmark_k_anonymity) fuer P6-Vorbereitung
B  — Cross-Tenant-Domain-Assertion im /findings-Endpoint (expected_domain
     Query-Param -> 403 bei Mismatch)
C  — Saving-Scan-Funnel: /api/compliance/agent/saving-scan/start mit
     Validierung + 24h-Rate-Limit pro Domain + Lead-Persistenz in
     saving_scan_leads + Auto-Discovery via _run_compliance_check; 6 Tests
D  — Risk-Badge im Email-Vendor-Row

Rechtliche Leitplanken (Memory feedback_oem_data_legal.md): nur eigene
Knapp-Bewertungen + Source-Pointer, keine 1:1-Kopien fremder CMP-Texte.
TDM-Opt-Out-Respect nach § 44b UrhG. KEINE Schema-Aenderungen — alles in
Sidecar-SQLite.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-18 23:48:34 +02:00

489 lines
20 KiB
Python

"""
Extras for the agent doc-check email report.
Split out from agent_doc_check_report.py to keep both files under the
500-line hard cap. Contains:
- build_scanned_urls_html (list of fetched URLs + cross-domain notice)
- build_provider_list_html (cookie banner + TCF vendor table)
"""
from __future__ import annotations
def build_scanned_urls_html(doc_entries: list[dict]) -> str:
"""Render the list of scanned URLs at the top of the report.
Transparent for the GF which sources were actually fetched/analysed.
Skips empty URLs (text-only uploads). Adds a cross-domain warning when
legal texts are distributed across multiple domains (e.g. BMW spreads
across bmw.de, bmwgroup.com, bmwgroup.jobs).
"""
from urllib.parse import urlparse
rows: list[str] = []
seen: set[str] = set()
domains: dict[str, list[str]] = {} # netloc -> list of doc_types
for entry in doc_entries:
url = (entry.get("url") or "").strip()
if not url or url in seen:
continue
seen.add(url)
label = _doc_type_label(entry.get("doc_type", ""))
words = entry.get("word_count") or 0
auto = entry.get("auto_discovered")
try:
netloc = urlparse(url).netloc.lower().lstrip("www.")
if netloc:
domains.setdefault(netloc, []).append(label)
except Exception:
pass
badge = ('<span style="display:inline-block;margin-left:6px;'
'background:#dbeafe;color:#1e40af;font-size:10px;'
'padding:1px 6px;border-radius:8px;font-family:sans-serif">'
'auto-entdeckt</span>') if auto else ""
rows.append(
f'<tr>'
f'<td style="padding:3px 12px 3px 0;color:#475569;font-size:12px">'
f'{label}{badge}</td>'
f'<td style="padding:3px 12px 3px 0;font-size:12px;'
f'font-family:ui-monospace,monospace;color:#1e293b;word-break:break-all">'
f'<a href="{url}" style="color:#2563eb;text-decoration:none">{url}</a></td>'
f'<td style="padding:3px 0;color:#94a3b8;font-size:11px;text-align:right;'
f'white-space:nowrap">{words} Woerter</td>'
f'</tr>'
)
if not rows:
return ""
cross_domain_html = _cross_domain_notice(domains) if len(domains) >= 2 else ""
return (
'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;'
'max-width:700px;margin:0 auto 16px;padding:12px 16px;'
'background:#fafafa;border:1px solid #e5e7eb;border-radius:8px">'
'<h3 style="margin:0 0 8px;font-size:14px;color:#334155">'
f'Gepruefte Quellen ({len(rows)})</h3>'
'<table style="width:100%;border-collapse:collapse">'
+ "".join(rows)
+ '</table>'
+ cross_domain_html
+ '</div>'
)
def _cross_domain_notice(domains: dict[str, list[str]]) -> str:
"""Warning box when legal texts are spread across multiple domains.
Relevant for big corporate groups (BMW Group: bmw.de / bmwgroup.com /
bmwgroup.jobs). Affects findability for data subjects and may indicate
incomplete disclosure on the main site.
"""
items = []
for netloc, labels in sorted(domains.items()):
labels_str = ", ".join(sorted(set(labels)))
items.append(
f'<li style="margin-bottom:2px"><strong>{netloc}</strong> '
f'<span style="color:#92400e;font-size:11px">&rarr; {labels_str}</span></li>'
)
return (
'<div style="margin-top:12px;padding:10px 12px;background:#fffbeb;'
'border-left:3px solid #f59e0b;border-radius:4px;font-size:12px;'
'color:#78350f">'
'<strong>Hinweis: Rechtstexte verteilt auf '
f'{len(domains)} Domains.</strong> '
'Erschwert die Auffindbarkeit fuer Betroffene (Art. 12 Abs. 1 DSGVO &mdash; '
'transparente Information). Pruefen Sie, ob alle Texte auch von der '
'Hauptdomain aus klar verlinkt sind.'
'<ul style="margin:6px 0 0 16px;padding-left:0">'
+ "".join(items) +
'</ul></div>'
)
def _doc_type_label(doc_type: str) -> str:
"""Lazy resolver — avoids circular import with agent_compliance_check_routes."""
labels = {
"dse": "Datenschutzerklaerung",
"datenschutz": "Datenschutzerklaerung",
"privacy": "Datenschutzerklaerung",
"impressum": "Impressum",
"agb": "AGB",
"widerruf": "Widerrufsbelehrung",
"cookie": "Cookie-Richtlinie",
"avv": "Auftragsverarbeitung",
"loeschkonzept": "Loeschkonzept",
"dsfa": "Datenschutz-Folgenabschaetzung",
"social_media": "Social Media Datenschutz",
"nutzungsbedingungen": "Nutzungsbedingungen",
"dsb": "DSB-Kontakt",
}
return labels.get(doc_type, doc_type.upper() if doc_type else "Dokument")
def build_provider_list_html(
banner_result: dict | None,
vvt_entries: list[dict] | None,
) -> str:
"""Render the cookie banner result + TCF vendor table for the email.
Sections:
1. Banner summary (provider, violations count)
2. Vendor table: Name | Kategorie | Zweck | Drittland | Rechtsgrundlage
"""
if not banner_result and not vvt_entries:
return ""
parts: list[str] = [
'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;'
'max-width:700px;margin:0 auto 16px;padding:12px 16px;'
'background:#fafafa;border:1px solid #e5e7eb;border-radius:8px">'
'<h3 style="margin:0 0 10px;font-size:14px;color:#334155">'
'Cookie-Banner &amp; Verarbeiter</h3>'
]
if banner_result:
detected = banner_result.get("banner_detected", False)
provider = banner_result.get("banner_provider") or "unbekannt"
violations = banner_result.get("banner_checks", {}).get("violations", [])
n_viol = len(violations) if isinstance(violations, list) else int(violations or 0)
status_color = "#16a34a" if detected and n_viol == 0 else (
"#d97706" if detected else "#6b7280"
)
parts.append(
f'<div style="font-size:13px;color:#374151;margin-bottom:10px">'
f'<span style="display:inline-block;width:8px;height:8px;'
f'border-radius:50%;background:{status_color};margin-right:8px"></span>'
f'Banner erkannt: <strong>{"Ja" if detected else "Nein"}</strong>'
f' &nbsp;&middot;&nbsp; Anbieter: <strong>{provider}</strong>'
f' &nbsp;&middot;&nbsp; Auffaelligkeiten: <strong>{n_viol}</strong>'
f'</div>'
)
vendors = vvt_entries or []
if vendors:
parts.append(
f'<div style="font-size:12px;color:#475569;margin:8px 0 6px">'
f'<strong>{len(vendors)} TCF-Verarbeiter ueber das Banner eingebunden:</strong>'
f'</div>'
'<table style="width:100%;border-collapse:collapse;font-size:11px">'
'<thead><tr style="background:#f1f5f9;color:#475569;text-align:left">'
'<th style="padding:5px 8px">Name</th>'
'<th style="padding:5px 8px">Kategorie</th>'
'<th style="padding:5px 8px">Zweck</th>'
'<th style="padding:5px 8px">Drittland</th>'
'<th style="padding:5px 8px">Rechtsgrundlage</th>'
'</tr></thead><tbody>'
)
for v in vendors[:50]:
parts.append(_render_vendor_row(v))
parts.append('</tbody></table>')
if len(vendors) > 50:
parts.append(
f'<div style="font-size:11px;color:#94a3b8;margin-top:4px">'
f'... und {len(vendors) - 50} weitere</div>'
)
elif banner_result and banner_result.get("banner_detected"):
parts.append(
'<div style="font-size:11px;color:#94a3b8">'
'Keine TCF-Verarbeiter erkannt (Banner nutzt kein TCF v2 Framework '
'oder Vendor-Liste konnte nicht ausgelesen werden).</div>'
)
parts.append('</div>')
return "".join(parts)
def _render_vendor_row(v: dict) -> str:
name = v.get("name") or "Unbekannt"
kategorie = _category_label(v.get("kategorie", ""))
zweck = v.get("zweck_kurz") or ", ".join((v.get("zweck") or [])[:2])
drittland = v.get("drittland")
land = v.get("land") or ""
if drittland is True:
drittland_str = (f'<span style="color:#dc2626">Ja ({land})</span>'
if land else '<span style="color:#dc2626">Ja</span>')
elif drittland is False:
drittland_str = (f'<span style="color:#16a34a">Nein ({land})</span>'
if land else '<span style="color:#16a34a">Nein</span>')
else:
drittland_str = '<span style="color:#94a3b8">unbekannt</span>'
rg = v.get("rechtsgrundlage", "")
rg_short = "Einwilligung" if "Einwilligung" in rg else (
"Berechtigtes Interesse" if "Berechtigtes" in rg else rg[:40]
)
return (
f'<tr style="border-top:1px solid #e2e8f0">'
f'<td style="padding:4px 8px;color:#1e293b">{name}</td>'
f'<td style="padding:4px 8px;color:#475569">{kategorie}</td>'
f'<td style="padding:4px 8px;color:#475569">{zweck}</td>'
f'<td style="padding:4px 8px">{drittland_str}</td>'
f'<td style="padding:4px 8px;color:#475569">{rg_short}</td>'
f'</tr>'
)
def _category_label(kat: str) -> str:
return {
"necessary": "Notwendig",
"functional": "Funktional",
"statistics": "Statistik",
"marketing": "Marketing",
"strictlyNecessary": "Notwendig",
"advertising": "Marketing",
}.get(kat, kat or "")
def build_vvt_table_html(vendors: list[dict]) -> str:
"""Render the per-vendor VVT-style table for the email report.
Splits vendors into 3-4 sections by recipient_type (Art. 30(1)(d)
DSGVO):
1. INTERNAL — own departments / own systems
2. GROUP_COMPANY — parent/subsidiary (if any)
3. PROCESSOR — Auftragsverarbeiter (AVV-pflichtig)
4. CONTROLLER — joint/independent controllers (Meta, Google,
LinkedIn — they build own profiles)
5. AUTHORITY / OTHER — rest
Within each section: rows sorted by compliance_score ascending so
the weakest entries surface first.
"""
if not vendors:
return ""
# Import here to avoid pulling backend service deps at module load
from compliance.services.vendor_classifier import RECIPIENT_TYPE_SECTIONS
# Bucket vendors by recipient_type
by_type: dict[str, list[dict]] = {}
for v in vendors:
rt = (v.get("recipient_type") or "OTHER").upper()
by_type.setdefault(rt, []).append(v)
# Top summary
n_total = len(vendors)
n_internal = sum(1 for v in vendors
if (v.get("recipient_type") or "").upper()
in ("INTERNAL", "GROUP_COMPANY"))
n_external = n_total - n_internal
n_critical = sum(1 for v in vendors if v.get("compliance_score", 0) < 50)
summary_parts = [f"{n_total} Verarbeitungen erfasst"]
if n_internal and n_external:
summary_parts.append(
f"&mdash; {n_internal} eigene + {n_external} externe Empfaenger"
)
if n_critical:
summary_parts.append(
f', <strong style="color:#dc2626">{n_critical} unter 50%</strong>'
)
else:
summary_parts.append("&mdash; alle ueber 50%")
summary = " ".join(summary_parts)
out: list[str] = [
'<div style="font-family:-apple-system,BlinkMacSystemFont,sans-serif;'
'max-width:760px;margin:0 auto 16px;padding:12px 16px;'
'background:#fafafa;border:1px solid #e5e7eb;border-radius:8px">',
'<h3 style="margin:0 0 4px;font-size:14px;color:#334155">'
'VVT-Vorschlag: Verarbeitungstaetigkeiten und Empfaenger aus der '
'Cookie-Richtlinie</h3>',
f'<p style="margin:0 0 10px;font-size:11px;color:#6b7280">{summary}. '
'Gruppiert nach Empfaengerkategorie (Art. 30(1)(d) DSGVO). Innerhalb '
'jeder Gruppe nach Compliance-Score sortiert. Bei eigenen '
'Verarbeitungen (INTERNAL/GROUP) werden Opt-Out und Privacy-Link '
'NICHT als Pflicht gewertet &mdash; der Widerruf erfolgt ueber das '
'Cookie-Banner, Privacy ist in der Haupt-DSI dokumentiert.</p>',
]
for rtype, section_label in RECIPIENT_TYPE_SECTIONS:
rows = by_type.get(rtype) or []
if not rows:
continue
rows = sorted(rows, key=lambda v: v.get("compliance_score", 0))
n = len(rows)
n_bad = sum(1 for v in rows if v.get("compliance_score", 0) < 50)
bad_hint = (f' <span style="color:#dc2626">({n_bad} unter 50%)</span>'
if n_bad else "")
out.append(
f'<h4 style="margin:14px 0 4px;font-size:12px;color:#1e293b;'
f'border-top:1px solid #e2e8f0;padding-top:8px">'
f'{section_label} <span style="color:#94a3b8;font-weight:400">'
f'({n}){bad_hint}</span></h4>'
)
out.append(_render_vendor_section(rows))
out.append('</div>')
return "".join(out)
def _render_vendor_section(rows: list[dict]) -> str:
body: list[str] = [
'<table style="width:100%;border-collapse:collapse;font-size:11px">'
'<thead><tr style="background:#f1f5f9;color:#475569;text-align:left">'
'<th style="padding:5px 8px">Name</th>'
'<th style="padding:5px 8px">Kategorie</th>'
'<th style="padding:5px 8px">Sitz</th>'
'<th style="padding:5px 8px;text-align:center">Cookies</th>'
'<th style="padding:5px 8px;text-align:center">Opt-Out</th>'
'<th style="padding:5px 8px;text-align:center">Privacy</th>'
'<th style="padding:5px 8px;text-align:right">Score</th>'
'</tr></thead><tbody>',
]
for v in rows:
body.append(_render_vendor_row_full(v))
body.append('</tbody></table>')
return "".join(body)
def _render_vendor_row_full(v: dict) -> str:
rtype = (v.get("recipient_type") or "OTHER").upper()
is_own = rtype in ("INTERNAL", "GROUP_COMPANY")
cat = (v.get("category") or "").lower()
is_necessary = cat in ("necessary", "strictlynecessary")
name = v.get("name") or "Unbekannt"
category = _category_label(v.get("category", ""))
country = v.get("country") or ("" if is_own else "")
cookies = v.get("cookies") or []
n_cookies = len(cookies)
score = int(v.get("compliance_score", 0))
flags = v.get("compliance_flags") or []
# Opt-Out: nicht erforderlich fuer eigene Verarbeitung oder
# technisch notwendige Cookies (§25 Abs. 2 TDDDG).
opt_na_reason = ("Nicht erforderlich (eigene Verarbeitung — "
"Widerruf ueber Cookie-Banner)") if is_own else (
"Nicht erforderlich (§25 Abs. 2 TDDDG — technisch notwendig)"
if is_necessary else None
)
opt_status = _link_status_badge(
v.get("opt_out_url"), v.get("opt_out_ok"), v.get("opt_out_status"),
na_label=opt_na_reason,
)
# Privacy: nicht erforderlich fuer eigene Verarbeitung (Haupt-DSI).
privacy_na_reason = (
"Nicht erforderlich (eigene Verarbeitung — durch Haupt-DSI abgedeckt)"
if is_own else None
)
privacy_status = _link_status_badge(
v.get("privacy_policy_url"), v.get("privacy_ok"),
v.get("privacy_status"), na_label=privacy_na_reason,
)
score_color = ("#16a34a" if score >= 80 else
"#d97706" if score >= 50 else "#dc2626")
# Score-Erklaerung: was wurde gewertet, was fehlt
# Annahme: Score = bestandene Kriterien / Gesamtkriterien * 100.
# Typisch 5 Kriterien fuer EXT: country, cookies, opt_out, privacy, scoring.
# Bei INTERNAL/GROUP: opt_out + privacy nicht gewertet (3 Kriterien).
n_criteria = 3 if is_own else 5
n_failed = len(flags) if flags else 0
score_tooltip = (
f"{n_criteria - n_failed} von {n_criteria} Kriterien erfuellt"
+ (f" — fehlt: {', '.join(_flag_short(f) for f in flags[:3])}"
if flags else "")
)
# Inline-Aktions-Anweisungen pro Flag
actions_html = ""
if flags:
from compliance.services.finding_action_recipes import recipe_for
action_items = []
for f in flags:
rec = recipe_for(f)
if not rec:
continue
action_items.append(
f'<li style="margin-bottom:6px"><strong>{_flag_short(f)}:</strong> '
f'{rec.get("what", "")}<br/>'
f'<span style="color:#475569"><strong>Was tun:</strong> '
f'{rec.get("fix_text", "").splitlines()[0][:200]}</span><br/>'
f'<span style="color:#94a3b8;font-size:9px">Quelle: '
f'{rec.get("why", "")[:160]}</span></li>'
)
if action_items:
actions_html = (
f'<details style="margin-top:4px"><summary style="cursor:pointer;'
f'color:#dc2626;font-size:10px">Was muss ich tun? '
f'({len(action_items)} Action{"s" if len(action_items) != 1 else ""})</summary>'
f'<ul style="margin:4px 0 0 14px;padding:0;font-size:10px;color:#1e293b">'
+ "".join(action_items)
+ '</ul></details>'
)
flag_str = ""
if flags:
flag_str = (
f'<div style="font-size:10px;color:#94a3b8;margin-top:2px">'
f'{", ".join(flags[:4])}</div>'
f'{actions_html}'
)
risk = v.get("compliance_risk") or {}
risk_label = risk.get("label") or ""
risk_badge = ""
if risk_label and risk_label != "unklar":
rc = {"kritisch": ("#dc2626", "#fff"), "hoch": ("#fecaca", "#991b1b"),
"mittel": ("#fde68a", "#92400e"), "gering": ("#d1fae5", "#065f46")}.get(risk_label, ("#e5e7eb", "#475569"))
risk_badge = (f'<span style="margin-left:6px;padding:1px 5px;border-radius:3px;font-size:9px;'
f'background:{rc[0]};color:{rc[1]}">Risk: {risk_label}</span>')
return (
f'<tr style="border-top:1px solid #e2e8f0">'
f'<td style="padding:6px 8px;color:#1e293b;font-size:11px">'
f'{name}{risk_badge}{flag_str}</td>'
f'<td style="padding:6px 8px;color:#475569;font-size:11px">{category}</td>'
f'<td style="padding:6px 8px;color:#475569;font-size:11px">{country}</td>'
f'<td style="padding:6px 8px;text-align:center;color:#475569;font-size:11px">'
f'{n_cookies}</td>'
f'<td style="padding:6px 8px;text-align:center">{opt_status}</td>'
f'<td style="padding:6px 8px;text-align:center">{privacy_status}</td>'
f'<td style="padding:6px 8px;text-align:right;font-weight:600;'
f'color:{score_color};font-size:11px" title="{score_tooltip}">'
f'{score}%<div style="font-size:9px;font-weight:400;color:#94a3b8">'
f'{n_criteria - n_failed}/{n_criteria}</div></td>'
f'</tr>'
)
def _flag_short(f: str) -> str:
"""Lesbare deutsche Form fuer einen Flag-Token."""
labels = {
"no_cookies_listed": "Cookies fehlen",
"no_country": "Sitzland fehlt",
"no_privacy_url": "Privacy-Link fehlt",
"broken_privacy_url": "Privacy-Link broken",
"no_opt_out_url": "Opt-Out fehlt",
"broken_opt_out": "Opt-Out broken",
}
return labels.get(f, f)
def _link_status_badge(
url: str | None,
ok: bool | None,
status: int | None,
na_label: str | None = None,
) -> str:
"""Render the link-status cell.
- url + ok -> green check
- url + broken -> red cross with status
- no url + na_label -> neutral em-dash with explanation tooltip
(used for INTERNAL/necessary rows where the field isn't required)
- no url + no na_label -> red cross (real gap)
"""
if not url:
if na_label:
return ('<span style="color:#94a3b8;font-size:11px" '
f'title="{na_label}">&mdash;</span>')
return ('<span style="color:#dc2626;font-size:11px" '
'title="Kein Link">&#10007;</span>')
if ok:
return ('<span style="color:#16a34a;font-size:11px" '
f'title="HTTP {status}">&#10003;</span>')
status_str = str(status) if status else "?"
return ('<span style="color:#dc2626;font-size:11px" '
f'title="HTTP {status_str}">&#10007; ({status_str})</span>')