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

"""Единый текст и stats для report.daily.ssh / report.daily.rdp (агент и SAC)."""

from __future__ import annotations

import html
import re
from datetime import datetime
from typing import Any, Literal

from app.models import Host

Platform = Literal["ssh", "windows"]

RDP_BAN_TYPES = frozenset({"rdp.ip.banned", "rdp.ip.ban"})
SSH_BAN_TYPES = frozenset({"ssh.ip.banned"})

_SERVER_LINE_RE = re.compile(r"(?m)^🖥️\s*Сервер\s*:")
_ACTIVE_USERS_HEADER_RE = re.compile(r"^👥\s*АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ")
_ACTIVE_USERS_EMPTY_LINE_RE = re.compile(
    r"^\(?нет данных|нет активных пользователей",
    re.IGNORECASE,
)
_AGENT_VERSION_LINE_RE = re.compile(r"(?m)^Agent version\s+", re.IGNORECASE)
_NOTIFICATION_SOURCE_RE = re.compile(r"(?m)^📡\s*Оповещение:\s*")
_LEGACY_SAC_SOURCE_RE = re.compile(
    r"(?m)^Источник:\s*Security Alert Center\b.*$",
    re.IGNORECASE,
)
_DAILY_REPORT_TITLE_MARKER = "ЕЖЕДНЕВНЫЙ ОТЧЕТ"
_SECTION_HEADER_RE = re.compile(r"^[📈🧾👥🖥️🕐]")
NOTIFICATION_SOURCE_PREFIX = "📡 Оповещение:"


def format_notification_source_plain(
    *,
    generated_by: str,
    product: str | None = None,
    product_version: str | None = None,
) -> str:
    gb = (generated_by or "agent").strip().lower()
    if gb == "sac":
        return f"{NOTIFICATION_SOURCE_PREFIX} SAC (Security Alert Center)"
    label = (product or "агент").strip()
    version = (product_version or "").strip()
    if version:
        return f"{NOTIFICATION_SOURCE_PREFIX} агент ({label} {version})"
    return f"{NOTIFICATION_SOURCE_PREFIX} агент ({label})"


def has_notification_source_line(text: str) -> bool:
    return bool(_NOTIFICATION_SOURCE_RE.search(text or ""))


def strip_legacy_sac_source_line(text: str) -> str:
    return _LEGACY_SAC_SOURCE_RE.sub("", text or "").strip()


def append_notification_source_plain(
    body: str,
    *,
    generated_by: str,
    product: str | None = None,
    product_version: str | None = None,
) -> str:
    text = strip_legacy_sac_source_line(body or "")
    if has_notification_source_line(text):
        return text
    line = format_notification_source_plain(
        generated_by=generated_by,
        product=product,
        product_version=product_version,
    )
    if text:
        return f"{text.rstrip()}\n\n{line}"
    return line


def append_notification_source_html(
    html_msg: str,
    *,
    generated_by: str,
    product: str | None = None,
    product_version: str | None = None,
) -> str:
    plain_line = format_notification_source_plain(
        generated_by=generated_by,
        product=product,
        product_version=product_version,
    )
    if NOTIFICATION_SOURCE_PREFIX in (html_msg or ""):
        return (html_msg or "").rstrip()
    line = html.escape(plain_line)
    msg = (html_msg or "").rstrip()
    if msg:
        return f"{msg}\n{line}"
    return line


def resolve_product_label(host: Host | None, platform: Platform | None = None) -> tuple[str | None, str | None]:
    if host is None:
        return None, None
    product = (host.product or "").strip()
    if not product:
        if platform == "ssh":
            product = "ssh-monitor"
        elif platform == "windows":
            product = "rdp-login-monitor"
    version = (host.product_version or "").strip() or None
    return (product or None), version


def host_server_line(host: Host | None) -> str:
    if host is None:
        return "unknown"
    label = (host.display_name or host.hostname or "unknown").strip()
    ip = (host.ipv4 or "").strip()
    if ip:
        return f"{label} ({ip})"
    return label


def collapse_blank_lines(text: str, *, max_run: int = 1) -> str:
    lines = text.replace("\r\n", "\n").replace("\r", "\n").split("\n")
    out: list[str] = []
    blank_run = 0
    for line in lines:
        if not line.strip():
            blank_run += 1
            if blank_run <= max_run:
                out.append("")
            continue
        blank_run = 0
        out.append(line.rstrip())
    return "\n".join(out).strip()


