security-alert-center/backend/tests/test_host_silence_scan.py

"""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_does_not_duplicate_after_correlation_window(db_session, scan_settings, monkeypatch):
    """host_silence — одна open-проблема на хост, даже если прошло > correlation window."""
    monkeypatch.setenv("SAC_PROBLEM_CORRELATION_WINDOW_MINUTES", "60")
    get_settings.cache_clear()

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

    t0 = datetime.now(timezone.utc)
    first = run_host_silence_scan(db_session, now=t0)
    assert first[0].created is True
    first[0].problem.last_seen_at = t0 - timedelta(hours=2)
    db_session.flush()

    later = run_host_silence_scan(db_session, now=t0 + timedelta(hours=3))
    assert len(later) == 1
    assert later[0].created is False
    assert later[0].problem.id == first[0].problem.id

    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_suppresses_after_manual_resolve_within_cooldown(db_session, scan_settings, monkeypatch):
    monkeypatch.setenv("SAC_HOST_SILENCE_MANUAL_RESOLVE_COOLDOWN_HOURS", "12")
    get_settings.cache_clear()

    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
    problem = first[0].problem
    problem.status = "resolved"
    problem.resolved_by = "manual"
    problem.updated_at = now
    db_session.flush()

    second = run_host_silence_scan(db_session, now=now + timedelta(minutes=10))
    assert second == []

    open_count = db_session.scalar(
        select(Problem).where(
            Problem.host_id == hb.host_id,
            Problem.rule_id == RULE_HOST_SILENCE,
            Problem.status == "open",
        )
    )
    assert open_count is None

    resolved_count = len(
        db_session.scalars(
            select(Problem).where(
                Problem.host_id == hb.host_id,
                Problem.rule_id == RULE_HOST_SILENCE,
                Problem.status == "resolved",
            )
        ).all()
    )
    assert resolved_count == 1


def test_scan_reopens_after_manual_cooldown_expires(db_session, scan_settings, monkeypatch):
    monkeypatch.setenv("SAC_HOST_SILENCE_MANUAL_RESOLVE_COOLDOWN_HOURS", "12")
    get_settings.cache_clear()

    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)
    problem = first[0].problem
    problem.status = "resolved"
    problem.resolved_by = "manual"
    problem.updated_at = now - timedelta(hours=13)
    db_session.flush()

    later = run_host_silence_scan(db_session, now=now)
    assert len(later) == 1
    assert later[0].created is True
    assert later[0].problem.id == problem.id
    assert later[0].problem.status == "open"
    assert later[0].problem.resolved_by is None


def test_scan_does_not_duplicate_acknowledged_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
    first[0].problem.status = "acknowledged"
    db_session.flush()

    second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5))
    assert len(second) == 1
    assert second[0].created is False
    assert second[0].problem.id == first[0].problem.id

    problems = db_session.scalars(
        select(Problem).where(
            Problem.host_id == hb.host_id,
            Problem.rule_id == RULE_HOST_SILENCE,
            Problem.status.in_(("open", "acknowledged")),
        )
    ).all()
    assert len(problems) == 1


def test_scan_dedupes_by_hostname_across_duplicate_hosts(db_session, scan_settings):
    from app.models import Host

    now = datetime.now(timezone.utc)
    host_a = Host(
        hostname="COMM-PC",
        os_family="windows",
        product="rdp-login-monitor",
        ipv4="192.168.162.33",
    )
    host_b = Host(
        hostname="COMM-PC",
        os_family="windows",
        product="rdp-login-monitor",
        ipv4="192.168.162.33",
    )
    db_session.add_all([host_a, host_b])
    db_session.flush()

    hb = _ingest(
        db_session,
        host={"hostname": host_a.hostname, "os_family": "windows", "ipv4": host_a.ipv4},
        type="agent.heartbeat",
        category="agent",
        severity="info",
        title="hb",
        summary="heartbeat",
    )
    hb.host_id = host_a.id
    hb.received_at = now - timedelta(hours=2)
    db_session.flush()

    first = run_host_silence_scan(db_session, now=now)
    assert len(first) == 1
    assert first[0].created is True

    hb_b = _ingest(
        db_session,
        event_id=str(uuid.uuid4()),
        host={"hostname": host_b.hostname, "os_family": "windows", "ipv4": host_b.ipv4},
        source={"product": "rdp-login-monitor", "product_version": "2.0.0-SAC"},
        type="agent.heartbeat",
        category="agent",
        severity="info",
        title="hb2",
        summary="heartbeat",
    )
    hb_b.host_id = host_b.id
    hb_b.received_at = now - timedelta(hours=2)
    db_session.flush()

    second = run_host_silence_scan(db_session, now=now + timedelta(minutes=5))
    assert len(second) == 2
    assert all(not item.created for item in second)
    assert {item.problem.id for item in second} == {first[0].problem.id}

    active = db_session.scalars(
        select(Problem).where(
            Problem.rule_id == RULE_HOST_SILENCE,
            Problem.status.in_(("open", "acknowledged")),
        )
    ).all()
    assert len(active) == 1


def test_scan_skipped_when_advisory_lock_not_acquired(db_session, scan_settings, monkeypatch):
    monkeypatch.setattr(
        "app.services.host_silence_scan._host_silence_scan_lock",
        lambda _db: False,
    )
    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()

    assert run_host_silence_scan(db_session) == []


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