PapaTramp
563b836acc
Adds mobile device registration by admin codes, refresh tokens, push channel in notification policy, and Settings section for managing Seaca clients. Co-authored-by: Cursor <cursoragent@cursor.com>
259 lines
8.0 KiB
Python
259 lines
8.0 KiB
Python
"""SAC daily report aggregation (F-NOT-05 / notif-32)."""
|
|
|
|
import uuid
|
|
from datetime import datetime, timedelta, timezone
|
|
from unittest.mock import patch
|
|
|
|
from sqlalchemy import select
|
|
|
|
from app.config import Settings
|
|
from app.models import Event, Host
|
|
from app.services import notify_dispatch
|
|
from app.services.daily_report import (
|
|
_aggregate_ssh,
|
|
generate_daily_report_for_host,
|
|
host_has_report_today,
|
|
run_daily_reports,
|
|
)
|
|
from app.services.notification_policy import NotificationPolicyConfig
|
|
from app.services.telegram_templates import format_event_telegram_html, sanitize_telegram_html
|
|
|
|
|
|
def _ssh_host(db) -> Host:
|
|
h = Host(
|
|
hostname="pilot-ssh",
|
|
os_family="linux",
|
|
product="ssh-monitor",
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
)
|
|
db.add(h)
|
|
db.flush()
|
|
return h
|
|
|
|
|
|
def _cfg(**overrides) -> Settings:
|
|
base = {
|
|
"sac_daily_report_enabled": True,
|
|
"sac_daily_report_hour": 9,
|
|
"sac_daily_report_timezone": "UTC",
|
|
"sac_daily_report_skip_if_agent_sent": False,
|
|
"sac_daily_report_require_activity": False,
|
|
}
|
|
base.update(overrides)
|
|
return Settings(**base)
|
|
|
|
|
|
def test_aggregate_ssh_counts_types(db_session):
|
|
now = datetime.now(timezone.utc)
|
|
events = [
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=1,
|
|
occurred_at=now,
|
|
category="auth",
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="ok",
|
|
summary="",
|
|
payload={},
|
|
),
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=1,
|
|
occurred_at=now,
|
|
category="auth",
|
|
type="ssh.login.failed",
|
|
severity="warning",
|
|
title="fail",
|
|
summary="",
|
|
details={"source_ip": "10.0.0.1"},
|
|
payload={},
|
|
),
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=1,
|
|
occurred_at=now,
|
|
category="privilege",
|
|
type="privilege.sudo.command",
|
|
severity="warning",
|
|
title="sudo",
|
|
summary="",
|
|
payload={},
|
|
),
|
|
]
|
|
stats = _aggregate_ssh(events)
|
|
assert stats["successful_ssh"] == 1
|
|
assert stats["failed_ssh"] == 1
|
|
assert stats["sudo_commands"] == 1
|
|
assert "10.0.0.1" in stats["top_failed_ips"][0]
|
|
|
|
|
|
def test_skip_when_agent_report_today(db_session):
|
|
h = _ssh_host(db_session)
|
|
cfg = _cfg(sac_daily_report_skip_if_agent_sent=True)
|
|
now = datetime.now(timezone.utc)
|
|
db_session.add(
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=h.id,
|
|
occurred_at=now,
|
|
received_at=now,
|
|
category="report",
|
|
type="report.daily.ssh",
|
|
severity="info",
|
|
title="agent",
|
|
summary="from agent",
|
|
details={"generated_by": "agent"},
|
|
payload={},
|
|
)
|
|
)
|
|
db_session.commit()
|
|
assert host_has_report_today(db_session, h.id, "report.daily.ssh", cfg) is True
|
|
res = generate_daily_report_for_host(db_session, h, cfg)
|
|
assert res is not None
|
|
assert res.created is False
|
|
assert res.skipped_reason == "agent_report_exists_today"
|
|
|
|
|
|
def test_generate_creates_report_and_notifies(db_session):
|
|
h = _ssh_host(db_session)
|
|
now = datetime.now(timezone.utc)
|
|
db_session.add(
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=h.id,
|
|
occurred_at=now - timedelta(hours=1),
|
|
category="auth",
|
|
type="ssh.login.failed",
|
|
severity="warning",
|
|
title="fail",
|
|
summary="x",
|
|
details={"source_ip": "1.2.3.4"},
|
|
payload={},
|
|
)
|
|
)
|
|
db_session.commit()
|
|
cfg = _cfg()
|
|
with patch("app.services.daily_report.notify_daily_report") as mock_notify:
|
|
res = generate_daily_report_for_host(db_session, h, cfg)
|
|
assert res is not None
|
|
assert res.created is True
|
|
assert res.report_type == "report.daily.ssh"
|
|
mock_notify.assert_called_once()
|
|
ev = db_session.scalar(select(Event).where(Event.type == "report.daily.ssh"))
|
|
assert ev is not None
|
|
assert ev.details.get("generated_by") == "sac"
|
|
|
|
|
|
def test_generate_rdp_report_unified_format(db_session):
|
|
h = Host(
|
|
hostname="win-srv",
|
|
display_name="RDCB",
|
|
ipv4="10.0.0.5",
|
|
os_family="windows",
|
|
product="rdp-login-monitor",
|
|
last_seen_at=datetime.now(timezone.utc),
|
|
)
|
|
db_session.add(h)
|
|
db_session.flush()
|
|
now = datetime.now(timezone.utc)
|
|
db_session.add(
|
|
Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=h.id,
|
|
occurred_at=now,
|
|
category="auth",
|
|
type="rdp.login.success",
|
|
severity="info",
|
|
title="ok",
|
|
summary="",
|
|
details={"user": "admin"},
|
|
payload={},
|
|
)
|
|
)
|
|
db_session.commit()
|
|
cfg = _cfg()
|
|
with patch("app.services.daily_report.notify_daily_report"):
|
|
res = generate_daily_report_for_host(db_session, h, cfg)
|
|
assert res and res.created
|
|
ev = db_session.scalar(select(Event).where(Event.type == "report.daily.rdp"))
|
|
body = ev.details.get("report_body", "")
|
|
assert "ЕЖЕДНЕВНЫЙ ОТЧЕТ МОНИТОРИНГА WINDOWS" in body
|
|
assert "RDCB (10.0.0.5)" in body
|
|
assert "АКТИВНЫЕ ПОЛЬЗОВАТЕЛИ" in body
|
|
|
|
|
|
def test_run_daily_reports_respects_hour(db_session):
|
|
h = _ssh_host(db_session)
|
|
db_session.commit()
|
|
cfg = _cfg(sac_daily_report_hour=23)
|
|
early = datetime(2026, 5, 29, 8, 0, tzinfo=timezone.utc)
|
|
with patch("app.services.daily_report._now_in_tz", return_value=early):
|
|
out = run_daily_reports(db_session, cfg, force=False)
|
|
assert out == []
|
|
|
|
|
|
def test_run_daily_reports_force_bypasses_hour(db_session):
|
|
_ssh_host(db_session)
|
|
db_session.commit()
|
|
cfg = _cfg(sac_daily_report_hour=23)
|
|
early = datetime(2026, 5, 29, 8, 0, tzinfo=timezone.utc)
|
|
with patch("app.services.daily_report._now_in_tz", return_value=early):
|
|
with patch("app.services.daily_report.generate_daily_report_for_host") as mock_gen:
|
|
mock_gen.return_value = None
|
|
run_daily_reports(db_session, cfg, force=True)
|
|
mock_gen.assert_called()
|
|
|
|
|
|
def test_notify_daily_report_bypasses_min_severity():
|
|
event = Event(
|
|
event_id="00000000-0000-4000-8000-000000000601",
|
|
host_id=1,
|
|
occurred_at=datetime(2026, 5, 29, 9, 0, tzinfo=timezone.utc),
|
|
category="report",
|
|
type="report.daily.ssh",
|
|
severity="info",
|
|
title="report",
|
|
summary="s",
|
|
dedup_key="sac|1|report.daily.ssh|2026-05-29",
|
|
payload={},
|
|
)
|
|
policy = NotificationPolicyConfig(
|
|
min_severity="warning",
|
|
use_telegram=True,
|
|
use_webhook=False,
|
|
use_email=False,
|
|
use_mobile=False,
|
|
source="db",
|
|
)
|
|
with patch.object(notify_dispatch, "get_effective_notification_policy", return_value=policy):
|
|
with patch.object(notify_dispatch, "should_notify_event", return_value=True):
|
|
with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
|
|
notify_dispatch.notify_daily_report(event)
|
|
mock_tg.notify_event.assert_called_once()
|
|
|
|
|
|
def test_daily_report_telegram_html_uses_report_html(db_session):
|
|
h = Host(hostname="h1", display_name="H1", os_family="linux")
|
|
event = Event(
|
|
event_id=str(uuid.uuid4()),
|
|
host_id=1,
|
|
host=h,
|
|
occurred_at=datetime(2026, 5, 29, 9, 0, tzinfo=timezone.utc),
|
|
category="report",
|
|
type="report.daily.ssh",
|
|
severity="info",
|
|
title="Отчёт",
|
|
summary="short",
|
|
details={"report_html": "<b>📊 OK</b><br>line"},
|
|
payload={},
|
|
)
|
|
assert format_event_telegram_html(event) == (
|
|
"<b>📊 OK</b>\nline\n📡 Оповещение: SAC (Security Alert Center)"
|
|
)
|
|
|
|
|
|
def test_sanitize_telegram_html_strips_div_and_br():
|
|
raw = '<div class="agent-report">a<br>b</div>'
|
|
assert sanitize_telegram_html(raw) == "a\nb"
|