From 98e8d11eb5e61d3cc660098cf7059b52674ad604 Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 12 Jun 2026 00:51:41 +0200 Subject: [PATCH] feat: initial implementation of website integrity scanner MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Python-Package zur unabhängigen Überwachung statischer Websites gegen SEO-Spam und unbefugte Manipulationen. Läuft außerhalb des Hosters. Kernfunktionen: - Reiner Python-Crawler (bs4+lxml), kein wget - Baseline-Management mit manuellem Freigabe-Workflow (niemals automatisch) - Text-/Link-/Meta-Diff mit Risiko-Scoring (grün/gelb/rot) - Hidden-Content-Erkennung (CSS inline, noscript, Kommentar-Links) - Whitelist für externe Domains und interne Pfade - E-Mail + Webhook Alarmierung - CLI: init, crawl, check, scan, approve, report, status - 79 Unit-Tests (pytest), alle grün Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 45 +++ CLAUDE.md | 83 ++++ Keller/PROMPT.md | 249 ++++++++++++ Keller/site_audit_tool_v4.py | 754 +++++++++++++++++++++++++++++++++++ README.md | 203 ++++++++++ config.yaml | 112 ++++++ config/allowed_external.yaml | 18 + config/allowed_paths.yaml | 17 + config/ignore_rules.yaml | 23 ++ requirements.txt | 5 + scanner/__init__.py | 0 scanner/__main__.py | 672 +++++++++++++++++++++++++++++++ scanner/alerter.py | 183 +++++++++ scanner/baseline.py | 231 +++++++++++ scanner/checker.py | 164 ++++++++ scanner/config.py | 110 +++++ scanner/crawler.py | 150 +++++++ scanner/differ.py | 338 ++++++++++++++++ scanner/extractor.py | 312 +++++++++++++++ tests/__init__.py | 0 tests/test_baseline.py | 161 ++++++++ tests/test_crawler.py | 63 +++ tests/test_differ.py | 250 ++++++++++++ tests/test_extractor.py | 152 +++++++ 24 files changed, 4295 insertions(+) create mode 100644 .gitignore create mode 100644 CLAUDE.md create mode 100644 Keller/PROMPT.md create mode 100755 Keller/site_audit_tool_v4.py create mode 100644 README.md create mode 100644 config.yaml create mode 100644 config/allowed_external.yaml create mode 100644 config/allowed_paths.yaml create mode 100644 config/ignore_rules.yaml create mode 100644 requirements.txt create mode 100644 scanner/__init__.py create mode 100644 scanner/__main__.py create mode 100644 scanner/alerter.py create mode 100644 scanner/baseline.py create mode 100644 scanner/checker.py create mode 100644 scanner/config.py create mode 100644 scanner/crawler.py create mode 100644 scanner/differ.py create mode 100644 scanner/extractor.py create mode 100644 tests/__init__.py create mode 100644 tests/test_baseline.py create mode 100644 tests/test_crawler.py create mode 100644 tests/test_differ.py create mode 100644 tests/test_extractor.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..1d0aaca --- /dev/null +++ b/.gitignore @@ -0,0 +1,45 @@ +# Python +__pycache__/ +*.py[cod] +*.pyo +*.pyd +.Python +*.egg-info/ +dist/ +build/ +.eggs/ +*.egg + +# Virtual environments +venv/ +.venv/ +env/ +.env/ + +# Tests / Coverage +.pytest_cache/ +.coverage +htmlcov/ +.tox/ + +# Runtime data (never commit — may contain crawled content or sensitive diffs) +data/snapshots/ +data/baseline/ +data/pending/ +reports/ +logs/ + +# Editor +.idea/ +.vscode/ +*.swp +*.swo +*~ + +# OS +.DS_Store +Thumbs.db + +# Secrets (extra safeguard) +*.env +.env.* diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..cc6ff25 --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1,83 @@ +# CLAUDE.md + +This file provides guidance to Claude Code (claude.ai/code) when working with code in this repository. + +## Commands + +```bash +# Install dependencies +pip install -r requirements.txt + +# Run all tests +pytest tests/ -v + +# Run a single test file +pytest tests/test_extractor.py -v + +# Run a single test +pytest tests/test_differ.py::TestScoreDiff::test_green_for_no_changes -v + +# CLI usage (always run from project root where config.yaml lives) +python -m scanner status +python -m scanner init +python -m scanner scan +python -m scanner approve --all --note "Reason" +python -m scanner approve --url https://bredelar.info/page/ +python -m scanner report +python -m scanner --verbose scan +``` + +## Architecture + +The scanner is a Python package (`scanner/`) invoked via `python -m scanner`. All subcommands share one data flow: + +``` +Crawler → Extractor → BaselineManager (snapshot) + ↓ + differ.compare_snapshots + ↓ + differ.score_diff → alerter.send_alert + ↓ + __main__._write_report (JSON + MD) +``` + +**Key design constraint**: The baseline is immutable except through an explicit `approve` call. `compare_snapshots` and `score_diff` are pure functions — they never write anything. + +### Module responsibilities + +- **`config.py`** — loads `config.yaml`, deep-merges onto `DEFAULT_CONFIG`. All paths (data_dir, reports_dir, logs_dir, config_dir) are resolved to absolute paths by `resolve_paths()` before being passed to other modules. +- **`crawler.py`** — breadth-first crawl using `requests`. `normalize_url()` is the canonical URL form used everywhere (strips fragment, lowercases host, removes noise query params, sorts remaining params). `should_crawl()` filters out binary assets and tracking URLs. +- **`extractor.py`** — parses HTML with bs4+lxml. Returns a single dict per page with keys: `text`, `links`, `metadata`, `hidden_content`, `inline_scripts`, `jsonld`, `comment_links`. The `links` dict has typed sub-lists (`a`, `link_rel`, `script`, `img`, `iframe`, `form`, `meta_refresh`, `canonical`). +- **`baseline.py`** — `BaselineManager` handles two separate stores: `data/snapshots/YYYYMMDD_HHMMSS/` (raw crawl output) and `data/baseline/` (approved reference). Each page is stored as `.json`. `approve_*` methods are the only path into the baseline. +- **`differ.py`** — `compare_snapshots(baseline_pages, current_pages, cfg)` returns a flat diff dict. `score_diff(diff, cfg)` maps it to `{score, level, reasons, exit_code}`. Thresholds: yellow ≥ 20, red ≥ 60 (configurable). A single new external domain = 50 pts (yellow); combined with hidden content = 90 pts (red). +- **`checker.py`** — stateless helper functions for whitelist enforcement, security-header auditing, canonical hijack detection. Called from `__main__.cmd_check`, not from `differ`. +- **`alerter.py`** — sends alert only when `level >= min_level`. SMTP password is read from the env var named in `smtp_password_env` (default: `SCANNER_SMTP_PASSWORD`). +- **`__main__.py`** — wires everything together. `cmd_scan` = `cmd_crawl` (crawl + save snapshot) + `cmd_check` (load baseline + snapshot, diff, score, report, alert). Exit codes: 0=green, 1=yellow, 2=red. + +### Data layout + +``` +data/baseline/manifest.json # approved_at, approved_by, note, urls[] +data/baseline/pages/<16hex>.json # one file per approved URL +data/snapshots/YYYYMMDD_HHMMSS/ # one dir per crawl run +reports/YYYYMMDD_HHMMSS/report.{json,md} +config/allowed_external.yaml # set of permitted external domains +config/allowed_paths.yaml # set of permitted internal paths (optional) +config/ignore_rules.yaml # normalization selectors + regex patterns +``` + +### Scoring weights (defaults, all configurable in `config.yaml`) + +| Event | Points | +|---|---| +| New external domain | 50 | +| Hidden content (CSS-hidden element with links/text) | 40 | +| Meta-Refresh | 40 | +| Unexpected canonical | 35 | +| Unexpected JSON-LD @type | 35 | +| New internal URL | 30 | +| New suspicious inline script | 30 | +| Comment-embedded link | 25 | +| Large text addition (>200 chars) | 20 | +| Missing internal URL | 20 | +| Missing security header | 5 | diff --git a/Keller/PROMPT.md b/Keller/PROMPT.md new file mode 100644 index 0000000..7e73f4e --- /dev/null +++ b/Keller/PROMPT.md @@ -0,0 +1,249 @@ +Erstelle bitte ein produktionsreifes, möglichst schlankes Kommandozeilen-Kontroll-Programm in Python zur unabhängigen Überwachung der kleinen, statischen Website https://bredelar.info (einstellbar) gegen Manipulationen. Es soll manuell oder als Cron-Programm einsetzbar sein. + +Ziel +Das Programm soll im Verantwortungsbereich des Betreibers laufen, nicht beim beauftragten Web-Dienstleister. Es soll regelmäßig, z. B. täglich per Cron um 03:00 Uhr (einstellbar) , die gesamte Web-Repräsentanz von bredelar.info überwachen und Alarm auslösen, wenn unerwartete Änderungen, neue Seiten, unerlaubte Links oder problematische Inhalte auftauchen. Der Fokus liegt auf robuster Erkennung bei möglichst geringem Betriebsaufwand. Die Website ist klein und soll klein bleiben. + +Wichtige Rahmenbedingungen +- Die Website ist klein, vermutlich mit wenigen Seiten. +- Es gab bereits einen SEO-Spam-/Parasite-SEO-artigen Vorfall. +- Angreifer können Texte thematisch passend tarnen; reine Keyword-Heuristik reicht daher nicht. +- Das System soll unabhängig vom Dienstleister arbeiten. +- Das System soll auf einem kleinen Linux-VHost oder Minimalserver laufen können. +- Bevorzugte Sprachen/Tools: Python 3 oder Shell, aber bitte pragmatisch entscheiden. +- Ergebnis soll einfach installierbar, wartbar und nachvollziehbar sein. +- Keine unnötig komplexe Infrastruktur, keine Kubernetes-, Cloud-, Queue- oder Datenbank-Overkill-Lösung. + +Die Lösung soll diese drei Kernprobleme adressieren + +1. Integritätsprüfung / Diff-Monitoring +- Prüfe regelmäßig alle relevanten Seiten von bredelar.info. +- Erstelle Snapshots einer „gesunden“ Referenzversion. +- Vergleiche spätere Abrufe mit dieser Referenz. +- Alarmiere bei jeder relevanten Abweichung. +- Alarm muss detailliert sein: betroffene URL, Art der Änderung, alter/neuer Zustand, neue/entfernte Links, neue/entfernte Textelemente, neue/entfernte HTML-Elemente, Risiko-Einstufung. +- Änderungen durch legitime Pflege müssen ebenfalls erkannt werden; sie dürfen erst nach manueller Freigabe zur neuen Referenz/Baseline werden. +- Wichtig: Das System darf kompromittierte Inhalte niemals automatisch als neue „gesunde“ Basis übernehmen. + +2. Link-Kontrolle / Whitelist +- Finde alle offenen und verdeckten Links auf der gesamten Web-Repräsentanz. +- Prüfe sichtbare Links, versteckte Links, Links in Metadaten, Canonical, hreflang, redirects, iframe/src, script/src, img/src, form action, meta refresh, JSON-LD, strukturierte Daten, noscript-Bereiche, Kommentare, data-Attribute, CSS-basierte Hinweise, falls sinnvoll. +- Führe eine Whitelist erlaubter externer Ziele und eine Whitelist erlaubter interner URL-Pfade. +- Jede neue externe Ziel-Domain oder unerwartete interne URL soll Alarm auslösen. +- SEO-Spam braucht Links; daher ist das ein zentraler Kontrollpunkt. + +3. Problematische Inhalte +- Erkenne auch problematische Inhalte ohne Links, z. B. Pornografie, Dirty Talk, politische Propaganda, diffamierende oder strafbare Inhalte, widersprüchliche Aussagen, versteckte Spam-Texte oder thematisch unpassende kommerzielle Inhalte. +- Nutze dafür einfache robuste Regeln plus optional eine zusätzliche semantische/KI-gestützte Analyse. +- Aber: KI/semantische Analyse ist Ergänzung, nicht Kern. Kern bleibt Integritäts- und Link-Monitoring. +- Gute Angreifer können Dorf-/Heimat-Begriffe einbauen; daher bitte nicht auf simple White-/Blacklist-Keywords vertrauen. + +Ganz wichtige Zusatzanforderung: „Gesunde“ Datenausgangslage und Pflege +Das System muss sauber mit einer Referenz-/Baseline-Verwaltung umgehen: + +A. Initiale gesunde Ausgangslage erzeugen +- Implementiere einen expliziten Initialisierungsmodus. +- Das System soll die Domain crawlen, bekannte Seiten erfassen und daraus eine erste Referenz erzeugen. +- Diese Referenz darf nicht blind übernommen werden; es muss einen Review-/Freigabeprozess geben. +- Liefere dafür einen klaren Workflow: + 1. Crawl der Domain + 2. Fundliste aller internen Seiten + 3. Extraktion aller externen Links + 4. Erkennung möglicher Auffälligkeiten + 5. Manuelle Prüfung + 6. Erst nach Freigabe: Speichern als „gesunde“ Baseline + +B. Pflege der gesunden Daten +- Änderungen an legitimen Inhalten sollen nicht automatisch in die Baseline einfließen. +- Stattdessen: + - Monitoring meldet Änderung + - Mensch prüft Änderung + - separater „approve baseline“-Schritt übernimmt die neue Version +- Baselines sollen versioniert und nachvollziehbar abgelegt werden. +- Es soll möglich sein, pro URL eine neue Referenz freizugeben. +- Es soll auch möglich sein, die gesamte Site-Referenz neu aufzubauen, wenn es große legitime Änderungen gibt. +- Es soll erkennbar bleiben, wer/was/wann eine Baseline freigegeben hat. + +Crawler- und Erkennungslogik +Bitte entwerfe das System so, dass es nicht nur bekannte Seiten prüft, sondern die Domain aktiv crawlt: +- Start mit einer Seed-Liste +- Danach internes Follow aller erlaubten Links +- Begrenzung auf bredelar.info +- Erkennung neuer interner URLs +- Erkennung von Seiten, die plötzlich verschwunden sind +- Optional: robots.txt respektieren konfigurierbar +- Schutz gegen Crawl-Explosion, Parameter-Müll und Kalender-/Session-URLs + +Das System soll auf mehreren Ebenen vergleichen: +- Roh-HTML-Diff +- bereinigter DOM-/Text-Diff +- Link-Diff +- URL-Diff der gefundenen internen Seiten +- Metadaten-Diff (title, meta description, canonical, hreflang, robots, og-tags) +- Hidden-Content-Indikatoren +- Optional Header-Checks + +Wichtig: Normalisierung gegen Fehlalarme +Bitte implementiere vor dem Diff sinnvolle Normalisierung, damit harmlose Änderungen nicht dauernd Alarm auslösen: +- Whitespace normalisieren +- irrelevante Timestamps, Cache-Buster, Nonces, zufällige IDs, Tracking-Parameter optional ignorieren +- HTML in eine stabile Form bringen +- konfigurierbare Ignore-Regeln für Seitenbereiche oder Selektoren +- getrennte Betrachtung von Roh-HTML und sichtbar extrahiertem Inhalt + +Hidden-/SEO-Spam-spezifische Checks +Bitte berücksichtige speziell typische SEO-Spam-/Parasite-SEO-Indikatoren: +- neue interne Unterseiten +- neue externe Links +- versteckte Links +- display:none / visibility:hidden / opacity:0 / off-screen positioning +- meta refresh +- unerwartete canonical-Tags +- unerwartete strukturierte Daten / JSON-LD mit fremden Entities +- starke Änderungen in Title/H1/H2 +- längere neue Textblöcke +- eingebettete kommerzielle Blöcke innerhalb eigentlich lokaler Inhalte +- auffällige Script-/Iframe-Einbindungen +- Kommentar- oder noscript-Missbrauch +- falls praktikabel: einfache Scoring-Logik für Verdachtsfälle + +Alarmierung +Das System soll Alarmmeldungen verschicken können, idealerweise modular: +- E-Mail als Mindeststandard +- optional Webhook/Signal/Matrix/Telegram/SMS als Erweiterung +- Alarmmeldungen sollen kompakt, aber forensisch brauchbar sein +- Beispielinhalt: + - Zeitpunkt + - URL + - Schweregrad + - neue interne URLs + - neue externe Domains + - Link-Diff + - Text-Diff + - Hidden-Content-Hinweise + - Empfehlung: nur beobachten / prüfen / sofort handeln + +Technische Zielarchitektur +Bitte schlage eine minimalistische, aber saubere Struktur vor, z. B.: +- ein Hauptprogramm mit Subcommands oder mehrere kleine Skripte +- Cron-tauglich +- Konfigurationsdateien in YAML oder JSON +- lokale Dateistruktur für: + - seeds + - erlaubte interne URLs + - erlaubte externe Links/Domains + - snapshots + - baselines + - reports + - state + - logs +- bitte ohne unnötige externe Dienste, wenn nicht zwingend erforderlich + +Gewünschte Artefakte +Ich möchte von dir eine vollständige, direkt nutzbare Lösung erzeugt bekommen. Bitte liefere: + +1. Architekturüberblick +- kurz und präzise +- welche Komponenten gibt es und warum + +2. Bedrohungsmodell +- was die Lösung erkennt +- was sie nicht erkennt +- welche Annahmen gelten + +3. Projektstruktur +- konkrete Dateibaum-Struktur + +4. Konfigurationsdateien +- vollständig, mit sinnvollen Beispielen +- z. B. config.yaml, seeds.yaml, allowed_external_domains.yaml, allowed_internal_paths.yaml, ignore_rules.yaml + +5. Vollständigen Quellcode +- möglichst in Python 3 +- sauber strukturiert +- modular +- mit CLI-Subcommands wie z. B.: + - init-baseline + - crawl + - check + - approve + - report +- robuste Fehlerbehandlung +- keine unnötige Magie +- gut lesbar + +6. HTML-Parsing und Extraktion +- stabile Extraktion von: + - sichtbarem Text + - Links + - Assets/Referenzen + - Metadaten + - Hidden-Content-Hinweisen +- bitte geeignete Bibliotheken wählen und begründen + +7. Diff-Logik +- wie Roh-HTML-, Text-, Link- und URL-Diffs erzeugt werden +- Ausgabe menschenlesbar und maschinenlesbar + +8. Alarmierungsmodul +- zunächst mindestens E-Mail +- optional Webhook-Schnittstelle abstrahiert +- Beispielkonfiguration + +9. Cron-Setup +- konkrete Cron-Beispiele +- z. B. täglicher Check 03:00 Uhr +- Log-Rotation oder Umgang mit Logs + +10. Workflow-Dokumentation +- Erstinitialisierung der „gesunden“ Daten +- Review/Freigabe +- täglicher Betrieb +- Umgang mit legitimen Änderungen +- Incident-Fall: was tun bei Alarm + +11. Sicherheits- und Qualitätsverbesserungen +- sinnvolle optionale Erweiterungen +- aber nur pragmatisch, nicht überengineert + +12. Tests +- mindestens ein paar einfache Unit- oder Integrations-Tests +- insbesondere für: + - URL-Normalisierung + - Link-Extraktion + - Hidden-Content-Erkennung + - Baseline-Freigabeprozess + +13. Beispielausgabe +- Beispiel für normalen Bericht ohne Auffälligkeiten +- Beispiel für Alarm bei neuer SEO-Spam-Seite +- Beispiel für Alarm bei neuem externen Link +- Beispiel für Alarm bei legitimer Änderung, die anschließend freigegeben wird + +Wichtige Designprinzipien +- So einfach wie möglich, aber nicht einfacher. +- Kleine Website, kleine Lösung. +- Hohe Nachvollziehbarkeit. +- Keine automatische Selbstheilung. +- Keine automatische Baseline-Aktualisierung. +- Lieber wenige robuste Regeln als viele fragile. +- Unabhängige Kontrolle des Dienstleisters. +- Gute Wartbarkeit für technisch versierte Betreiber. + +Wichtige technische Präferenzen +- Linux-tauglich +- Python 3 bevorzugt +- Wenige, verbreitete Bibliotheken +- Keine schweren externen Datenbanken, wenn Dateispeicherung reicht +- UTF-8 überall +- Saubere Trennung zwischen Konfiguration, Zustand, Referenzdaten und Reports + +Bitte beginne mit: +1. einer kurzen Zielarchitektur, +2. danach der konkreten Projektstruktur, +3. dann allen Dateien mit vollständigem Inhalt, +4. und zum Schluss einer knappen Betriebsanleitung. + +Wichtig: +- Bitte keine abstrakte Beratung allein, sondern vollständige umsetzbare Artefakte. +- Falls du Annahmen treffen musst, triff pragmatische Annahmen und dokumentiere sie. +- Der Code soll direkt als Ausgangspunkt verwendbar sein. diff --git a/Keller/site_audit_tool_v4.py b/Keller/site_audit_tool_v4.py new file mode 100755 index 0000000..236634c --- /dev/null +++ b/Keller/site_audit_tool_v4.py @@ -0,0 +1,754 @@ +#!/usr/bin/env python3 + +import argparse +import hashlib +import json +import os +import re +import subprocess +import sys +from collections import defaultdict +from html.parser import HTMLParser +from pathlib import Path +from urllib.parse import urljoin, urlparse + +import requests + +DEFAULT_ALLOWED = { + 'html','htm','css','js','mjs','json','txt','xml','svg', + 'png','jpg','jpeg','gif','webp','ico','map', + 'woff','woff2','ttf','otf','eot','pdf','webmanifest' +} + +SUSPICIOUS_NAME_RE = re.compile( + r'(shell|backdoor|cmd|upload|gate|mailer|wso|c99|r57|b374k)', + re.I +) + +SUSPICIOUS_CONTENT_PATTERNS = [ + r'eval\s*\(', + r'base64_decode', + r'shell_exec\s*\(', + r'passthru\s*\(', + r'system\s*\(', + r'exec\s*\(', + r'cmd\.exe', + r'powershell', + r'document\.createElement\s*\(\s*[\"\']script[\"\']\s*\)', + r'new\s+Image\s*\(\s*\)\.src', + r'fromCharCode', + r'atob\s*\(', + r'fetch\s*\(', + r'XMLHttpRequest' +] + +HEADERS = {"User-Agent": "site-audit/4.0 (+security review)"} + + +class AssetParser(HTMLParser): + def __init__(self): + super().__init__() + self.links = [] + self.inline_scripts = [] + self.inline_script = [] + self.in_script = False + + def handle_starttag(self, tag, attrs): + attrs = dict(attrs) + if tag == 'a' and attrs.get('href'): + self.links.append(('a', attrs['href'])) + elif tag == 'link' and attrs.get('href'): + self.links.append(('link', attrs['href'])) + elif tag == 'script' and attrs.get('src'): + self.links.append(('script', attrs['src'])) + elif tag == 'img' and attrs.get('src'): + self.links.append(('img', attrs['src'])) + elif tag == 'source' and attrs.get('src'): + self.links.append(('source', attrs['src'])) + elif tag == 'iframe' and attrs.get('src'): + self.links.append(('iframe', attrs['src'])) + elif tag == 'script' and not attrs.get('src'): + self.in_script = True + self.inline_script = [] + + def handle_endtag(self, tag): + if tag == 'script' and self.in_script: + self.inline_scripts.append(''.join(self.inline_script)) + self.in_script = False + + def handle_data(self, data): + if self.in_script: + self.inline_script.append(data) + + +def sha256_file(path: Path): + h = hashlib.sha256() + with path.open('rb') as f: + for chunk in iter(lambda: f.read(65536), b''): + h.update(chunk) + return h.hexdigest() + + +def ensure_dir(p: Path): + p.mkdir(parents=True, exist_ok=True) + + +def is_same_host(base, target): + b = urlparse(base) + t = urlparse(target) + return (not t.netloc) or (t.netloc == b.netloc) + + +def clean_url(url): + parsed = urlparse(url) + return parsed._replace(fragment='').geturl() + + +def mirror_with_wget(url: str, outdir: Path): + ensure_dir(outdir) + cmd = [ + 'wget', + '--mirror', + '--convert-links', + '--adjust-extension', + '--page-requisites', + '--no-parent', + '--directory-prefix', + str(outdir), + url, + ] + proc = subprocess.run(cmd, capture_output=True, text=True) + return { + 'returncode': proc.returncode, + 'stdout': proc.stdout[-4000:], + 'stderr': proc.stderr[-4000:], + } + + +def collect_files(root: Path): + return sorted([p for p in root.rglob('*') if p.is_file()]) + + +def scan_tree(root: Path, allowed_exts): + findings = defaultdict(list) + for p in collect_files(root): + rel = str(p.relative_to(root)) + ext = p.suffix.lower().lstrip('.') + if ext and ext not in allowed_exts: + findings['unexpected_extensions'].append(rel) + + if SUSPICIOUS_NAME_RE.search(p.name): + findings['suspicious_filenames'].append(rel) + + try: + data = p.read_text(encoding='utf-8', errors='ignore') + except Exception: + continue + + for pat in SUSPICIOUS_CONTENT_PATTERNS: + if re.search(pat, data, re.I): + findings['suspicious_content'].append( + {'file': rel, 'pattern': pat} + ) + + if '(.*?)', r.text, re.I | re.S) + if m: + title = re.sub(r'\s+', ' ', m.group(1)).strip() + + internal_links = [] + external_links = [] + + for kind, ref in parser.links: + full = normalize_internal_url(final_url, ref) + if not full: + continue + item = {'kind': kind, 'url': full} + if is_same_host(start_url, full): + internal_links.append(item) + if ( + kind == 'a' + and should_crawl(full, start_url) + and full not in seen + and full not in queue + ): + queue.append(full) + else: + external_links.append(item) + externals.append({'from': final_url, **item}) + + inline_hits = [] + for idx, script in enumerate(parser.inline_scripts, 1): + for pat in SUSPICIOUS_CONTENT_PATTERNS: + if re.search(pat, script, re.I): + inline_hits.append({ + 'script_index': idx, + 'pattern': pat + }) + + pages.append({ + 'url': url, + 'final_url': final_url, + 'status': status, + 'content_type': ctype, + 'title': title, + 'inline_hits': inline_hits, + 'internal_links': internal_links, + 'external_links': external_links, + }) + except Exception as e: + page_errors.append({'url': url, 'error': str(e)}) + + return {'pages': pages, 'externals': externals, 'page_errors': page_errors} + + +def extract_remote_assets(start_url: str): + r = requests.get(start_url, headers=HEADERS, timeout=20) + r.raise_for_status() + parser = AssetParser() + parser.feed(r.text) + + assets = [] + externals = [] + + for kind, ref in parser.links: + full = clean_url(urljoin(start_url, ref)) + if is_same_host(start_url, full): + assets.append({'kind': kind, 'url': full}) + else: + externals.append({'kind': kind, 'url': full}) + + inline_hits = [] + for idx, script in enumerate(parser.inline_scripts, 1): + for pat in SUSPICIOUS_CONTENT_PATTERNS: + if re.search(pat, script, re.I): + inline_hits.append({'script_index': idx, 'pattern': pat}) + + return { + 'page_url': start_url, + 'assets': assets, + 'externals': externals, + 'inline_hits': inline_hits, + } + + +def fetch_headers(url: str): + try: + r = requests.get(url, headers=HEADERS, timeout=20) + headers = {k: v for k, v in r.headers.items()} + interesting_keys = [ + 'Content-Security-Policy', + 'Strict-Transport-Security', + 'X-Content-Type-Options', + 'X-Frame-Options', + 'Referrer-Policy', + 'Permissions-Policy', + 'Cache-Control', + 'Content-Type', + 'Server', + ] + interesting = { + k: headers.get(k) + for k in interesting_keys + if headers.get(k) is not None + } + missing = [ + k for k in [ + 'Content-Security-Policy', + 'Strict-Transport-Security', + 'X-Content-Type-Options', + 'Referrer-Policy', + ] + if k not in interesting + ] + return { + 'url': r.url, + 'status': r.status_code, + 'headers': interesting, + 'missing_recommended': missing, + } + except Exception as e: + return { + 'url': url, + 'error': str(e), + 'headers': {}, + 'missing_recommended': [], + } + + +def scan_dom_text_for_indicators(html: str): + indicators = [] + checks = { + 'base_tag_present': r']+http-equiv=[\"\']refresh[\"\']', + 'iframe_present': r' 0: + score += 30 + reasons.append('unerwartete Dateiendungen gefunden') + + if report['summary'].get('suspicious_filenames', 0) > 0: + score += 40 + reasons.append('verdächtige Dateinamen gefunden') + + if report['summary'].get('suspicious_content', 0) > 0: + score += 50 + reasons.append('verdächtige Inhaltsmuster gefunden') + + if report['summary'].get('dom_indicators', 0) > 0: + score += 10 + reasons.append('DOM-Indikatoren erkannt') + + if report['summary'].get('header_missing', 0) >= 2: + score += 10 + reasons.append('mehrere empfohlene Security-Header fehlen') + + if report.get('baseline_diff'): + delta = len(report['baseline_diff'].get('added', [])) + \ + len(report['baseline_diff'].get('changed', [])) + if delta > 0: + score += min(40, delta * 5) + reasons.append('Abweichungen zur Baseline gefunden') + + if report.get('local_diff'): + delta = len(report['local_diff'].get('added_on_server', [])) + \ + len(report['local_diff'].get('changed', [])) + if delta > 0: + score += min(40, delta * 5) + reasons.append('Abweichungen zum lokalen Build gefunden') + + if score >= 70: + level = 'red' + exit_code = 2 + elif score >= 20: + level = 'yellow' + exit_code = 1 + else: + level = 'green' + exit_code = 0 + + return { + 'score': score, + 'level': level, + 'exit_code': exit_code, + 'reasons': reasons, + } + + +def send_webhook(url: str, payload: dict): + try: + r = requests.post(url, json=payload, headers=HEADERS, timeout=20) + return { + 'status_code': r.status_code, + 'ok': r.ok, + 'text': r.text[:500], + } + except Exception as e: + return {'ok': False, 'error': str(e)} + + +def write_report(outdir: Path, report: dict): + ensure_dir(outdir) + + json_path = outdir / 'audit_report.json' + md_path = outdir / 'audit_report.md' + + json_path.write_text( + json.dumps(report, indent=2, ensure_ascii=False), + encoding='utf-8' + ) + + lines = [] + lines.append("# Website-Audit\n") + lines.append(f"- Ziel: {report['target']}\n") + lines.append(f"- Mirror-Verzeichnis: `{report['mirror_dir']}`\n") + lines.append( + f"- Dateien im Mirror: {report['summary']['file_count']}\n" + ) + lines.append( + f"- Unerwartete Endungen: {report['summary']['unexpected_extensions']}\n" + ) + lines.append( + f"- Verdächtige Dateinamen: {report['summary']['suspicious_filenames']}\n" + ) + lines.append( + f"- Verdächtige Inhaltsmuster: {report['summary']['suspicious_content']}\n" + ) + lines.append( + f"- Externe Ressourcen auf Startseite: {report['summary']['external_assets']}\n" + ) + lines.append( + f"- Treffer in Inline-Skripten: {report['summary']['inline_hits']}\n" + ) + lines.append( + f"- Gecrawlte Seiten: {report['summary']['crawled_pages']}\n" + ) + lines.append( + f"- Externe Ressourcen im Crawl: {report['summary']['crawl_externals']}\n" + ) + lines.append( + f"- Crawl-Fehler: {report['summary']['crawl_errors']}\n" + ) + lines.append( + f"- Header-Warnungen: {report['summary']['header_missing']}\n" + ) + lines.append( + f"- DOM-Indikatoren auf Startseite: {report['summary']['dom_indicators']}\n" + ) + lines.append( + f"- Ampelstatus: {report['assessment']['level']}\n" + ) + lines.append( + f"- Risikoscore: {report['assessment']['score']}\n" + ) + + if report.get('assessment'): + lines.append("\n## Bewertung\n") + lines.append( + f"- Level: `{report['assessment']['level']}`" + ) + lines.append( + f"- Score: `{report['assessment']['score']}`" + ) + lines.append( + f"- Exit-Code: `{report['assessment']['exit_code']}`" + ) + for item in report['assessment'].get('reasons', []): + lines.append(f"- Grund: {item}") + + if report.get('headers'): + lines.append("\n## Header-Check\n") + if report['headers'].get('error'): + lines.append(f"- Fehler: {report['headers']['error']}") + else: + lines.append(f"- URL: {report['headers']['url']}") + lines.append(f"- Status: {report['headers']['status']}") + for k, v in report['headers']['headers'].items(): + lines.append(f"- {k}: `{v}`") + if report['headers']['missing_recommended']: + lines.append( + f"- Fehlend empfohlen: " + f"{', '.join(report['headers']['missing_recommended'])}" + ) + + if report['findings']['unexpected_extensions']: + lines.append("\n## Unerwartete Endungen\n") + for x in report['findings']['unexpected_extensions'][:200]: + lines.append(f"- `{x}`") + + if report['findings']['suspicious_filenames']: + lines.append("\n## Verdächtige Dateinamen\n") + for x in report['findings']['suspicious_filenames'][:200]: + lines.append(f"- `{x}`") + + if report['findings']['suspicious_content']: + lines.append("\n## Verdächtige Inhaltsmuster\n") + for x in report['findings']['suspicious_content'][:200]: + lines.append(f"- `{x['file']}` → `{x['pattern']}`") + + if report['remote']['externals']: + lines.append("\n## Externe Ressourcen\n") + for x in report['remote']['externals'][:200]: + lines.append(f"- {x['kind']}: {x['url']}") + + if report['remote']['inline_hits']: + lines.append("\n## Inline-Skript-Treffer\n") + for x in report['remote']['inline_hits'][:200]: + lines.append( + f"- Inline-Skript {x['script_index']} → `{x['pattern']}`" + ) + + if report.get('crawl', {}).get('pages'): + lines.append("\n## Gecrawlte Seiten\n") + for x in report['crawl']['pages'][:200]: + title = f" — {x['title']}" if x.get('title') else "" + lines.append( + f"- {x['status']} {x['final_url']}{title}" + ) + + if report.get('crawl', {}).get('externals'): + lines.append("\n## Externe Ressourcen im Crawl\n") + for x in report['crawl']['externals'][:200]: + lines.append( + f"- {x['from']} -> {x['kind']}: {x['url']}" + ) + + if report.get('dom_indicators'): + lines.append("\n## DOM-Indikatoren\n") + for item in report['dom_indicators']: + lines.append(f"- {item}") + + if report.get('baseline_diff'): + lines.append("\n## Baseline-Diff\n") + lines.append( + f"- Baseline: `{report['baseline_diff']['baseline_source']}`" + ) + for key in ('added', 'removed', 'changed'): + if report['baseline_diff'][key]: + lines.append( + f"- {key}: {len(report['baseline_diff'][key])}" + ) + for item in report['baseline_diff'][key][:200]: + lines.append(f" - `{item}`") + + if report.get('local_diff'): + lines.append("\n## Vergleich mit lokalem Build\n") + lines.append( + f"- Lokal: `{report['local_diff']['local_dir']}`" + ) + for key in ( + 'added_on_server', + 'missing_on_server', + 'changed', + ): + if report['local_diff'][key]: + lines.append( + f"- {key}: {len(report['local_diff'][key])}" + ) + for item in report['local_diff'][key][:200]: + lines.append(f" - `{item}`") + + md_path.write_text('\n'.join(lines), encoding='utf-8') + baseline_path = export_baseline(outdir, report) + return json_path, md_path, baseline_path + +def main(): + ap = argparse.ArgumentParser( + description='Mirror + Scan für statische Websites' + ) + ap.add_argument('url') + ap.add_argument('--out', default='output/site_audit') + ap.add_argument('--allow-ext', action='append', default=[]) + ap.add_argument('--max-pages', type=int, default=200) + ap.add_argument('--baseline-json', default=None) + ap.add_argument('--local-dir', default=None) + ap.add_argument('--webhook-url', default=None) + ap.add_argument('--fail-on-yellow', action='store_true') + args = ap.parse_args() + + outdir = Path(args.out) + mirror = outdir / 'mirror' + + allowed = set(DEFAULT_ALLOWED) + for item in args.allow_ext: + for part in item.split(','): + part = part.strip().lower().lstrip('.') + if part: + allowed.add(part) + + wget_result = mirror_with_wget(args.url, mirror) + remote = extract_remote_assets(args.url) + crawl = crawl_site(args.url, max_pages=args.max_pages) + headers_report = fetch_headers(args.url) + html = requests.get(args.url, headers=HEADERS, timeout=20).text + dom_indicators = scan_dom_text_for_indicators(html) + + file_root_candidates = [ + p for p in mirror.rglob('*') + if p.is_dir() and p.name == urlparse(args.url).netloc + ] + file_root = file_root_candidates[0] if file_root_candidates else mirror + + findings = scan_tree(file_root, allowed) + files = collect_files(file_root) + hashes = { + str(p.relative_to(file_root)): sha256_file(p) + for p in files + } + + baseline_diff = ( + compare_with_baseline(hashes, Path(args.baseline_json)) + if args.baseline_json else None + ) + local_diff = ( + compare_with_local_dir(file_root, Path(args.local_dir)) + if args.local_dir else None + ) + + report = { + 'target': args.url, + 'mirror_dir': str(file_root), + 'wget': wget_result, + 'summary': { + 'file_count': len(files), + 'unexpected_extensions': + len(findings['unexpected_extensions']), + 'suspicious_filenames': + len(findings['suspicious_filenames']), + 'suspicious_content': + len(findings['suspicious_content']), + 'external_assets': len(remote['externals']), + 'inline_hits': len(remote['inline_hits']), + 'crawled_pages': len(crawl['pages']), + 'crawl_externals': len(crawl['externals']), + 'crawl_errors': len(crawl['page_errors']), + 'header_missing': + len(headers_report.get('missing_recommended', [])), + 'dom_indicators': len(dom_indicators), + }, + 'findings': findings, + 'remote': remote, + 'crawl': crawl, + 'headers': headers_report, + 'dom_indicators': dom_indicators, + 'baseline_diff': baseline_diff, + 'local_diff': local_diff, + 'hashes': hashes, + } + + assessment = score_report(report) + if args.fail_on_yellow and assessment['level'] == 'yellow': + assessment['exit_code'] = 2 + report['assessment'] = assessment + + if args.webhook_url: + report['webhook_result'] = send_webhook( + args.webhook_url, + { + 'target': report['target'], + 'summary': report['summary'], + 'assessment': report['assessment'], + }, + ) + + write_report(outdir, report) + sys.exit(report['assessment']['exit_code']) + + +if __name__ == '__main__': + main() + diff --git a/README.md b/README.md new file mode 100644 index 0000000..01fb2be --- /dev/null +++ b/README.md @@ -0,0 +1,203 @@ +# Website Integrity Scanner + +Unabhängiges Überwachungswerkzeug für statische Websites gegen SEO-Spam +und unbefugte Manipulationen. Läuft im Verantwortungsbereich des Betreibers, +nicht beim Webhoster. + +## Schnellstart + +```bash +cd integrity_scanner_fuer_statische_Webseiten +python -m venv venv +source venv/bin/activate +pip install -r requirements.txt + +# 1. Erste Baseline anlegen +python -m scanner init + +# 2. Ergebnis prüfen, dann freigeben +python -m scanner approve --all --note "Initiale Baseline nach Sichtprüfung" + +# 3. Täglicher Betrieb (oder per Cron) +python -m scanner scan +``` + +--- + +## Subcommands + +| Befehl | Funktion | +|---|---| +| `init` | Erstcrawl + Vorschau, kein Baseline-Update | +| `crawl` | Crawlt die Website, speichert Snapshot | +| `check` | Vergleicht letzten Snapshot mit Baseline | +| `scan` | `crawl` + `check` in einem (für Cron) | +| `approve --all` | Alle Snapshot-URLs als neue Baseline freigeben | +| `approve --url URL` | Einzelne URL freigeben | +| `approve --rebuild` | Gesamte Baseline ersetzen (nach großer Änderung) | +| `report` | Letzten Report anzeigen | +| `status` | Baseline-Datum, letzter Scan, Risiko-Level | + +Alle Befehle akzeptieren `--config path/to/config.yaml` und `--verbose`. + +--- + +## Workflow: Erstinitialisierung + +``` +python -m scanner init +``` + +Ausgabe zeigt: +- alle gefundenen internen URLs +- alle externen Links (mit Whitelist-Status) +- potenzielle Auffälligkeiten (Hidden Content, Meta-Refresh, ...) + +**Wichtig**: Die Ausgabe manuell prüfen, bevor die Baseline gesetzt wird. +Insbesondere externe Links in `config/allowed_external.yaml` eintragen. + +``` +python -m scanner approve --all --note "Initiale Baseline, geprüft am $(date)" +``` + +--- + +## Workflow: Täglicher Betrieb (Cron) + +``` +python -m scanner scan +``` + +Exit-Codes: +- `0` = Grün (keine Auffälligkeiten) +- `1` = Gelb (Warnung, manuelle Prüfung empfohlen) +- `2` = Rot (Alarm, sofortige Prüfung notwendig) + +--- + +## Workflow: Legitime Änderung freigeben + +```bash +# 1. Scan durchgeführt, Warnung auf /impressum/ +python -m scanner report # Report ansehen +python -m scanner approve --url https://bredelar.info/impressum/ --note "Impressum aktualisiert" + +# 2. Oder alle Änderungen auf einmal (wenn alles geprüft) +python -m scanner approve --all --note "Redesign vom 2026-06-12 freigegeben" + +# 3. Nach großem Redesign: komplette neue Baseline +python -m scanner approve --rebuild --note "Komplettes Redesign, alle Seiten neu" +``` + +--- + +## Workflow: Incident (Alarm) + +1. Report lesen: `python -m scanner report` +2. Verdächtige URLs manuell im Browser prüfen +3. Quelltext der betroffenen Seiten ansehen (insb. `display:none`-Bereiche) +4. Dienstleister kontaktieren, Zugriff auf Hosting prüfen +5. Nach Bereinigung: `python -m scanner scan` — muss grün sein +6. Dann: `python -m scanner approve --rebuild --note "Nach Incident bereinigt"` + +**Wichtig**: Niemals `approve` auf eine kompromittierte Seite anwenden. + +--- + +## Cron-Setup + +```cron +# Täglich 03:00 Uhr +0 3 * * * cd /opt/scanner && /opt/scanner/venv/bin/python -m scanner scan >> /opt/scanner/logs/cron.log 2>&1 +``` + +Log-Rotation (`/etc/logrotate.d/scanner`): +``` +/opt/scanner/logs/*.log { + daily + rotate 30 + compress + missingok + notifempty +} +``` + +--- + +## Konfiguration + +### E-Mail aktivieren + +In `config.yaml`: +```yaml +alerting: + email: + enabled: true + smtp_host: mail.example.com + smtp_port: 587 + smtp_user: scanner@example.com + smtp_password_env: SCANNER_SMTP_PASSWORD + from: scanner@bredelar.info + to: + - admin@bredelar.info +``` + +Passwort als Umgebungsvariable setzen (nicht in config.yaml): +```bash +export SCANNER_SMTP_PASSWORD="geheimesPasswort" +``` + +### Webhook aktivieren + +```yaml +alerting: + webhook: + enabled: true + url: "https://hooks.slack.com/services/..." +``` + +### Externe Domains whitelisten + +Nach dem ersten `init` alle legitimen externen Domains in +`config/allowed_external.yaml` eintragen: + +```yaml +- fonts.googleapis.com +- maps.googleapis.com +``` + +--- + +## Verzeichnisstruktur + +``` +data/ + baseline/ Aktuell freigegebene Referenz (NIEMALS automatisch aktualisiert) + manifest.json Wer/was/wann freigegeben hat + pages/ Pro URL eine JSON-Datei + snapshots/ Zeitgestempelte Crawl-Ergebnisse + YYYYMMDD_HHMMSS/ +reports/ Generierte Reports (JSON + Markdown) +logs/ scanner.log +config/ + allowed_external.yaml + allowed_paths.yaml + ignore_rules.yaml +``` + +--- + +## Tests + +```bash +pytest tests/ -v +``` + +--- + +## Erweiterungen (optional, nicht implementiert) + +- Ollama-Integration: Neue Textblöcke semantisch auf Spam-Kategorien prüfen +- Telegram-Bot: direkter Alert-Kanal +- Diff-Anzeige im Terminal: `python -m scanner report --diff-only` +- Automatische Archivierung alter Snapshots nach N Tagen diff --git a/config.yaml b/config.yaml new file mode 100644 index 0000000..25ec957 --- /dev/null +++ b/config.yaml @@ -0,0 +1,112 @@ +# Integrity Scanner — Hauptkonfiguration +# Alle Werte sind optional; fehlende werden aus den Defaults übernommen. + +target: "https://bredelar.info" +user_agent: "integrity-scanner/1.0 (+security-monitoring)" +request_timeout: 20 + +data_dir: "data" +reports_dir: "reports" +logs_dir: "logs" +config_dir: "config" + +crawl: + max_pages: 200 + delay_seconds: 1.0 + respect_robots_txt: false + skip_extensions: + - ".jpg" + - ".jpeg" + - ".png" + - ".gif" + - ".webp" + - ".svg" + - ".ico" + - ".pdf" + - ".zip" + - ".woff" + - ".woff2" + - ".ttf" + - ".otf" + - ".eot" + - ".mp4" + - ".mp3" + - ".webm" + - ".avi" + - ".mov" + +scoring: + new_external_domain: 50 + new_internal_url: 30 + missing_internal_url: 20 + hidden_content: 40 + unexpected_canonical: 35 + meta_refresh: 40 + unexpected_jsonld: 35 + large_text_addition: 20 + new_inline_script_suspicious: 30 + missing_security_header: 5 + suspicious_content_pattern: 50 + comment_links: 25 + noscript_links: 25 + new_external_link_unlisted: 40 + +thresholds: + yellow: 20 # ab diesem Score: Warnung (gelb) + red: 60 # ab diesem Score: Alarm (rot) + large_text_block_chars: 200 # Textblock ab dieser Länge wird als „groß" gewertet + +alerting: + min_level: "yellow" # "yellow" oder "red" + email: + enabled: false + smtp_host: "localhost" + smtp_port: 587 + smtp_tls: true + smtp_user: "" + smtp_password_env: "SCANNER_SMTP_PASSWORD" # Passwort aus Umgebungsvariable + from: "scanner@bredelar.info" + to: + - "admin@bredelar.info" + webhook: + enabled: false + url: "" # z.B. Telegram-Bot-Webhook, Slack-Incoming-Webhook + +normalization: + strip_html_comments: true + collapse_whitespace: true + # Selektoren, die vor dem Diff ignoriert werden (z.B. Datums-/Cache-Elemente) + ignore_selectors: + # - "#last-modified" + # - ".timestamp" + # Regex-Muster im Text, die vor dem Diff entfernt werden + ignore_patterns: + # - '\d{1,2}\.\d{1,2}\.\d{4}' # Datumsangaben + # - 'Stand:\s+\S+' + +# Erlaubte @type-Werte in JSON-LD (alles andere löst Alarm aus) +allowed_jsonld_types: + - "Organization" + - "WebSite" + - "WebPage" + - "Article" + - "BreadcrumbList" + - "ItemList" + - "LocalBusiness" + - "Place" + - "Person" + - "Event" + - "FAQPage" + - "Question" + - "Answer" + - "ImageObject" + - "SiteLinksSearchBox" + - "ContactPage" + - "AboutPage" + +# Security-Header, die auf jeder Seite erwartet werden +security_headers: + - "Content-Security-Policy" + - "Strict-Transport-Security" + - "X-Content-Type-Options" + - "Referrer-Policy" diff --git a/config/allowed_external.yaml b/config/allowed_external.yaml new file mode 100644 index 0000000..ae0a9e4 --- /dev/null +++ b/config/allowed_external.yaml @@ -0,0 +1,18 @@ +# Whitelist erlaubter externer Domains +# +# Jede externe Domain, die auf der Website verlinkt ist, muss hier eingetragen sein. +# Neue externe Domains, die NICHT hier stehen, lösen einen Alarm aus. +# +# Format: eine Domain pro Zeile (ohne Protokoll) +# Beispiele: + +# Häufig verwendete externe Ressourcen — nur eintragen, was wirklich genutzt wird: +# - fonts.googleapis.com +# - fonts.gstatic.com +# - cdnjs.cloudflare.com +# - maps.googleapis.com +# - www.google-analytics.com +# - www.googletagmanager.com + +# Für bredelar.info: Bitte nach erstem `init` prüfen und befüllen. +# Vorerst leer = alle externen Links lösen Alarm aus. diff --git a/config/allowed_paths.yaml b/config/allowed_paths.yaml new file mode 100644 index 0000000..464924b --- /dev/null +++ b/config/allowed_paths.yaml @@ -0,0 +1,17 @@ +# Whitelist erlaubter interner URL-Pfade (optional) +# +# Wenn diese Liste befüllt ist, lösen interne Links auf nicht gelistete +# Pfade einen Alarm aus. +# Leer lassen = keine Pfad-Whitelist-Prüfung (nur Baseline-Diff zählt). +# +# Format: ein Pfad pro Zeile (absoluter Pfad, ohne Domain) +# Beispiel für bredelar.info: +# +# - / +# - /index.html +# - /kontakt/ +# - /kontakt/index.html +# - /impressum/ +# - /datenschutz/ +# - /ueber-uns/ +# - /aktuelles/ diff --git a/config/ignore_rules.yaml b/config/ignore_rules.yaml new file mode 100644 index 0000000..81b9ec8 --- /dev/null +++ b/config/ignore_rules.yaml @@ -0,0 +1,23 @@ +# Normalisierungsregeln für den Diff +# +# Diese Einstellungen reduzieren Fehlalarme durch harmlose, automatisch +# generierte Inhaltsänderungen. + +# CSS-Selektoren, die vor dem Diff aus dem DOM entfernt werden. +# Nützlich für Elemente mit automatischen Zeitstempeln oder Zählern. +ignore_selectors: + # - "#last-modified" + # - ".timestamp" + # - "[data-cache-buster]" + # - ".cookie-banner" + +# Reguläre Ausdrücke, die im normalisierten Text durch [IGNORED] ersetzt werden. +# Nützlich für Datumsangaben, Versionsnummern, etc. +ignore_patterns: + # Datumsformate (deutsch: 12.06.2026 oder 12. Juni 2026) + # - '\b\d{1,2}\.\s*(?:Januar|Februar|März|April|Mai|Juni|Juli|August|September|Oktober|November|Dezember)\s+\d{4}\b' + # - '\b\d{1,2}\.\d{1,2}\.\d{4}\b' + # "Stand: ..." Angaben + # - 'Stand:\s+\S+' + # Cache-Buster in URLs (werden im HTML-Text sichtbar) + # - '\?v=[\w.]+\b' diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..f206e24 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,5 @@ +requests>=2.31 +beautifulsoup4>=4.12 +lxml>=5.0 +PyYAML>=6.0 +pytest>=7.0 diff --git a/scanner/__init__.py b/scanner/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/scanner/__main__.py b/scanner/__main__.py new file mode 100644 index 0000000..82e3035 --- /dev/null +++ b/scanner/__main__.py @@ -0,0 +1,672 @@ +""" +integrity-scanner — CLI entry point. + +Usage: + python -m scanner [options] + +Subcommands: + init First-time crawl + baseline setup workflow + crawl Crawl site, save snapshot (no comparison) + check Compare latest snapshot against baseline, report + alert + scan crawl + check in one step (use this for cron) + approve Approve snapshot as new baseline + report Display / export latest report + status Show baseline age, last scan, open changes +""" +import argparse +import json +import logging +import sys +from datetime import datetime, timezone +from pathlib import Path +from urllib.parse import urlparse + +from .alerter import send_alert +from .baseline import BaselineManager +from .checker import ( + canonical_is_hijacked, + check_jsonld_types, + check_links_against_whitelist, + check_security_headers, + is_suspicious_url, + load_whitelist, +) +from .config import load_config, resolve_paths +from .crawler import Crawler +from .differ import compare_snapshots, normalize_text, score_diff +from .extractor import extract_page + + +# --------------------------------------------------------------------------- +# Logging +# --------------------------------------------------------------------------- + +def _setup_logging(logs_dir: str, verbose: bool) -> None: + level = logging.DEBUG if verbose else logging.INFO + logs_path = Path(logs_dir) + logs_path.mkdir(parents=True, exist_ok=True) + log_file = logs_path / "scanner.log" + + fmt = "%(asctime)s %(levelname)-8s %(name)s: %(message)s" + handlers: list[logging.Handler] = [ + logging.StreamHandler(sys.stderr), + logging.FileHandler(log_file, encoding="utf-8"), + ] + logging.basicConfig(level=level, format=fmt, handlers=handlers, force=True) + + +# --------------------------------------------------------------------------- +# Crawl + extract helpers +# --------------------------------------------------------------------------- + +def _crawl_and_extract(cfg: dict) -> tuple[list[dict], list[dict]]: + """Run crawler, extract each HTML page. Returns (pages, errors).""" + logger = logging.getLogger(__name__) + crawler = Crawler(cfg) + raw = crawler.crawl() + + pages: list[dict] = [] + for raw_page in raw["pages"]: + if raw_page.get("html"): + extracted = extract_page(raw_page["html"], raw_page["url"], cfg) + page = {**raw_page, **extracted} + else: + page = dict(raw_page) + pages.append(page) + + logger.info("Extracted %d pages", len(pages)) + return pages, raw["errors"] + + +# --------------------------------------------------------------------------- +# Whitelist loading +# --------------------------------------------------------------------------- + +def _load_whitelists(cfg: dict) -> tuple[set[str], set[str]]: + config_dir = Path(cfg["config_dir"]) + allowed_ext = load_whitelist(config_dir / "allowed_external.yaml") + allowed_int = load_whitelist(config_dir / "allowed_paths.yaml") + return allowed_ext, allowed_int + + +# --------------------------------------------------------------------------- +# Report building +# --------------------------------------------------------------------------- + +def _build_report( + target: str, + diff: dict, + assessment: dict, + whitelist_violations: list[dict], + missing_sec_headers: dict, + crawl_errors: list[dict], + snap_dir: Path, +) -> dict: + return { + "generated_at": datetime.now(timezone.utc).isoformat(), + "target": target, + "snapshot_dir": str(snap_dir), + "assessment": assessment, + "diff": diff, + "whitelist_violations": whitelist_violations, + "missing_security_headers": missing_sec_headers, + "crawl_errors": crawl_errors, + } + + +def _write_report(report: dict, reports_dir: str) -> tuple[Path, Path]: + ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") + out = Path(reports_dir) / ts + out.mkdir(parents=True, exist_ok=True) + + json_path = out / "report.json" + md_path = out / "report.md" + + json_path.write_text( + json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8" + ) + md_path.write_text(_render_markdown(report), encoding="utf-8") + return json_path, md_path + + +def _render_markdown(r: dict) -> str: + a = r.get("assessment", {}) + d = r.get("diff", {}) + level = a.get("level", "?").upper() + score = a.get("score", 0) + level_badge = {"GREEN": "[OK]", "YELLOW": "[WARNUNG]", "RED": "[ALARM]"}.get(level, level) + + lines = [ + f"# Integrity Report — {r.get('target', '?')}", + "", + f"**Erstellt:** {r.get('generated_at', '?')} ", + f"**Level:** {level_badge} ", + f"**Score:** {score}", + "", + "## Zusammenfassung", + "", + f"| Kennzahl | Wert |", + f"|---|---|", + f"| Geprüfte Seiten | {d.get('total_pages_checked', '?')} |", + f"| Geänderte Seiten | {d.get('changed_pages', 0)} |", + f"| Neue interne URLs | {len(d.get('new_internal_urls', []))} |", + f"| Fehlende URLs | {len(d.get('missing_internal_urls', []))} |", + f"| Neue externe Domains | {len(d.get('new_external_domains', []))} |", + f"| Whitelist-Verstösse | {len(r.get('whitelist_violations', []))} |", + f"| Crawl-Fehler | {len(r.get('crawl_errors', []))} |", + "", + "## Risikobewertung", + "", + ] + for reason in a.get("reasons", []): + lines.append(f"- {reason}") + if not a.get("reasons"): + lines.append("- Keine Auffälligkeiten.") + + if d.get("new_external_domains"): + lines += ["", "## Neue externe Domains", ""] + for dom in d["new_external_domains"]: + lines.append(f"- `{dom}`") + + if d.get("new_internal_urls"): + lines += ["", "## Neue interne URLs", ""] + for url in d["new_internal_urls"]: + lines.append(f"- {url}") + + if d.get("missing_internal_urls"): + lines += ["", "## Fehlende URLs (waren in Baseline)", ""] + for url in d["missing_internal_urls"]: + lines.append(f"- {url}") + + if d.get("page_diffs"): + lines += ["", "## Geänderte Seiten", ""] + for pd in d["page_diffs"]: + lines += [f"### {pd['url']}", ""] + if pd.get("added_chars"): + lines.append(f"- Neuer Text: +{pd['added_chars']} Zeichen") + if pd.get("meta_diff"): + for field, change in pd["meta_diff"].items(): + if isinstance(change, dict) and "old" in change: + lines.append(f"- {field}: `{change['old']}` → `{change['new']}`") + else: + lines.append(f"- {field}: {change}") + if pd.get("canonical_changed"): + lines.append( + f"- Canonical geändert: `{pd.get('old_canonical')}` → `{pd.get('new_canonical')}`" + ) + if pd.get("new_hidden_content"): + for h in pd["new_hidden_content"]: + lines.append(f"- Hidden Content ({h.get('type')}): `{', '.join(h.get('indicators', []))}`") + if pd.get("new_meta_refresh"): + for mr in pd["new_meta_refresh"]: + lines.append(f"- Meta-Refresh → `{mr}`") + if pd.get("new_comment_links"): + for cl in pd["new_comment_links"]: + lines.append(f"- Kommentar-Link: `{cl}`") + if pd.get("new_inline_scripts"): + for s in pd["new_inline_scripts"]: + lines.append(f"- Inline-Script: `{', '.join(s.get('patterns', []))}`") + if pd.get("link_diff"): + for ltype, ld in pd["link_diff"].items(): + for added in ld.get("added", []): + lines.append(f"- {ltype} hinzugefügt: `{added}`") + for removed in ld.get("removed", []): + lines.append(f"- {ltype} entfernt: `{removed}`") + if pd.get("text_diff"): + lines += ["", "```diff"] + lines += pd["text_diff"][:40] + if len(pd["text_diff"]) > 40: + lines.append(f"... ({len(pd['text_diff']) - 40} weitere Zeilen)") + lines.append("```") + lines.append("") + + if d.get("new_pages_analysis"): + lines += ["", "## Neue Seiten (Erstanalyse)", ""] + for npa in d["new_pages_analysis"]: + problems = ( + npa.get("hidden_content") + or npa.get("inline_scripts") + or npa.get("meta_refresh") + or npa.get("unexpected_jsonld") + or npa.get("comment_links") + ) + flag = " [VERDAECHTIG]" if problems else "" + lines.append(f"- {npa['url']}{flag}") + + if r.get("whitelist_violations"): + lines += ["", "## Whitelist-Verstösse", ""] + for v in r["whitelist_violations"][:50]: + lines.append(f"- [{v.get('type')}] {v.get('url', '')} (Domain: {v.get('domain', '')})") + + if r.get("missing_security_headers"): + lines += ["", "## Fehlende Security-Header", ""] + for url, hdrs in r["missing_security_headers"].items(): + for h in hdrs: + lines.append(f"- `{url}`: {h}") + + if r.get("crawl_errors"): + lines += ["", "## Crawl-Fehler", ""] + for e in r["crawl_errors"][:20]: + lines.append(f"- `{e.get('url', '?')}`: {e.get('error', '?')}") + + lines += [ + "", + "---", + "", + "## Nächste Schritte", + "", + ] + lev = a.get("level", "green") + if lev == "red": + lines += [ + "1. **Sofort**: Website-Backend überprüfen, Dateien mit Backup vergleichen.", + "2. Dienstleister kontaktieren.", + "3. Nach Bereinigung: `python -m scanner approve --rebuild`", + ] + elif lev == "yellow": + lines += [ + "1. Alle markierten URLs manuell prüfen.", + "2. Legitime Änderungen freigeben: `python -m scanner approve --url `", + "3. Alle freigeben (wenn geprüft): `python -m scanner approve --all`", + ] + else: + lines += ["Keine Aktion erforderlich."] + + return "\n".join(lines) + "\n" + + +def _latest_report_path(reports_dir: str) -> Path | None: + rd = Path(reports_dir) + if not rd.exists(): + return None + dirs = sorted(d for d in rd.iterdir() if d.is_dir()) + if not dirs: + return None + return dirs[-1] / "report.md" + + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + +def cmd_init(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.init") + print("=== Initialisierung: Erster Crawl ===") + print(f"Ziel: {cfg['target']}") + + pages, errors = _crawl_and_extract(cfg) + if not pages: + print("FEHLER: Kein Seiten gecrawlt.", file=sys.stderr) + return 1 + + bm = BaselineManager(cfg["data_dir"]) + snap_dir = bm.save_snapshot(pages) + print(f"Snapshot gespeichert: {snap_dir}") + print(f" Seiten: {len(pages)}") + print(f" Fehler: {len(errors)}") + print() + + allowed_ext, allowed_int = _load_whitelists(cfg) + base_netloc = urlparse(cfg["target"]).netloc + + print("=== Gefundene interne URLs ===") + for p in pages: + print(f" {p['url']}") + + print() + print("=== Externe Links (alle) ===") + seen_domains: set[str] = set() + for p in pages: + for ltype, links in p.get("links", {}).items(): + if not isinstance(links, list): + continue + for l in links: + if isinstance(l, dict) and l.get("class") == "external": + domain = urlparse(l.get("url", "")).netloc + if domain and domain not in seen_domains: + seen_domains.add(domain) + in_wl = " [in Whitelist]" if domain in allowed_ext else " [NICHT in Whitelist]" + print(f" {domain}{in_wl}") + + print() + auffaelligkeiten = 0 + for p in pages: + if p.get("hidden_content"): + print(f" [!] Hidden Content auf {p['url']}: {len(p['hidden_content'])} Fund(e)") + auffaelligkeiten += len(p["hidden_content"]) + if p.get("inline_scripts"): + print(f" [!] Verdächtige Inline-Scripts auf {p['url']}: {len(p['inline_scripts'])} Fund(e)") + auffaelligkeiten += len(p["inline_scripts"]) + if p.get("links", {}).get("meta_refresh"): + print(f" [!] Meta-Refresh auf {p['url']}") + auffaelligkeiten += 1 + if p.get("comment_links"): + print(f" [!] Links in HTML-Kommentaren auf {p['url']}") + auffaelligkeiten += len(p["comment_links"]) + + if errors: + print() + print("=== Crawl-Fehler ===") + for e in errors: + print(f" {e['url']}: {e['error']}") + + print() + if auffaelligkeiten: + print(f"[WARNUNG] {auffaelligkeiten} potenzielle Auffälligkeit(en) gefunden.") + print(" Bitte alle URLs oben manuell prüfen, bevor Sie freigeben.") + else: + print("[OK] Keine offensichtlichen Auffälligkeiten gefunden.") + + print() + print("=== Naechste Schritte ===") + print(" 1. Prüfen Sie alle oben aufgelisteten externen Links und Auffälligkeiten.") + print(" 2. Wenn alles sauber ist, Baseline freigeben:") + print(" python -m scanner approve --all") + print(" 3. Danach ist der tägliche Betrieb möglich:") + print(" python -m scanner scan") + return 0 + + +def cmd_crawl(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.crawl") + print(f"Crawle {cfg['target']} ...") + pages, errors = _crawl_and_extract(cfg) + bm = BaselineManager(cfg["data_dir"]) + snap_dir = bm.save_snapshot(pages) + print(f"Snapshot: {snap_dir}") + print(f" Seiten: {len(pages)}") + print(f" Fehler: {len(errors)}") + return 0 + + +def cmd_check(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.check") + bm = BaselineManager(cfg["data_dir"]) + + if not bm.baseline_exists(): + print( + "FEHLER: Keine Baseline vorhanden. Zuerst `python -m scanner init` ausführen.", + file=sys.stderr, + ) + return 1 + + snap = bm.load_snapshot() + if not snap: + print( + "FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.", + file=sys.stderr, + ) + return 1 + + baseline = bm.load_baseline() + logger.info( + "Vergleiche %d Baseline-Seiten mit %d Snapshot-Seiten ...", + len(baseline["pages"]), + len(snap["pages"]), + ) + + diff = compare_snapshots(baseline["pages"], snap["pages"], cfg) + + # Whitelist checks on all current pages + allowed_ext, allowed_int = _load_whitelists(cfg) + base_netloc = urlparse(cfg["target"]).netloc + wl_violations: list[dict] = [] + missing_sec: dict = {} + + for page in snap["pages"].values(): + violations = check_links_against_whitelist(page, allowed_ext, base_netloc) + if violations: + wl_violations.extend({"page": page["url"], **v} for v in violations) + + hdrs = page.get("response_headers", {}) + missing = check_security_headers(hdrs, cfg.get("security_headers", [])) + if missing: + missing_sec[page["url"]] = missing + + # Unexpected JSON-LD types (absolute check, not just diff-based) + unexp = check_jsonld_types(page, cfg.get("allowed_jsonld_types", [])) + for j in unexp: + if not any( + pd.get("url") == page["url"] and j in pd.get("new_jsonld", []) + for pd in diff.get("page_diffs", []) + ): + # Add as new_jsonld to the page_diff if not already there + pd_for_url = next( + (pd for pd in diff.get("page_diffs", []) if pd["url"] == page["url"]), + None, + ) + if pd_for_url: + pd_for_url.setdefault("new_jsonld", []).append({**j, "unexpected": True}) + + # Score with whitelist-violation contribution + for v in wl_violations: + if v.get("type") == "unlisted_external_domain": + existing = diff.get("new_external_domains", []) + domain = v.get("domain", "") + if domain and domain not in existing: + existing.append(domain) + diff["new_external_domains"] = existing + + assessment = score_diff(diff, cfg) + + crawl_errors: list[dict] = [] + if getattr(args, "with_errors", False): + # errors were not persisted in snapshot, skip + pass + + report = _build_report( + cfg["target"], diff, assessment, wl_violations, missing_sec, + crawl_errors, snap["dir"], + ) + json_path, md_path = _write_report(report, cfg["reports_dir"]) + logger.info("Report: %s", md_path) + + _print_summary(assessment, diff) + + send_alert(report, cfg) + + return assessment["exit_code"] + + +def cmd_scan(args: argparse.Namespace, cfg: dict) -> int: + """crawl + check.""" + logger = logging.getLogger("scanner.scan") + print(f"Scanne {cfg['target']} ...") + pages, errors = _crawl_and_extract(cfg) + bm = BaselineManager(cfg["data_dir"]) + snap_dir = bm.save_snapshot(pages) + logger.info("Snapshot: %s (%d Seiten, %d Fehler)", snap_dir, len(pages), len(errors)) + + if not bm.baseline_exists(): + print() + print("Noch keine Baseline vorhanden — Scan abgeschlossen, kein Vergleich.") + print("Führen Sie aus: python -m scanner approve --all") + return 0 + + return cmd_check(args, cfg) + + +def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: + logger = logging.getLogger("scanner.approve") + bm = BaselineManager(cfg["data_dir"]) + snap_dir = Path(args.snapshot) if getattr(args, "snapshot", None) else None + note = getattr(args, "note", "") or "" + + if getattr(args, "rebuild", False): + approved = bm.rebuild_baseline(snap_dir=snap_dir, note=note) + print(f"Baseline komplett neu aufgebaut: {len(approved)} URLs freigegeben.") + return 0 + + if getattr(args, "all", False): + approved = bm.approve_all(snap_dir=snap_dir, note=note) + print(f"Alle {len(approved)} URLs freigegeben.") + for u in approved: + print(f" + {u}") + return 0 + + url = getattr(args, "url", None) + if url: + ok = bm.approve_url(url, snap_dir=snap_dir, note=note) + if ok: + print(f"Freigegeben: {url}") + return 0 + else: + print(f"FEHLER: URL nicht im Snapshot gefunden: {url}", file=sys.stderr) + return 1 + + print("FEHLER: --url, --all oder --rebuild angeben.", file=sys.stderr) + return 1 + + +def cmd_report(args: argparse.Namespace, cfg: dict) -> int: + fmt = getattr(args, "format", "md") + rd = Path(cfg["reports_dir"]) + if not rd.exists(): + print("Noch keine Reports vorhanden.") + return 0 + + dirs = sorted(d for d in rd.iterdir() if d.is_dir()) + if not dirs: + print("Noch keine Reports vorhanden.") + return 0 + + latest = dirs[-1] + if fmt == "json": + p = latest / "report.json" + else: + p = latest / "report.md" + + if not p.exists(): + print(f"Report-Datei nicht gefunden: {p}", file=sys.stderr) + return 1 + + print(p.read_text(encoding="utf-8")) + return 0 + + +def cmd_status(args: argparse.Namespace, cfg: dict) -> int: + bm = BaselineManager(cfg["data_dir"]) + + print(f"Ziel: {cfg['target']}") + + if bm.baseline_exists(): + manifest = json.loads( + (bm.baseline_dir / "manifest.json").read_text(encoding="utf-8") + ) + print(f"Baseline: {manifest.get('approved_at', '?')} (von: {manifest.get('approved_by', '?')})") + print(f" Seiten: {len(manifest.get('urls', []))}") + if manifest.get("note"): + print(f" Notiz: {manifest['note']}") + else: + print("Baseline: KEINE — 'python -m scanner init' ausführen") + + snap_dir = bm.latest_snapshot_dir() + if snap_dir: + snap_manifest = json.loads((snap_dir / "manifest.json").read_text(encoding="utf-8")) + print(f"Letzter Scan: {snap_manifest.get('timestamp', '?')} ({snap_manifest.get('page_count', '?')} Seiten)") + else: + print("Letzter Scan: KEINER") + + rd = Path(cfg["reports_dir"]) + if rd.exists(): + dirs = sorted(d for d in rd.iterdir() if d.is_dir()) + if dirs: + rp = dirs[-1] / "report.json" + if rp.exists(): + r = json.loads(rp.read_text(encoding="utf-8")) + a = r.get("assessment", {}) + print(f"Letzter Report: {dirs[-1].name} — Level: {a.get('level','?').upper()} (Score: {a.get('score','?')})") + else: + print(f"Letzter Report: {dirs[-1].name}") + return 0 + + +# --------------------------------------------------------------------------- +# Output helper +# --------------------------------------------------------------------------- + +def _print_summary(assessment: dict, diff: dict) -> None: + level = assessment.get("level", "?").upper() + score = assessment.get("score", 0) + print() + print(f"=== Ergebnis: {level} (Score {score}) ===") + for reason in assessment.get("reasons", []): + print(f" * {reason}") + if not assessment.get("reasons"): + print(" Keine Auffälligkeiten.") + print(f" Neue URLs: {len(diff.get('new_internal_urls', []))}") + print(f" Fehlende URLs: {len(diff.get('missing_internal_urls', []))}") + print(f" Neue externe Domains:{len(diff.get('new_external_domains', []))}") + print(f" Geänderte Seiten: {diff.get('changed_pages', 0)}") + + +# --------------------------------------------------------------------------- +# Argument parser +# --------------------------------------------------------------------------- + +def _build_parser() -> argparse.ArgumentParser: + p = argparse.ArgumentParser( + prog="scanner", + description="Website-Integrity-Scanner für statische Webseiten", + ) + p.add_argument("--config", default="config.yaml", help="Pfad zur config.yaml") + p.add_argument("--verbose", "-v", action="store_true", help="Debug-Ausgabe") + + sub = p.add_subparsers(dest="command", required=True) + + sub.add_parser("init", help="Erstcrawl + Baseline-Workflow starten") + + sub.add_parser("crawl", help="Nur crawlen, Snapshot speichern") + + sub.add_parser("check", help="Snapshot vs. Baseline vergleichen") + + sub.add_parser("scan", help="crawl + check (für Cron)") + + ap = sub.add_parser("approve", help="Snapshot als neue Baseline freigeben") + ap.add_argument("--all", action="store_true", help="Alle URLs freigeben") + ap.add_argument("--url", help="Einzelne URL freigeben") + ap.add_argument("--rebuild", action="store_true", help="Baseline komplett ersetzen") + ap.add_argument("--snapshot", help="Pfad zu einem bestimmten Snapshot-Verzeichnis") + ap.add_argument("--note", default="", help="Freitext-Notiz für die Baseline-Versionierung") + + rp = sub.add_parser("report", help="Letzten Report anzeigen") + rp.add_argument("--format", choices=["md", "json"], default="md") + + sub.add_parser("status", help="Übersichtsstatus anzeigen") + + return p + + +# --------------------------------------------------------------------------- +# Main +# --------------------------------------------------------------------------- + +def main() -> None: + parser = _build_parser() + args = parser.parse_args() + + cfg = load_config(args.config) + base_dir = Path(args.config).parent.resolve() + resolve_paths(cfg, base_dir) + + _setup_logging(cfg["logs_dir"], args.verbose) + + commands = { + "init": cmd_init, + "crawl": cmd_crawl, + "check": cmd_check, + "scan": cmd_scan, + "approve": cmd_approve, + "report": cmd_report, + "status": cmd_status, + } + handler = commands.get(args.command) + if not handler: + parser.print_help() + sys.exit(1) + + sys.exit(handler(args, cfg)) + + +if __name__ == "__main__": + main() diff --git a/scanner/alerter.py b/scanner/alerter.py new file mode 100644 index 0000000..e137c1b --- /dev/null +++ b/scanner/alerter.py @@ -0,0 +1,183 @@ +""" +Alert dispatch: e-mail (smtplib) and HTTP webhook. + +Alerts are only sent when the assessment level meets or exceeds min_level. +""" +import logging +import os +import smtplib +from datetime import datetime, timezone +from email.message import EmailMessage + +import requests + +logger = logging.getLogger(__name__) + +_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2} + + +def send_alert(report: dict, cfg: dict) -> None: + """Send alert if the report's level is at or above min_level.""" + alerting = cfg.get("alerting", {}) + level = report.get("assessment", {}).get("level", "green") + min_level = alerting.get("min_level", "yellow") + + if _LEVEL_ORDER.get(level, 0) < _LEVEL_ORDER.get(min_level, 1): + logger.debug("Alert suppressed: level=%s < min_level=%s", level, min_level) + return + + subject = _make_subject(report) + body = _format_body(report) + + email_cfg = alerting.get("email", {}) + if email_cfg.get("enabled"): + _send_email(subject, body, email_cfg) + + webhook_cfg = alerting.get("webhook", {}) + if webhook_cfg.get("enabled") and webhook_cfg.get("url"): + _send_webhook(webhook_cfg["url"], report, subject) + + +# --------------------------------------------------------------------------- +# Formatting +# --------------------------------------------------------------------------- + +def _make_subject(report: dict) -> str: + level = report.get("assessment", {}).get("level", "?").upper() + target = report.get("target", "?") + ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") + return f"[{level}] Integrity Alert: {target} — {ts}" + + +def _format_body(report: dict) -> str: + lines = [ + "=" * 60, + "WEBSITE INTEGRITY ALERT", + "=" * 60, + f"Ziel: {report.get('target', '?')}", + f"Zeitpunkt: {datetime.now(timezone.utc).isoformat()}", + ] + + assessment = report.get("assessment", {}) + level = assessment.get("level", "?").upper() + score = assessment.get("score", "?") + lines += [ + f"Level: {level}", + f"Score: {score}", + "", + "--- Gruende ---", + ] + for reason in assessment.get("reasons", []): + lines.append(f" * {reason}") + + diff = report.get("diff", {}) + + _section(lines, "Neue externe Domains", diff.get("new_external_domains", []), "+") + _section(lines, "Neue interne URLs", diff.get("new_internal_urls", []), "+") + _section(lines, "Fehlende URLs", diff.get("missing_internal_urls", []), "-") + + for pd in diff.get("page_diffs", [])[:10]: + lines += ["", f"--- Aenderung: {pd['url']} ---"] + if pd.get("text_diff"): + lines.append(f" Text-Diff: {pd['added_chars']} Zeichen zugefügt") + lines += [f" {l}" for l in pd["text_diff"][:30]] + if pd.get("new_hidden_content"): + lines.append(f" Hidden Content: {len(pd['new_hidden_content'])} neue(r) Fund(e)") + if pd.get("new_meta_refresh"): + lines.append(f" Meta-Refresh: {pd['new_meta_refresh']}") + if pd.get("canonical_changed"): + lines.append(f" Canonical geaendert: {pd.get('old_canonical')} -> {pd.get('new_canonical')}") + if pd.get("new_comment_links"): + lines.append(f" Kommentar-Links: {pd['new_comment_links']}") + + lines += [ + "", + "--- Empfehlung ---", + ] + level_lower = assessment.get("level", "green") + if level_lower == "red": + lines += [ + " SOFORT HANDELN: Website moeglicherweise kompromittiert.", + " Zugriff auf Hosting-Backend prüfen, Dateien vergleichen.", + ] + elif level_lower == "yellow": + lines += [ + " PRÜFEN: Unerwartete Aenderungen gefunden.", + " Alle markierten URLs manuell kontrollieren.", + " Falls legitim: python -m scanner approve --all", + ] + else: + lines.append(" OK — nur zur Information.") + + lines += [ + "", + "--- Workflow ---", + " Details: python -m scanner report", + " Freigabe: python -m scanner approve --url ", + " Rebuild: python -m scanner approve --rebuild", + "=" * 60, + ] + return "\n".join(lines) + + +def _section(lines: list, title: str, items: list, prefix: str) -> None: + if not items: + return + lines += ["", f"--- {title} ---"] + for item in items[:50]: + lines.append(f" {prefix} {item}") + if len(items) > 50: + lines.append(f" ... und {len(items) - 50} weitere") + + +# --------------------------------------------------------------------------- +# Transport +# --------------------------------------------------------------------------- + +def _send_email(subject: str, body: str, cfg: dict) -> None: + password = os.environ.get(cfg.get("smtp_password_env", "SCANNER_SMTP_PASSWORD"), "") + recipients = cfg.get("to", []) + if not recipients: + logger.warning("E-Mail alert enabled but no recipients configured.") + return + + msg = EmailMessage() + msg["Subject"] = subject + msg["From"] = cfg.get("from", "scanner@localhost") + msg["To"] = ", ".join(recipients) + msg.set_content(body) + + host = cfg.get("smtp_host", "localhost") + port = cfg.get("smtp_port", 587) + use_tls = cfg.get("smtp_tls", True) + + try: + with smtplib.SMTP(host, port) as server: + if use_tls: + server.starttls() + user = cfg.get("smtp_user", "") + if user and password: + server.login(user, password) + server.send_message(msg) + logger.info("Alert e-mail sent to %s", recipients) + except Exception as exc: + logger.error("E-mail alert failed: %s", exc) + + +def _send_webhook(url: str, report: dict, subject: str) -> None: + payload = { + "text": subject, + "target": report.get("target"), + "level": report.get("assessment", {}).get("level"), + "score": report.get("assessment", {}).get("score"), + "reasons": report.get("assessment", {}).get("reasons", []), + "new_external_domains": report.get("diff", {}).get("new_external_domains", []), + "new_internal_urls": report.get("diff", {}).get("new_internal_urls", []), + "missing_internal_urls": report.get("diff", {}).get("missing_internal_urls", []), + "changed_pages": report.get("diff", {}).get("changed_pages", 0), + } + try: + resp = requests.post(url, json=payload, timeout=10) + logger.info("Webhook delivered: HTTP %s", resp.status_code) + except Exception as exc: + logger.error("Webhook alert failed: %s", exc) diff --git a/scanner/baseline.py b/scanner/baseline.py new file mode 100644 index 0000000..3e37162 --- /dev/null +++ b/scanner/baseline.py @@ -0,0 +1,231 @@ +""" +Snapshot and baseline management. + +Snapshots are timestamped crawl results. The baseline is a manually +approved reference state. Compromised content must never flow into the +baseline automatically — every update requires an explicit `approve` call. +""" +import hashlib +import json +import logging +from datetime import datetime, timezone +from pathlib import Path +from typing import Optional + +logger = logging.getLogger(__name__) + + +def url_key(url: str) -> str: + """Stable filename-safe key for a URL (first 16 hex chars of SHA-256).""" + return hashlib.sha256(url.encode("utf-8")).hexdigest()[:16] + + +class BaselineManager: + def __init__(self, data_dir: str | Path): + self.data_dir = Path(data_dir) + self.baseline_dir = self.data_dir / "baseline" + self.pages_dir = self.baseline_dir / "pages" + self.snapshots_dir = self.data_dir / "snapshots" + + # ------------------------------------------------------------------ + # Snapshot operations + # ------------------------------------------------------------------ + + def save_snapshot(self, pages: list[dict]) -> Path: + """ + Persist a crawl+extraction result as a timestamped snapshot. + Returns the snapshot directory path. + """ + ts = _utc_stamp() + snap_dir = self.snapshots_dir / ts + snap_pages_dir = snap_dir / "pages" + snap_pages_dir.mkdir(parents=True, exist_ok=True) + + manifest = { + "timestamp": ts, + "page_count": len(pages), + "urls": [p["url"] for p in pages], + } + (snap_dir / "manifest.json").write_text( + json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8" + ) + for page in pages: + fname = url_key(page["url"]) + ".json" + (snap_pages_dir / fname).write_text( + json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + logger.info("Snapshot saved: %s (%d pages)", snap_dir, len(pages)) + return snap_dir + + def latest_snapshot_dir(self) -> Optional[Path]: + """Return the most recent snapshot directory, or None.""" + if not self.snapshots_dir.exists(): + return None + dirs = sorted( + (d for d in self.snapshots_dir.iterdir() if d.is_dir()), + key=lambda d: d.name, + ) + return dirs[-1] if dirs else None + + def load_snapshot(self, snap_dir: Optional[Path] = None) -> dict: + """ + Load a snapshot (default: latest). + Returns {manifest, pages: {url: page_dict}, dir: Path}. + """ + if snap_dir is None: + snap_dir = self.latest_snapshot_dir() + if snap_dir is None or not snap_dir.exists(): + return {} + + manifest = json.loads((snap_dir / "manifest.json").read_text(encoding="utf-8")) + pages: dict[str, dict] = {} + pages_dir = snap_dir / "pages" + for f in pages_dir.glob("*.json"): + page = json.loads(f.read_text(encoding="utf-8")) + pages[page["url"]] = page + + return {"manifest": manifest, "pages": pages, "dir": snap_dir} + + # ------------------------------------------------------------------ + # Baseline operations + # ------------------------------------------------------------------ + + def baseline_exists(self) -> bool: + return (self.baseline_dir / "manifest.json").exists() + + def load_baseline(self) -> dict: + """ + Load the current approved baseline. + Returns {manifest, pages: {url: page_dict}}. + """ + if not self.baseline_exists(): + return {"manifest": {}, "pages": {}} + + manifest = json.loads( + (self.baseline_dir / "manifest.json").read_text(encoding="utf-8") + ) + pages: dict[str, dict] = {} + if self.pages_dir.exists(): + for f in self.pages_dir.glob("*.json"): + page = json.loads(f.read_text(encoding="utf-8")) + pages[page["url"]] = page + + return {"manifest": manifest, "pages": pages} + + # ------------------------------------------------------------------ + # Approve operations (the only way baseline content is ever updated) + # ------------------------------------------------------------------ + + def approve_all( + self, + snap_dir: Optional[Path] = None, + note: str = "", + approved_by: str = "cli", + ) -> list[str]: + """ + Approve every page in the snapshot as the new baseline. + Existing baseline pages NOT present in snapshot are kept unless + --rebuild is requested (caller responsibility to wipe baseline_dir first). + """ + snap = self.load_snapshot(snap_dir) + if not snap: + raise RuntimeError("No snapshot found to approve.") + + self.pages_dir.mkdir(parents=True, exist_ok=True) + approved: list[str] = [] + + for url, page in snap["pages"].items(): + fname = url_key(url) + ".json" + (self.pages_dir / fname).write_text( + json.dumps(page, indent=2, ensure_ascii=False), encoding="utf-8" + ) + approved.append(url) + + self._write_manifest(approved, note, approved_by, snap.get("dir")) + logger.info("Approved %d URLs into baseline.", len(approved)) + return approved + + def approve_url( + self, + url: str, + snap_dir: Optional[Path] = None, + note: str = "", + approved_by: str = "cli", + ) -> bool: + """Approve a single URL from snapshot into the baseline.""" + snap = self.load_snapshot(snap_dir) + if url not in snap.get("pages", {}): + logger.error("URL not found in snapshot: %s", url) + return False + + self.pages_dir.mkdir(parents=True, exist_ok=True) + fname = url_key(url) + ".json" + (self.pages_dir / fname).write_text( + json.dumps(snap["pages"][url], indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + # Update manifest incrementally + if self.baseline_exists(): + manifest = json.loads( + (self.baseline_dir / "manifest.json").read_text(encoding="utf-8") + ) + else: + manifest = {"urls": []} + + if url not in manifest.get("urls", []): + manifest.setdefault("urls", []).append(url) + + manifest["last_updated"] = datetime.now(timezone.utc).isoformat() + manifest["last_note"] = note + manifest["last_approved_by"] = approved_by + + (self.baseline_dir / "manifest.json").write_text( + json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8" + ) + logger.info("Approved single URL into baseline: %s", url) + return True + + def rebuild_baseline( + self, + snap_dir: Optional[Path] = None, + note: str = "", + approved_by: str = "cli", + ) -> list[str]: + """ + Wipe the entire baseline and replace it with the given snapshot. + Use when a major legitimate redesign has taken place. + """ + import shutil + if self.baseline_dir.exists(): + shutil.rmtree(self.baseline_dir) + self.pages_dir.mkdir(parents=True, exist_ok=True) + return self.approve_all(snap_dir=snap_dir, note=note, approved_by=approved_by) + + # ------------------------------------------------------------------ + # Internal helpers + # ------------------------------------------------------------------ + + def _write_manifest( + self, + urls: list[str], + note: str, + approved_by: str, + source_snap: Optional[Path], + ) -> None: + self.baseline_dir.mkdir(parents=True, exist_ok=True) + manifest = { + "approved_at": datetime.now(timezone.utc).isoformat(), + "approved_by": approved_by, + "note": note, + "source_snapshot": str(source_snap) if source_snap else "", + "urls": urls, + } + (self.baseline_dir / "manifest.json").write_text( + json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + +def _utc_stamp() -> str: + return datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S") diff --git a/scanner/checker.py b/scanner/checker.py new file mode 100644 index 0000000..1fe4afe --- /dev/null +++ b/scanner/checker.py @@ -0,0 +1,164 @@ +""" +Additional integrity checks beyond baseline diff: +- link whitelist enforcement +- security-header audit +- canonical hijack detection +- suspicious URL filenames (from v4 heritage) +""" +import re +from pathlib import Path +from urllib.parse import urlparse + +import yaml + +_SUSPICIOUS_FILENAME_RE = re.compile( + r"(shell|backdoor|cmd|upload|gate|mailer|wso|c99|r57|b374k)", + re.I, +) + + +# --------------------------------------------------------------------------- +# Whitelist checks +# --------------------------------------------------------------------------- + +def load_whitelist(path: str | Path) -> set[str]: + """Load a YAML list file into a set. Returns empty set if missing.""" + p = Path(path) + if not p.exists(): + return set() + data = yaml.safe_load(p.read_text(encoding="utf-8")) or [] + if isinstance(data, list): + return {str(s).strip() for s in data if s} + return set() + + +def check_links_against_whitelist( + page: dict, + allowed_external_domains: set[str], + base_netloc: str, +) -> list[dict]: + """ + Return violations: external links to domains not in allowed_external_domains. + Also flags noscript links and comment links to external domains. + """ + violations: list[dict] = [] + + for link_type, links in page.get("links", {}).items(): + if not isinstance(links, list): + continue + for link in links: + if not isinstance(link, dict): + continue + if link.get("class") != "external": + continue + domain = urlparse(link.get("url", "")).netloc + if domain and domain not in allowed_external_domains: + violations.append({ + "type": "unlisted_external_domain", + "link_type": link_type, + "url": link.get("url", ""), + "domain": domain, + }) + + # Comment links + for href in page.get("comment_links", []): + domain = urlparse(href).netloc + if domain and domain != base_netloc and domain not in allowed_external_domains: + violations.append({ + "type": "comment_link_to_unlisted_domain", + "link_type": "comment", + "url": href, + "domain": domain, + }) + + # noscript links + for hc in page.get("hidden_content", []): + if hc.get("type") == "noscript_links": + for href in hc.get("links", []): + domain = urlparse(href).netloc + if domain and domain != base_netloc and domain not in allowed_external_domains: + violations.append({ + "type": "noscript_link_to_unlisted_domain", + "link_type": "noscript", + "url": href, + "domain": domain, + }) + + return violations + + +def check_internal_paths( + page: dict, + allowed_paths: set[str], +) -> list[dict]: + """ + Flag internal links to paths not in the allowed_paths whitelist. + If allowed_paths is empty, the check is skipped. + """ + if not allowed_paths: + return [] + violations: list[dict] = [] + for link in page.get("links", {}).get("a", []): + if link.get("class") != "internal": + continue + path = urlparse(link.get("url", "")).path + if path and path not in allowed_paths: + violations.append({ + "type": "unlisted_internal_path", + "url": link.get("url", ""), + "path": path, + }) + return violations + + +# --------------------------------------------------------------------------- +# Security headers +# --------------------------------------------------------------------------- + +def check_security_headers(response_headers: dict, required: list[str]) -> list[str]: + """Return list of missing recommended security headers.""" + # Header lookup is case-insensitive + present = {k.lower() for k in response_headers} + return [h for h in required if h.lower() not in present] + + +# --------------------------------------------------------------------------- +# Canonical hijack +# --------------------------------------------------------------------------- + +def canonical_is_hijacked(page: dict, base_netloc: str) -> bool: + """ + Return True if the canonical URL points to a different domain. + Canonical appearing in both and metadata is checked. + """ + canonical = ( + page.get("links", {}).get("canonical") + or page.get("metadata", {}).get("canonical") + ) + if not canonical: + return False + netloc = urlparse(canonical).netloc + return bool(netloc) and netloc != base_netloc + + +# --------------------------------------------------------------------------- +# Suspicious filenames +# --------------------------------------------------------------------------- + +def is_suspicious_url(url: str) -> bool: + """Flag URLs with names typical for webshells / backdoors.""" + filename = urlparse(url).path.rsplit("/", 1)[-1] + return bool(_SUSPICIOUS_FILENAME_RE.search(filename)) + + +# --------------------------------------------------------------------------- +# JSON-LD type check +# --------------------------------------------------------------------------- + +def check_jsonld_types(page: dict, allowed_types: list[str]) -> list[dict]: + """Return JSON-LD entries whose @type is not in the allowed list.""" + allowed = set(allowed_types) + return [ + j for j in page.get("jsonld", []) + if j.get("type") not in allowed + ] diff --git a/scanner/config.py b/scanner/config.py new file mode 100644 index 0000000..c9b56eb --- /dev/null +++ b/scanner/config.py @@ -0,0 +1,110 @@ +"""Config loading with deep-merge against defaults.""" +from pathlib import Path +import yaml + +DEFAULT_CONFIG: dict = { + "target": "https://bredelar.info", + "user_agent": "integrity-scanner/1.0 (+security-monitoring)", + "request_timeout": 20, + "data_dir": "data", + "reports_dir": "reports", + "logs_dir": "logs", + "config_dir": "config", + "crawl": { + "max_pages": 200, + "delay_seconds": 1.0, + "respect_robots_txt": False, + "skip_extensions": [ + ".jpg", ".jpeg", ".png", ".gif", ".webp", ".svg", ".ico", + ".pdf", ".zip", ".woff", ".woff2", ".ttf", ".otf", ".eot", + ".mp4", ".mp3", ".webm", ".avi", ".mov", + ], + }, + "scoring": { + "new_external_domain": 50, + "new_internal_url": 30, + "missing_internal_url": 20, + "hidden_content": 40, + "unexpected_canonical": 35, + "meta_refresh": 40, + "unexpected_jsonld": 35, + "large_text_addition": 20, + "new_inline_script_suspicious": 30, + "missing_security_header": 5, + "suspicious_content_pattern": 50, + "comment_links": 25, + "noscript_links": 25, + "new_external_link_unlisted": 40, + }, + "thresholds": { + "yellow": 20, + "red": 60, + "large_text_block_chars": 200, + }, + "alerting": { + "min_level": "yellow", + "email": { + "enabled": False, + "smtp_host": "localhost", + "smtp_port": 587, + "smtp_tls": True, + "smtp_user": "", + "smtp_password_env": "SCANNER_SMTP_PASSWORD", + "from": "scanner@example.com", + "to": [], + }, + "webhook": { + "enabled": False, + "url": "", + }, + }, + "normalization": { + "strip_html_comments": True, + "collapse_whitespace": True, + "ignore_selectors": [], + "ignore_patterns": [], + }, + "allowed_jsonld_types": [ + "Organization", "WebSite", "WebPage", "Article", "BreadcrumbList", + "ItemList", "LocalBusiness", "Place", "Person", "Event", + "FAQPage", "Question", "Answer", "ImageObject", "SiteLinksSearchBox", + "ContactPage", "AboutPage", + ], + "security_headers": [ + "Content-Security-Policy", + "Strict-Transport-Security", + "X-Content-Type-Options", + "Referrer-Policy", + ], +} + + +def load_config(config_file: str = "config.yaml") -> dict: + """Load config from YAML, deep-merged on top of defaults.""" + cfg = _deep_copy(DEFAULT_CONFIG) + p = Path(config_file) + if p.exists(): + with p.open(encoding="utf-8") as f: + user_cfg = yaml.safe_load(f) or {} + _deep_update(cfg, user_cfg) + return cfg + + +def resolve_paths(cfg: dict, base_dir: Path) -> dict: + """Resolve relative paths in config against base_dir.""" + for key in ("data_dir", "reports_dir", "logs_dir", "config_dir"): + cfg[key] = str(base_dir / cfg[key]) + return cfg + + +def _deep_copy(d: dict) -> dict: + import copy + return copy.deepcopy(d) + + +def _deep_update(base: dict, override: dict) -> None: + for k, v in override.items(): + if isinstance(v, dict) and isinstance(base.get(k), dict): + _deep_update(base[k], v) + else: + base[k] = v diff --git a/scanner/crawler.py b/scanner/crawler.py new file mode 100644 index 0000000..c8bb177 --- /dev/null +++ b/scanner/crawler.py @@ -0,0 +1,150 @@ +"""Pure-Python crawler — no wget dependency.""" +import logging +import re +import time +from urllib.parse import parse_qs, urlencode, urljoin, urlparse, urlunparse + +import requests +from bs4 import BeautifulSoup + +logger = logging.getLogger(__name__) + +# Query parameters that indicate calendar/session/tracking noise +_NOISE_PARAM_RE = re.compile( + r"^(year|month|day|date|week|session|token|sid|nonce|_|v|cache|utm_|fbclid)", + re.I, +) + + +def normalize_url(url: str) -> str | None: + """ + Normalise a URL: lowercase scheme+host, drop fragment, sort and filter + query params that look like cache-busters or tracking noise. + Returns None for non-http(s) URLs. + """ + try: + p = urlparse(url.strip()) + if p.scheme not in ("http", "https"): + return None + netloc = p.netloc.lower() + path = p.path or "/" + # Remove fragment + p = p._replace(netloc=netloc, path=path, fragment="") + # Keep only non-noise query params, sorted for stable comparison + if p.query: + clean_params = sorted( + (k, v) + for k, vs in parse_qs(p.query, keep_blank_values=True).items() + for v in vs + if not _NOISE_PARAM_RE.match(k) + ) + p = p._replace(query=urlencode(clean_params)) + return urlunparse(p) + except Exception: + return None + + +def should_crawl(url: str, base_netloc: str, skip_extensions: list[str]) -> bool: + """Return True if url should be followed during a crawl.""" + p = urlparse(url) + if p.netloc != base_netloc: + return False + path_lower = p.path.lower() + for ext in skip_extensions: + if path_lower.endswith(ext): + return False + # Skip URLs whose path looks like calendar/session noise + if _NOISE_PARAM_RE.search(p.query): + return False + return True + + +class Crawler: + def __init__(self, cfg: dict): + self.target = cfg["target"].rstrip("/") + self.base_netloc = urlparse(self.target).netloc + self.max_pages: int = cfg["crawl"]["max_pages"] + self.delay: float = cfg["crawl"]["delay_seconds"] + self.skip_ext: list[str] = cfg["crawl"]["skip_extensions"] + self.timeout: int = cfg["request_timeout"] + self.headers = {"User-Agent": cfg["user_agent"]} + + def crawl(self) -> dict: + """ + Crawl the target site breadth-first. + + Returns a dict with: + - pages: list of page-dicts (url, final_url, status, html, headers, ...) + - errors: list of {url, error} + """ + seen: set[str] = set() + queue: list[str] = [normalize_url(self.target + "/") or self.target + "/"] + pages: list[dict] = [] + errors: list[dict] = [] + + while queue and len(pages) < self.max_pages: + url = queue.pop(0) + if url in seen: + continue + seen.add(url) + + if self.delay > 0 and pages: + time.sleep(self.delay) + + logger.debug("Crawling %s", url) + try: + resp = requests.get( + url, + headers=self.headers, + timeout=self.timeout, + allow_redirects=True, + ) + final_url = normalize_url(resp.url) or resp.url + content_type = resp.headers.get("content-type", "") + is_html = "text/html" in content_type and resp.status_code == 200 + + page: dict = { + "url": url, + "final_url": final_url, + "status": resp.status_code, + "content_type": content_type, + "response_headers": dict(resp.headers), + "html": resp.text if is_html else None, + "redirected": final_url != url, + } + pages.append(page) + + if is_html: + for link_url in self._extract_follow_links(resp.text, final_url): + if link_url not in seen and link_url not in queue: + queue.append(link_url) + + except requests.exceptions.RequestException as exc: + errors.append({"url": url, "error": str(exc)}) + logger.warning("Error crawling %s: %s", url, exc) + + logger.info( + "Crawl finished: %d pages, %d errors, %d queued but skipped", + len(pages), + len(errors), + len(queue), + ) + return { + "target": self.target, + "pages": pages, + "errors": errors, + } + + def _extract_follow_links(self, html: str, base_url: str) -> list[str]: + """Extract internal links worth following.""" + result = [] + try: + soup = BeautifulSoup(html, "lxml") + for tag in soup.find_all("a", href=True): + href = tag["href"].strip() + abs_url = normalize_url(urljoin(base_url, href)) + if abs_url and should_crawl(abs_url, self.base_netloc, self.skip_ext): + result.append(abs_url) + except Exception as exc: + logger.debug("Link extraction failed on %s: %s", base_url, exc) + return result diff --git a/scanner/differ.py b/scanner/differ.py new file mode 100644 index 0000000..7b6d100 --- /dev/null +++ b/scanner/differ.py @@ -0,0 +1,338 @@ +""" +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", []): + 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.""" + 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) + + # 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 + ] + + 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, + "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', []))}]") + + for j in pd.get("new_jsonld", []): + if j.get("unexpected"): + add(sc.get("unexpected_jsonld", 35), + f"{u}: Unerwartetes JSON-LD @type={j.get('type')}") + + 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") + + # 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") + if npa.get("unexpected_jsonld"): + add(sc.get("unexpected_jsonld", 35), f"{u} (neu): Unerwartetes JSON-LD") + + 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() diff --git a/scanner/extractor.py b/scanner/extractor.py new file mode 100644 index 0000000..cde2d58 --- /dev/null +++ b/scanner/extractor.py @@ -0,0 +1,312 @@ +""" +HTML extraction with BeautifulSoup4+lxml. + +Extracts: visible text, links (all types), metadata, hidden-content indicators, +inline script patterns, JSON-LD, and links buried in HTML comments. +""" +import json +import re +from urllib.parse import urljoin, urlparse + +from bs4 import BeautifulSoup, Comment + +# Inline-script patterns that warrant flagging +_SUSPICIOUS_INLINE = [ + (r"eval\s*\(", "eval()"), + (r"atob\s*\(", "atob()"), + (r"fromCharCode", "fromCharCode"), + (r"document\.createElement\s*\(\s*[\"']script[\"']", "createElement(script)"), + (r"\bfetch\s*\(", "fetch()"), + (r"XMLHttpRequest", "XMLHttpRequest"), + (r"base64_decode", "base64_decode"), +] + +# CSS-based hiding patterns (applied to inline style attributes) +_HIDDEN_STYLE_CHECKS = [ + (r"display\s*:\s*none", "display:none"), + (r"visibility\s*:\s*hidden", "visibility:hidden"), + (r"opacity\s*:\s*0(?:\.0+)?(?!\d)", "opacity:0"), + (r"font-size\s*:\s*0", "font-size:0"), + (r"color\s*:\s*(?:white|#fff(?:fff)?|rgba?\(\s*255\s*,\s*255\s*,\s*255)", "white-text"), + (r"width\s*:\s*0(?:px)?[;\s]", "width:0"), + (r"height\s*:\s*0(?:px)?[;\s]", "height:0"), + (r"position\s*:\s*absolute", "position:absolute"), # combined with off-screen check below + (r"(?:left|top)\s*:\s*-\d{3,}", "off-screen-position"), + (r"clip\s*:\s*rect\s*\(\s*0\s*,?\s*0\s*,?\s*0\s*,?\s*0\s*\)", "clip:rect(0)"), + (r"text-indent\s*:\s*-\d{3,}", "text-indent-offscreen"), +] + + +def extract_page(html: str, url: str, cfg: dict | None = None) -> dict: + """ + Full extraction from page HTML. + + Returns a structured dict with all data needed for diffing and checking. + """ + cfg = cfg or {} + soup = BeautifulSoup(html, "lxml") + + # Remove ignored selectors before text extraction + ignore_selectors = cfg.get("normalization", {}).get("ignore_selectors", []) + for sel in ignore_selectors: + for el in soup.select(sel): + el.decompose() + + return { + "url": url, + "text": _visible_text(soup), + "links": _extract_links(soup, url), + "metadata": _extract_metadata(soup), + "hidden_content": _find_hidden_content(soup), + "inline_scripts": _check_inline_scripts(soup), + "jsonld": _extract_jsonld(soup), + "comment_links": _find_comment_links(html, url), + } + + +# --------------------------------------------------------------------------- +# Text +# --------------------------------------------------------------------------- + +def _visible_text(soup: BeautifulSoup) -> str: + """Extract visible text — strips scripts, styles, noscript, head.""" + work = soup.__copy__() + for tag in work(["script", "style", "noscript", "head", "meta", "link"]): + tag.decompose() + text = work.get_text(separator=" ") + return re.sub(r"\s+", " ", text).strip() + + +# --------------------------------------------------------------------------- +# Links +# --------------------------------------------------------------------------- + +def _extract_links(soup: BeautifulSoup, base_url: str) -> dict: + """ + Extract all link categories: + a, link, script[src], img[src], iframe[src], form[action], + meta_refresh, canonical + Each entry: {url, class: 'internal'|'external', ...extra} + """ + base_netloc = urlparse(base_url).netloc + + def classify(u: str) -> str: + return "internal" if urlparse(u).netloc == base_netloc else "external" + + def resolve(href: str | None) -> str | None: + if not href: + return None + href = href.strip() + if href.startswith(("javascript:", "mailto:", "tel:", "data:", "#")): + return None + abs_url = urljoin(base_url, href) + p = urlparse(abs_url) + if p.scheme not in ("http", "https"): + return None + # Drop fragment + return abs_url.split("#")[0] + + result: dict = { + "a": [], + "link_rel": [], + "script": [], + "img": [], + "iframe": [], + "form": [], + "meta_refresh": [], + "canonical": None, + } + + for tag in soup.find_all("a", href=True): + u = resolve(tag["href"]) + if u: + result["a"].append({ + "url": u, + "text": tag.get_text(" ", strip=True)[:200], + "class": classify(u), + "rel": " ".join(tag.get("rel", [])), + }) + + for tag in soup.find_all("link", href=True): + u = resolve(tag["href"]) + if not u: + continue + rel = " ".join(tag.get("rel", [])) + if "canonical" in rel: + result["canonical"] = u + result["link_rel"].append({"url": u, "rel": rel, "class": classify(u)}) + + for tag in soup.find_all("script", src=True): + u = resolve(tag["src"]) + if u: + result["script"].append({"url": u, "class": classify(u)}) + + for tag in soup.find_all("img", src=True): + u = resolve(tag["src"]) + if u: + result["img"].append({"url": u, "class": classify(u)}) + + for tag in soup.find_all("iframe", src=True): + u = resolve(tag["src"]) + if u: + result["iframe"].append({"url": u, "class": classify(u)}) + + for tag in soup.find_all("form", action=True): + u = resolve(tag["action"]) + if u: + result["form"].append({"url": u, "class": classify(u)}) + + for tag in soup.find_all("meta"): + if tag.get("http-equiv", "").lower() == "refresh": + content = tag.get("content", "") + m = re.search(r"url\s*=\s*(.+)", content, re.I) + if m: + u = resolve(m.group(1).strip().strip("\"'")) + if u: + result["meta_refresh"].append(u) + + return result + + +# --------------------------------------------------------------------------- +# Metadata +# --------------------------------------------------------------------------- + +def _extract_metadata(soup: BeautifulSoup) -> dict: + """Extract title, description, canonical, og-tags, robots, hreflang, H1/H2.""" + meta: dict = {} + + t = soup.find("title") + meta["title"] = t.get_text(strip=True) if t else None + + canonical_tag = soup.find("link", rel="canonical") + if canonical_tag: + meta["canonical"] = canonical_tag.get("href", "").strip() + + for tag in soup.find_all("meta"): + name = (tag.get("name") or "").lower().strip() + prop = (tag.get("property") or "").lower().strip() + content = tag.get("content", "") + + if name == "description": + meta["description"] = content + elif name == "robots": + meta["robots"] = content + elif name == "keywords": + meta["keywords"] = content + elif prop.startswith("og:"): + meta[prop] = content + elif prop.startswith("twitter:"): + meta[prop] = content + + hreflang = [] + for tag in soup.find_all("link", rel="alternate"): + hl = tag.get("hreflang") + if hl: + hreflang.append({"lang": hl, "href": tag.get("href", "")}) + meta["hreflang"] = hreflang + + meta["h1"] = [h.get_text(" ", strip=True) for h in soup.find_all("h1")] + meta["h2"] = [h.get_text(" ", strip=True) for h in soup.find_all("h2")] + + return meta + + +# --------------------------------------------------------------------------- +# Hidden content +# --------------------------------------------------------------------------- + +def _find_hidden_content(soup: BeautifulSoup) -> list[dict]: + """ + Detect elements that are visually hidden but contain text or links. + Catches common SEO-spam hiding techniques. + """ + findings: list[dict] = [] + + for tag in soup.find_all(style=True): + style = tag.get("style", "") + matched_indicators = [] + for pattern, label in _HIDDEN_STYLE_CHECKS: + if re.search(pattern, style, re.I): + matched_indicators.append(label) + if not matched_indicators: + continue + + text = tag.get_text(" ", strip=True)[:300] + links = [a.get("href", "") for a in tag.find_all("a", href=True)] + if text or links: + findings.append({ + "type": "inline_style_hidden", + "indicators": matched_indicators, + "tag": tag.name, + "text_preview": text, + "links": links, + }) + + #