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 <cursoragent@cursor.com>

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
 

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()

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())

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)
 
 

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

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"})

backend/app/services/telegram_templates.py

@@ -543,6 +543,8 @@ def format_problem_telegram_html(problem: Problem, event: Event | None = None) -
     msg += f"\n<b>{html_escape(problem.title)}</b>\n{html_escape(problem.summary)}"
     if event is not None:
         msg += f"\n\n<i>Триггер:</i> {html_escape(event.type)} ({html_escape(event.severity)})"
+    elif problem.rule_id == "rule:host_silence":
+        msg += "\n\n<i>Триггер:</i> периодическая проверка SAC (host_silence scan)"
         if event.type in (
             "rdp.login.success",
             "rdp.login.failed",

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}"

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()