PTah
364 lines
15 KiB
Python
364 lines
15 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 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_generated_by(event: Event) -> str:
|
|
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 _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:
|
|
product, version = _event_product_version(event)
|
|
return append_notification_source_html(
|
|
html_msg,
|
|
generated_by=_event_generated_by(event),
|
|
product=product,
|
|
product_version=version,
|
|
)
|
|
|
|
|
|
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"<b>{header}</b>\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"<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_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()
|
|
|
|
|
|
def _format_event_body_html(event: Event) -> str:
|
|
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 != "-":
|
|
return html.escape(body)
|
|
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)
|
|
return format_generic_event_html(event)
|
|
|
|
|
|
def format_event_telegram_html(event: Event) -> str:
|
|
return _append_event_source(_format_event_body_html(event), event)
|
|
|
|
|
|
def _format_rdg_html(event: Event) -> str:
|
|
details = _details_dict(event)
|
|
ok = event.type.endswith(".success")
|
|
header = "✅ RD Gateway: подключение" if ok else "❌ RD Gateway: ошибка"
|
|
msg = f"<b>{header}</b>\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))
|
|
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",
|
|
):
|
|
msg += "\n" + _format_event_body_html(event)
|
|
return append_notification_source_html(msg, generated_by="sac")
|