PTah
7ebbb9a9d1
Agent-style report body for SAC aggregation; shared stats cards; RDP ban note in docs Co-authored-by: Cursor <cursoragent@cursor.com>
126 lines
4.8 KiB
Python
126 lines
4.8 KiB
Python
"""Единый текст и stats для report.daily.ssh / report.daily.rdp (агент и SAC)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
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"})
|
|
|
|
|
|
def host_server_line(host: Host) -> str:
|
|
label = (host.display_name or host.hostname or "unknown").strip()
|
|
ip = (host.ipv4 or "").strip()
|
|
if ip:
|
|
return f"{label} ({ip})"
|
|
return label
|
|
|
|
|
|
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 = list(stats.get("active_users") or [])
|
|
|
|
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,
|
|
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,
|
|
f"🖥️ Сервер: {server}",
|
|
f"🕐 Время отчета: {time_str}",
|
|
"",
|
|
"📈 СТАТИСТИКА ЗА ПОСЛЕДНИЕ 24 ЧАСА:",
|
|
f" ✅ Успешных RDP подключений: {ok}",
|
|
f" ❌ Неудачных попыток RDP: {fail}",
|
|
f" 🚫 Активных банов: {bans}",
|
|
]
|
|
|
|
lines.extend(_section_top_ips(top))
|
|
lines.extend(_section_active_users(active_users, sac_generated=sac_generated))
|
|
|
|
if sac_generated:
|
|
lines.append("")
|
|
lines.append("Источник: Security Alert Center (агрегация ingest за 24 ч).")
|
|
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]
|
|
return out
|