PapaTramp/security-alert-center
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: reconcile daily report active users count and list (0.3.5)

Browse Source 
Normalize empty-session placeholders; sync stats with report body after ingest.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
PTah
2026-06-23 14:41:47 +10:00
parent 8b52c5fe3c
commit 64b942447b
5 changed files with 128 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/app/services/daily_report_format.py
+68 -2
View File
@@ -16,6 +16,10 @@ 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(
@@ -130,10 +134,21 @@ def collapse_blank_lines(text: str, *, max_run: int = 1) -> str:
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:
if not entry or is_active_users_empty_line(entry):
return []
if entry.count("👤") <= 1:
line = entry if entry.startswith("👤") else f"👤 {entry}"
@@ -231,21 +246,71 @@ def normalize_active_users_in_body(body: str) -> str:
break
if _SECTION_HEADER_RE.match(stripped) and not stripped.startswith("👤"):
break
if stripped.startswith("(нет данных"):
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"))
@@ -422,5 +487,6 @@ def normalize_daily_report_details(
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
backend/app/version.py
+1 -1
View File
@@ -1,5 +1,5 @@
"""Единый источник версии SAC (API, health, логи, OpenAPI)."""
APP_NAME = "Security Alert Center"
APP_VERSION = "0.3.4"
APP_VERSION = "0.3.5"
APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}"
backend/tests/test_daily_report_format.py
+56
View File
@@ -134,3 +134,59 @@ def test_normalize_report_body_adds_server_and_collapses_blanks():
user_lines = [ln for ln in body.split("\n") if ln.strip().startswith("👤")]
assert len(user_lines) == 2
assert " 👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (2)" in body
def test_normalize_ssh_active_users_mismatch_count_and_empty_body():
host = Host(hostname="srv", display_name="SSH", ipv4="10.0.0.1", product="ssh-monitor")
raw = "\n".join(
[
"📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА",
"👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (2):",
" (нет данных)",
]
)
body = normalize_report_body(raw, host, "ssh")
assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in body
assert "(нет данных)" in body
assert not any(ln.strip().startswith("👤") for ln in body.split("\n"))
def test_normalize_rdp_empty_active_users_placeholder():
host = Host(hostname="gw", display_name="K6A-DC3", ipv4="192.168.160.40", product="rdp-login-monitor")
raw = "\n".join(
[
"📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS",
"👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0):",
" (нет активных пользователей / RDP-сессий)",
]
)
body = normalize_report_body(raw, host, "windows")
assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in body
assert "нет активных пользователей" in body
assert "👤 (нет активных" not in body
def test_normalize_daily_report_reconciles_stats_from_body():
from app.services.daily_report_format import normalize_daily_report_details
host = Host(hostname="srv", display_name="SSH", ipv4="10.0.0.2", product="ssh-monitor")
details = {
"report_body": "\n".join(
[
"📊 ЕЖЕДНЕВНЫЙ ОТЧЕТ SSH МОНИТОРИНГА",
"👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (9):",
" (нет данных)",
]
),
"stats": {
"successful_ssh": 1,
"active_sessions": 9,
"active_users": [],
"generated_by": "agent",
},
}
out = normalize_daily_report_details(details, host, "report.daily.ssh")
assert out is not None
assert out["stats"]["active_sessions"] == 0
assert out["stats"]["active_users"] == []
assert "👥 АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ (0)" in out["report_body"]
backend/tests/test_health.py
+2 -2
View File
@@ -4,6 +4,6 @@ from app.version import APP_NAME, APP_VERSION, APP_VERSION_LABEL
def test_version_constants():
assert APP_VERSION == "0.3.4"
assert APP_VERSION == "0.3.5"
assert APP_NAME == "Security Alert Center"
assert APP_VERSION_LABEL == "Security Alert Center v.0.3.4"
assert APP_VERSION_LABEL == "Security Alert Center v.0.3.5"