integrity_scanner_fuer_statische_Webseiten/scanner/differ.py

"""
Diff logic: text, link, metadata, and URL-set comparison.
Scoring maps diff results to a risk level (green/yellow/red).
"""
import difflib
import hashlib
import re
from urllib.parse import urlparse


# ---------------------------------------------------------------------------
# Normalisation
# ---------------------------------------------------------------------------

def normalize_text(text: str, cfg: dict) -> str:
    """Apply noise-reduction before diffing."""
    norm_cfg = cfg.get("normalization", {})
    if norm_cfg.get("collapse_whitespace", True):
        text = re.sub(r"\s+", " ", text).strip()
    for pattern in (norm_cfg.get("ignore_patterns") or []):
        text = re.sub(pattern, "[IGNORED]", text)
    return text


def text_hash(text: str) -> str:
    return hashlib.sha256(text.encode("utf-8")).hexdigest()


# ---------------------------------------------------------------------------
# Atomic diffs
# ---------------------------------------------------------------------------

def diff_text(old: str, new: str, context: int = 2) -> list[str]:
    """Unified diff of two normalised texts."""
    return list(difflib.unified_diff(
        old.splitlines(),
        new.splitlines(),
        fromfile="baseline",
        tofile="current",
        lineterm="",
        n=context,
    ))


def diff_link_set(old_list: list[dict], new_list: list[dict]) -> dict:
    """Compare two link lists by URL.

    Cache-buster query params are stripped at extraction time
    (extractor.strip_cache_params), so stored URLs are already stable here.
    """
    old_urls = {l["url"] for l in old_list if isinstance(l, dict) and "url" in l}
    new_urls = {l["url"] for l in new_list if isinstance(l, dict) and "url" in l}
    return {
        "added": sorted(new_urls - old_urls),
        "removed": sorted(old_urls - new_urls),
    }


def diff_metadata(old: dict, new: dict) -> dict:
    """Compare metadata fields that matter for integrity monitoring."""
    changes: dict = {}
    for key in ("title", "description", "canonical", "robots"):
        if old.get(key) != new.get(key):
            changes[key] = {"old": old.get(key), "new": new.get(key)}

    for heading in ("h1", "h2"):
        old_set = set(old.get(heading, []))
        new_set = set(new.get(heading, []))
        if old_set != new_set:
            changes[heading] = {
                "added": sorted(new_set - old_set),
                "removed": sorted(old_set - new_set),
            }
    return changes


def _all_external_domains(pages: dict) -> set[str]:
    """Collect every external domain referenced across all pages."""
    domains: set[str] = set()
    for page in pages.values():
        for link_type, links in page.get("links", {}).items():
            if not isinstance(links, list):
                continue
            for l in links:
                if isinstance(l, dict) and l.get("class") == "external":
                    netloc = urlparse(l.get("url", "")).netloc
                    if netloc:
                        domains.add(netloc)
    return domains


# ---------------------------------------------------------------------------
# Full snapshot comparison
# ---------------------------------------------------------------------------

