"""HTML Telegram message templates (RDP/SSH style, notif-30).""" from __future__ import annotations import html from datetime import datetime from typing import Any from app.models import Event, Host, Problem LOGON_TYPE_NAMES: dict[int, str] = { 2: "Интерактивный (консоль)", 3: "Сеть/RDP (Network)", 4: "Пакетный (Batch)", 5: "Сервис (Service)", 7: "Разблокировка (Unlock)", 8: "Сетевой с явными данными", 9: "Новые учетные данные", 10: "Удаленный интерактивный (RDP)", } def html_escape(value: Any) -> str: if value is None: return "" return html.escape(str(value), quote=False) def _details_dict(event: Event | None) -> dict[str, Any]: if event is None or not isinstance(event.details, dict): return {} return event.details def _detail(details: dict[str, Any], *keys: str, default: str = "-") -> str: for key in keys: val = details.get(key) if val is not None and str(val).strip() not in ("", "-"): return str(val).strip() return default def logon_type_label(logon_type: Any) -> str: try: code = int(logon_type) except (TypeError, ValueError): return html_escape(logon_type) if logon_type not in (None, "") else "-" name = LOGON_TYPE_NAMES.get(code, f"Тип {code}") return f"{html_escape(name)} ({code})" def host_label(host: Host | None, *, fallback: str = "unknown") -> str: if host is None: return html_escape(fallback) if host.display_name and host.display_name.strip(): return html_escape(host.display_name.strip()) return html_escape(host.hostname) def format_time(dt: datetime | None) -> str: if dt is None: return "-" try: return html_escape(dt.strftime("%d.%m.%Y %H:%M:%S")) except Exception: return html_escape(str(dt)) def _line(emoji: str, label: str, value: str) -> str: return f"{emoji} {label}: {value}\n" def format_rdp_login_html(event: Event) -> str: details = _details_dict(event) is_success = event.type == "rdp.login.success" header = "✅ УСПЕШНЫЙ ВХОД" if is_success else "❌ НЕУДАЧНАЯ ПОПЫТКА" win_id = _detail(details, "event_id_windows", default="") if not win_id and event.type == "rdp.login.failed": win_id = "4625" elif not win_id and is_success: win_id = "4624" user = html_escape(_detail(details, "user", "username")) ip = html_escape(_detail(details, "ip_address", "source_ip", "ip")) workstation = html_escape(_detail(details, "workstation_name", "computer_name")) process = html_escape(_detail(details, "process_name", "process", default="")) logon = logon_type_label(details.get("logon_type")) msg = f"{header}\n" msg += _line("👤", "Пользователь", user) msg += _line("🏢", "Сервер", host_label(event.host)) msg += _line("🖥️", "Рабочая станция", workstation) msg += _line("🌐", "IP адрес", ip) if process and process != "-": msg += _line("⚙️", "Процесс", process) msg += _line("🔑", "Тип входа", logon) msg += _line("🕐", "Время", format_time(event.occurred_at)) if win_id: msg += f"🔢 Event ID: {html_escape(win_id)}" return msg.rstrip() def format_ssh_auth_html(event: Event) -> str: details = _details_dict(event) is_success = event.type == "ssh.login.success" header = "✅ SSH: успешный вход" if is_success else "❌ SSH: неудачная попытка" user = html_escape(_detail(details, "user", "username")) ip = html_escape(_detail(details, "source_ip", "ip_address", "ip")) port = html_escape(_detail(details, "port", default="")) attempt = _detail(details, "attempt_number", default="") max_attempts = _detail(details, "max_attempts", default="") msg = f"{header}\n" msg += _line("👤", "Пользователь", user) msg += _line("🏢", "Хост", host_label(event.host)) msg += _line("🌐", "IP", ip) if port and port != "-": msg += _line("🔌", "Порт", port) if attempt != "-" and max_attempts != "-": msg += _line("🔢", "Попытка", f"{html_escape(attempt)} / {html_escape(max_attempts)}") msg += _line("🕐", "Время", format_time(event.occurred_at)) msg += f"📋 {html_escape(event.type)}" return msg.rstrip() def format_sudo_html(event: Event) -> str: details = _details_dict(event) risk = _detail(details, "risk_level", default="") header = "⚠️ SUDO" if risk and risk.lower() == "critical": header = "🔥 SUDO (critical)" msg = f"{header}\n" msg += _line("👤", "Пользователь", html_escape(_detail(details, "user", "username"))) msg += _line("🏢", "Хост", host_label(event.host)) msg += _line("▶️", "Команда", html_escape(_detail(details, "command", default=event.summary[:200]))) run_as = _detail(details, "run_as", default="") if run_as != "-": msg += _line("👥", "От имени", html_escape(run_as)) pwd = _detail(details, "pwd", default="") if pwd != "-": msg += _line("📁", "Каталог", html_escape(pwd)) if risk != "-": msg += _line("⚡", "Risk", html_escape(risk)) msg += _line("🕐", "Время", format_time(event.occurred_at)) return msg.rstrip() def format_generic_event_html(event: Event) -> str: sev = event.severity.upper() msg = f"🚨 SAC: {html_escape(event.title)}\n" msg += _line("🏢", "Хост", host_label(event.host)) msg += _line("📋", "Тип", html_escape(event.type)) msg += _line("⚡", "Severity", html_escape(sev)) msg += _line("🕐", "Время", format_time(event.occurred_at)) if event.summary: msg += f"\n{html_escape(event.summary)}" return msg.rstrip() def format_daily_report_html(event: Event) -> str: details = _details_dict(event) report_html = details.get("report_html") if isinstance(report_html, str) and report_html.strip(): # Уже HTML от агента или SAC return report_html.strip() body = _detail(details, "report_body", default=event.summary) if body != "-": escaped = html.escape(body) return ( f"📊 {html_escape(event.title)}\n" f'