def is_active_users_empty_line(text: str) -> bool:
    """Строка-заглушка вместо списка сессий (агент или SAC)."""
    s = text.strip()
    if not s:
        return True
    inner = s.lstrip("👤").strip()
    if _ACTIVE_USERS_EMPTY_LINE_RE.search(inner):
        return True
    return False


def split_active_user_tokens(entry: str) -> list[str]:
    """Разбивает «👤 u1 👤 u2» на отдельные строки (как в отчёте агента)."""
    entry = entry.strip()
    if not entry or is_active_users_empty_line(entry):
        return []
    if entry.count("👤") <= 1:
        line = entry if entry.startswith("👤") else f"👤 {entry}"
        return [f" {line}"]
    parts = re.split(r"(?=👤)", entry)
    result: list[str] = []
    for part in parts:
        part = part.strip()
        if not part:
            continue
        if not part.startswith("👤"):
            part = f"👤 {part}"
        result.append(f" {part}")
    return result


def normalize_active_users_list(users: list[Any] | None) -> list[str]:
    out: list[str] = []
    seen: set[str] = set()
    for raw in users or []:
        for line in split_active_user_tokens(str(raw)):
            key = line.strip().lower()
            if key and key not in seen:
                seen.add(key)
                out.append(line)
    return out


def format_agent_version_line(version: str) -> str:
    v = (version or "").strip()
    if not v:
        return ""
    return f"Agent version {v}"


def ensure_agent_version_line(body: str, version: str | None) -> str:
    line = format_agent_version_line(version or "")
    if not line:
        return body
    if _AGENT_VERSION_LINE_RE.search(body):
        return body
    lines = body.replace("\r\n", "\n").split("\n")
    for i, ln in enumerate(lines):
        if _DAILY_REPORT_TITLE_MARKER in ln:
            lines.insert(i + 1, line)
            return "\n".join(lines)
    return body


def _fix_active_users_header_count(header_line: str, user_count: int) -> str:
    stripped = header_line.rstrip()
    if _ACTIVE_USERS_HEADER_RE.match(stripped.strip()):
        return re.sub(
            r"(\👥\s*АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ\s*\()[^)]+(\))",
            rf"\g<1>{user_count}\2",
            stripped,
            count=1,
        )
    return stripped


def ensure_server_line(body: str, host: Host | None) -> str:
    server = host_server_line(host)
    if not server or server == "unknown":
        return body
    if _SERVER_LINE_RE.search(body):
        return body
    lines = body.replace("\r\n", "\n").split("\n")
    for i, line in enumerate(lines):
        if "ЕЖЕДНЕВНЫЙ ОТЧЕТ" in line:
            insert_at = i + 1
            if insert_at < len(lines) and _AGENT_VERSION_LINE_RE.match(lines[insert_at].strip()):
                insert_at += 1
            lines.insert(insert_at, f"🖥️ Сервер: {server}")
            return "\n".join(lines)
    return f"🖥️ Сервер: {server}\n{body}"


def normalize_active_users_in_body(body: str) -> str:
    lines = body.replace("\r\n", "\n").split("\n")
    out: list[str] = []
    i = 0
    while i < len(lines):
        line = lines[i]
        if _ACTIVE_USERS_HEADER_RE.match(line.strip()):
            out.append(line)
            i += 1
            user_lines: list[str] = []
            while i < len(lines):
                cur = lines[i]
                stripped = cur.strip()
                if not stripped:
                    break
                if stripped.startswith("Источник:") or stripped.startswith(NOTIFICATION_SOURCE_PREFIX):
                    break
                if _SECTION_HEADER_RE.match(stripped) and not stripped.startswith("👤"):
                    break
                if is_active_users_empty_line(stripped):
                    out[-1] = _fix_active_users_header_count(out[-1], 0)
                    out.append(cur)
                    i += 1
                    break
                user_lines.extend(split_active_user_tokens(cur))
                i += 1
            user_lines = [ln for ln in user_lines if not is_active_users_empty_line(ln.strip())]
            if user_lines:
                out[-1] = _fix_active_users_header_count(out[-1], len(user_lines))
                out.extend(user_lines)
            else:
                out[-1] = _fix_active_users_header_count(out[-1], 0)
            continue
        out.append(line)
        i += 1
    return "\n".join(out)


