From aafb80fa317ac9dab0e4bd5da06c4f198cf8ab82 Mon Sep 17 00:00:00 2001 From: PTah Date: Wed, 10 Jun 2026 09:07:32 +1000 Subject: [PATCH] feat: proactive host_silence scan for stale agent.heartbeat (0.8.3) Background scan every 5 min opens rule:host_silence and notifies without waiting for the next ingest event. CLI job + optional systemd timer. Co-authored-by: Cursor --- backend/app/config.py | 4 + backend/app/jobs/host_silence_background.py | 51 ++++++++ backend/app/jobs/host_silence_scan.py | 38 ++++++ backend/app/main.py | 18 ++- backend/app/services/host_silence_scan.py | 119 ++++++++++++++++++ backend/app/services/problem_rules.py | 32 +++-- backend/app/services/telegram_templates.py | 2 + backend/app/version.py | 2 +- backend/tests/test_host_silence_scan.py | 122 +++++++++++++++++++ deploy/env.native.example | 4 + deploy/systemd/sac-host-silence-scan.service | 16 +++ deploy/systemd/sac-host-silence-scan.timer | 11 ++ docs/agent-integration.md | 2 + 13 files changed, 411 insertions(+), 10 deletions(-) create mode 100644 backend/app/jobs/host_silence_background.py create mode 100644 backend/app/jobs/host_silence_scan.py create mode 100644 backend/app/services/host_silence_scan.py create mode 100644 backend/tests/test_host_silence_scan.py create mode 100644 deploy/systemd/sac-host-silence-scan.service create mode 100644 deploy/systemd/sac-host-silence-scan.timer diff --git a/backend/app/config.py b/backend/app/config.py index 90680c6..bda05ef 100644 --- a/backend/app/config.py +++ b/backend/app/config.py @@ -85,6 +85,10 @@ class Settings(BaseSettings): # Порог «живости» агента sac_heartbeat_stale_minutes: int = 780 + # Периодическое сканирование stale heartbeat (proactive host_silence) + sac_host_silence_scan_enabled: bool = True + sac_host_silence_scan_interval_minutes: int = 5 + # Окно корреляции Problems: host + type + rule в одном open Problem sac_problem_correlation_window_minutes: int = 60 diff --git a/backend/app/jobs/host_silence_background.py b/backend/app/jobs/host_silence_background.py new file mode 100644 index 0000000..8a09983 --- /dev/null +++ b/backend/app/jobs/host_silence_background.py @@ -0,0 +1,51 @@ +"""Фоновый asyncio-цикл сканирования stale heartbeat (в процессе sac-api).""" + +from __future__ import annotations + +import asyncio +import logging + +from app.config import get_settings +from app.database import SessionLocal +from app.services.host_silence_scan import dispatch_host_silence_notifications, run_host_silence_scan + +logger = logging.getLogger("sac.host_silence_scan") + + +async def host_silence_scan_loop(stop_event: asyncio.Event) -> None: + settings = get_settings() + interval_sec = max(60, int(settings.sac_host_silence_scan_interval_minutes) * 60) + logger.info( + "host_silence background scan enabled (interval=%s min, stale=%s min)", + settings.sac_host_silence_scan_interval_minutes, + settings.sac_heartbeat_stale_minutes, + ) + while not stop_event.is_set(): + try: + await asyncio.to_thread(_run_scan_once) + except Exception: + logger.exception("host_silence background scan error") + try: + await asyncio.wait_for(stop_event.wait(), timeout=interval_sec) + except asyncio.TimeoutError: + continue + + +def _run_scan_once() -> None: + db = SessionLocal() + try: + results = run_host_silence_scan(db) + notified = dispatch_host_silence_notifications(db, results) + db.commit() + if results: + logger.info( + "host_silence background: stale=%s created=%s notified=%s", + len(results), + sum(1 for r in results if r.created), + notified, + ) + except Exception: + db.rollback() + raise + finally: + db.close() diff --git a/backend/app/jobs/host_silence_scan.py b/backend/app/jobs/host_silence_scan.py new file mode 100644 index 0000000..eea06b8 --- /dev/null +++ b/backend/app/jobs/host_silence_scan.py @@ -0,0 +1,38 @@ +"""CLI: python -m app.jobs.host_silence_scan""" + +from __future__ import annotations + +import logging +import sys + +from app.database import SessionLocal +from app.services.host_silence_scan import dispatch_host_silence_notifications, run_host_silence_scan + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger("sac.host_silence_scan") + + +def main() -> int: + db = SessionLocal() + try: + results = run_host_silence_scan(db) + notified = dispatch_host_silence_notifications(db, results) + db.commit() + created = sum(1 for r in results if r.created) + logger.info( + "host_silence scan done stale_hosts=%s created=%s notified=%s", + len(results), + created, + notified, + ) + return 0 + except Exception: + db.rollback() + logger.exception("host_silence scan failed") + return 1 + finally: + db.close() + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backend/app/main.py b/backend/app/main.py index 4858002..221cb8b 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -1,5 +1,6 @@ +import asyncio import logging -from contextlib import asynccontextmanager +from contextlib import asynccontextmanager, suppress from pathlib import Path from fastapi import FastAPI, HTTPException @@ -57,7 +58,22 @@ async def lifespan(_app: FastAPI): logger.info("%s — application startup (version %s)", APP_VERSION_LABEL, APP_VERSION) bootstrap_api_key() bootstrap_users() + + stop_scan = asyncio.Event() + scan_task: asyncio.Task | None = None + settings = get_settings() + if settings.sac_host_silence_scan_enabled: + from app.jobs.host_silence_background import host_silence_scan_loop + + scan_task = asyncio.create_task(host_silence_scan_loop(stop_scan)) + yield + + if scan_task is not None: + stop_scan.set() + scan_task.cancel() + with suppress(asyncio.CancelledError): + await scan_task logger.info("%s — application shutdown", APP_VERSION_LABEL) diff --git a/backend/app/services/host_silence_scan.py b/backend/app/services/host_silence_scan.py new file mode 100644 index 0000000..c104e62 --- /dev/null +++ b/backend/app/services/host_silence_scan.py @@ -0,0 +1,119 @@ +"""Периодическое сканирование хостов с устаревшим agent.heartbeat (proactive host_silence).""" + +from __future__ import annotations + +import logging +from dataclasses import dataclass +from datetime import datetime, timezone + +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models import Host, Problem +from app.services.problem_rules import RuleMatch, build_fingerprint, host_silence_match_for_host +from app.services.problems import _correlation_cutoff + +logger = logging.getLogger(__name__) + + +@dataclass(frozen=True) +class HostSilenceScanResult: + host_id: int + hostname: str + problem: Problem + created: bool + + +def open_or_refresh_host_silence_problem( + db: Session, + host_id: int, + match: RuleMatch, + *, + now: datetime | None = None, +) -> tuple[Problem, bool]: + """Открыть или обновить open Problem host_silence без ingest-события.""" + ref = now or datetime.now(timezone.utc) + fingerprint = build_fingerprint( + host_id, + match.correlation_type, + match.rule_id, + match.fingerprint_suffix, + ) + cutoff = _correlation_cutoff(ref) + + open_problem = db.scalar( + select(Problem) + .where( + Problem.status == "open", + Problem.fingerprint == fingerprint, + Problem.host_id == host_id, + Problem.last_seen_at >= cutoff, + ) + .order_by(Problem.last_seen_at.desc()) + .limit(1) + ) + if open_problem: + open_problem.summary = match.summary + open_problem.title = match.title + open_problem.last_seen_at = ref + open_problem.updated_at = ref + db.flush() + return open_problem, False + + problem = Problem( + host_id=host_id, + title=match.title, + summary=match.summary, + severity=match.severity, + status="open", + rule_id=match.rule_id, + fingerprint=fingerprint, + event_count=0, + last_seen_at=ref, + ) + db.add(problem) + db.flush() + return problem, True + + +def run_host_silence_scan(db: Session, *, now: datetime | None = None) -> list[HostSilenceScanResult]: + """Найти stale-хосты и создать/обновить Problems rule:host_silence.""" + ref = now or datetime.now(timezone.utc) + hosts = db.scalars(select(Host).order_by(Host.id)).all() + results: list[HostSilenceScanResult] = [] + + for host in hosts: + match = host_silence_match_for_host(db, host.id, hostname=host.hostname, now=ref) + if match is None: + continue + problem, created = open_or_refresh_host_silence_problem(db, host.id, match, now=ref) + results.append( + HostSilenceScanResult( + host_id=host.id, + hostname=host.hostname, + problem=problem, + created=created, + ) + ) + if created: + logger.info( + "host_silence scan: new problem host=%s id=%s problem_id=%s", + host.hostname, + host.id, + problem.id, + ) + + return results + + +def dispatch_host_silence_notifications(db: Session, results: list[HostSilenceScanResult]) -> int: + """Telegram/webhook/email только для вновь созданных Problems.""" + from app.services.notify_dispatch import notify_problem + + sent = 0 + for item in results: + if not item.created: + continue + notify_problem(item.problem, event=None, db=db) + sent += 1 + return sent diff --git a/backend/app/services/problem_rules.py b/backend/app/services/problem_rules.py index b939094..40bcc65 100644 --- a/backend/app/services/problem_rules.py +++ b/backend/app/services/problem_rules.py @@ -122,18 +122,26 @@ def evaluate_privilege_spike(db: Session, event: Event) -> RuleMatch | None: ) -def evaluate_host_silence(db: Session, event: Event) -> RuleMatch | None: - if event.type == HEARTBEAT_TYPE: - return None +def host_silence_match_for_host( + db: Session, + host_id: int, + *, + hostname: str | None = None, + now: datetime | None = None, +) -> RuleMatch | None: + """Сформировать RuleMatch host_silence, если последний heartbeat устарел.""" settings = get_settings() - hb = last_heartbeat_at(db, event.host_id) + ref = now or _now() + hb = last_heartbeat_at(db, host_id) if hb is None: return None - if agent_status(hb, stale_minutes=settings.sac_heartbeat_stale_minutes, now=_now()) != "stale": + if agent_status(hb, stale_minutes=settings.sac_heartbeat_stale_minutes, now=ref) != "stale": return None - host = db.get(Host, event.host_id) - hostname = host.hostname if host else f"host#{event.host_id}" - age_min = int((_now() - hb.replace(tzinfo=timezone.utc if hb.tzinfo is None else hb.tzinfo)).total_seconds() // 60) + if hostname is None: + host = db.get(Host, host_id) + hostname = host.hostname if host else f"host#{host_id}" + hb_utc = hb.replace(tzinfo=timezone.utc) if hb.tzinfo is None else hb.astimezone(timezone.utc) + age_min = int((ref - hb_utc).total_seconds() // 60) return RuleMatch( rule_id=RULE_HOST_SILENCE, correlation_type="host.silence", @@ -146,6 +154,14 @@ def evaluate_host_silence(db: Session, event: Event) -> RuleMatch | None: ) +def evaluate_host_silence(db: Session, event: Event) -> RuleMatch | None: + if event.type == HEARTBEAT_TYPE: + return None + host = db.get(Host, event.host_id) + hostname = host.hostname if host else None + return host_silence_match_for_host(db, event.host_id, hostname=hostname) + + def evaluate_immediate_high(event: Event) -> RuleMatch | None: """Одиночные high/critical и известные типы (ban, mass bruteforce).""" problem_types = frozenset({"ssh.ip.banned", "ssh.bruteforce.mass"}) diff --git a/backend/app/services/telegram_templates.py b/backend/app/services/telegram_templates.py index e628aa0..9e4e2fb 100644 --- a/backend/app/services/telegram_templates.py +++ b/backend/app/services/telegram_templates.py @@ -543,6 +543,8 @@ def format_problem_telegram_html(problem: Problem, event: Event | None = None) - msg += f"\n{html_escape(problem.title)}\n{html_escape(problem.summary)}" if event is not None: msg += f"\n\nТриггер: {html_escape(event.type)} ({html_escape(event.severity)})" + elif problem.rule_id == "rule:host_silence": + msg += "\n\nТриггер: периодическая проверка SAC (host_silence scan)" if event.type in ( "rdp.login.success", "rdp.login.failed", diff --git a/backend/app/version.py b/backend/app/version.py index b4c2790..07c33a7 100644 --- a/backend/app/version.py +++ b/backend/app/version.py @@ -1,5 +1,5 @@ """Единый источник версии SAC (API, health, логи, OpenAPI).""" APP_NAME = "Security Alert Center" -APP_VERSION = "0.8.2" +APP_VERSION = "0.8.3" APP_VERSION_LABEL = f"{APP_NAME} v.{APP_VERSION}" diff --git a/backend/tests/test_host_silence_scan.py b/backend/tests/test_host_silence_scan.py new file mode 100644 index 0000000..43b864a --- /dev/null +++ b/backend/tests/test_host_silence_scan.py @@ -0,0 +1,122 @@ +"""Proactive host_silence scan (stale heartbeat without waiting for ingest).""" + +import uuid +from datetime import datetime, timedelta, timezone +from unittest.mock import patch + +import pytest +from sqlalchemy import select + +from app.config import get_settings +from app.models import Problem +from app.services.host_silence_scan import run_host_silence_scan +from app.services.problem_rules import RULE_HOST_SILENCE +from tests.test_problem_rules import _ingest + + +@pytest.fixture +def scan_settings(monkeypatch): + monkeypatch.setenv("SAC_HEARTBEAT_STALE_MINUTES", "60") + monkeypatch.setenv("SAC_HOST_SILENCE_SCAN_ENABLED", "true") + monkeypatch.setenv("SAC_HOST_SILENCE_SCAN_INTERVAL_MINUTES", "5") + get_settings.cache_clear() + yield + get_settings.cache_clear() + + +def test_scan_creates_problem_for_stale_host(db_session, scan_settings): + hb = _ingest( + db_session, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2) + db_session.flush() + + now = datetime.now(timezone.utc) + results = run_host_silence_scan(db_session, now=now) + assert len(results) == 1 + assert results[0].created is True + assert results[0].problem.rule_id == RULE_HOST_SILENCE + + problem = db_session.scalar( + select(Problem).where( + Problem.host_id == hb.host_id, + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status == "open", + ) + ) + assert problem is not None + + +def test_scan_does_not_duplicate_open_problem(db_session, scan_settings): + hb = _ingest( + db_session, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2) + db_session.flush() + + now = datetime.now(timezone.utc) + first = run_host_silence_scan(db_session, now=now) + assert first[0].created is True + + second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5)) + assert len(second) == 1 + assert second[0].created is False + + problems = db_session.scalars( + select(Problem).where( + Problem.host_id == hb.host_id, + Problem.rule_id == RULE_HOST_SILENCE, + Problem.status == "open", + ) + ).all() + assert len(problems) == 1 + + +def test_scan_skips_host_without_heartbeat(db_session, scan_settings): + _ingest( + db_session, + event_id=str(uuid.uuid4()), + type="ssh.login.success", + severity="info", + title="login", + summary="ok", + ) + results = run_host_silence_scan(db_session) + assert results == [] + + +def test_scan_notifies_only_on_create(db_session, scan_settings): + hb = _ingest( + db_session, + type="agent.heartbeat", + category="agent", + severity="info", + title="hb", + summary="heartbeat", + ) + hb.received_at = datetime.now(timezone.utc) - timedelta(hours=2) + db_session.flush() + + from app.services.host_silence_scan import dispatch_host_silence_notifications + + with patch("app.services.notify_dispatch.notify_problem") as mock_notify: + results = run_host_silence_scan(db_session) + notified = dispatch_host_silence_notifications(db_session, results) + assert notified == 1 + mock_notify.assert_called_once() + + with patch("app.services.notify_dispatch.notify_problem") as mock_notify: + results2 = run_host_silence_scan(db_session) + notified2 = dispatch_host_silence_notifications(db_session, results2) + assert notified2 == 0 + mock_notify.assert_not_called() diff --git a/deploy/env.native.example b/deploy/env.native.example index a34cbfc..6408ee7 100644 --- a/deploy/env.native.example +++ b/deploy/env.native.example @@ -63,6 +63,10 @@ SAC_DAILY_REPORT_REQUIRE_ACTIVITY=true # Статус хоста по agent.heartbeat SAC_HEARTBEAT_STALE_MINUTES=780 +# Проактивный алерт host_silence (без ожидания нового события с хоста) +SAC_HOST_SILENCE_SCAN_ENABLED=true +SAC_HOST_SILENCE_SCAN_INTERVAL_MINUTES=5 + # Корреляция Problems: host + type + rule в одном open Problem (минуты) SAC_PROBLEM_CORRELATION_WINDOW_MINUTES=60 diff --git a/deploy/systemd/sac-host-silence-scan.service b/deploy/systemd/sac-host-silence-scan.service new file mode 100644 index 0000000..86d1fa8 --- /dev/null +++ b/deploy/systemd/sac-host-silence-scan.service @@ -0,0 +1,16 @@ +[Unit] +Description=Security Alert Center host silence scan (stale heartbeat) +After=network-online.target postgresql.service +Wants=network-online.target + +[Service] +Type=oneshot +User=sac +Group=sac +WorkingDirectory=/opt/security-alert-center/backend +EnvironmentFile=/opt/security-alert-center/config/sac-api.env +ExecStart=/opt/security-alert-center/backend/.venv/bin/python -m app.jobs.host_silence_scan +Nice=10 + +[Install] +WantedBy=multi-user.target diff --git a/deploy/systemd/sac-host-silence-scan.timer b/deploy/systemd/sac-host-silence-scan.timer new file mode 100644 index 0000000..7801614 --- /dev/null +++ b/deploy/systemd/sac-host-silence-scan.timer @@ -0,0 +1,11 @@ +[Unit] +Description=Periodic SAC host silence scan (stale agent.heartbeat) + +[Timer] +OnBootSec=3min +OnUnitActiveSec=5min +RandomizedDelaySec=30 +Persistent=true + +[Install] +WantedBy=timers.target diff --git a/docs/agent-integration.md b/docs/agent-integration.md index a3ce7e8..6ad1c68 100644 --- a/docs/agent-integration.md +++ b/docs/agent-integration.md @@ -101,6 +101,8 @@ UI **Настройки** (`/settings`) — правило + параметры | `rule:privilege_spike` | `privilege.sudo.command` на хосте | `SAC_PRIVILEGE_SPIKE_WINDOW_MINUTES=10`, `SAC_PRIVILEGE_SPIKE_THRESHOLD=10` | | `rule:host_silence` | был heartbeat, но устарел (`agent.heartbeat`) | `SAC_HEARTBEAT_STALE_MINUTES` (как для UI «Хосты») | +**Proactive host_silence:** фоновый scan в `sac-api` (`SAC_HOST_SILENCE_SCAN_ENABLED`, интервал `SAC_HOST_SILENCE_SCAN_INTERVAL_MINUTES`, по умолчанию каждые 5 мин) создаёт Problem и шлёт Telegram **без ожидания** нового ingest с хоста. Альтернатива: `systemd sac-host-silence-scan.timer` + `python -m app.jobs.host_silence_scan` (отключите in-process scan при нескольких воркерах uvicorn). + Свежий `agent.heartbeat` **автоматически resolve** open/ack проблему `rule:host_silence`. Одиночные `high`/`critical` (ban, mass bruteforce) — как раньше. ---