def compare_snapshots(baseline_pages: dict, current_pages: dict, cfg: dict) -> dict:
    """
    Compare two page sets (url → page_dict) and produce a structured diff.

    Returns a dict with:
      new_internal_urls, missing_internal_urls,
      new_external_domains, removed_external_domains,
      page_diffs (list of per-page change records),
      new_pages_analysis (freshly seen pages checked for immediate issues)
    """
    b_urls = set(baseline_pages)
    c_urls = set(current_pages)

    new_urls = sorted(c_urls - b_urls)
    missing_urls = sorted(b_urls - c_urls)
    common_urls = sorted(b_urls & c_urls)

    page_diffs: list[dict] = []

    for url in common_urls:
        old = baseline_pages[url]
        new = current_pages[url]

        old_text = normalize_text(old.get("text", ""), cfg)
        new_text = normalize_text(new.get("text", ""), cfg)
        text_changed = text_hash(old_text) != text_hash(new_text)

        # Quick-skip if nothing changed (text + links + meta all identical)
        if (
            not text_changed
            and old.get("links") == new.get("links")
            and old.get("metadata") == new.get("metadata")
            and old.get("hidden_content") == new.get("hidden_content")
            and old.get("inline_scripts") == new.get("inline_scripts")
            and old.get("jsonld") == new.get("jsonld")
        ):
            continue

        text_diff = diff_text(old_text, new_text) if text_changed else []

        # Per-type link diff
        link_diff: dict = {}
        for ltype in ("a", "script", "iframe", "link_rel", "form"):
            old_links = old.get("links", {}).get(ltype, [])
            new_links = new.get("links", {}).get(ltype, [])
            ld = diff_link_set(old_links, new_links)
            if ld["added"] or ld["removed"]:
                link_diff[ltype] = ld

        meta_diff = diff_metadata(old.get("metadata", {}), new.get("metadata", {}))

        # New hidden content (compare by fingerprint to avoid dupe noise)
        old_hc_fps = {_fingerprint(h) for h in old.get("hidden_content", [])}
        new_hidden = [
            h for h in new.get("hidden_content", [])
            if _fingerprint(h) not in old_hc_fps
        ]

        # New suspicious inline scripts
        old_scripts_fps = {_fingerprint(s) for s in old.get("inline_scripts", [])}
        new_scripts = [
            s for s in new.get("inline_scripts", [])
            if _fingerprint(s) not in old_scripts_fps
        ]

        # New JSON-LD
        old_jld_fps = {_fingerprint(j) for j in old.get("jsonld", [])}
        new_jsonld = [
            j for j in new.get("jsonld", [])
            if _fingerprint(j) not in old_jld_fps
        ]

        # New meta-refresh targets
        old_mr = set(old.get("links", {}).get("meta_refresh", []))
        new_mr = [u for u in new.get("links", {}).get("meta_refresh", []) if u not in old_mr]

        # New canonical
        old_canonical = old.get("links", {}).get("canonical") or old.get("metadata", {}).get("canonical")
        new_canonical = new.get("links", {}).get("canonical") or new.get("metadata", {}).get("canonical")
        canonical_changed = old_canonical != new_canonical

        # New comment links
        old_cl = set(old.get("comment_links", []))
        new_cl = [u for u in new.get("comment_links", []) if u not in old_cl]

        if any([
            text_diff, link_diff, meta_diff, new_hidden, new_scripts,
            new_jsonld, new_mr, canonical_changed, new_cl,
        ]):
            added_chars = sum(
                len(line[1:]) for line in text_diff
                if line.startswith("+") and not line.startswith("+++")
            )
            page_diffs.append({
                "url": url,
                "text_diff": text_diff,
                "added_chars": added_chars,
                "link_diff": link_diff,
                "meta_diff": meta_diff,
                "new_hidden_content": new_hidden,
                "new_inline_scripts": new_scripts,
                "new_jsonld": new_jsonld,
                "new_meta_refresh": new_mr,
                "canonical_changed": canonical_changed,
                "old_canonical": old_canonical,
                "new_canonical": new_canonical,
                "new_comment_links": new_cl,
            })

    # Site-wide external domain diff
    old_ext = _all_external_domains(baseline_pages)
    new_ext = _all_external_domains(current_pages)

    # Broken link diff (4xx/5xx pages)
    baseline_broken = {
        url for url, p in baseline_pages.items() if p.get("status", 0) >= 400
    }
    current_broken_map = {
        url: p.get("status", 0)
        for url, p in current_pages.items()
        if p.get("status", 0) >= 400
    }
    current_broken = set(current_broken_map)
    new_broken = [
        {"url": url, "status": current_broken_map[url]}
        for url in sorted(current_broken - baseline_broken)
    ]
    known_broken = [
        {"url": url, "status": current_broken_map[url]}
        for url in sorted(current_broken & baseline_broken)
    ]

    # Analysis of brand-new pages (no baseline to compare against)
    new_pages_analysis = [
        _quick_check_new_page(current_pages[url])
        for url in new_urls
        if url in current_pages
    ]

    # Unexpected JSON-LD @types — independent of page_diffs.
    # Only flag types that are NOT allowed AND were not already in the baseline
    # for that URL (avoids re-alerting on a pre-approved page).
    allowed_jsonld = set(cfg.get("allowed_jsonld_types", []))
    unexpected_jsonld: list[dict] = []
    for url, page in current_pages.items():
        base_types = {
            j.get("type") for j in baseline_pages.get(url, {}).get("jsonld", [])
        }
        for j in page.get("jsonld", []):
            t = j.get("type")
            if t not in allowed_jsonld and t not in base_types:
                unexpected_jsonld.append({"url": url, "type": t})

    return {
        "new_internal_urls": new_urls,
        "missing_internal_urls": missing_urls,
        "new_external_domains": sorted(new_ext - old_ext),
        "removed_external_domains": sorted(old_ext - new_ext),
        "page_diffs": page_diffs,
        "new_pages_analysis": new_pages_analysis,
        "unexpected_jsonld": unexpected_jsonld,
        "new_broken_urls": new_broken,
        "known_broken_urls": known_broken,
        "total_pages_checked": len(common_urls),
        "changed_pages": len(page_diffs),
    }


