PTah
2501ecc4fe
Ensure server line, compact spacing, and one user per row when ingesting agent reports or sending SAC-generated daily reports. Co-authored-by: Cursor <cursoragent@cursor.com>
269 lines
9.6 KiB
Python
269 lines
9.6 KiB
Python
"""Единый текст и 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*АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ")
|
|
_SECTION_HEADER_RE = re.compile(r"^[📈🧾👥🖥️🕐]")
|
|
|
|
|
|
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 split_active_user_tokens(entry: str) -> list[str]:
|
|
"""Разбивает «👤 u1 👤 u2» на отдельные строки (как в отчёте агента)."""
|
|
entry = entry.strip()
|
|
if not 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 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:
|
|
lines.insert(i + 1, 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("Источник:"):
|
|
break
|
|
if _SECTION_HEADER_RE.match(stripped) and not stripped.startswith("👤"):
|
|
break
|
|
if stripped.startswith("(нет данных"):
|
|
out.append(cur)
|
|
i += 1
|
|
break
|
|
user_lines.extend(split_active_user_tokens(cur))
|
|
i += 1
|
|
if user_lines:
|
|
out.extend(user_lines)
|
|
continue
|
|
out.append(line)
|
|
i += 1
|
|
return "\n".join(out)
|
|
|
|
|
|
def normalize_report_body(body: str, host: Host | None, platform: Platform) -> str:
|
|
"""Приводит текст отчёта (агент/SAC) к единому компактному виду."""
|
|
text = collapse_blank_lines(body.replace("\r\n", "\n"))
|
|
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 []))
|
|
|
|
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.append("")
|
|
lines.extend(_section_top_ips(top))
|
|
lines.append("")
|
|
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]
|
|
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)
|
|
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))
|
|
out["stats"] = stats
|
|
return out
|