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