security-alert-center/backend/app/services/telegram_templates.py

"""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"<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_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_telegram_html(event: Event) -> str:
    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)
    return format_generic_event_html(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"):
            msg += "\n" + format_event_telegram_html(event)
    return msg.rstrip()