dschlueter/integrity_scanner_fuer_statische_Webseiten
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

feat: Laientauglichkeit — Auto-Modus, Klartext-Ausgabe, einfachere Doku

Browse source 
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 <noreply@anthropic.com>
This commit is contained in:
Your Name 2026-06-12 11:14:05 +02:00
commit bed20db8bd
11 changed files with 856 additions and 202 deletions
Download patch file Download diff file Expand all files Collapse all files

55
BEDIENUNGSANLEITUNG.md
View file

@ -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/<ts>_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/<ts>_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.

39
README.md
View file

@ -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

9
config.yaml
View file

@ -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:

505
scanner/__main__.py
View file

@ -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<yellow<red)."""
worst = "green"
for lvl in levels:
if _LEVEL_ORDER.get(lvl, 0) > _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 <a>-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 <Adresse> 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 <a> 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")

53
scanner/alerter.py
View file

@ -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", []), "-")

44
scanner/baseline.py
View file

@ -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
# ------------------------------------------------------------------

7
scanner/config.py
View file

@ -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": {

179
scanner/plain.py Normal file
View file

@ -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."]

41
tests/test_assets.py
View file

@ -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() == {}

85
tests/test_plain.py Normal file
View file

@ -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))

47
tests/test_state.py Normal file
View file

@ -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