def _quick_check_new_page(page: dict) -> dict:
    """Summarise a new (not in baseline) page for immediate risk indicators."""
    return {
        "url": page["url"],
        "hidden_content": page.get("hidden_content", []),
        "inline_scripts": page.get("inline_scripts", []),
        "unexpected_jsonld": [j for j in page.get("jsonld", []) if j.get("unexpected")],
        "meta_refresh": page.get("links", {}).get("meta_refresh", []),
        "comment_links": page.get("comment_links", []),
        "external_link_count": sum(
            1 for links in page.get("links", {}).values()
            if isinstance(links, list)
            for l in links
            if isinstance(l, dict) and l.get("class") == "external"
        ),
    }


# ---------------------------------------------------------------------------
# Scoring
# ---------------------------------------------------------------------------

def score_diff(diff: dict, cfg: dict) -> dict:
    """
    Map a diff result to a risk score and level.

    Returns {score, level, reasons, exit_code}.
    """
    sc = cfg.get("scoring", {})
    thr = cfg.get("thresholds", {"yellow": 20, "red": 60})
    large_text = thr.get("large_text_block_chars", 200)

    score = 0
    reasons: list[str] = []

    def add(pts: int, msg: str) -> None:
        nonlocal score
        score += pts
        reasons.append(f"{msg} (+{pts})")

    for domain in diff.get("new_external_domains", []):
        add(sc.get("new_external_domain", 50), f"Neue externe Domain: {domain}")

    for url in diff.get("new_internal_urls", []):
        add(sc.get("new_internal_url", 30), f"Neue interne URL: {url}")

    for url in diff.get("missing_internal_urls", []):
        add(sc.get("missing_internal_url", 20), f"Fehlende URL: {url}")

    for pd in diff.get("page_diffs", []):
        u = pd["url"]

        for h in pd.get("new_hidden_content", []):
            add(sc.get("hidden_content", 40),
                f"{u}: Hidden Content [{', '.join(h.get('indicators', []))}]")

        for mr in pd.get("new_meta_refresh", []):
            add(sc.get("meta_refresh", 40), f"{u}: Meta-Refresh → {mr}")

        for s in pd.get("new_inline_scripts", []):
            add(sc.get("new_inline_script_suspicious", 30),
                f"{u}: Verdächtiges Inline-Script [{', '.join(s.get('patterns', []))}]")

        if pd.get("canonical_changed"):
            add(sc.get("unexpected_canonical", 35),
                f"{u}: Canonical geändert → {pd.get('new_canonical')}")

        for cl in pd.get("new_comment_links", []):
            add(sc.get("comment_links", 25), f"{u}: Link in HTML-Kommentar: {cl}")

        if pd.get("added_chars", 0) > large_text:
            add(sc.get("large_text_addition", 20),
                f"{u}: +{pd['added_chars']} Zeichen neuer Text")

    for entry in diff.get("new_broken_urls", []):
        add(sc.get("new_broken_link", 20),
            f"Neuer kaputte Link: {entry['url']} (HTTP {entry['status']})")
    # known_broken_urls are reported but not scored — they were already in the baseline

    for entry in diff.get("unexpected_jsonld", []):
        add(sc.get("unexpected_jsonld", 35),
            f"{entry['url']}: Unerwartetes JSON-LD @type={entry['type']}")

    for entry in diff.get("suspicious_filenames", []):
        add(sc.get("suspicious_filename", 60),
            f"{entry['page']}: Verdächtiger Dateiname {entry['url']}")

    # New pages: immediate suspicious content counts too
    for npa in diff.get("new_pages_analysis", []):
        u = npa["url"]
        if npa.get("hidden_content"):
            add(sc.get("hidden_content", 40), f"{u} (neu): Hidden Content")
        if npa.get("inline_scripts"):
            add(sc.get("new_inline_script_suspicious", 30), f"{u} (neu): Verdächtiges Script")
        if npa.get("meta_refresh"):
            add(sc.get("meta_refresh", 40), f"{u} (neu): Meta-Refresh")
        if npa.get("comment_links"):
            add(sc.get("comment_links", 25), f"{u} (neu): Kommentar-Link")
        # Unexpected JSON-LD on new pages is covered by the unexpected_jsonld channel.

    yellow = thr.get("yellow", 20)
    red = thr.get("red", 60)

    if score >= red:
        level = "red"
        exit_code = 2
    elif score >= yellow:
        level = "yellow"
        exit_code = 1
    else:
        level = "green"
        exit_code = 0

    return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}


