From bed20db8bd76366e76dff01035a777b1e1fb4d2c Mon Sep 17 00:00:00 2001 From: Your Name Date: Fri, 12 Jun 2026 11:14:05 +0200 Subject: [PATCH] =?UTF-8?q?feat:=20Laientauglichkeit=20=E2=80=94=20Auto-Mo?= =?UTF-8?q?dus,=20Klartext-Ausgabe,=20einfachere=20Doku?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Ziel: ein einziger täglicher Befehl genügt, jede Meldung ist auch ohne IT-Wissen verständlich. Auto-Modus (ein Cron-Job genügt für alles): - data/state.json merkt sich, wann die Wochen-Prüfungen zuletzt liefen (baseline.py: load_state/save_state/is_check_due/mark_check_run) - `scan` führt fällige Zusatzprüfungen (Tarnung/externe Links/Dateien) automatisch mit aus; Gesamt-Ampel = schlechtestes Teilergebnis; EIN kombinierter Report + EIN Alert - neuer Config-Abschnitt periodic_checks (Default: wöchentlich, an) Klartext (scanner/plain.py): - Ampel 🟢/🟡/🔴 + Sätze ohne Fachbegriffe, aus den strukturierten Befunden - Report: Klartext oben, "Technische Details (für Ihren Dienstleister)" unten - Terminal-Ausgabe und E-Mail (alerter.py) ebenso umgestellt Komfort: - approve --all/--rebuild legt die Datei-Überwachung automatisch mit an (Opt-out: --skip-assets); approve-assets bleibt für den Sonderfall - status zeigt Ampel zuoberst + wann die Wochen-Prüfungen zuletzt liefen - Onboarding-Text in einfacher Sprache, wenn noch kein Vergleichsstand existiert Doku: README + BEDIENUNGSANLEITUNG mit "In 3 Schritten"-Einstieg. Tests: test_state.py, test_plain.py (Fachbegriff-Assertion), Auto-Asset- Baseline-Test. 165 Tests grün. Co-Authored-By: Claude Opus 4.8 --- BEDIENUNGSANLEITUNG.md | 55 ++++- README.md | 39 ++-- config.yaml | 9 + scanner/__main__.py | 505 +++++++++++++++++++++++++++-------------- scanner/alerter.py | 53 +++-- scanner/baseline.py | 44 ++++ scanner/config.py | 7 + scanner/plain.py | 179 +++++++++++++++ tests/test_assets.py | 41 ++++ tests/test_plain.py | 85 +++++++ tests/test_state.py | 47 ++++ 11 files changed, 859 insertions(+), 205 deletions(-) create mode 100644 scanner/plain.py create mode 100644 tests/test_plain.py create mode 100644 tests/test_state.py diff --git a/BEDIENUNGSANLEITUNG.md b/BEDIENUNGSANLEITUNG.md index 4d12a3d..23e06fb 100644 --- a/BEDIENUNGSANLEITUNG.md +++ b/BEDIENUNGSANLEITUNG.md @@ -5,6 +5,46 @@ unbefugte Manipulationen. Der Scanner läuft lokal im Verantwortungsbereich des --- +## Für Einsteiger – in 3 Schritten + +Sie brauchen kein IT-Wissen. Das Programm zeigt eine **Ampel** und erklärt alles in +einfachen Worten. + +**Schritt 1 — Erststand speichern** (einmalig, wenn die Website in Ordnung ist): +``` +python -m scanner approve --all --note "Erststand geprüft" +``` + +**Schritt 2 — täglich prüfen** (am besten automatisch per Cron, siehe Abschnitt 10): +``` +python -m scanner scan +``` + +**Schritt 3 — bei 🟡 oder 🔴 nachschauen:** +``` +python -m scanner report +``` + +Was die Ampel bedeutet: + +| Ampel | Bedeutung | Was tun? | +|---|---|---| +| 🟢 | Alles in Ordnung | Nichts. | +| 🟡 | Bitte nachschauen | Bericht lesen; wenn die Änderung gewollt war, mit `approve --all` bestätigen. | +| 🔴 | Achtung | Website ansehen und die Person/Firma informieren, die Ihre Website betreut. | + +> **Wichtig:** Die selteneren Prüfungen (Tarnung, externe Links, Dateien) laufen +> **automatisch einmal pro Woche** beim täglichen `scan` mit. Sie müssen dafür **nichts** +> extra einrichten — ein einziger täglicher Aufruf genügt für alles. +> +> Beim ersten `approve --all` wird zusätzlich die **Datei-Überwachung** automatisch +> eingerichtet (Bilder, Skripte, Stylesheets). Auch dafür ist kein weiterer Schritt nötig. + +Alles Weitere in dieser Anleitung richtet sich an **Fortgeschrittene** und an +**Ihren Dienstleister**. + +--- + ## Inhaltsverzeichnis 1. [Grundprinzip](#1-grundprinzip) @@ -607,8 +647,9 @@ das richtige Werkzeug (URL-Prüfung → Live testen). ### Wann ausführen? -Nicht täglich — der Doppel-Crawl belastet den Server doppelt. Empfehlung: monatlich -oder bei konkretem Verdacht. +**Läuft automatisch wöchentlich** beim täglichen `scan` mit (steuerbar über +`periodic_checks` in der `config.yaml`). Der manuelle Aufruf ist nur bei konkretem +Verdacht nötig. --- @@ -643,8 +684,8 @@ Der Report liegt unter `reports/_ext-links/report.md`. ### Wann ausführen? -Wöchentlich oder monatlich reicht. Nicht im täglichen Cron — zu viele Requests an -externe Server. +**Läuft automatisch wöchentlich** beim täglichen `scan` mit (steuerbar über +`periodic_checks` in der `config.yaml`). Ein manueller Aufruf ist nur bei Bedarf nötig. --- @@ -697,5 +738,7 @@ Der Report liegt unter `reports/_assets/report.md` und zeigt pro geänderter ### Wann ausführen? -Wöchentlich oder nach bekannten Website-Updates. Nicht täglich — lädt alle Assets herunter -und ist entsprechend langsam. +**Läuft automatisch wöchentlich** beim täglichen `scan` mit (steuerbar über +`periodic_checks` in der `config.yaml`). Die Datei-Vergleichsgrundlage wird zudem beim +ersten `approve --all` automatisch angelegt — `approve-assets` ist nur für den manuellen +Sonderfall gedacht. diff --git a/README.md b/README.md index 01fb2be..b831bb2 100644 --- a/README.md +++ b/README.md @@ -1,27 +1,40 @@ # 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. +## Was macht dieses Programm? -## Schnellstart +Es überwacht Ihre Website und meldet Ihnen, wenn sich etwas verändert hat — +besonders heimliche Manipulationen wie versteckten Spam, fremde Links oder +ausgetauschte Dateien. So merken Sie früh, wenn jemand unbefugt in Ihre Seite +eingegriffen hat. + +Das Programm läuft bei Ihnen, nicht beim Webhoster. Sie brauchen kein IT-Wissen: +Es zeigt eine **Ampel** (🟢 / 🟡 / 🔴) und erklärt jeden Fund in einfachen Worten. + +## In 3 Schritten startklar ```bash +# Einmalig: Programm einrichten 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 ``` +```bash +# Schritt 1: Wenn Ihre Website gerade in Ordnung ist, speichern Sie diesen Zustand: +python -m scanner approve --all --note "Erststand, Website geprüft" + +# Schritt 2: Ab jetzt täglich prüfen (am besten automatisch, siehe unten): +python -m scanner scan + +# Schritt 3: Bei Gelb/Rot den Bericht ansehen: +python -m scanner report +``` + +**Gut zu wissen:** Die selteneren Prüfungen (Tarnung, externe Links, Dateien) +laufen **automatisch einmal pro Woche** beim täglichen `scan` mit. Sie brauchen +dafür **nichts** extra einzurichten — ein täglicher Aufruf genügt für alles. + --- ## Subcommands diff --git a/config.yaml b/config.yaml index e234505..f8d9aef 100644 --- a/config.yaml +++ b/config.yaml @@ -60,6 +60,15 @@ thresholds: red: 60 # ab diesem Score: Alarm (rot) large_text_block_chars: 200 # Textblock ab dieser Länge wird als „groß" gewertet +# Wöchentliche Zusatzprüfungen laufen automatisch beim täglichen `scan` mit. +# So genügt EIN Cron-Job für alles — kein separater Aufruf nötig. +periodic_checks: + enabled: true # Zusatzprüfungen automatisch mitlaufen lassen + interval_days: 7 # wie oft (in Tagen) die Zusatzprüfungen fällig werden + cloak_check: true # Tarnungs-Prüfung (sieht Google einen anderen Inhalt?) + ext_links: true # externe Links auf Erreichbarkeit prüfen + assets: true # Bilder/Skripte/Stylesheets auf Veränderung prüfen + alerting: min_level: "yellow" # "yellow" oder "red" email: diff --git a/scanner/__main__.py b/scanner/__main__.py index 4069988..c20087e 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -42,6 +42,19 @@ from .differ import ( score_diff, ) from .extractor import extract_page +from .plain import ampel, ampel_zeile, klartext_befunde, was_tun + +# Reihenfolge der Risiko-Level, um das schlechteste zu bestimmen +_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2} + + +def _worst_level(levels: list[str]) -> str: + """Gibt das schwerwiegendste Level aus einer Liste zurück (green _LEVEL_ORDER.get(worst, 0): + worst = lvl + return worst # --------------------------------------------------------------------------- @@ -109,8 +122,11 @@ def _build_report( missing_sec_headers: dict, crawl_errors: list[dict], snap_dir: Path, + cloak_diff: dict | None = None, + asset_diff: dict | None = None, + ext_links: dict | None = None, ) -> dict: - return { + report = { "generated_at": datetime.now(timezone.utc).isoformat(), "target": target, "snapshot_dir": str(snap_dir), @@ -120,6 +136,14 @@ def _build_report( "missing_security_headers": missing_sec_headers, "crawl_errors": crawl_errors, } + # Optionale Ergebnisse der wöchentlichen Zusatzprüfungen (nur wenn gelaufen) + if cloak_diff is not None: + report["cloak_diff"] = cloak_diff + if asset_diff is not None: + report["asset_diff"] = asset_diff + if ext_links is not None: + report["ext_links"] = ext_links + return report def _write_report(report: dict, reports_dir: str) -> tuple[Path, Path]: @@ -140,14 +164,38 @@ def _write_report(report: dict, reports_dir: str) -> tuple[Path, Path]: def _render_markdown(r: dict) -> str: a = r.get("assessment", {}) d = r.get("diff", {}) - level = a.get("level", "?").upper() + lvl = a.get("level", "green") + level = lvl.upper() score = a.get("score", 0) level_badge = {"GREEN": "[OK]", "YELLOW": "[WARNUNG]", "RED": "[ALARM]"}.get(level, level) + symbol, headline = ampel(lvl) + # ---- Klartext-Teil (für den Betreiber) ---- lines = [ - f"# Integrity Report — {r.get('target', '?')}", + f"# Prüfbericht für Ihre Website — {r.get('target', '?')}", + "", + f"**Erstellt:** {r.get('generated_at', '?')}", + "", + f"## {symbol} {headline}", + "", + "### Was bedeutet das?", + "", + ] + for satz in klartext_befunde( + d, r.get("cloak_diff"), r.get("asset_diff"), r.get("ext_links") + ): + lines.append(f"- {satz}") + lines += ["", "### Was sollten Sie tun?", ""] + for schritt in was_tun(lvl): + lines.append(f"- {schritt}") + + # ---- Technischer Teil (für den Dienstleister) ---- + lines += [ + "", + "---", + "", + "# Technische Details (für Ihren Dienstleister)", "", - f"**Erstellt:** {r.get('generated_at', '?')} ", f"**Level:** {level_badge} ", f"**Score:** {score}", "", @@ -280,6 +328,44 @@ def _render_markdown(r: dict) -> str: for e in r["crawl_errors"][:20]: lines.append(f"- `{e.get('url', '?')}`: {e.get('error', '?')}") + # ---- Ergebnisse der wöchentlichen Zusatzprüfungen (falls gelaufen) ---- + cd = r.get("cloak_diff") + if cd is not None: + lines += ["", "## Tarnungs-Prüfung (Cloaking)", ""] + if cd.get("findings"): + for f in cd["findings"]: + lines.append(f"- `{f['url']}`") + if f.get("extra_links"): + for link in f["extra_links"]: + lines.append(f" - Nur für Google sichtbarer Link: `{link}`") + if f.get("extra_text_chars", 0) > 50: + lines.append(f" - Google sieht +{f['extra_text_chars']} Zeichen mehr Text") + if f.get("vary_ua"): + lines.append(" - Server liefert je nach Besucher anderen Inhalt (Vary: User-Agent)") + else: + lines.append("- Keine Auffälligkeiten.") + + ad = r.get("asset_diff") + if ad is not None: + lines += ["", "## Datei-Prüfung (Bilder/Skripte/Stylesheets)", ""] + if ad.get("changed"): + for entry in ad["changed"]: + size_info = f" ({entry['size_diff']:+d} Bytes)" if entry.get("size_diff") else "" + lines.append(f"- Geändert [{entry['type']}]: `{entry['url']}`{size_info}") + else: + lines.append("- Keine Datei-Änderungen.") + if ad.get("fetch_errors"): + for entry in ad["fetch_errors"]: + lines.append(f"- Nicht prüfbar: `{entry['url']}` ({entry['error']})") + + el = r.get("ext_links") + if el is not None: + lines += ["", "## Externe Links (Erreichbarkeit)", ""] + lines.append(f"- Geprüft: {el.get('total', 0)}, kaputt: {el.get('broken', 0)}, " + f"nicht erreichbar: {el.get('errors', 0)}") + for entry in el.get("broken_links", []): + lines.append(f"- HTTP {entry['status']}: `{entry['url']}`") + lines += [ "", "---", @@ -679,32 +765,12 @@ def cmd_crawl(args: argparse.Namespace, cfg: dict) -> int: 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"]), - ) - +def _run_content_check(cfg: dict, snap: dict, baseline: dict) -> tuple[dict, dict, list, dict]: + """ + Kern der Inhaltsprüfung: vergleicht Snapshot mit Baseline, prüft Whitelist, + Security-Header und verdächtige Dateinamen. Schreibt/alarmiert NICHT. + Returns (diff, assessment, whitelist_violations, missing_security_headers). + """ diff = compare_snapshots(baseline["pages"], snap["pages"], cfg) # Whitelist / header / webshell checks — only on successfully fetched HTML @@ -728,7 +794,6 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: if missing: missing_sec[page["url"]] = missing - # Webshell / backdoor filename in any referenced link for ltype, links in page.get("links", {}).items(): if not isinstance(links, list): continue @@ -742,7 +807,6 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: diff["suspicious_filenames"] = suspicious_filenames - # Score with whitelist-violation contribution for v in wl_violations: if v.get("type") == "unlisted_external_domain": existing = diff.get("new_external_domains", []) @@ -752,39 +816,167 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int: diff["new_external_domains"] = existing assessment = score_diff(diff, cfg) + return diff, assessment, wl_violations, missing_sec - crawl_errors = snap.get("errors", []) + +def _run_cloak_core(cfg: dict) -> tuple[dict, dict]: + """Doppel-Crawl Browser-UA vs. Googlebot-UA. Returns (cloak_diff, assessment).""" + pages_normal, _ = _crawl_and_extract(cfg) + bot_cfg = {**cfg, "user_agent": GOOGLEBOT_UA} + pages_bot, _ = _crawl_and_extract(bot_cfg) + normal_map = {p["url"]: p for p in pages_normal} + bot_map = {p["url"]: p for p in pages_bot} + cloak_diff = compare_cloaking(normal_map, bot_map, cfg) + return cloak_diff, score_cloak_diff(cloak_diff, cfg) + + +def _run_ext_links_core(cfg: dict, snap: dict) -> dict: + """Prüft externe -Links auf Erreichbarkeit. Returns ext_links-Report-Dict.""" + ext_urls: dict[str, list[str]] = {} + for page in snap["pages"].values(): + if page.get("status") != 200: + continue + for link in page.get("links", {}).get("a", []): + if isinstance(link, dict) and link.get("class") == "external": + url = link.get("url", "") + if url: + ext_urls.setdefault(url, []).append(page["url"]) + + if not ext_urls: + return {"total": 0, "ok": 0, "broken": 0, "errors": 0, + "broken_links": [], "error_links": []} + + results = check_external_links(list(ext_urls.keys()), timeout=cfg.get("request_timeout", 10)) + broken = {url: r for url, r in results.items() if r.get("status") and r["status"] >= 400} + conn_errors = {url: r for url, r in results.items() if r.get("error")} + return { + "total": len(results), + "ok": len(results) - len(broken) - len(conn_errors), + "broken": len(broken), + "errors": len(conn_errors), + "broken_links": [ + {"url": url, "status": r["status"], "pages": ext_urls[url]} + for url, r in sorted(broken.items()) + ], + "error_links": [ + {"url": url, "error": r["error"], "pages": ext_urls[url]} + for url, r in sorted(conn_errors.items()) + ], + } + + +def _run_assets_core(cfg: dict, bm: BaselineManager, snap: dict) -> tuple[dict, dict, bool]: + """Hasht Assets und vergleicht mit Baseline. Returns (asset_diff, assessment, has_baseline).""" + asset_urls = _collect_asset_urls(snap) + if not asset_urls: + empty = {"changed": [], "new": [], "missing": [], "fetch_errors": [], "total_checked": 0} + return empty, score_asset_diff(empty, cfg), bool(bm.load_asset_hashes()) + + raw_hashes = fetch_asset_hashes(list(asset_urls.keys()), timeout=cfg.get("request_timeout", 15)) + current_hashes = {url: {**r, "type": asset_urls[url]} for url, r in raw_hashes.items()} + baseline_hashes = bm.load_asset_hashes() + asset_diff = compare_asset_hashes(baseline_hashes, current_hashes) + return asset_diff, score_asset_diff(asset_diff, cfg), bool(baseline_hashes) + + +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_kein_baseline_hinweis() + return 1 + + snap = bm.load_snapshot() + if not snap: + print("Es gibt noch keinen aktuellen Stand zum Vergleichen.") + print("Bitte zuerst ausführen: python -m scanner scan") + return 1 + + baseline = bm.load_baseline() + diff, assessment, wl_violations, missing_sec = _run_content_check(cfg, snap, baseline) report = _build_report( cfg["target"], diff, assessment, wl_violations, missing_sec, - crawl_errors, snap["dir"], + snap.get("errors", []), snap["dir"], ) - json_path, md_path = _write_report(report, cfg["reports_dir"]) - logger.info("Report: %s", md_path) - + _write_report(report, cfg["reports_dir"]) _print_summary(assessment, diff) - send_alert(report, cfg) - return assessment["exit_code"] def cmd_scan(args: argparse.Namespace, cfg: dict) -> int: - """crawl + check.""" + """crawl + check. Führt fällige Wochen-Prüfungen automatisch mit aus.""" logger = logging.getLogger("scanner.scan") - print(f"Scanne {cfg['target']} ...") + print(f"Prüfe Ihre Website {cfg['target']} ...") pages, errors = _crawl_and_extract(cfg) bm = BaselineManager(cfg["data_dir"]) - snap_dir = bm.save_snapshot(pages, errors) - logger.info("Snapshot: %s (%d Seiten, %d Fehler)", snap_dir, len(pages), len(errors)) + snap = bm.load_snapshot(bm.save_snapshot(pages, errors)) + logger.info("Snapshot: %d Seiten, %d Fehler", 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") + _print_kein_baseline_hinweis() return 0 - return cmd_check(args, cfg) + baseline = bm.load_baseline() + diff, assessment, wl_violations, missing_sec = _run_content_check(cfg, snap, baseline) + levels = [assessment["level"]] + + # --- Fällige Wochen-Prüfungen automatisch mitlaufen lassen --- + cloak_diff = asset_diff = ext_links = None + pc = cfg.get("periodic_checks", {}) + if pc.get("enabled", True): + interval = pc.get("interval_days", 7) + if pc.get("cloak_check", True) and bm.is_check_due("cloak_check", interval): + print(" Wöchentliche Tarnungs-Prüfung läuft mit ...") + cloak_diff, cloak_assess = _run_cloak_core(cfg) + levels.append(cloak_assess["level"]) + bm.mark_check_run("cloak_check") + if pc.get("assets", True) and bm.is_check_due("assets", interval): + print(" Wöchentliche Datei-Prüfung läuft mit ...") + asset_diff, asset_assess, _ = _run_assets_core(cfg, bm, snap) + levels.append(asset_assess["level"]) + bm.mark_check_run("assets") + if pc.get("ext_links", True) and bm.is_check_due("ext_links", interval): + print(" Wöchentliche Prüfung externer Links läuft mit ...") + ext_links = _run_ext_links_core(cfg, snap) # informativ, hebt Level nicht an + bm.mark_check_run("ext_links") + + # Gesamt-Level = schlechtestes aus Inhalt + Cloaking + Datei-Check + overall = _worst_level(levels) + assessment = {**assessment, "level": overall, + "exit_code": _LEVEL_ORDER[overall]} + + report = _build_report( + cfg["target"], diff, assessment, wl_violations, missing_sec, + snap.get("errors", []), snap["dir"], + cloak_diff=cloak_diff, asset_diff=asset_diff, ext_links=ext_links, + ) + _write_report(report, cfg["reports_dir"]) + _print_summary(assessment, diff, cloak_diff, asset_diff, ext_links) + send_alert(report, cfg) + return assessment["exit_code"] + + +def _establish_asset_baseline(cfg: dict, bm: BaselineManager, snap_dir: Path | None) -> None: + """Hasht die Dateien des freigegebenen Stands und legt sie als Vergleich an.""" + snap = bm.load_snapshot(snap_dir) + if not snap: + return + asset_urls = _collect_asset_urls(snap) + if not asset_urls: + return + print(f"Richte Datei-Überwachung ein ({len(asset_urls)} Dateien werden geprüft) ...") + raw_hashes = fetch_asset_hashes(list(asset_urls.keys()), timeout=cfg.get("request_timeout", 15)) + hashes_to_save = { + url: {**r, "type": asset_urls[url]} + for url, r in raw_hashes.items() + if r.get("sha256") + } + bm.save_asset_hashes(hashes_to_save) + print(f"Datei-Überwachung eingerichtet: {len(hashes_to_save)} Dateien.") def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: @@ -792,30 +984,35 @@ def cmd_approve(args: argparse.Namespace, cfg: dict) -> int: bm = BaselineManager(cfg["data_dir"]) snap_dir = Path(args.snapshot) if getattr(args, "snapshot", None) else None note = getattr(args, "note", "") or "" + skip_assets = getattr(args, "skip_assets", False) 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.") + print(f"Ihre Website wurde als neuer Vergleichsstand gespeichert ({len(approved)} Seiten).") + if not skip_assets: + _establish_asset_baseline(cfg, bm, snap_dir) 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}") + print(f"Ihre Website wurde als Vergleichsstand gespeichert ({len(approved)} Seiten).") + print("Ab jetzt meldet das Programm jede spätere Veränderung.") + if not skip_assets: + _establish_asset_baseline(cfg, bm, snap_dir) 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}") + print(f"Diese Seite wurde als in Ordnung bestätigt: {url}") return 0 else: - print(f"FEHLER: URL nicht im Snapshot gefunden: {url}", file=sys.stderr) + print(f"Diese Adresse wurde im letzten Stand nicht gefunden: {url}", file=sys.stderr) return 1 - print("FEHLER: --url, --all oder --rebuild angeben.", file=sys.stderr) + print("Bitte geben Sie an, was freigegeben werden soll: --all, --url oder --rebuild.", + file=sys.stderr) return 1 @@ -848,37 +1045,52 @@ def cmd_report(args: argparse.Namespace, cfg: dict) -> int: def cmd_status(args: argparse.Namespace, cfg: dict) -> int: bm = BaselineManager(cfg["data_dir"]) - print(f"Ziel: {cfg['target']}") + # Ampel zuoberst: letzter bekannter Zustand + rd = Path(cfg["reports_dir"]) + last_report = None + if rd.exists(): + dirs = sorted(d for d in rd.iterdir() if d.is_dir()) + for d in reversed(dirs): + rp = d / "report.json" + if rp.exists(): + last_report = json.loads(rp.read_text(encoding="utf-8")) + break + if last_report: + lvl = last_report.get("assessment", {}).get("level", "green") + print(ampel_zeile(lvl)) + else: + print("Noch keine Prüfung durchgeführt.") + print() + + print(f"Ihre Website: {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', []))}") + print(f"Vergleichsstand: vom {manifest.get('approved_at', '?')[:10]} " + f"({len(manifest.get('urls', []))} Seiten)") if manifest.get("note"): - print(f" Notiz: {manifest['note']}") + print(f" Notiz: {manifest['note']}") else: - print("Baseline: KEINE — 'python -m scanner init' ausführen") + print("Vergleichsstand: noch keiner — bitte einmal `approve --all` 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)") + print(f"Letzte Prüfung: {snap_manifest.get('timestamp', '?')}") else: - print("Letzter Scan: KEINER") + print("Letzte Prüfung: noch keine") - 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}") + # Wann laufen die Wochen-Prüfungen das nächste Mal? + state = bm.load_state().get("last_runs", {}) + if state: + labels = {"cloak_check": "Tarnungs-Prüfung", "assets": "Datei-Prüfung", + "ext_links": "Externe Links"} + print("Wöchentliche Zusatzprüfungen — zuletzt gelaufen:") + for key, label in labels.items(): + ts = state.get(key, "noch nie") + print(f" {label}: {ts[:10] if ts != 'noch nie' else ts}") return 0 @@ -888,19 +1100,10 @@ def cmd_status(args: argparse.Namespace, cfg: dict) -> int: def cmd_cloak_check(args: argparse.Namespace, cfg: dict) -> int: logger = logging.getLogger("scanner.cloak") - print(f"Cloaking-Check: {cfg['target']}") - print(" Schritt 1/2: Crawl mit Browser-UA ...") - pages_normal, _ = _crawl_and_extract(cfg) - - print(" Schritt 2/2: Crawl mit Googlebot-UA ...") - bot_cfg = {**cfg, "user_agent": GOOGLEBOT_UA} - pages_bot, _ = _crawl_and_extract(bot_cfg) - - normal_map = {p["url"]: p for p in pages_normal} - bot_map = {p["url"]: p for p in pages_bot} - - cloak_diff = compare_cloaking(normal_map, bot_map, cfg) - assessment = score_cloak_diff(cloak_diff, cfg) + print(f"Tarnungs-Prüfung für {cfg['target']} (zwei Durchläufe) ...") + cloak_diff, assessment = _run_cloak_core(cfg) + bm = BaselineManager(cfg["data_dir"]) + bm.mark_check_run("cloak_check") report = { "generated_at": datetime.now(timezone.utc).isoformat(), @@ -926,60 +1129,26 @@ def cmd_check_ext_links(args: argparse.Namespace, cfg: dict) -> int: snap = bm.load_snapshot() if not snap: - print( - "FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.", - file=sys.stderr, - ) + print("Es gibt noch keinen aktuellen Stand. Bitte zuerst: python -m scanner scan") return 1 - # Collect unique external links → pages that link to them - ext_urls: dict[str, list[str]] = {} - for page in snap["pages"].values(): - if page.get("status") != 200: - continue - for link in page.get("links", {}).get("a", []): - if isinstance(link, dict) and link.get("class") == "external": - url = link.get("url", "") - if url: - ext_urls.setdefault(url, []).append(page["url"]) + print("Prüfe die externen Links Ihrer Website ...") + result = _run_ext_links_core(cfg, snap) + bm.mark_check_run("ext_links") - if not ext_urls: - print("Keine externen Links im letzten Snapshot gefunden.") + if result["total"] == 0: + print("Ihre Website verweist nicht auf andere Websites.") return 0 - print(f"Prüfe {len(ext_urls)} externe Links ...") - results = check_external_links( - list(ext_urls.keys()), - timeout=cfg.get("request_timeout", 10), - ) - - broken = {url: r for url, r in results.items() if r.get("status") and r["status"] >= 400} - conn_errors = {url: r for url, r in results.items() if r.get("error")} - ok_count = len(results) - len(broken) - len(conn_errors) - report = { "generated_at": datetime.now(timezone.utc).isoformat(), "target": cfg["target"], "type": "ext-links", - "total": len(results), - "ok": ok_count, - "broken": len(broken), - "errors": len(conn_errors), - "broken_links": [ - {"url": url, "status": r["status"], "pages": ext_urls[url]} - for url, r in sorted(broken.items()) - ], - "error_links": [ - {"url": url, "error": r["error"], "pages": ext_urls[url]} - for url, r in sorted(conn_errors.items()) - ], + **result, } - - json_path, md_path = _write_ext_links_report(report, cfg["reports_dir"]) - logger.info("Ext-Links-Report: %s", md_path) - + _write_ext_links_report(report, cfg["reports_dir"]) _print_ext_links_summary(report) - return 0 if not broken and not conn_errors else 1 + return 0 if not result["broken"] and not result["errors"] else 1 # --------------------------------------------------------------------------- @@ -992,34 +1161,16 @@ def cmd_check_assets(args: argparse.Namespace, cfg: dict) -> int: snap = bm.load_snapshot() if not snap: - print( - "FEHLER: Kein Snapshot vorhanden. Zuerst `python -m scanner crawl` ausführen.", - file=sys.stderr, - ) + print("Es gibt noch keinen aktuellen Stand. Bitte zuerst: python -m scanner scan") return 1 - asset_urls = _collect_asset_urls(snap) - if not asset_urls: - print("Keine Assets im letzten Snapshot gefunden.") + if not _collect_asset_urls(snap): + print("Ihre Website enthält keine prüfbaren Dateien (Bilder/Skripte/Stylesheets).") return 0 - print(f"Hashe {len(asset_urls)} Assets (script: {sum(1 for t in asset_urls.values() if t=='script')}, " - f"css: {sum(1 for t in asset_urls.values() if t=='link_rel')}, " - f"img: {sum(1 for t in asset_urls.values() if t=='img')}) ...") - - raw_hashes = fetch_asset_hashes( - list(asset_urls.keys()), - timeout=cfg.get("request_timeout", 15), - ) - # Attach link type to each result for scoring - current_hashes = { - url: {**r, "type": asset_urls[url]} - for url, r in raw_hashes.items() - } - - baseline_hashes = bm.load_asset_hashes() - asset_diff = compare_asset_hashes(baseline_hashes, current_hashes) - assessment = score_asset_diff(asset_diff, cfg) + print("Prüfe die Dateien Ihrer Website (Bilder/Skripte/Stylesheets) ...") + asset_diff, assessment, has_baseline = _run_assets_core(cfg, bm, snap) + bm.mark_check_run("assets") report = { "generated_at": datetime.now(timezone.utc).isoformat(), @@ -1028,15 +1179,13 @@ def cmd_check_assets(args: argparse.Namespace, cfg: dict) -> int: "assessment": assessment, "asset_diff": asset_diff, } - json_path, md_path = _write_asset_report(report, cfg["reports_dir"]) - logger.info("Asset-Report: %s", md_path) - + _write_asset_report(report, cfg["reports_dir"]) _print_asset_summary(assessment, asset_diff) - if not baseline_hashes: + if not has_baseline: print() - print("Noch keine Asset-Baseline vorhanden.") - print("Führen Sie aus: python -m scanner approve-assets") + print("Hinweis: Es war noch keine Datei-Vergleichsgrundlage vorhanden.") + print("Diese wird automatisch angelegt, sobald Sie mit `approve --all` freigeben.") return assessment["exit_code"] @@ -1080,19 +1229,35 @@ def cmd_approve_assets(args: argparse.Namespace, cfg: dict) -> int: # Output helper # --------------------------------------------------------------------------- -def _print_summary(assessment: dict, diff: dict) -> None: - level = assessment.get("level", "?").upper() - score = assessment.get("score", 0) +def _print_kein_baseline_hinweis() -> None: + """Klartext-Anleitung, wenn noch kein Vergleichsstand (Baseline) existiert.""" + print("Es gibt noch keinen gespeicherten Vergleichsstand Ihrer Website.") + print("Wenn Ihre Website gerade in Ordnung ist, speichern Sie diesen Zustand einmalig als") + print("Vergleichsgrundlage. Ab dann meldet das Programm jede spätere Veränderung.") 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)}") + print("So geht's: python -m scanner approve --all") + + +def _print_summary(assessment: dict, diff: dict, + cloak_diff: dict | None = None, + asset_diff: dict | None = None, + ext_links: dict | None = None) -> None: + level = assessment.get("level", "green") + print() + print(ampel_zeile(level)) + print() + for satz in klartext_befunde(diff, cloak_diff, asset_diff, ext_links): + print(f" • {satz}") + print() + for schritt in was_tun(level): + print(f" → {schritt}") + # Knappe technische Kennzahlen darunter (für den Dienstleister) + print() + print(f" — Technische Details: Score {assessment.get('score', 0)}, " + f"{diff.get('changed_pages', 0)} geänderte Seite(n), " + f"{len(diff.get('new_internal_urls', []))} neue / " + f"{len(diff.get('missing_internal_urls', []))} fehlende Seite(n). " + f"Vollständiger Bericht: python -m scanner report") # --------------------------------------------------------------------------- @@ -1123,6 +1288,8 @@ def _build_parser() -> argparse.ArgumentParser: 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") + ap.add_argument("--skip-assets", action="store_true", + help="Datei-Überwachung NICHT automatisch mit anlegen") rp = sub.add_parser("report", help="Letzten Report anzeigen") rp.add_argument("--format", choices=["md", "json"], default="md") diff --git a/scanner/alerter.py b/scanner/alerter.py index e137c1b..968438d 100644 --- a/scanner/alerter.py +++ b/scanner/alerter.py @@ -11,6 +11,8 @@ from email.message import EmailMessage import requests +from .plain import ampel, klartext_befunde, was_tun + logger = logging.getLogger(__name__) _LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2} @@ -43,35 +45,52 @@ def send_alert(report: dict, cfg: dict) -> None: # --------------------------------------------------------------------------- def _make_subject(report: dict) -> str: - level = report.get("assessment", {}).get("level", "?").upper() + level = report.get("assessment", {}).get("level", "green") target = report.get("target", "?") - ts = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M UTC") - return f"[{level}] Integrity Alert: {target} — {ts}" + symbol, _ = ampel(level) + if level == "red": + return f"{symbol} Ihre Website {target} braucht Aufmerksamkeit" + if level == "yellow": + return f"{symbol} Bitte einmal auf Ihrer Website {target} nachschauen" + return f"{symbol} Ihre Website {target}: alles in Ordnung" 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", "?") + level = assessment.get("level", "green") + diff = report.get("diff", {}) + symbol, headline = ampel(level) + + # ---- Klartext-Teil (für den Betreiber) ---- + lines = [ + f"Prüfbericht für Ihre Website: {report.get('target', '?')}", + f"Zeitpunkt: {datetime.now(timezone.utc).strftime('%d.%m.%Y %H:%M Uhr')} (UTC)", + "", + f"{symbol} {headline}", + "", + "Was bedeutet das?", + ] + for satz in klartext_befunde( + diff, report.get("cloak_diff"), report.get("asset_diff"), report.get("ext_links") + ): + lines.append(f" • {satz}") + lines += ["", "Was sollten Sie tun?"] + for schritt in was_tun(level): + lines.append(f" → {schritt}") + + # ---- Technischer Teil (für den Dienstleister) ---- lines += [ - f"Level: {level}", - f"Score: {score}", + "", + "=" * 60, + "Technische Details (für Ihren Dienstleister)", + "=" * 60, + f"Level: {level.upper()} Score: {assessment.get('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", []), "-") diff --git a/scanner/baseline.py b/scanner/baseline.py index 1cc574b..3f9b7a4 100644 --- a/scanner/baseline.py +++ b/scanner/baseline.py @@ -233,6 +233,50 @@ class BaselineManager: ) logger.info("Asset hashes saved: %d entries", len(hashes)) + # ------------------------------------------------------------------ + # Run-state (drives auto-scheduling of the weekly checks) + # ------------------------------------------------------------------ + + @property + def _state_path(self) -> Path: + return self.data_dir / "state.json" + + def load_state(self) -> dict: + """Load the run-state file. Returns {'last_runs': {}} if none exists.""" + if not self._state_path.exists(): + return {"last_runs": {}} + try: + data = json.loads(self._state_path.read_text(encoding="utf-8")) + except (json.JSONDecodeError, ValueError): + return {"last_runs": {}} + data.setdefault("last_runs", {}) + return data + + def save_state(self, state: dict) -> None: + """Persist the run-state file.""" + self.data_dir.mkdir(parents=True, exist_ok=True) + self._state_path.write_text( + json.dumps(state, indent=2, ensure_ascii=False), encoding="utf-8" + ) + + def is_check_due(self, name: str, interval_days: int) -> bool: + """True if a periodic check has never run or last ran >= interval_days ago.""" + last = self.load_state().get("last_runs", {}).get(name) + if not last: + return True + try: + last_dt = datetime.fromisoformat(last) + except ValueError: + return True + age = datetime.now(timezone.utc) - last_dt + return age.total_seconds() >= interval_days * 86400 + + def mark_check_run(self, name: str) -> None: + """Record that a periodic check just ran (now, UTC).""" + state = self.load_state() + state["last_runs"][name] = datetime.now(timezone.utc).isoformat() + self.save_state(state) + # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ diff --git a/scanner/config.py b/scanner/config.py index 3e87ba2..c31cb3b 100644 --- a/scanner/config.py +++ b/scanner/config.py @@ -45,6 +45,13 @@ DEFAULT_CONFIG: dict = { "red": 60, "large_text_block_chars": 200, }, + "periodic_checks": { + "enabled": True, + "interval_days": 7, + "cloak_check": True, + "ext_links": True, + "assets": True, + }, "alerting": { "min_level": "yellow", "email": { diff --git a/scanner/plain.py b/scanner/plain.py new file mode 100644 index 0000000..698fae9 --- /dev/null +++ b/scanner/plain.py @@ -0,0 +1,179 @@ +""" +Klartext-Schicht: übersetzt technische Befunde in Sätze, die auch ein +Website-Betreiber ohne IT-Wissen versteht. + +Alle Texte hier sind bewusst laienverständlich. Fachbegriffe (Score, Canonical, +JSON-LD, Meta-Refresh, Hash) gehören NICHT in dieses Modul — die technischen +Details stehen weiterhin im Detail-Abschnitt des Reports. +""" + +# Ampel-Überschrift je Risiko-Level +_AMPEL = { + "green": ("🟢", "Alles in Ordnung."), + "yellow": ("🟡", "Bitte einmal nachschauen."), + "red": ("🔴", "Achtung — bitte zeitnah prüfen oder Ihren Dienstleister informieren."), +} + + +def ampel(level: str) -> tuple[str, str]: + """Symbol und Überschrift für ein Risiko-Level (green/yellow/red).""" + return _AMPEL.get(level, ("⚪", "Status unbekannt.")) + + +def ampel_zeile(level: str) -> str: + """Einzeilige Ampel-Darstellung, z. B. '🟡 Bitte einmal nachschauen.'""" + symbol, satz = ampel(level) + return f"{symbol} {satz}" + + +def _zaehle_seiten_befunde(diff: dict) -> dict: + """Zählt die Befund-Kategorien über alle geänderten Seiten zusammen.""" + z = { + "hidden": 0, "meta_refresh": 0, "canonical": 0, + "comment_links": 0, "scripts": 0, "grosser_text": 0, + } + for pd in diff.get("page_diffs", []): + if pd.get("new_hidden_content"): + z["hidden"] += 1 + if pd.get("new_meta_refresh"): + z["meta_refresh"] += 1 + if pd.get("canonical_changed"): + z["canonical"] += 1 + if pd.get("new_comment_links"): + z["comment_links"] += 1 + if pd.get("new_inline_scripts"): + z["scripts"] += 1 + if pd.get("added_chars", 0) > 200: + z["grosser_text"] += 1 + return z + + +def klartext_befunde( + diff: dict | None = None, + cloak_diff: dict | None = None, + asset_diff: dict | None = None, + ext_links: dict | None = None, +) -> list[str]: + """ + Erzeugt eine Liste laienverständlicher Sätze aus den strukturierten Befunden. + Leere Befunde → eine beruhigende Meldung. + """ + diff = diff or {} + saetze: list[str] = [] + + # --- Eindeutige Angriffszeichen zuerst --- + if diff.get("suspicious_filenames"): + saetze.append( + "Es wurde auf eine Datei mit verdächtigem Namen verlinkt, wie sie Angreifer " + "für Schadprogramme verwenden. Das ist ein ernstes Warnzeichen." + ) + + if cloak_diff: + bot_links = sum(len(f.get("extra_links", [])) for f in cloak_diff.get("findings", [])) + bot_text = any(f.get("extra_text_chars", 0) > 100 for f in cloak_diff.get("findings", [])) + if bot_links: + saetze.append( + "Ihre Website zeigt der Google-Suchmaschine andere Links als normalen " + "Besuchern. Das ist eine typische Methode von Spam-Angriffen." + ) + elif bot_text: + saetze.append( + "Ihre Website zeigt der Google-Suchmaschine zusätzlichen Text, den normale " + "Besucher nicht sehen. Bitte prüfen lassen." + ) + + # --- Verdächtige Inhaltsänderungen --- + z = _zaehle_seiten_befunde(diff) + if z["hidden"]: + saetze.append( + f"Auf {z['hidden']} Seite(n) wurde unsichtbarer Text oder unsichtbare Links " + "gefunden. So versteckt man üblicherweise Spam vor dem Auge des Besuchers." + ) + if z["meta_refresh"]: + saetze.append( + f"{z['meta_refresh']} Seite(n) leiten Besucher jetzt automatisch auf eine " + "andere Adresse weiter. Wenn Sie das nicht selbst eingerichtet haben, ist es verdächtig." + ) + if z["scripts"]: + saetze.append( + f"Auf {z['scripts']} Seite(n) wurde neuer, auffälliger Programmcode eingebaut. " + "Bitte von Ihrem Dienstleister prüfen lassen." + ) + if z["comment_links"]: + saetze.append( + f"Auf {z['comment_links']} Seite(n) wurden versteckte Links im Quelltext gefunden." + ) + if z["canonical"]: + saetze.append( + f"Bei {z['canonical']} Seite(n) wurde die hinterlegte 'Original-Adresse' geändert. " + "Damit versuchen Angreifer manchmal, Ihre Seite bei Google umzuleiten." + ) + + # --- Fremde Adressen / neue Seiten --- + neue_domains = diff.get("new_external_domains", []) + if neue_domains: + beispiele = ", ".join(neue_domains[:3]) + mehr = f" (und {len(neue_domains) - 3} weitere)" if len(neue_domains) > 3 else "" + saetze.append( + f"Es tauchen Verweise auf neue fremde Internet-Adressen auf: {beispiele}{mehr}. " + "Das kann harmlos sein (z. B. ein neuer Partner-Link) oder ein Hinweis auf einen Eingriff." + ) + + neue_seiten = diff.get("new_internal_urls", []) + if neue_seiten: + saetze.append( + f"Es sind {len(neue_seiten)} neue Unterseite(n) hinzugekommen. Haben Sie oder Ihr " + "Dienstleister die angelegt?" + ) + fehlende = diff.get("missing_internal_urls", []) + if fehlende: + saetze.append( + f"{len(fehlende)} Seite(n), die es früher gab, sind nicht mehr auffindbar." + ) + + if z["grosser_text"]: + saetze.append( + f"Auf {z['grosser_text']} Seite(n) wurde ein größerer Textblock neu hinzugefügt." + ) + + # --- Datei-Veränderungen (Assets) --- + if asset_diff and asset_diff.get("changed"): + n = len(asset_diff["changed"]) + saetze.append( + f"{n} Datei(en) (z. B. Bilder, Skripte oder Stylesheets) haben sich verändert. " + "Nach einem geplanten Update ist das normal — sonst bitte prüfen lassen." + ) + + # --- Kaputte Links (rein informativ) --- + neue_kaputt = diff.get("new_broken_urls", []) + if neue_kaputt: + saetze.append( + f"{len(neue_kaputt)} interne Link(s) führen ins Leere (Seite nicht gefunden)." + ) + if ext_links and ext_links.get("broken"): + saetze.append( + f"{ext_links['broken']} Link(s) zu anderen Websites funktionieren nicht mehr. " + "Das ist meist harmlos — die fremde Seite wurde gelöscht oder umbenannt." + ) + + if not saetze: + saetze.append("Es hat sich nichts Verdächtiges verändert.") + + return saetze + + +def was_tun(level: str) -> list[str]: + """Konkrete, laienverständliche Handlungsempfehlung je Level.""" + if level == "red": + return [ + "Öffnen Sie Ihre Website im Browser und schauen Sie, ob etwas Fremdes zu sehen ist.", + "Informieren Sie die Person oder Firma, die Ihre Website betreut.", + "Geben Sie nichts frei, solange der Verdacht nicht geklärt ist.", + ] + if level == "yellow": + return [ + "Schauen Sie sich die unten genannten Punkte einmal an.", + "Wenn die Änderung von Ihnen oder Ihrem Dienstleister gewollt war, ist alles in Ordnung.", + "Gewollte Änderungen bestätigen Sie mit: python -m scanner approve --all", + ] + return ["Sie müssen nichts tun."] diff --git a/tests/test_assets.py b/tests/test_assets.py index 8585748..9db2ca5 100644 --- a/tests/test_assets.py +++ b/tests/test_assets.py @@ -4,6 +4,7 @@ from unittest.mock import MagicMock, patch import pytest import requests +from scanner.baseline import BaselineManager from scanner.checker import fetch_asset_hashes from scanner.differ import compare_asset_hashes, score_asset_diff @@ -142,3 +143,43 @@ class TestScoreAssetDiff: assert result["score"] == 80 assert result["level"] == "red" assert result["exit_code"] == 2 + + +class TestAutoAssetBaselineOnApprove: + """approve --all soll die Datei-Baseline automatisch mit anlegen.""" + + def test_establish_asset_baseline_saves_hashes(self, tmp_path): + from scanner.__main__ import _establish_asset_baseline + + bm = BaselineManager(tmp_path) + page = { + "url": "https://x.de/", + "status": 200, + "links": {"script": [{"url": "https://x.de/app.js", "class": "internal"}], + "link_rel": [], "img": []}, + } + snap_dir = bm.save_snapshot([page]) + + with patch("scanner.__main__.fetch_asset_hashes") as mock_fetch: + mock_fetch.return_value = { + "https://x.de/app.js": {"sha256": "abc123", "size": 10, "error": None} + } + _establish_asset_baseline({"request_timeout": 15}, bm, snap_dir) + + saved = bm.load_asset_hashes() + assert "https://x.de/app.js" in saved + assert saved["https://x.de/app.js"]["sha256"] == "abc123" + + def test_no_assets_writes_nothing(self, tmp_path): + from scanner.__main__ import _establish_asset_baseline + + bm = BaselineManager(tmp_path) + page = {"url": "https://x.de/", "status": 200, + "links": {"script": [], "link_rel": [], "img": []}} + snap_dir = bm.save_snapshot([page]) + + with patch("scanner.__main__.fetch_asset_hashes") as mock_fetch: + _establish_asset_baseline({"request_timeout": 15}, bm, snap_dir) + mock_fetch.assert_not_called() + + assert bm.load_asset_hashes() == {} diff --git a/tests/test_plain.py b/tests/test_plain.py new file mode 100644 index 0000000..03e9d32 --- /dev/null +++ b/tests/test_plain.py @@ -0,0 +1,85 @@ +"""Tests für die Klartext-Schicht (laienverständliche Ausgabe).""" +import pytest + +from scanner.plain import ampel, ampel_zeile, klartext_befunde, was_tun + +# Fachbegriffe, die in der Klartext-Ausgabe NICHT vorkommen dürfen +JARGON = ["Score", "Canonical", "JSON-LD", "Meta-Refresh", "Hash", "SHA-256", + "Whitelist", "Hidden Content", "Exit-Code"] + + +def _no_jargon(saetze): + text = " ".join(saetze) + treffer = [w for w in JARGON if w.lower() in text.lower()] + assert not treffer, f"Fachbegriff(e) in Klartext gefunden: {treffer}" + + +class TestAmpel: + @pytest.mark.parametrize("level", ["green", "yellow", "red"]) + def test_each_level_has_symbol_and_sentence(self, level): + symbol, satz = ampel(level) + assert symbol + assert satz.endswith(".") or satz.endswith(")") + + def test_unknown_level_safe(self): + symbol, satz = ampel("bogus") + assert symbol and satz + + def test_ampel_zeile_combines(self): + zeile = ampel_zeile("green") + assert zeile.startswith("🟢") + + +class TestKlartextBefunde: + def test_empty_diff_says_all_good(self): + saetze = klartext_befunde({}) + assert any("nichts" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_new_external_domain_plain(self): + saetze = klartext_befunde({"new_external_domains": ["spam.example.com"]}) + assert any("fremde" in s.lower() or "adresse" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_hidden_content_plain(self): + diff = {"page_diffs": [{"url": "x", "new_hidden_content": [{"type": "x"}]}]} + saetze = klartext_befunde(diff) + assert any("unsichtbar" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_suspicious_filename_is_serious(self): + saetze = klartext_befunde({"suspicious_filenames": [{"page": "x", "url": "shell.php"}]}) + assert any("warnzeichen" in s.lower() or "verdächtig" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_cloaking_extra_link_plain(self): + cloak = {"findings": [{"url": "x", "extra_links": ["https://spam.com/x"], "extra_text_chars": 0}]} + saetze = klartext_befunde({}, cloak_diff=cloak) + assert any("google" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_asset_change_plain(self): + asset = {"changed": [{"url": "x", "type": "script"}]} + saetze = klartext_befunde({}, asset_diff=asset) + assert any("datei" in s.lower() for s in saetze) + _no_jargon(saetze) + + def test_broken_ext_links_plain(self): + ext = {"broken": 3} + saetze = klartext_befunde({}, ext_links=ext) + assert any("funktionieren" in s.lower() or "website" in s.lower() for s in saetze) + _no_jargon(saetze) + + +class TestWasTun: + def test_red_has_urgent_steps(self): + schritte = was_tun("red") + assert any("betreut" in s.lower() or "informieren" in s.lower() for s in schritte) + + def test_green_needs_nothing(self): + schritte = was_tun("green") + assert any("nichts" in s.lower() for s in schritte) + + def test_no_jargon_anywhere(self): + for lvl in ("green", "yellow", "red"): + _no_jargon(was_tun(lvl)) diff --git a/tests/test_state.py b/tests/test_state.py new file mode 100644 index 0000000..d7799f5 --- /dev/null +++ b/tests/test_state.py @@ -0,0 +1,47 @@ +"""Tests für die Zustandsdatei, die die Wochen-Prüfungen steuert.""" +from datetime import datetime, timedelta, timezone + +import pytest + +from scanner.baseline import BaselineManager + + +@pytest.fixture +def bm(tmp_path): + return BaselineManager(tmp_path) + + +class TestRunState: + def test_never_run_is_due(self, bm): + assert bm.is_check_due("cloak_check", 7) is True + + def test_just_run_is_not_due(self, bm): + bm.mark_check_run("cloak_check") + assert bm.is_check_due("cloak_check", 7) is False + + def test_old_run_is_due(self, bm): + # Lauf vor 10 Tagen → bei Intervall 7 fällig + old = (datetime.now(timezone.utc) - timedelta(days=10)).isoformat() + bm.save_state({"last_runs": {"assets": old}}) + assert bm.is_check_due("assets", 7) is True + + def test_recent_run_within_interval_not_due(self, bm): + recent = (datetime.now(timezone.utc) - timedelta(days=3)).isoformat() + bm.save_state({"last_runs": {"assets": recent}}) + assert bm.is_check_due("assets", 7) is False + + def test_save_load_roundtrip(self, bm): + bm.mark_check_run("ext_links") + state = bm.load_state() + assert "ext_links" in state["last_runs"] + + def test_corrupt_state_file_recovers(self, bm): + bm._state_path.parent.mkdir(parents=True, exist_ok=True) + bm._state_path.write_text("{ not valid json", encoding="utf-8") + # Darf nicht crashen, sondern als "nie gelaufen" gelten + assert bm.load_state() == {"last_runs": {}} + assert bm.is_check_due("cloak_check", 7) is True + + def test_invalid_timestamp_is_due(self, bm): + bm.save_state({"last_runs": {"assets": "kein-datum"}}) + assert bm.is_check_due("assets", 7) is True