""" 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) # Reverse index: broken_url → pages that contain a link to it link_sources: dict[str, list[str]] = {} for page_url, page in current_pages.items(): for link_list in page.get("links", {}).values(): if not isinstance(link_list, list): continue for link in link_list: if isinstance(link, dict): href = link.get("url", "") if href: link_sources.setdefault(href, []).append(page_url) new_broken = [ {"url": url, "status": current_broken_map[url], "pages": sorted(set(link_sources.get(url, [])))} for url in sorted(current_broken - baseline_broken) ] known_broken = [ {"url": url, "status": current_broken_map[url], "pages": sorted(set(link_sources.get(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()