# ---------------------------------------------------------------------------
# Cloaking detection
# ---------------------------------------------------------------------------

def compare_cloaking(normal_pages: dict, bot_pages: dict, cfg: dict) -> dict:
    """
    Compare pages crawled with a normal browser UA vs. Googlebot UA.
    Returns findings where the bot version contains extra content.
    """
    findings: list[dict] = []
    common = sorted(set(normal_pages) & set(bot_pages))

    for url in common:
        normal = normal_pages[url]
        bot = bot_pages[url]

        # Vary: User-Agent signals server-side content negotiation
        vary = bot.get("response_headers", {}).get("Vary", "")
        has_vary_ua = "user-agent" in vary.lower()

        # Text diff
        normal_text = normalize_text(normal.get("text", ""), cfg)
        bot_text = normalize_text(bot.get("text", ""), cfg)
        text_changed = text_hash(normal_text) != text_hash(bot_text)
        text_diff = diff_text(normal_text, bot_text) if text_changed else []
        extra_chars = sum(
            len(line[1:]) for line in text_diff
            if line.startswith("+") and not line.startswith("+++")
        )

        # Links only in bot version
        extra_links: list[str] = []
        for ltype in ("a", "script", "iframe"):
            ld = diff_link_set(
                normal.get("links", {}).get(ltype, []),
                bot.get("links", {}).get(ltype, []),
            )
            extra_links.extend(ld["added"])

        if extra_links or extra_chars > 50 or has_vary_ua:
            findings.append({
                "url": url,
                "vary_ua": has_vary_ua,
                "extra_links": extra_links,
                "extra_text_chars": extra_chars,
                "text_diff": text_diff[:40],
            })

    return {
        "findings": findings,
        "pages_checked": len(common),
    }