def extract_active_users_from_report_body(body: str) -> list[str]:
    lines = body.replace("\r\n", "\n").split("\n")
    in_section = False
    users: list[str] = []
    for line in lines:
        stripped = line.strip()
        if _ACTIVE_USERS_HEADER_RE.match(stripped):
            in_section = True
            continue
        if not in_section:
            continue
        if not stripped:
            break
        if is_active_users_empty_line(stripped):
            break
        if stripped.startswith("Источник:") or stripped.startswith(NOTIFICATION_SOURCE_PREFIX):
            break
        if _SECTION_HEADER_RE.match(stripped) and not stripped.startswith("👤"):
            break
        users.extend(split_active_user_tokens(stripped))
    return normalize_active_users_list(users)


def reconcile_stats_from_report_body(
    stats: dict[str, Any],
    body: str,
    platform: Platform,
) -> dict[str, Any]:
    """Синхронизирует stats.active_users и счётчики с текстом отчёта после нормализации."""
    out = dict(stats)
    users = extract_active_users_from_report_body(body)
    out["active_users"] = users
    count = len(users)
    if platform == "ssh":
        out["active_sessions"] = count
    else:
        out["active_sessions_rdp"] = count
        names: list[str] = []
        for raw in users:
            name = re.sub(r"^👤\s*", "", raw.strip()).strip()
            if name and not is_active_users_empty_line(name):
                names.append(name)
        out["unique_users"] = sorted(set(names))
    return out


def normalize_report_body(body: str, host: Host | None, platform: Platform) -> str:
    """Приводит текст отчёта (агент/SAC) к единому компактному виду."""
    text = collapse_blank_lines(body.replace("\r\n", "\n"))
    agent_version = host.product_version if host else None
    text = ensure_agent_version_line(text, agent_version)
    text = ensure_server_line(text, host)
    text = normalize_active_users_in_body(text)
    return collapse_blank_lines(text)


def body_to_report_html(body: str) -> str:
    escaped = html.escape(body)
    return f'<div class="agent-report">{escaped.replace(chr(10), "<br>")}</div>'


def _section_top_ips(top: list[str]) -> list[str]:
    lines = [" 🧾 ТОП-5 IP ПО НЕУДАЧНЫМ ПОПЫТКАМ:"]
    if top:
        lines.extend(f" {line}" if not line.startswith(" ") else line for line in top[:5])
    else:
        lines.append(" (нет данных)")
    return lines


def _section_active_users(users: list[str], *, sac_generated: bool) -> list[str]:
    count = len(users)
    lines = [f" 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ ({count}):"]
    if users:
        for line in users:
            lines.append(f" {line}" if not line.startswith(" ") else line)
    elif sac_generated:
        lines.append(
            " (нет данных — полный список сессий только в отчёте агента; "
            "в SAC — пользователи из событий входа за 24 ч, если были)"
        )
    else:
        lines.append(" (нет данных)")
    return lines


