493 lines
17 KiB
Python
493 lines
17 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*АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ")
|
|
_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
|