def score_cloak_diff(cloak_diff: dict, cfg: dict) -> dict:
    """Score the cloaking check result."""
    sc = cfg.get("scoring", {})
    thr = cfg.get("thresholds", {"yellow": 20, "red": 60})
    score = 0
    reasons: list[str] = []

    def add(pts: int, msg: str) -> None:
        nonlocal score
        score += pts
        reasons.append(f"{msg} (+{pts})")

    for page in cloak_diff.get("findings", []):
        url = page["url"]
        for link in page.get("extra_links", []):
            add(sc.get("cloaking_extra_link", 60), f"{url}: Bot-Only-Link: {link}")
        if page.get("extra_text_chars", 0) > 100:
            add(sc.get("cloaking_extra_text", 40),
                f"{url}: Bot sieht +{page['extra_text_chars']} Zeichen mehr Text")
        # Vary: User-Agent ohne sonstigen Unterschied — informativer Hinweis
        if page.get("vary_ua") and not page.get("extra_links") and page.get("extra_text_chars", 0) <= 100:
            add(sc.get("cloaking_vary_ua", 25), f"{url}: Vary: User-Agent Header")

    yellow = thr.get("yellow", 20)
    red = thr.get("red", 60)

    if score >= red:
        level, exit_code = "red", 2
    elif score >= yellow:
        level, exit_code = "yellow", 1
    else:
        level, exit_code = "green", 0

    return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}


# ---------------------------------------------------------------------------
# Asset hash comparison
# ---------------------------------------------------------------------------

def compare_asset_hashes(baseline: dict, current: dict) -> dict:
    """
    Compare asset hash dicts {url: {sha256, size, type, error}}.
    Returns {changed, new, missing, fetch_errors, total_checked}.
    """
    b_hashed = {url for url, r in baseline.items() if r.get("sha256")}
    c_hashed = {url for url, r in current.items() if r.get("sha256")}

    changed = [
        {
            "url": url,
            "type": current[url].get("type", "unknown"),
            "old_sha256": baseline[url]["sha256"][:16] + "…",
            "new_sha256": current[url]["sha256"][:16] + "…",
            "size_diff": (current[url].get("size") or 0) - (baseline[url].get("size") or 0),
        }
        for url in sorted(b_hashed & c_hashed)
        if baseline[url]["sha256"] != current[url]["sha256"]
    ]

    return {
        "changed": changed,
        "new": sorted(c_hashed - b_hashed),
        "missing": sorted(b_hashed - set(current)),
        "fetch_errors": [
            {"url": url, "error": r["error"]}
            for url, r in sorted(current.items())
            if r.get("error")
        ],
        "total_checked": len(c_hashed),
    }


def score_asset_diff(asset_diff: dict, cfg: dict) -> dict:
    """Score asset hash changes by file type."""
    sc = cfg.get("scoring", {})
    thr = cfg.get("thresholds", {"yellow": 20, "red": 60})
    score = 0
    reasons: list[str] = []

    def add(pts: int, msg: str) -> None:
        nonlocal score
        score += pts
        reasons.append(f"{msg} (+{pts})")

    for entry in asset_diff.get("changed", []):
        asset_type = entry.get("type", "unknown")
        if asset_type == "script":
            pts = sc.get("changed_script", 40)
        elif asset_type == "link_rel":
            pts = sc.get("changed_stylesheet", 30)
        else:
            pts = sc.get("changed_asset", 20)
        size_info = f", {entry['size_diff']:+d} Bytes" if entry.get("size_diff") else ""
        add(pts, f"Geänderte Datei [{asset_type}]{size_info}: {entry['url']}")

    yellow = thr.get("yellow", 20)
    red = thr.get("red", 60)
    if score >= red:
        level, exit_code = "red", 2
    elif score >= yellow:
        level, exit_code = "yellow", 1
    else:
        level, exit_code = "green", 0

    return {"score": score, "level": level, "reasons": reasons, "exit_code": exit_code}


def _fingerprint(obj: dict | list | str) -> str:
    """Stable hash for deduplicating diff entries."""
    import json as _json
    return hashlib.md5(  # noqa: S324 — not crypto, just dedup
        _json.dumps(obj, sort_keys=True, ensure_ascii=False).encode()
    ).hexdigest()