PTah
985d51c4ba
Add session_duration_sec to EventSummary and a Duration column (HH:MM:SS / Nd HH:MM:SS).
605 lines
25 KiB
Python
605 lines
25 KiB
Python
"""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"<br\s*/?>", re.I)
|
||
_TELEGRAM_DROP_TAGS = re.compile(
|
||
r"</?(?:div|p|ul|ol|li|hr|h[1-6]|table|tr|td|th|thead|tbody|body|html|head)\b[^>]*>",
|
||
re.I,
|
||
)
|
||
_TELEGRAM_SPAN_OPEN = re.compile(r"<span\b(?![^>]*\btg-spoiler\b)[^>]*>", re.I)
|
||
_TELEGRAM_SPAN_CLOSE = re.compile(r"</span>", 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"<b>{header}</b>\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"<b>{header}</b>\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"<b>{header}</b>\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"<b>{header}</b>\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 = "<b>⚠️ WinRM / Enter-PSSession — удалённая shell</b>\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 = "<b>📁 Доступ к админ-шару (5140)</b>\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"• <b>{field}</b>: {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 = "<b>⚠️ Изменилось железо хоста</b>\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 = "<b>🖥️ Инвентаризация хоста</b>\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"<b>🚨 SAC: {html_escape(event.title)}</b>\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"<b>{esc.strip()}</b>")
|
||
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"<b>{header}</b>\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"<b>{esc.strip()}</b>")
|
||
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"<b>{html.escape(header)}</b>\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"</?(?:b|i|u|s|code|pre|tg-spoiler)(\s[^>]*)?>", re.I)
|
||
_TELEGRAM_ANCHOR_OPEN = re.compile(r"<a\b[^>]*>", re.I)
|
||
_TELEGRAM_ANCHOR_CLOSE = re.compile(r"</a>", 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:
|
||
from app.services.rdg_display import build_rdg_display
|
||
|
||
details = _details_dict(event)
|
||
display = build_rdg_display(event)
|
||
if event.type.endswith(".success"):
|
||
header = "✅ RDS через RD Gateway"
|
||
elif event.type.endswith(".disconnected"):
|
||
header = "ℹ️ RDS отключение (RD Gateway)"
|
||
else:
|
||
header = "❌ RDS ошибка (RD Gateway)"
|
||
msg = f"<b>{header}</b>\n"
|
||
if display is not None and display.access_path:
|
||
msg += _line("🔀", "Путь", html_escape(display.access_path))
|
||
msg += _line("👤", "Пользователь", html_escape(_detail(details, "user", "username")))
|
||
msg += _line("🚪", "Шлюз RDG", host_label(event.host))
|
||
msg += _line("🌐", "Внешний IP", html_escape(_detail(details, "external_ip", "ip_address")))
|
||
msg += _line("🖥️", "Рабочий ПК", html_escape(_detail(details, "internal_ip", default="-")))
|
||
err = _detail(details, "gateway_error_code", "error_code", default="")
|
||
if err != "-":
|
||
msg += _line("⚠️", "Код ошибки", html_escape(err))
|
||
from app.services.session_duration import extract_session_duration_sec, format_session_duration
|
||
|
||
dur_sec = extract_session_duration_sec(details if isinstance(details, dict) else None)
|
||
if dur_sec is not None and dur_sec > 0:
|
||
msg += _line("⏱️", "Длительность", html_escape(format_session_duration(dur_sec)))
|
||
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 = "<b>🔥 SAC Problem</b>\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<b>{html_escape(problem.title)}</b>\n{html_escape(problem.summary)}"
|
||
if event is not None:
|
||
msg += f"\n\n<i>Триггер:</i> {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<i>Триггер:</i> периодическая проверка SAC (host_silence scan)"
|
||
return append_notification_source_html(msg, generated_by="sac")
|