""" 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']}") 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") # --------------------------------------------------------------------------- # 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)