fix: Report und Alarm-Mail konsistent (gemeinsamer Renderer)

Behebt vier vom Nutzer gemeldete Inkonsistenzen zwischen Report und E-Mail:

1. E-Mail ⊊ Report: Es gab zwei divergierende Renderer (alerter._format_body
   und __main__._render_markdown). Jetzt EIN gemeinsamer Renderer
   (scanner/report.py:render_markdown), den beide nutzen → identischer Inhalt,
   keine Divergenz mehr möglich.

2. KI im Report unsichtbar: Der Report rief klartext_befunde OHNE ai_result auf
   (KI-Satz fehlte im "Was bedeutet das?") und die Zusammenfassungs-Tabelle
   hatte keine KI-Zeile. Jetzt: KI-Satz im Klartext + Zeilen "KI-Funde" und
   "KI nicht geprüft" in der Tabelle.

3. Fehlende URL in Kommandos: approve/report/ai-dismiss standen ohne Target da
   (nicht ausführbar). Neuer Helfer plain.scanner_cmd(target, rest) injiziert
   das Target überall → "python -m scanner https://site approve --all".

4. "Level: YELLOW" aber "Score: 0": Der angezeigte Score war nur der Inhalts-
   Diff-Score; KI-/Cloak-/Asset-Scores flossen ins Level, aber nicht in den
   Score. cmd_scan/cmd_check aggregieren jetzt alle Teil-Scores → Score und
   Level passen zusammen.

Tests: 254 grün (+4: Body==Report, Kommandos-mit-Target, Score-Anzeige,
Level-Badge). alerter._format_body delegiert nur noch an render_markdown.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Dieter Schlüter 2026-06-13 03:35:05 +02:00
commit 3ab313e733
5 changed files with 327 additions and 398 deletions

View file

