Dieter Schlüter
d06e3d3dfd
Bisher: Konnte die KI kein Verdikt liefern (alle Modellstufen scheitern),
ging der Inhalt still als grün durch — für ein Sicherheitswerkzeug ein
Falsch-Negativ-Risiko ("konnte nicht prüfen" ≠ "sauber").
Zwei Maßnahmen:
1. Retry: _classify wiederholt die ganze Modell-Kette bei Komplettausfall
bis max_retries (Default 1) mit Backoff — fängt transiente Aussetzer ab.
2. Tracking: nicht prüfbare Inhalte landen in ai_result["unchecked"] (statt
stillem skip) und werden sichtbar in Terminal, Report (Markdown), diff-only
und E-Mail — jeweils mit URL. score_ai_findings nimmt sie als Grund auf;
mit unchecked_level="yellow" heben sie das Level auf mindestens Gelb an
(Default "warn" = nur Hinweis, Ampel unberührt).
Config: max_retries, retry_backoff_seconds, unchecked_level (ai_analysis).
Tests: 230 grün (+ Retry-Recovery, unchecked-Tracking, unchecked_level).
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
225 lines
8.4 KiB
Python
225 lines
8.4 KiB
Python
"""
|
|
Alert dispatch: e-mail (smtplib) and HTTP webhook.
|
|
|
|
Alerts are only sent when the assessment level meets or exceeds min_level.
|
|
"""
|
|
import logging
|
|
import os
|
|
import smtplib
|
|
from datetime import datetime, timezone
|
|
from email.message import EmailMessage
|
|
|
|
import requests
|
|
|
|
from .plain import ampel, klartext_befunde, was_tun
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
_LEVEL_ORDER = {"green": 0, "yellow": 1, "red": 2}
|
|
|
|
|
|
def send_alert(report: dict, cfg: dict) -> None:
|
|
"""Send alert if the report's level is at or above min_level."""
|
|
alerting = cfg.get("alerting", {})
|
|
level = report.get("assessment", {}).get("level", "green")
|
|
min_level = alerting.get("min_level", "yellow")
|
|
|
|
if _LEVEL_ORDER.get(level, 0) < _LEVEL_ORDER.get(min_level, 1):
|
|
logger.debug("Alert suppressed: level=%s < min_level=%s", level, min_level)
|
|
return
|
|
|
|
subject = _make_subject(report)
|
|
body = _format_body(report)
|
|
|
|
email_cfg = alerting.get("email", {})
|
|
if email_cfg.get("enabled"):
|
|
_send_email(subject, body, email_cfg)
|
|
|
|
webhook_cfg = alerting.get("webhook", {})
|
|
if webhook_cfg.get("enabled") and webhook_cfg.get("url"):
|
|
_send_webhook(webhook_cfg["url"], report, subject)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Formatting
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _make_subject(report: dict) -> str:
|
|
level = report.get("assessment", {}).get("level", "green")
|
|
target = report.get("target", "?")
|
|
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:
|
|
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")
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Transport
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _send_email(subject: str, body: str, cfg: dict) -> None:
|
|
password = os.environ.get(cfg.get("smtp_password_env", "SCANNER_SMTP_PASSWORD"), "")
|
|
recipients = cfg.get("to", [])
|
|
if not recipients:
|
|
logger.warning("E-Mail alert enabled but no recipients configured.")
|
|
return
|
|
|
|
msg = EmailMessage()
|
|
msg["Subject"] = subject
|
|
msg["From"] = cfg.get("from", "scanner@localhost")
|
|
msg["To"] = ", ".join(recipients)
|
|
msg.set_content(body)
|
|
|
|
host = cfg.get("smtp_host", "localhost")
|
|
port = cfg.get("smtp_port", 587)
|
|
use_tls = cfg.get("smtp_tls", True)
|
|
|
|
try:
|
|
with smtplib.SMTP(host, port) as server:
|
|
if use_tls:
|
|
server.starttls()
|
|
user = cfg.get("smtp_user", "")
|
|
if user and password:
|
|
server.login(user, password)
|
|
server.send_message(msg)
|
|
logger.info("Alert e-mail sent to %s", recipients)
|
|
except Exception as exc:
|
|
logger.error("E-mail alert failed: %s", exc)
|
|
|
|
|
|
def _send_webhook(url: str, report: dict, subject: str) -> None:
|
|
payload = {
|
|
"text": subject,
|
|
"target": report.get("target"),
|
|
"level": report.get("assessment", {}).get("level"),
|
|
"score": report.get("assessment", {}).get("score"),
|
|
"reasons": report.get("assessment", {}).get("reasons", []),
|
|
"new_external_domains": report.get("diff", {}).get("new_external_domains", []),
|
|
"new_internal_urls": report.get("diff", {}).get("new_internal_urls", []),
|
|
"missing_internal_urls": report.get("diff", {}).get("missing_internal_urls", []),
|
|
"changed_pages": report.get("diff", {}).get("changed_pages", 0),
|
|
}
|
|
try:
|
|
resp = requests.post(url, json=payload, timeout=10)
|
|
logger.info("Webhook delivered: HTTP %s", resp.status_code)
|
|
except Exception as exc:
|
|
logger.error("Webhook alert failed: %s", exc)
|