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