diff --git a/scanner/__main__.py b/scanner/__main__.py index 2748999..0cba4fc 100644 --- a/scanner/__main__.py +++ b/scanner/__main__.py @@ -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 `", - "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: diff --git a/scanner/alerter.py b/scanner/alerter.py index 48f01ba..fe6ca52 100644 --- a/scanner/alerter.py +++ b/scanner/alerter.py @@ -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 ", - " 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) # --------------------------------------------------------------------------- diff --git a/scanner/plain.py b/scanner/plain.py index 512f5fe..b55137e 100644 --- a/scanner/plain.py +++ b/scanner/plain.py @@ -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."] diff --git a/scanner/report.py b/scanner/report.py new file mode 100644 index 0000000..f1f5655 --- /dev/null +++ b/scanner/report.py @@ -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 ')}`", + f"3. Alle freigeben (wenn geprüft): `{scanner_cmd(target, 'approve --all')}`", + ] + else: + lines += ["Keine Aktion erforderlich."] + + return "\n".join(lines) + "\n" diff --git a/tests/test_alerter.py b/tests/test_alerter.py index cf93bd5..ec187b9 100644 --- a/tests/test_alerter.py +++ b/tests/test_alerter.py @@ -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"))