def build_report_body(
    platform: Platform,
    host: Host,
    stats: dict[str, Any],
    when_local: datetime,
    *,
    sac_generated: bool = False,
) -> str:
    server = host_server_line(host)
    time_str = when_local.strftime("%d.%m.%Y %H:%M:%S")
    top = list(stats.get("top_failed_ips") or [])[:5]
    active_users = normalize_active_users_list(list(stats.get("active_users") or []))
    agent_version = (
        str(stats.get("agent_version") or stats.get("product_version") or host.product_version or "")
    ).strip()
    if sac_generated and not agent_version:
        agent_version = "sac"

    if platform == "ssh":
        title = "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА"
        ok = int(stats.get("successful_logins", stats.get("successful_ssh", 0)))
        fail = int(stats.get("failed_logins", stats.get("failed_ssh", 0)))
        sudo = int(stats.get("sudo_commands", 0))
        bans = int(stats.get("active_bans", 0))
        lines = [title]
        if agent_version:
            lines.append(format_agent_version_line(agent_version))
        lines.extend([
            f"🖥️ Сервер: {server}",
            f"🕐 Время отчета: {time_str}",
            "",
            " 📈 СТАТИСТИКА ЗА ПОСЛЕДНИЕ 24 ЧАСА:",
            f" ✅ Успешных SSH подключений: {ok}",
            f" ❌ Неудачных попыток SSH: {fail}",
            f" ⚠️ Команд через sudo: {sudo}",
            f" 🚫 Активных банов (ipset): {bans}",
        ])
    else:
        title = "📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS"
        ok = int(stats.get("successful_logins", stats.get("rdp_success", 0)))
        fail = int(stats.get("failed_logins", stats.get("rdp_failed", 0)))
        bans = int(stats.get("active_bans", 0))
        lines = [title]
        if agent_version:
            lines.append(format_agent_version_line(agent_version))
        lines.extend([
            f"🖥️ Сервер: {server}",
            f"🕐 Время отчета: {time_str}",
            "",
            " 📈 СТАТИСТИКА ЗА ПОСЛЕДНИЕ 24 ЧАСА:",
            f" ✅ Успешных RDP подключений: {ok}",
            f" ❌ Неудачных попыток RDP: {fail}",
            f" 🚫 Активных банов: {bans}",
        ])

    lines.append("")
    lines.extend(_section_top_ips(top))
    lines.append("")
    lines.extend(_section_active_users(active_users, sac_generated=sac_generated))

    generated_by = "sac" if sac_generated else "agent"
    product, version = resolve_product_label(host, platform)
    if not sac_generated and agent_version:
        version = agent_version
    lines.append("")
    lines.append(
        format_notification_source_plain(
            generated_by=generated_by,
            product=product,
            product_version=version,
        )
    )
    return "\n".join(lines)


def enrich_stats_for_storage(platform: Platform, stats: dict[str, Any]) -> dict[str, Any]:
    """Канонические поля + legacy-ключи для UI и старых отчётов."""
    out = dict(stats)
    out["platform"] = platform
    if platform == "ssh":
        ok = int(out.get("successful_logins", out.get("successful_ssh", 0)))
        fail = int(out.get("failed_logins", out.get("failed_ssh", 0)))
        out["successful_logins"] = ok
        out["failed_logins"] = fail
        out["successful_ssh"] = ok
        out["failed_ssh"] = fail
    else:
        ok = int(out.get("successful_logins", out.get("rdp_success", 0)))
        fail = int(out.get("failed_logins", out.get("rdp_failed", 0)))
        out["successful_logins"] = ok
        out["failed_logins"] = fail
        out["rdp_success"] = ok
        out["rdp_failed"] = fail
        if "unique_users" in out and "active_users" not in out:
            users = out.get("unique_users") or []
            out["active_users"] = [f"👤 {u}" if not str(u).startswith("👤") else str(u) for u in users]
    if "active_users" in out:
        out["active_users"] = normalize_active_users_list(list(out.get("active_users") or []))
    return out


def normalize_daily_report_details(
    details: dict[str, Any] | None,
    host: Host | None,
    report_type: str,
) -> dict[str, Any] | None:
    if not isinstance(details, dict):
        return details
    platform: Platform = "windows" if report_type == "report.daily.rdp" else "ssh"
    body = details.get("report_body")
    if not isinstance(body, str) or not body.strip():
        return details

    normalized_body = normalize_report_body(body, host, platform)
    gb = str(details.get("generated_by") or "agent").strip().lower()
    if gb not in ("agent", "sac"):
        gb = "agent"
    product, version = resolve_product_label(host, platform)
    stats_raw = details.get("stats")
    if isinstance(stats_raw, dict) and gb == "agent":
        av = stats_raw.get("agent_version") or stats_raw.get("product_version")
        if av:
            version = str(av).strip()
    normalized_body = append_notification_source_plain(
        normalized_body,
        generated_by=gb,
        product=product,
        product_version=version,
    )
    out = dict(details)
    out["report_body"] = normalized_body
    out["report_html"] = body_to_report_html(normalized_body)

    stats = out.get("stats")
    if isinstance(stats, dict):
        stats = enrich_stats_for_storage(platform, dict(stats))
        stats = reconcile_stats_from_report_body(stats, normalized_body, platform)
        out["stats"] = stats
    return out