"""HTML Telegram message templates (RDP/SSH style, notif-30).""" from __future__ import annotations import html import re from datetime import datetime from typing import Any from app.models import Event, Host, Problem from app.services.daily_report_format import ( append_notification_source_html, normalize_report_body, resolve_product_label, ) LOGON_TYPE_NAMES: dict[int, str] = { 2: "Интерактивный (консоль)", 3: "Сеть/RDP (Network)", 4: "Пакетный (Batch)", 5: "Сервис (Service)", 7: "Разблокировка (Unlock)", 8: "Сетевой с явными данными", 9: "Новые учетные данные", 10: "Удаленный интерактивный (RDP)", } # Telegram parse_mode=HTML: только теги из Bot API (без div/br/p и т.д.) _TELEGRAM_BR = re.compile(r"", re.I) _TELEGRAM_DROP_TAGS = re.compile( r"]*>", re.I, ) _TELEGRAM_SPAN_OPEN = re.compile(r"]*\btg-spoiler\b)[^>]*>", re.I) _TELEGRAM_SPAN_CLOSE = re.compile(r"", re.I) def sanitize_telegram_html(text: str) -> str: """Приводит HTML отчёта (UI/агент) к тегам, допустимым в Telegram.""" out = _TELEGRAM_BR.sub("\n", text) out = _TELEGRAM_DROP_TAGS.sub("", out) out = _TELEGRAM_SPAN_OPEN.sub("", out) out = _TELEGRAM_SPAN_CLOSE.sub("", out) return out.strip() 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 _host_display_name_base(display_name: str | None) -> str: if not display_name or not display_name.strip(): return "" return display_name.strip().split("(", 1)[0].strip() def _rdp_workstation_is_redundant(workstation: str, host: Host | None) -> bool: """Скрыть Workstation Name, если пусто/«-» или совпадает с hostname/display_name сервера.""" ws = (workstation or "").strip() if not ws or ws in ("-", "N/A", "?"): return True if host is None: return False ws_key = ws.casefold() if host.hostname and ws_key == host.hostname.strip().casefold(): return True dn = (host.display_name or "").strip() if dn and ws_key == dn.casefold(): return True base = _host_display_name_base(host.display_name) if base and ws_key == base.casefold(): return True return False 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 _event_content_author(event: Event) -> str: """Кто сгенерировал содержимое события (ingest / отчёт).""" details = _details_dict(event) gb = details.get("generated_by") if isinstance(gb, str) and gb.strip().lower() in ("agent", "sac"): return gb.strip().lower() stats = details.get("stats") if isinstance(stats, dict) and stats.get("sac_generated"): return "sac" if event.type in ("report.daily.rdp", "report.daily.ssh") and details.get("report_body"): return "agent" return "agent" def _notification_delivered_by(event: Event) -> str: """Кто доставил сообщение в Telegram (footer «📡 Оповещение»). Шаблоны здесь используются только при отправке из SAC. Агент добавляет footer сам. """ details = _details_dict(event) via = details.get("telegram_via") if isinstance(via, str) and via.strip().lower() in ("agent", "sac"): return via.strip().lower() if _event_content_author(event) == "sac": return "sac" # ingest от агента, канал Telegram — SAC (exclusive / notify_lifecycle) return "sac" def _event_product_version(event: Event) -> tuple[str | None, str | None]: details = _details_dict(event) platform = None if event.type == "report.daily.ssh": platform = "ssh" elif event.type == "report.daily.rdp": platform = "windows" elif event.host is not None: if event.host.os_family == "linux": platform = "ssh" elif event.host.os_family == "windows": platform = "windows" product, version = resolve_product_label(event.host, platform) stats = details.get("stats") if isinstance(stats, dict): av = stats.get("agent_version") or stats.get("product_version") if av: version = str(av).strip() payload = event.payload if isinstance(event.payload, dict) else {} source = payload.get("source") if isinstance(source, dict): if source.get("product"): product = str(source["product"]).strip() if source.get("product_version"): version = str(source["product_version"]).strip() return product, version def _append_event_source(html_msg: str, event: Event) -> str: delivered = _notification_delivered_by(event) product, version = _event_product_version(event) return append_notification_source_html( html_msg, generated_by=delivered, product=product if delivered == "agent" else None, product_version=version if delivered == "agent" else None, ) 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_raw = _detail(details, "workstation_name", "computer_name") workstation = html_escape(workstation_raw) 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)) if not _rdp_workstation_is_redundant(workstation_raw, 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_rdp_shadow_html(event: Event) -> str: details = _details_dict(event) action = _detail(details, "shadow_action", default="") if action == "control_stopped" or event.type.endswith(".stopped"): header = "🎭 RDS SHADOW CONTROL — остановлено" elif action == "control_permission" or event.type.endswith(".permission"): header = "🎭 RDS SHADOW CONTROL — разрешение" else: header = "🎭 RDS SHADOW CONTROL — начато" msg = f"{header}\n" msg += _line("🏢", "Сервер", host_label(event.host)) msg += _line("👤", "Администратор", html_escape(_detail(details, "shadower_user", "user"))) msg += _line("🎯", "Сессия пользователя", html_escape(_detail(details, "target_user", default="-"))) sid = _detail(details, "session_id", "target_session_id", default="") if sid != "-": msg += _line("🔢", "Session ID", html_escape(sid)) msg += _line("🕐", "Время", format_time(event.occurred_at)) win_id = _detail(details, "event_id_windows", default="") if win_id != "-": msg += f"🔢 Event ID: {html_escape(win_id)} (RemoteConnectionManager)" return msg.rstrip() def format_winrm_session_html(event: Event) -> str: details = _details_dict(event) msg = "⚠️ WinRM / Enter-PSSession — удалённая shell\n" msg += _line("🏢", "Сервер", host_label(event.host)) msg += _line("👤", "Пользователь", html_escape(_detail(details, "user", "username"))) ip = _detail(details, "source_ip", "ip_address", "ip") if ip != "-": msg += _line("🌐", "IP источника", html_escape(ip)) uri = _detail(details, "resource_uri", default="") if uri != "-": msg += _line("🔗", "ResourceUri", html_escape(uri)) msg += _line("🕐", "Время", format_time(event.occurred_at)) win_id = _detail(details, "event_id_windows", default="") if win_id != "-": msg += f"🔢 Event ID: {html_escape(win_id)} (WinRM Operational)" return msg.rstrip() def format_smb_admin_share_html(event: Event) -> str: details = _details_dict(event) msg = "📁 Доступ к админ-шару (5140)\n" msg += _line("🏢", "Сервер", host_label(event.host)) msg += _line("👤", "Пользователь", html_escape(_detail(details, "user", "username"))) ip = _detail(details, "source_ip", "ip_address", "ip") if ip != "-": msg += _line("🌐", "IP клиента", html_escape(ip)) share = _detail(details, "share_name", default="") if share != "-": msg += _line("📂", "Share", html_escape(share)) path = _detail(details, "share_path", default="") if path != "-": msg += _line("📍", "Путь", html_escape(path)) target = _detail(details, "relative_target", default="") if target != "-": msg += _line("📄", "Объект", html_escape(target)) msg += _line("🕐", "Время", format_time(event.occurred_at)) win_id = _detail(details, "event_id_windows", default="") if win_id != "-": msg += f"🔢 Event ID: {html_escape(win_id)} (File Share)" return msg.rstrip() def _format_inventory_change_line(change: dict[str, Any]) -> str: field = html_escape(str(change.get("field") or "?")) old = change.get("old") new = change.get("new") return f"• {field}: {html_escape(old)} → {html_escape(new)}" def format_agent_inventory_html(event: Event) -> str: details = _details_dict(event) changes = details.get("hardware_changes") if isinstance(changes, list) and changes: msg = "⚠️ Изменилось железо хоста\n" msg += _line("🏢", "Хост", host_label(event.host)) msg += _line("🕐", "Время", format_time(event.occurred_at)) msg += "\n" for item in changes: if isinstance(item, dict): msg += _format_inventory_change_line(item) + "\n" if event.summary: msg += f"\n{html_escape(event.summary)}" return msg.rstrip() inv = details.get("inventory") if not isinstance(inv, dict): inv = {} msg = "🖥️ Инвентаризация хоста\n" msg += _line("🏢", "Хост", host_label(event.host)) windows = inv.get("windows") if isinstance(inv.get("windows"), dict) else {} os_name = windows.get("product_name") or event.host.os_version if event.host else None if os_name: msg += _line("💿", "ОС", html_escape(os_name)) mem = inv.get("memory_gb") if mem is not None: msg += _line("🧠", "RAM", f"{html_escape(mem)} GB") procs = inv.get("processor") if isinstance(procs, list) and procs: names = [str(p.get("name") or "").strip() for p in procs if isinstance(p, dict)] names = [n for n in names if n] if names: msg += _line("⚙️", "CPU", html_escape("; ".join(names))) 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() _LIFECYCLE_HEADERS: dict[str, str] = { "started": "✅ Агент запущен", "stopped": "⚠️ Агент остановлен", "settings_reloaded": "🔄 Настройки агента перечитаны", } _LIFECYCLE_TRIGGER_LABELS: dict[str, str] = { "boot": "загрузка ОС / задача планировщика", "deploy_recycle": "обновление скрипта (recycle)", "settings_reload": "graceful restart (settings)", "manual_recycle": "ручной recycle", "shutdown": "остановка процесса", } def _format_lifecycle_body_html(body: str) -> str: text = body.replace("\r\n", "\n").strip() if not text: return "" parts: list[str] = [] for i, line in enumerate(text.split("\n")): esc = html.escape(line) if i == 0 and line.strip(): parts.append(f"{esc.strip()}") else: parts.append(esc) return "\n".join(parts).strip() def format_lifecycle_html(event: Event) -> str: details = _details_dict(event) notification_body = details.get("notification_body") if isinstance(notification_body, str) and notification_body.strip(): return _format_lifecycle_body_html(notification_body) lifecycle = _detail(details, "lifecycle", default="started").strip().lower() trigger = _detail(details, "trigger", default="").strip().lower() header = _LIFECYCLE_HEADERS.get(lifecycle, f"📋 Lifecycle: {html_escape(lifecycle)}") msg = f"{header}\n" msg += _line("🏢", "Хост", host_label(event.host)) product = "" version = "" payload = event.payload if isinstance(event.payload, dict) else {} source = payload.get("source") if isinstance(source, dict): product = str(source.get("product") or "").strip() version = str(source.get("product_version") or "").strip() if product or version: label = " ".join(x for x in (product, version) if x).strip() if label: msg += _line("📦", "Версия", html_escape(label)) if trigger and trigger != "-": trigger_human = _LIFECYCLE_TRIGGER_LABELS.get(trigger, trigger) msg += _line("🔁", "Причина", html_escape(trigger_human)) msg += _line("🕐", "Время", format_time(event.occurred_at)) if event.summary: msg += f"\n{html_escape(event.summary)}" return msg.rstrip() def _format_event_body_html(event: Event) -> str: if event.type == "agent.lifecycle": return format_lifecycle_html(event) if event.type in ("report.daily.ssh", "report.daily.rdp"): details = _details_dict(event) platform = "windows" if event.type == "report.daily.rdp" else "ssh" body = _detail(details, "report_body", default="") if body and body != "-": body = normalize_report_body(body, event.host, platform) parts: list[str] = [] for i, line in enumerate(body.split("\n")): esc = html.escape(line) if i == 0 and "ЕЖЕДНЕВНЫЙ ОТЧЕТ" in line: parts.append(f"{esc.strip()}") else: parts.append(esc) return "\n".join(parts) report_html = details.get("report_html") if isinstance(report_html, str) and report_html.strip(): text = sanitize_telegram_html(report_html.strip()) return re.sub(r"\n{3,}", "\n\n", text) body = _detail(details, "report_body", default=event.summary) if body != "-": header = ( "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS" if event.type == "report.daily.rdp" else "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА" ) msg = f"{html.escape(header)}\n" msg += _line("🏢", "Сервер", host_label(event.host)) msg += f"\n{html.escape(body)}" return msg return format_generic_event_html(event) if event.type in ("rdp.login.success", "rdp.login.failed"): return format_rdp_login_html(event) if event.type in ("ssh.login.success", "ssh.login.failed"): return format_ssh_auth_html(event) if event.type == "privilege.sudo.command": return format_sudo_html(event) if event.type.startswith("rdg."): return _format_rdg_html(event) if event.type.startswith("rdp.shadow."): return format_rdp_shadow_html(event) if event.type.startswith("winrm."): return format_winrm_session_html(event) if event.type.startswith("smb."): return format_smb_admin_share_html(event) if event.type == "agent.inventory": return format_agent_inventory_html(event) return format_generic_event_html(event) def format_event_telegram_html(event: Event) -> str: return _append_event_source(_format_event_body_html(event), event) _TELEGRAM_INLINE_TAGS = re.compile(r"]*)?>", re.I) _TELEGRAM_ANCHOR_OPEN = re.compile(r"]*>", re.I) _TELEGRAM_ANCHOR_CLOSE = re.compile(r"", re.I) _MOBILE_PUSH_BODY_MAX = 3500 def telegram_html_to_plaintext(text: str) -> str: """Plain text for Seaca push from the same HTML templates as Telegram.""" out = sanitize_telegram_html(text) out = _TELEGRAM_INLINE_TAGS.sub("", out) out = _TELEGRAM_ANCHOR_OPEN.sub("", out) out = _TELEGRAM_ANCHOR_CLOSE.sub("", out) out = html.unescape(out) out = re.sub(r"\n{3,}", "\n\n", out) return out.strip() def format_event_mobile_push(event: Event) -> tuple[str, str]: """Title + body for FCM data payload (mirrors Telegram wording).""" plain = telegram_html_to_plaintext(format_event_telegram_html(event)) lines = [line.strip() for line in plain.split("\n") if line.strip()] if lines: title = lines[0] body = "\n".join(lines[1:]).strip() if not body: body = (event.summary or "").strip() else: title = f"[{event.severity}] {event.title}" body = (event.summary or "").strip() if len(title) > 120: title = title[:117] + "…" if len(body) > _MOBILE_PUSH_BODY_MAX: body = body[: _MOBILE_PUSH_BODY_MAX - 1] + "…" return title, body def _format_rdg_html(event: Event) -> str: details = _details_dict(event) if event.type.endswith(".success"): header = "✅ RD Gateway: подключение" elif event.type.endswith(".disconnected"): header = "ℹ️ RD Gateway: отключение" else: header = "❌ RD Gateway: ошибка" msg = f"{header}\n" msg += _line("👤", "Пользователь", html_escape(_detail(details, "user", "username"))) msg += _line("🏢", "Хост", host_label(event.host)) msg += _line("🌐", "Внешний IP", html_escape(_detail(details, "external_ip", "ip_address"))) msg += _line("🏠", "Внутренний IP", html_escape(_detail(details, "internal_ip", default="-"))) err = _detail(details, "gateway_error_code", "error_code", default="") if err != "-": msg += _line("⚠️", "Код ошибки", html_escape(err)) dur = _detail(details, "session_duration_sec", default="") if dur not in ("-", "0", ""): msg += _line("⏱️", "Длительность", f"{html_escape(dur)} с") msg += _line("🕐", "Время", format_time(event.occurred_at)) win_id = _detail(details, "event_id_windows", default="") if win_id != "-": msg += f"\n🔢 Event ID: {html_escape(win_id)}" return msg.rstrip() def format_problem_telegram_html(problem: Problem, event: Event | None = None) -> str: msg = "🔥 SAC Problem\n" msg += _line("🏢", "Хост", host_label(problem.host)) msg += _line("📏", "Severity", html_escape(problem.severity.upper())) if problem.rule_id: msg += _line("📐", "Правило", html_escape(problem.rule_id)) msg += _line("🔢", "Событий", html_escape(problem.event_count)) msg += _line("🕐", "Last seen", format_time(problem.last_seen_at)) msg += f"\n{html_escape(problem.title)}\n{html_escape(problem.summary)}" if event is not None: msg += f"\n\nТриггер: {html_escape(event.type)} ({html_escape(event.severity)})" if event.type in ( "rdp.login.success", "rdp.login.failed", "ssh.login.success", "ssh.login.failed", "rdp.shadow.control.started", "rdp.shadow.control.stopped", "rdp.shadow.control.permission", "winrm.session.started", "smb.admin_share.access", ): msg += "\n" + _format_event_body_html(event) elif problem.rule_id == "rule:host_silence": msg += "\n\nТриггер: периодическая проверка SAC (host_silence scan)" return append_notification_source_html(msg, generated_by="sac")