@ -46,7 +46,8 @@ from .differ import (
score_diff,
)
from .extractor import extract_page
from .plain import ampel, ampel_zeile, klartext_befunde, was_tun
from .plain import ampel, ampel_zeile, klartext_befunde, scanner_cmd, was_tun
from .report import render_markdown
# Reihenfolge der Risiko-Level, um das schlechteste zu bestimmen
_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2}
@ -185,279 +186,10 @@ def _write_report(report: dict, reports_dir: str) -> tuple[Path, Path]:
json_path.write_text(
json.dumps(report, indent=2, ensure_ascii=False), encoding="utf-8"
)
md_path.write_text(_render_markdown(report), encoding="utf-8")
md_path.write_text(render_markdown(report), encoding="utf-8")
return json_path, md_path
def _render_markdown(r: dict) -> str:
a = r.get("assessment", {})
d = r.get("diff", {})
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"# 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"**Level:** {level_badge} ",
f"**Score:** {score}",
"",
"## Zusammenfassung",
"",
f"| Kennzahl | Wert |",
f"|---|---|",
f"| Geprüfte Seiten | {d.get('total_pages_checked', '?')} |",
f"| Geänderte Seiten | {d.get('changed_pages', 0)} |",
f"| Neue interne URLs | {len(d.get('new_internal_urls', []))} |",
f"| Fehlende URLs | {len(d.get('missing_internal_urls', []))} |",
f"| Neue externe Domains | {len(d.get('new_external_domains', []))} |",
f"| Kaputte Links (neu) | {len(d.get('new_broken_urls', []))} |",
f"| Kaputte Links (bekannt) | {len(d.get('known_broken_urls', []))} |",
f"| Unerwartetes JSON-LD | {len(d.get('unexpected_jsonld', []))} |",
f"| Verdächtige Dateinamen | {len(d.get('suspicious_filenames', []))} |",
f"| Whitelist-Verstösse | {len(r.get('whitelist_violations', []))} |",
f"| Crawl-Fehler | {len(r.get('crawl_errors', []))} |",
"",
"## Risikobewertung",
"",
]
for reason in a.get("reasons", []):
lines.append(f"- {reason}")
if not a.get("reasons"):
lines.append("- Keine Auffälligkeiten.")
if d.get("new_external_domains"):
lines += ["", "## Neue externe Domains", ""]
for dom in d["new_external_domains"]:
lines.append(f"- `{dom}`")
if d.get("new_internal_urls"):
lines += ["", "## Neue interne URLs", ""]
for url in d["new_internal_urls"]:
lines.append(f"- {url}")
if d.get("missing_internal_urls"):
lines += ["", "## Fehlende URLs (waren in Baseline)", ""]
for url in d["missing_internal_urls"]:
lines.append(f"- {url}")
if d.get("page_diffs"):
lines += ["", "## Geänderte Seiten", ""]
for pd in d["page_diffs"]:
lines += [f"### {pd['url']}", ""]
if pd.get("added_chars"):
lines.append(f"- Neuer Text: +{pd['added_chars']} Zeichen")
if pd.get("meta_diff"):
for field, change in pd["meta_diff"].items():
if isinstance(change, dict) and "old" in change:
lines.append(f"- {field}: `{change['old']}` → `{change['new']}`")
else:
lines.append(f"- {field}: {change}")
if pd.get("canonical_changed"):
lines.append(
f"- Canonical geändert: `{pd.get('old_canonical')}` → `{pd.get('new_canonical')}`"
)
if pd.get("new_hidden_content"):
for h in pd["new_hidden_content"]:
lines.append(f"- Hidden Content ({h.get('type')}): `{', '.join(h.get('indicators', []))}`")
if pd.get("new_meta_refresh"):
for mr in pd["new_meta_refresh"]:
lines.append(f"- Meta-Refresh → `{mr}`")
if pd.get("new_comment_links"):
for cl in pd["new_comment_links"]:
lines.append(f"- Kommentar-Link: `{cl}`")
if pd.get("new_inline_scripts"):
for s in pd["new_inline_scripts"]:
lines.append(f"- Inline-Script: `{', '.join(s.get('patterns', []))}`")
if pd.get("link_diff"):
for ltype, ld in pd["link_diff"].items():
for added in ld.get("added", []):
lines.append(f"- {ltype} hinzugefügt: `{added}`")
for removed in ld.get("removed", []):
lines.append(f"- {ltype} entfernt: `{removed}`")
if pd.get("text_diff"):
lines += ["", "```diff"]
lines += pd["text_diff"][:40]
if len(pd["text_diff"]) > 40:
lines.append(f"... ({len(pd['text_diff']) - 40} weitere Zeilen)")
lines.append("```")
lines.append("")
if d.get("new_pages_analysis"):
lines += ["", "## Neue Seiten (Erstanalyse)", ""]
for npa in d["new_pages_analysis"]:
problems = (
npa.get("hidden_content")
or npa.get("inline_scripts")
or npa.get("meta_refresh")
or npa.get("unexpected_jsonld")
or npa.get("comment_links")
)
flag = " [VERDAECHTIG]" if problems else ""
lines.append(f"- {npa['url']}{flag}")
new_broken = d.get("new_broken_urls", [])
known_broken = d.get("known_broken_urls", [])
if new_broken or known_broken:
lines += ["", "## Kaputte Links", ""]
for entry in new_broken:
lines.append(f"- [NEU] HTTP {entry['status']} `{entry['url']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
for entry in known_broken:
lines.append(f"- [bekannt, noch nicht behoben] HTTP {entry['status']} `{entry['url']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
if d.get("unexpected_jsonld"):
lines += ["", "## Unerwartetes JSON-LD", ""]
for entry in d["unexpected_jsonld"]:
lines.append(f"- `{entry['url']}`: @type={entry['type']}")
if d.get("suspicious_filenames"):
lines += ["", "## Verdächtige Dateinamen", ""]
for entry in d["suspicious_filenames"]:
lines.append(f"- `{entry['page']}` → `{entry['url']}` ({entry['link_type']})")
if r.get("whitelist_violations"):
lines += ["", "## Whitelist-Verstösse", ""]
for v in r["whitelist_violations"][:50]:
lines.append(f"- [{v.get('type')}] {v.get('url', '')} (Domain: {v.get('domain', '')})")
if r.get("missing_security_headers"):
lines += ["", "## Fehlende Security-Header", ""]
for url, hdrs in r["missing_security_headers"].items():
for h in hdrs:
lines.append(f"- `{url}`: {h}")
if r.get("crawl_errors"):
lines += ["", "## Crawl-Fehler", ""]
for e in r["crawl_errors"][:20]:
lines.append(f"- `{e.get('url', '?')}`: {e.get('error', '?')}")
# ---- 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']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
for entry in el.get("error_links", []):
lines.append(f"- Nicht erreichbar: `{entry['url']}` ({entry.get('error', '?')})")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
ai = r.get("ai_result")
if ai is not None:
lines += ["", "## KI-Inhaltsanalyse", ""]
if ai.get("skipped"):
lines.append(f"- Übersprungen: {ai['skipped']}")
else:
lines.append(
f"- {ai.get('checked', 0)} Inhalt(e) geprüft, "
f"{ai.get('cache_hits', 0)} aus Cache, {ai.get('api_calls', 0)} API-Abfrage(n)."
)
if ai.get("unchecked"):
lines.append(f"- ⚠ {len(ai['unchecked'])} Inhalt(e) **NICHT geprüft** "
"(KI-Dienst nicht erreichbar) — „nicht geprüft“ ≠ „sauber“:")
for u in ai["unchecked"][:20]:
ziel = u.get("asset_url") or u.get("url", "?")
lines.append(f" - [{u.get('kind', '?')}] `{ziel}`")
for f in ai.get("findings", []):
asset = f.get("asset_url")
ziel = f"`{asset}` (auf {f['url']})" if asset else f"`{f['url']}`"
lines.append(
f"- [{f['kind']}] {ziel}: **{f['category']}** "
f"(Schwere {f['severity']}, Konfidenz {f['confidence']:.0%})"
)
if f.get("explanation"):
lines.append(f" - Begründung: {f['explanation']}")
if f.get("fingerprint"):
lines.append(f" - Quittieren bei Fehlalarm: "
f"`python -m scanner ai-dismiss --hash {f['fingerprint']}`")
lines += [
"",
"---",
"",
"## Nächste Schritte",
"",
]
lev = a.get("level", "green")
if lev == "red":
lines += [
"1. **Sofort**: Website-Backend überprüfen, Dateien mit Backup vergleichen.",
"2. Dienstleister kontaktieren.",
"3. Nach Bereinigung: `python -m scanner approve --rebuild`",
]
elif lev == "yellow":
lines += [
"1. Alle markierten URLs manuell prüfen.",
"2. Legitime Änderungen freigeben: `python -m scanner approve --url <URL>`",
"3. Alle freigeben (wenn geprüft): `python -m scanner approve --all`",
]
else:
lines += ["Keine Aktion erforderlich."]
return "\n".join(lines) + "\n"
def _write_cloak_report(report: dict, reports_dir: str) -> tuple[Path, Path]:
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
@ -968,7 +700,7 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int:
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")
print(f"Bitte zuerst ausführen: {scanner_cmd(cfg['target'], 'scan')}")
return 1
baseline = bm.load_baseline()
@ -981,6 +713,7 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int:
ai_assess = score_ai_findings(ai_result, cfg)
overall = _worst_level([assessment["level"], ai_assess["level"]])
assessment = {**assessment, "level": overall, "exit_code": _LEVEL_ORDER[overall],
"score": assessment.get("score", 0) + ai_assess.get("score", 0),
"reasons": assessment.get("reasons", []) + ai_assess["reasons"]}
report = _build_report(
@ -988,7 +721,7 @@ def cmd_check(args: argparse.Namespace, cfg: dict) -> int:
snap.get("errors", []), snap["dir"], ai_result=ai_result,
)
_write_report(report, cfg["reports_dir"])
_print_summary(assessment, diff, ai_result=ai_result)
_print_summary(assessment, diff, ai_result=ai_result, target=cfg["target"])
send_alert(report, cfg)
return assessment["exit_code"]
@ -1010,6 +743,7 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int:
baseline = bm.load_baseline()
diff, assessment, wl_violations, missing_sec = _run_content_check(cfg, snap, baseline)
levels = [assessment["level"]]
score_total = assessment.get("score", 0) # alle Teil-Scores aufaddieren
# --- Fällige Wochen-Prüfungen automatisch mitlaufen lassen ---
cloak_diff = asset_diff = ext_links = None
@ -1020,11 +754,13 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int:
print(" Wöchentliche Tarnungs-Prüfung läuft mit ...")
cloak_diff, cloak_assess = _run_cloak_core(cfg, pages_normal=pages)
levels.append(cloak_assess["level"])
score_total += cloak_assess.get("score", 0)
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"])
score_total += asset_assess.get("score", 0)
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 ...")
@ -1040,11 +776,13 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int:
f"{ai_result['cache_hits']} aus Cache.")
ai_assess = score_ai_findings(ai_result, cfg)
levels.append(ai_assess["level"])
score_total += ai_assess.get("score", 0)
assessment["reasons"] = assessment.get("reasons", []) + ai_assess["reasons"]
# Gesamt-Level = schlechtestes aus Inhalt + Cloaking + Datei-Check + KI
# Gesamt-Level = schlechtestes aus Inhalt + Cloaking + Datei-Check + KI;
# Gesamt-Score = Summe aller Teil-Scores (damit Level und Score zusammenpassen).
overall = _worst_level(levels)
assessment = {**assessment, "level": overall,
assessment = {**assessment, "level": overall, "score": score_total,
"exit_code": _LEVEL_ORDER[overall]}
report = _build_report(
@ -1054,7 +792,7 @@ def cmd_scan(args: argparse.Namespace, cfg: dict) -> int:
ai_result=ai_result,
)
_write_report(report, cfg["reports_dir"])
_print_summary(assessment, diff, cloak_diff, asset_diff, ext_links, ai_result)
_print_summary(assessment, diff, cloak_diff, asset_diff, ext_links, ai_result, cfg["target"])
send_alert(report, cfg)
return assessment["exit_code"]
@ -1481,7 +1219,8 @@ def _print_summary(assessment: dict, diff: dict,
cloak_diff: dict | None = None,
asset_diff: dict | None = None,
ext_links: dict | None = None,
ai_result: dict | None = None) -> None:
ai_result: dict | None = None,
target: str = "") -> None:
level = assessment.get("level", "green")
print()
print(ampel_zeile(level))
@ -1489,7 +1228,7 @@ def _print_summary(assessment: dict, diff: dict,
for satz in klartext_befunde(diff, cloak_diff, asset_diff, ext_links, ai_result):
print(f"{satz}")
print()
for schritt in was_tun(level):
for schritt in was_tun(level, target):
print(f"{schritt}")
# Knappe technische Kennzahlen darunter (für den Dienstleister)
print()
@ -1497,7 +1236,7 @@ def _print_summary(assessment: dict, diff: dict,
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")
f"Vollständiger Bericht: {scanner_cmd(target, 'report')}")
def cmd_test_alert(args: argparse.Namespace, cfg: dict) -> int:

View file

@ -6,12 +6,12 @@ Alerts are only sent when the assessment level meets or exceeds min_level.
import logging
import os
import smtplib
from datetime import datetime, timezone
from email.message import EmailMessage
import requests
from .plain import ampel, klartext_befunde, was_tun
from .plain import ampel
from .report import render_markdown
logger = logging.getLogger(__name__)
@ -56,120 +56,9 @@ def _make_subject(report: dict) -> str:
def _format_body(report: dict) -> str:
assessment = report.get("assessment", {})
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"), report.get("ai_result"),
):
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 += [
"",
"=" * 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}")
_section(lines, "Neue externe Domains", diff.get("new_external_domains", []), "+")
_section(lines, "Neue interne URLs", diff.get("new_internal_urls", []), "+")
_section(lines, "Fehlende URLs", diff.get("missing_internal_urls", []), "-")
for pd in diff.get("page_diffs", [])[:10]:
lines += ["", f"--- Aenderung: {pd['url']} ---"]
if pd.get("text_diff"):
lines.append(f" Text-Diff: {pd['added_chars']} Zeichen zugefügt")
lines += [f" {l}" for l in pd["text_diff"][:30]]
if pd.get("new_hidden_content"):
lines.append(f" Hidden Content: {len(pd['new_hidden_content'])} neue(r) Fund(e)")
if pd.get("new_meta_refresh"):
lines.append(f" Meta-Refresh: {pd['new_meta_refresh']}")
if pd.get("canonical_changed"):
lines.append(f" Canonical geaendert: {pd.get('old_canonical')} -> {pd.get('new_canonical')}")
if pd.get("new_comment_links"):
lines.append(f" Kommentar-Links: {pd['new_comment_links']}")
# ---- KI-Inhaltsanalyse: jede beanstandete Datei mit URL + Begründung ----
ai = report.get("ai_result") or {}
if ai.get("findings"):
lines += ["", "--- KI-Inhaltsanalyse: beanstandete Inhalte ---"]
for f in ai["findings"]:
asset = f.get("asset_url")
ziel = f"{asset} (auf Seite {f['url']})" if asset else f["url"]
lines.append(f" [{f['kind']}] {ziel}")
lines.append(f" Befund: {f['category']} "
f"(Schwere {f['severity']}, Konfidenz {f.get('confidence', 0):.0%})")
if f.get("explanation"):
lines.append(f" Begründung: {f['explanation']}")
if f.get("fingerprint"):
lines.append(f" Quittieren: python -m scanner ai-dismiss --hash {f['fingerprint']}")
if ai.get("unchecked"):
lines += ["", "--- KI-Inhaltsanalyse: NICHT geprüfte Inhalte ---",
" Achtung: Der Prüfdienst war nicht erreichbar. „Nicht geprüft“ ≠ „sauber“ —"
" bitte beim nächsten Lauf erneut prüfen."]
for u in ai["unchecked"]:
ziel = u.get("asset_url") or u.get("url", "?")
lines.append(f" [{u.get('kind', '?')}] {ziel}")
lines += [
"",
"--- Empfehlung ---",
]
level_lower = assessment.get("level", "green")
if level_lower == "red":
lines += [
" SOFORT HANDELN: Website moeglicherweise kompromittiert.",
" Zugriff auf Hosting-Backend prüfen, Dateien vergleichen.",
]
elif level_lower == "yellow":
lines += [
" PRÜFEN: Unerwartete Aenderungen gefunden.",
" Alle markierten URLs manuell kontrollieren.",
" Falls legitim: python -m scanner approve --all",
]
else:
lines.append(" OK — nur zur Information.")
lines += [
"",
"--- Workflow ---",
" Details: python -m scanner report",
" Freigabe: python -m scanner approve --url <URL>",
" Rebuild: python -m scanner approve --rebuild",
"=" * 60,
]
return "\n".join(lines)
def _section(lines: list, title: str, items: list, prefix: str) -> None:
if not items:
return
lines += ["", f"--- {title} ---"]
for item in items[:50]:
lines.append(f" {prefix} {item}")
if len(items) > 50:
lines.append(f" ... und {len(items) - 50} weitere")
"""E-Mail-Body = derselbe Renderer wie der Dateibericht (eine Quelle der
Wahrheit). So enthalten Report und Mail garantiert denselben Inhalt."""
return render_markdown(report)
# ---------------------------------------------------------------------------

View file

@ -20,6 +20,12 @@ def ampel(level: str) -> tuple[str, str]:
return _AMPEL.get(level, ("", "Status unbekannt."))
def scanner_cmd(target: str, rest: str) -> str:
"""Baut ein direkt ausführbares Kommando mit Target, z. B.
'python -m scanner https://meine-seite.de approve --all'."""
return f"python -m scanner {target} {rest}".strip() if target else f"python -m scanner {rest}"
def ampel_zeile(level: str) -> str:
"""Einzeilige Ampel-Darstellung, z. B. '🟡 Bitte einmal nachschauen.'"""
symbol, satz = ampel(level)
@ -205,7 +211,7 @@ def klartext_befunde(
return saetze
def was_tun(level: str) -> list[str]:
def was_tun(level: str, target: str = "") -> list[str]:
"""Konkrete, laienverständliche Handlungsempfehlung je Level."""
if level == "red":
return [
@ -217,6 +223,6 @@ def was_tun(level: str) -> list[str]:
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",
"Gewollte Änderungen bestätigen Sie mit: " + scanner_cmd(target, "approve --all"),
]
return ["Sie müssen nichts tun."]

277
scanner/report.py Normal file
View file

@ -0,0 +1,277 @@
"""
Gemeinsamer Report-Renderer.
EINE Quelle der Wahrheit für den Dateibericht (report.md) UND den E-Mail-Body.
So können Report und Alarm-Mail nicht mehr auseinanderlaufen. Alle Kommandos
enthalten das Target (direkt ausführbar); der angezeigte Score ist der bereits
aggregierte Gesamt-Score (Inhalt + KI + Zusatzprüfungen).
"""
from .plain import ampel, klartext_befunde, scanner_cmd, was_tun
def render_markdown(r: dict) -> str:
a = r.get("assessment", {})
d = r.get("diff", {})
target = r.get("target", "")
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)
ai = r.get("ai_result")
# ---- Klartext-Teil (für den Betreiber) ----
lines = [
f"# Prüfbericht für Ihre Website — {target or '?'}",
"",
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"), ai):
lines.append(f"- {satz}")
lines += ["", "### Was sollten Sie tun?", ""]
for schritt in was_tun(lvl, target):
lines.append(f"- {schritt}")
# ---- Technischer Teil (für den Dienstleister) ----
ai_funde = len(ai.get("findings", [])) if ai else 0
ai_ungeprueft = len(ai.get("unchecked", [])) if ai else 0
lines += [
"",
"---",
"",
"# Technische Details (für Ihren Dienstleister)",
"",
f"**Level:** {level_badge} ",
f"**Score:** {score}",
"",
"## Zusammenfassung",
"",
"| Kennzahl | Wert |",
"|---|---|",
f"| Geprüfte Seiten | {d.get('total_pages_checked', '?')} |",
f"| Geänderte Seiten | {d.get('changed_pages', 0)} |",
f"| Neue interne URLs | {len(d.get('new_internal_urls', []))} |",
f"| Fehlende URLs | {len(d.get('missing_internal_urls', []))} |",
f"| Neue externe Domains | {len(d.get('new_external_domains', []))} |",
f"| Kaputte Links (neu) | {len(d.get('new_broken_urls', []))} |",
f"| Kaputte Links (bekannt) | {len(d.get('known_broken_urls', []))} |",
f"| Unerwartetes JSON-LD | {len(d.get('unexpected_jsonld', []))} |",
f"| Verdächtige Dateinamen | {len(d.get('suspicious_filenames', []))} |",
f"| KI-Funde (Text/Bild) | {ai_funde} |",
f"| KI nicht geprüft | {ai_ungeprueft} |",
f"| Whitelist-Verstösse | {len(r.get('whitelist_violations', []))} |",
f"| Crawl-Fehler | {len(r.get('crawl_errors', []))} |",
"",
"## Risikobewertung",
"",
]
for reason in a.get("reasons", []):
lines.append(f"- {reason}")
if not a.get("reasons"):
lines.append("- Keine Auffälligkeiten.")
if d.get("new_external_domains"):
lines += ["", "## Neue externe Domains", ""]
for dom in d["new_external_domains"]:
lines.append(f"- `{dom}`")
if d.get("new_internal_urls"):
lines += ["", "## Neue interne URLs", ""]
for url in d["new_internal_urls"]:
lines.append(f"- {url}")
if d.get("missing_internal_urls"):
lines += ["", "## Fehlende URLs (waren in Baseline)", ""]
for url in d["missing_internal_urls"]:
lines.append(f"- {url}")
if d.get("page_diffs"):
lines += ["", "## Geänderte Seiten", ""]
for pd in d["page_diffs"]:
lines += [f"### {pd['url']}", ""]
if pd.get("added_chars"):
lines.append(f"- Neuer Text: +{pd['added_chars']} Zeichen")
if pd.get("meta_diff"):
for field, change in pd["meta_diff"].items():
if isinstance(change, dict) and "old" in change:
lines.append(f"- {field}: `{change['old']}` → `{change['new']}`")
else:
lines.append(f"- {field}: {change}")
if pd.get("canonical_changed"):
lines.append(
f"- Canonical geändert: `{pd.get('old_canonical')}` → `{pd.get('new_canonical')}`"
)
if pd.get("new_hidden_content"):
for h in pd["new_hidden_content"]:
lines.append(f"- Hidden Content ({h.get('type')}): `{', '.join(h.get('indicators', []))}`")
if pd.get("new_meta_refresh"):
for mr in pd["new_meta_refresh"]:
lines.append(f"- Meta-Refresh → `{mr}`")
if pd.get("new_comment_links"):
for cl in pd["new_comment_links"]:
lines.append(f"- Kommentar-Link: `{cl}`")
if pd.get("new_inline_scripts"):
for s in pd["new_inline_scripts"]:
lines.append(f"- Inline-Script: `{', '.join(s.get('patterns', []))}`")
if pd.get("link_diff"):
for ltype, ld in pd["link_diff"].items():
for added in ld.get("added", []):
lines.append(f"- {ltype} hinzugefügt: `{added}`")
for removed in ld.get("removed", []):
lines.append(f"- {ltype} entfernt: `{removed}`")
if pd.get("text_diff"):
lines += ["", "```diff"]
lines += pd["text_diff"][:40]
if len(pd["text_diff"]) > 40:
lines.append(f"... ({len(pd['text_diff']) - 40} weitere Zeilen)")
lines.append("```")
lines.append("")
if d.get("new_pages_analysis"):
lines += ["", "## Neue Seiten (Erstanalyse)", ""]
for npa in d["new_pages_analysis"]:
problems = (
npa.get("hidden_content")
or npa.get("inline_scripts")
or npa.get("meta_refresh")
or npa.get("unexpected_jsonld")
or npa.get("comment_links")
)
flag = " [VERDAECHTIG]" if problems else ""
lines.append(f"- {npa['url']}{flag}")
new_broken = d.get("new_broken_urls", [])
known_broken = d.get("known_broken_urls", [])
if new_broken or known_broken:
lines += ["", "## Kaputte Links", ""]
for entry in new_broken:
lines.append(f"- [NEU] HTTP {entry['status']} `{entry['url']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
for entry in known_broken:
lines.append(f"- [bekannt, noch nicht behoben] HTTP {entry['status']} `{entry['url']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
if d.get("unexpected_jsonld"):
lines += ["", "## Unerwartetes JSON-LD", ""]
for entry in d["unexpected_jsonld"]:
lines.append(f"- `{entry['url']}`: @type={entry['type']}")
if d.get("suspicious_filenames"):
lines += ["", "## Verdächtige Dateinamen", ""]
for entry in d["suspicious_filenames"]:
lines.append(f"- `{entry['page']}` → `{entry['url']}` ({entry['link_type']})")
if r.get("whitelist_violations"):
lines += ["", "## Whitelist-Verstösse", ""]
for v in r["whitelist_violations"][:50]:
lines.append(f"- [{v.get('type')}] {v.get('url', '')} (Domain: {v.get('domain', '')})")
if r.get("missing_security_headers"):
lines += ["", "## Fehlende Security-Header", ""]
for url, hdrs in r["missing_security_headers"].items():
for h in hdrs:
lines.append(f"- `{url}`: {h}")
if r.get("crawl_errors"):
lines += ["", "## Crawl-Fehler", ""]
for e in r["crawl_errors"][:20]:
lines.append(f"- `{e.get('url', '?')}`: {e.get('error', '?')}")
# ---- 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']}`")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
for entry in el.get("error_links", []):
lines.append(f"- Nicht erreichbar: `{entry['url']}` ({entry.get('error', '?')})")
for page in entry.get("pages", [])[:3]:
lines.append(f" - gefunden auf: {page}")
if ai is not None:
lines += ["", "## KI-Inhaltsanalyse", ""]
if ai.get("skipped"):
lines.append(f"- Übersprungen: {ai['skipped']}")
else:
lines.append(
f"- {ai.get('checked', 0)} Inhalt(e) geprüft, "
f"{ai.get('cache_hits', 0)} aus Cache, {ai.get('api_calls', 0)} API-Abfrage(n)."
)
if ai.get("unchecked"):
lines.append(f"- ⚠ {len(ai['unchecked'])} Inhalt(e) **NICHT geprüft** "
"(KI-Dienst nicht erreichbar) — „nicht geprüft“ ≠ „sauber“:")
for u in ai["unchecked"][:20]:
ziel = u.get("asset_url") or u.get("url", "?")
lines.append(f" - [{u.get('kind', '?')}] `{ziel}`")
for f in ai.get("findings", []):
asset = f.get("asset_url")
ziel = f"`{asset}` (auf {f['url']})" if asset else f"`{f['url']}`"
lines.append(
f"- [{f['kind']}] {ziel}: **{f['category']}** "
f"(Schwere {f['severity']}, Konfidenz {f['confidence']:.0%})"
)
if f.get("explanation"):
lines.append(f" - Begründung: {f['explanation']}")
if f.get("fingerprint"):
lines.append(" - Quittieren bei Fehlalarm: "
f"`{scanner_cmd(target, 'ai-dismiss --hash ' + f['fingerprint'])}`")
lines += ["", "---", "", "## Nächste Schritte", ""]
lev = a.get("level", "green")
if lev == "red":
lines += [
"1. **Sofort**: Website-Backend überprüfen, Dateien mit Backup vergleichen.",
"2. Dienstleister kontaktieren.",
f"3. Nach Bereinigung: `{scanner_cmd(target, 'approve --rebuild')}`",
]
elif lev == "yellow":
lines += [
"1. Alle markierten URLs manuell prüfen.",
f"2. Legitime Änderungen freigeben: `{scanner_cmd(target, 'approve --url <URL>')}`",
f"3. Alle freigeben (wenn geprüft): `{scanner_cmd(target, 'approve --all')}`",
]
else:
lines += ["Keine Aktion erforderlich."]
return "\n".join(lines) + "\n"

View file

@ -93,7 +93,25 @@ class TestFormatBody:
def test_body_contains_level(self):
body = _format_body(_report("red"))
assert "RED" in body
assert "[ALARM]" in body # rotes Level-Badge im technischen Teil
assert "🔴" in body # Ampel-Symbol im Klartext-Teil
def test_body_equals_report_renderer(self):
"""E-Mail-Body muss exakt der Dateibericht sein (eine Quelle der Wahrheit)."""
from scanner.report import render_markdown
rep = _report("yellow")
assert _format_body(rep) == render_markdown(rep)
def test_commands_include_target_url(self):
"""Alle Scanner-Kommandos in der Mail enthalten das Target (direkt ausführbar)."""
body = _format_body(_report("yellow", "https://meine-seite.de"))
assert "python -m scanner https://meine-seite.de approve --all" in body
def test_score_reflects_findings_not_just_zero(self):
"""Bei YELLOW darf der angezeigte Score nicht 0 sein, wenn Gründe vorliegen."""
rep = _report("yellow")
rep["assessment"]["score"] = 40 # aggregierter Gesamt-Score (z. B. aus KI)
assert "Score:** 40" in _format_body(rep)
def test_body_contains_reasons(self):
body = _format_body(_report("yellow"))