"""Pro-Cookie-Abgleich gegen die Cookie-Knowledge-Library. Vergleicht die DEKLARIERTEN Angaben aus dem CMP/Snapshot (Kategorie, Zweck, Laufzeit) mit dem, was unsere Library (`cookie_knowledge_db`) über den Cookie weiß — und leitet pro Befund eine Abstellmaßnahme ab. Befund-Typen: tracker_as_necessary — als notwendig deklariert, laut Library kein techn. Zweck missing_purpose — kein Zweck deklariert, Library kennt ihn excessive_lifetime — deklarierte Speicherdauer >> typische (Art. 5(1)(e)) vague_duration — Speicherdauer nicht konkret (Art. 5(1)(e)+13) [je Cookie] missing_retention — Verarbeitung deklariert, aber keine Speicherdauer/ Löschfrist + keine Cookies gelistet [je Vendor] third_country — Drittland-Transfer (Schrems II, Art. 44 ff.) [je Vendor] eu_alternative — EU-Ersatz verfügbar (kommerziell) [je Vendor] """ from __future__ import annotations import re from sqlalchemy import text from compliance.services.cookie_knowledge_db import lookup_cookie _TRACKER_CATS = {"marketing", "statistics", "social_media", "targeting"} # A — auditfeste Verdrahtung: jeder Befund-Typ → echter Control (control_id aus # doc_check_controls) + legal_basis. Die Controls tragen regulation/article noch # NULL, daher liefern wir die Rechtsgrundlage hier strukturiert mit (bis sie in # den Controls gepflegt ist). Kette: Regulation → Article → Control → Finding. _CONTROL_MAP = { "vague_duration": {"control_id": "AUTH-2051-A03", "regulation": "DSGVO", "article": "Art. 5 Abs. 1 lit. e + Art. 13"}, "missing_retention": {"control_id": "AUTH-2051-A03", "regulation": "DSGVO", "article": "Art. 5 Abs. 1 lit. e + Art. 13 Abs. 2 lit. a"}, "excessive_lifetime": {"control_id": "AUTH-2051-A02", "regulation": "DSGVO", "article": "Art. 5 Abs. 1 lit. e"}, "tracker_as_necessary": {"control_id": "DATA-2851-A05", "regulation": "TDDDG", "article": "§ 25 Abs. 1"}, "missing_purpose": {"control_id": "AUTH-2053-A05", "regulation": "DSGVO", "article": "Art. 13"}, "missing_opt_out": {"control_id": "DATA-2851-A05", "regulation": "DSGVO", "article": "Art. 7 Abs. 3 + Art. 21"}, "third_country": {"control_id": "DATA-1624-A04", "regulation": "DSGVO", "article": "Art. 44 ff."}, "eu_alternative": {"control_id": None, "regulation": "—", "article": "kommerzielle Empfehlung"}, } # Advisory-Typen: keine bestätigten Verstöße, sondern Hinweise, die der # Cross-Finding-Agent gegen die DSE abgleicht (Drittland kann dort bereits via # SCC/Art. 49/Angemessenheit abgedeckt sein → dann unterdrücken). _HINWEIS_TYPES = {"third_country", "eu_alternative"} # Trennzeichen, an denen ein Runtime-Suffix abgeschnitten werden darf # (z.B. '_ga_GTM-XYZ' → '_ga', 'AMCVS_1234@AdobeOrg' → 'AMCVS'). _SEP_RE = re.compile(r"[_\-.:$%@\[]") def _candidate_keys(name: str) -> list[str]: """Library-Match-Kandidaten: voller (entwildcardeter) Name + Präfixe an Trennzeichen. Fängt Per-Instanz-Suffixe (GTM-Container, @AdobeOrg, Hash-IDs), ohne kurze generische Namen zu über-matchen (Mindestlänge 3).""" from compliance.services.cookie_library_lookup import _strip_wildcards base = _strip_wildcards(name) keys: list[str] = [] if base: keys.append(base) cur = base while True: seps = list(_SEP_RE.finditer(cur)) if not seps: break cur = cur[:seps[-1].start()].rstrip("_-.:$%@") if len(cur) >= 3 and cur not in keys: keys.append(cur) else: break return keys def _match_lib(name: str, lib_bases: dict) -> dict | None: """Erster Treffer eines Kandidaten-Schlüssels in der (entwildcardeten) Library-Basis-Map. Pure + testbar.""" for k in _candidate_keys(name): if len(k) >= 3 and k in lib_bases: return lib_bases[k] return None def load_big_library(db, names: list[str]) -> dict: """Präfix-bewusster Lookup gegen die Open-Cookie-Database (compliance.cookie_library, ~2287). Lädt die Library einmal, entwildcardet die Namen zu Basen und matcht jeden Cookie über _candidate_keys (exact + Runtime-Suffix-Präfix). Schlüssel = ORIGINAL-Cookiename (lower) → Library-Row, damit der Aufrufer wie gewohnt big_lib.get(name.lower()) nutzen kann.""" from compliance.services.cookie_library_lookup import _strip_wildcards uniq = {(n or "").lower() for n in names if n} if not uniq: return {} rows = db.execute( text( "SELECT lower(cookie_name) AS n, actual_category, " "typical_max_age_seconds, vendor_name, purpose_de, purpose_en, " "is_pii FROM compliance.cookie_library" ) ).mappings().fetchall() lib_bases: dict[str, dict] = {} for r in rows: base = _strip_wildcards(r["n"]) if base and base not in lib_bases: lib_bases[base] = dict(r) out: dict[str, dict] = {} for low in uniq: hit = _match_lib(low, lib_bases) if hit: out[low] = hit return out _NECESSARY_CATS = { "necessary", "notwendig", "essential", "essenziell", "funktional", "functional", } _EEA = { "DE", "FR", "IE", "NL", "AT", "BE", "BG", "HR", "CY", "CZ", "DK", "EE", "FI", "GR", "HU", "IT", "LV", "LT", "LU", "MT", "PL", "PT", "RO", "SK", "SI", "ES", "SE", "IS", "LI", "NO", } # Unbekannte/leere Herkunft ist KEIN Drittland (z.B. First-Party-Session-Cookies # PHPSESSID/JSESSIONID mit vendor_country 'N/A'). _UNKNOWN_COUNTRY = {"", "N/A", "NA", "N.A.", "UNKNOWN", "UNBEKANNT", "?"} # Einwilligungspflichtige Kategorien (für Opt-Out-/Widerspruchs-Pflicht). _CONSENT_CATS = {"marketing", "statistics", "targeting", "social_media", "tracking", "werbung", "advertising"} _SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2} # Vage Laufzeit-Formulierungen — keine konkrete Speicherdauer i.S.v. # Art. 5(1)(e) + Art. 13 DSGVO (User-Domain-Vorgabe 2026-06-11). _VAGUE_DURATION = ( "dauerhaft", "bis zur löschung", "bis zur loeschung", "deaktiviert", "unbegrenzt", "unbefristet", "solange erforderlich", "solange benötigt", "unendlich", "permanent", "auf unbestimmte zeit", "kein ablauf", "no expir", "persistent", "until deleted", ) def _is_vague_duration(expiry: str) -> bool: """True, wenn die Angabe vage ist (keine konkrete Dauer/Session/Kriterium).""" e = (expiry or "").strip().lower() if not e: return False has_concrete = ( bool(re.search(r"\d+\s*(tag|woche|monat|jahr|stunde|minute|day|week|" r"month|year|hour|min)", e)) or "session" in e or "browser schließ" in e or "browser schliess" in e or "schließen des browser" in e or "schliessen des browser" in e ) return not has_concrete and any(p in e for p in _VAGUE_DURATION) def _duration_days(s: str) -> int: """Grobe Normalisierung einer Laufzeit-Angabe in Tage (0 = Session).""" s = (s or "").lower() if not s or "session" in s: return 0 m = re.search(r"(\d+)", s) n = int(m.group(1)) if m else 0 if "jahr" in s or "year" in s: return n * 365 if "monat" in s or "month" in s: return n * 30 if "woche" in s or "week" in s: return n * 7 if "tag" in s or "day" in s: return n if "stunde" in s or "hour" in s: return 1 return n def analyze_cookies(vendors: list[dict], big_lib: dict | None = None) -> dict: """Gleiche alle Cookies gegen BEIDE Libraries ab: die 2287er Open-Cookie-DB (`big_lib`, breite Abdeckung: Kategorie/Retention) + die 35er rich-DB (`lookup_cookie`, tiefe Rechtsfelder).""" big_lib = big_lib or {} findings: list[dict] = [] checked = 0 in_library = 0 seen_third: set[str] = set() seen_alt: set[str] = set() # name_lower → tatsächliche Kategorie laut Library (für die Banner-Sicht: # zeigt, wo ein Cookie eigentlich hingehört, falls falsch einsortiert). cookie_cats: dict[str, str] = {} for v in vendors or []: vcat = (v.get("category") or "").lower() vcat_label = v.get("category") or "—" vname = v.get("name") or "?" for c in v.get("cookies") or []: checked += 1 name = c.get("name", "") # Vage Speicherdauer — OHNE Library, gilt fuer ALLE Cookies. if _is_vague_duration(c.get("expiry", "")): findings.append({ "vendor": vname, "cookie": name, "type": "vague_duration", "severity": "MEDIUM", "declared": c.get("expiry", ""), "library_purpose": "", "remediation": ( f"Speicherdauer von '{name}' ist nicht konkret angegeben " f"('{c.get('expiry', '')}'). Art. 5 Abs. 1 lit. e + Art. 13 " f"DSGVO verlangen eine konkrete Dauer oder nachvollziehbare " f"Kriterien (z.B. '13 Monate', 'Session', 'bis Widerruf, " f"max. 13 Monate')." ), }) rich = lookup_cookie(name) or {} big = big_lib.get(name.lower(), {}) if big.get("actual_category"): cookie_cats[name.lower()] = big["actual_category"] if not rich and not big: continue in_library += 1 necessity = rich.get("technical_necessity", "") actual_cat = (big.get("actual_category") or "").lower() purpose = (rich.get("exact_purpose") or big.get("purpose_de") or big.get("purpose_en") or "") alt = rich.get("eu_alternative_vendor", "") country = (rich.get("vendor_country") or "").upper() schrems = rich.get("schrems_ii_status", "") is_tracker = necessity in ("none", "partial") or actual_cat in _TRACKER_CATS # 1) Als notwendig deklariert, laut Library aber Tracker. if vcat in _NECESSARY_CATS and is_tracker: rem = ( f"'{name}' ({vname}) ist als '{vcat_label}' eingestuft, ist laut " f"Library aber kein rein technischer Cookie" + (f" ({purpose})" if purpose else "") + ". Als einwilligungspflichtig nach § 25 Abs. 1 TDDDG einstufen" ) if alt: rem += f"; EU-Alternative: {alt}" findings.append({ "vendor": vname, "cookie": name, "type": "tracker_as_necessary", "severity": "HIGH" if rich.get("reid_risk") == "high" else "MEDIUM", "declared": vcat_label, "library_purpose": purpose, "remediation": rem + ".", }) # 2) Kein Zweck deklariert, Library kennt ihn. elif not (c.get("purpose") or "").strip() and purpose: findings.append({ "vendor": vname, "cookie": name, "type": "missing_purpose", "severity": "MEDIUM", "declared": "(kein Zweck angegeben)", "library_purpose": purpose, "remediation": f"Zweck für '{name}' ergänzen. Laut Library: {purpose}", }) # 3) Speicherdauer deutlich über typischer Laufzeit. decl_days = _duration_days(c.get("expiry", "")) max_age = big.get("typical_max_age_seconds") if max_age: lib_days = int(max_age) // 86400 typ = f"{lib_days} Tage" else: lib_days = _duration_days(rich.get("typical_lifetime", "")) typ = rich.get("typical_lifetime", "") if lib_days > 0 and decl_days - lib_days > 180: findings.append({ "vendor": vname, "cookie": name, "type": "excessive_lifetime", "severity": "LOW", "declared": c.get("expiry", "") or "—", "library_purpose": f"typisch: {typ}", "remediation": ( f"Speicherdauer von '{name}' ({c.get('expiry', '')}) " f"überschreitet die typische ({typ}) deutlich — Art. 5 Abs. 1 " f"lit. e DSGVO (Speicherbegrenzung) prüfen." ), }) # 4) Drittland-Transfer (je Vendor einmal). Nur bei BEKANNTEM # Nicht-EWR-Land — 'N/A'/unbekannt ist KEIN Drittland (First-Party- # Session-Cookies); Self-Hosting laut Library = kein Transfer. country_third = (country not in _UNKNOWN_COUNTRY and country not in _EEA and "SELF-HOST" not in country) if (country_third or schrems) and vname not in seen_third: seen_third.add(vname) findings.append({ "vendor": vname, "cookie": name, "type": "third_country", "severity": "MEDIUM", "declared": country or "—", "library_purpose": schrems or f"Anbieter-Sitz {country}", "remediation": ( f"Neutrales Finding: {vname} kann Daten außerhalb der EU " f"({country or 'Drittland'}) verarbeiten. Für jeden solchen " f"Verarbeiter geeignete Garantien konkret nachweisen (SCC Art. 46 / " f"Angemessenheitsbeschluss / Art. 49) und ggf. eine " f"Transfer-Folgenabschätzung (TIA). Pauschale DSE-Formulierungen " f"('in der Regel SCC') genügen nicht — pro Verarbeiter prüfen " f"(Art. 44 ff. DSGVO). Interne Verträge können wir nicht einsehen." ), }) # 8) EU-Alternative (je Vendor einmal, kommerziell). if alt and (vname + alt) not in seen_alt: seen_alt.add(vname + alt) findings.append({ "vendor": vname, "cookie": name, "type": "eu_alternative", "severity": "LOW", "declared": vname, "library_purpose": f"EU-Ersatz: {alt}", "remediation": ( f"EU-Alternative für {vname}: {alt} — gleiche Funktion, kein " f"Drittland-Transfer, häufig Lizenzkosten-Ersparnis." ), }) # Vendor-Ebene: Verarbeitung deklariert, aber KEINE Speicherdauer. Greift, # wenn keine Cookies gelistet UND keine persistence (z.B. Nayoki GmbH: # 'necessary' Auftragsverarbeiter ohne Löschfrist) — Art. 5(1)(e)+13(2)(a). v_persist = (v.get("persistence") or "").strip() v_purpose = (v.get("purpose") or "").strip() if not (v.get("cookies") or []) and not v_persist and (v_purpose or vcat): findings.append({ "vendor": vname, "cookie": "(keine Cookies gelistet)", "type": "missing_retention", "severity": "MEDIUM", "declared": f"{vcat_label} / keine Speicherdauer", "library_purpose": v_purpose, "remediation": ( f"Für '{vname}' ist eine Datenverarbeitung deklariert " f"(Kategorie '{vcat_label}'), aber keine Speicherdauer/Löschfrist " f"angegeben und keine Cookies gelistet. Art. 5 Abs. 1 lit. e + " f"Art. 13 Abs. 2 lit. a DSGVO verlangen eine konkrete " f"Speicherdauer bzw. Löschfrist — bitte für '{vname}' eine " f"Löschfrist festlegen und in der Cookie-Richtlinie ausweisen." ), }) # Vendor-Ebene: einwilligungspflichtiger Anbieter (Marketing/Tracking) # mit Cookies, aber ohne Opt-Out-/Widerspruchs-Link. if (vcat in _CONSENT_CATS and (v.get("cookies") or []) and not (v.get("opt_out_url") or "").strip()): findings.append({ "vendor": vname, "cookie": "(Vendor-Ebene)", "type": "missing_opt_out", "severity": "LOW", "declared": vcat_label, "library_purpose": "", "remediation": ( f"Für den einwilligungspflichtigen Anbieter '{vname}' " f"({vcat_label}) ist kein Opt-Out-/Widerspruchs-Link " f"hinterlegt. Eine einfache Widerrufs-/Widerspruchs-Möglichkeit " f"angeben (Art. 7 Abs. 3 + Art. 21 DSGVO, § 25 TDDDG) — so " f"einfach wie die Einwilligung." ), }) # A: jeden Befund an Control + Rechtsgrundlage haengen + als echtes Finding # (zu beheben) oder Hinweis (advisory, gegen DSE abzugleichen) klassifizieren. for f in findings: f["control"] = _CONTROL_MAP.get(f["type"], {}) f["kind"] = "hinweis" if f["type"] in _HINWEIS_TYPES else "finding" findings.sort(key=lambda f: _SEV_ORDER.get(f["severity"], 3)) return { "summary": { "checked": checked, "in_library": in_library, "findings": len(findings), }, "findings": findings, "cookie_categories": cookie_cats, }