breakpilot-compliance/backend-compliance/compliance/services/cookie_library_check.py

"""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))
  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"}


def load_big_library(db, names: list[str]) -> dict:
    """Batch-Lookup der grossen Open-Cookie-Database (compliance.cookie_library,
    ~2287 Cookies) fuer die gegebenen Namen. Breite Abdeckung: Kategorie,
    Retention, Vendor."""
    uniq = sorted({(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 "
            "WHERE lower(cookie_name) = ANY(:names)"
        ),
        {"names": uniq},
    ).mappings().fetchall()
    return {r["n"]: dict(r) for r in rows}

_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",
}
_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()

    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 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).
            if (country and country not in _EEA 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"{vname} überträgt in ein Drittland ({country or 'außerhalb EWR'}) — "
                        f"SCC (Art. 46) oder DPF-Zertifizierung prüfen und in der "
                        f"Datenschutzerklärung benennen (Art. 44 ff. DSGVO)."
                    ),
                })

            # 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."
                    ),
                })

    findings.sort(key=lambda f: _SEV_ORDER.get(f["severity"], 3))
    return {
        "summary": {
            "checked": checked,
            "in_library": in_library,
            "findings": len(findings),
        },
        "findings": findings,
    }