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

"""Unit tests for problem rules v1."""

import uuid
from datetime import datetime, timedelta, timezone

import pytest
from sqlalchemy import select

from app.config import get_settings
from app.models import Event, Problem
from app.services.ingest import ingest_event
from app.services.problem_rules import (
    RULE_BRUTE_FORCE,
    RULE_HOST_SILENCE,
    RULE_PRIVILEGE_SPIKE,
    evaluate_brute_force_burst,
    evaluate_host_silence,
    evaluate_privilege_spike,
    last_heartbeat_at,
)
from app.services.problems import maybe_create_problem
from tests.test_ingest import VALID_EVENT


def _payload(**overrides):
    now = datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")
    base = {
        **VALID_EVENT,
        "event_id": str(uuid.uuid4()),
        "occurred_at": now,
    }
    base.update(overrides)
    return base


def _ingest(db, **overrides):
    event, _ = ingest_event(db, _payload(**overrides))
    db.flush()
    return event


@pytest.fixture
def rule_settings(monkeypatch):
    monkeypatch.setenv("SAC_BRUTE_FORCE_THRESHOLD", "3")
    monkeypatch.setenv("SAC_BRUTE_FORCE_WINDOW_MINUTES", "15")
    monkeypatch.setenv("SAC_PRIVILEGE_SPIKE_THRESHOLD", "3")
    monkeypatch.setenv("SAC_PRIVILEGE_SPIKE_WINDOW_MINUTES", "10")
    monkeypatch.setenv("SAC_HEARTBEAT_STALE_MINUTES", "60")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


def test_brute_force_burst_at_threshold(db_session, rule_settings):
    for _ in range(3):
        _ingest(
            db_session,
            type="ssh.login.failed",
            severity="warning",
            title="failed",
            summary="fail",
            details={"source_ip": "10.0.0.9"},
        )
    event = _ingest(
        db_session,
        type="ssh.login.failed",
        severity="warning",
        title="failed",
        summary="fail",
        details={"source_ip": "10.0.0.9"},
    )
    match = evaluate_brute_force_burst(db_session, event)
    assert match is not None
    assert match.rule_id == RULE_BRUTE_FORCE
    problem, created = maybe_create_problem(db_session, event)
    assert created is True
    assert problem.rule_id == RULE_BRUTE_FORCE


def test_brute_force_below_threshold(db_session, rule_settings):
    event = _ingest(
        db_session,
        type="ssh.login.failed",
        severity="warning",
        title="failed",
        summary="fail",
        details={"source_ip": "10.0.0.1"},
    )
    assert evaluate_brute_force_burst(db_session, event) is None


def test_privilege_spike_at_threshold(db_session, rule_settings):
    for _ in range(2):
        _ingest(
            db_session,
            type="privilege.sudo.command",
            severity="warning",
            title="sudo",
            summary="sudo cmd",
        )
    event = _ingest(
        db_session,
        type="privilege.sudo.command",
        severity="warning",
        title="sudo",
        summary="sudo cmd",
    )
    match = evaluate_privilege_spike(db_session, event)
    assert match is not None
    assert match.rule_id == RULE_PRIVILEGE_SPIKE
    problem, created = maybe_create_problem(db_session, event)
    assert created is True
    assert problem.rule_id == RULE_PRIVILEGE_SPIKE


def test_host_silence_when_heartbeat_stale(db_session, rule_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()

    assert last_heartbeat_at(db_session, hb.host_id) is not None
    event = _ingest(
        db_session,
        type="ssh.login.success",
        severity="info",
        title="ok",
        summary="login",
    )
    match = evaluate_host_silence(db_session, event)
    assert match is not None
    assert match.rule_id == RULE_HOST_SILENCE
    problem, created = maybe_create_problem(db_session, event)
    assert created is True
    assert problem.rule_id == RULE_HOST_SILENCE


def test_heartbeat_resolves_host_silence(db_session, rule_settings):
    hb_old = _ingest(
        db_session,
        type="agent.heartbeat",
        category="agent",
        severity="info",
        title="hb",
        summary="heartbeat",
    )
    hb_old.received_at = datetime.now(timezone.utc) - timedelta(hours=2)
    db_session.flush()

    trigger = _ingest(
        db_session,
        type="ssh.login.success",
        severity="info",
        title="ok",
        summary="login",
    )
    maybe_create_problem(db_session, trigger)
    db_session.flush()

    open_silence = db_session.scalar(
        select(Problem).where(
            Problem.rule_id == RULE_HOST_SILENCE,
            Problem.status == "open",
        )
    )
    assert open_silence is not None

    fresh_hb = _ingest(
        db_session,
        type="agent.heartbeat",
        category="agent",
        severity="info",
        title="hb fresh",
        summary="heartbeat ok",
    )
    maybe_create_problem(db_session, fresh_hb)
    db_session.refresh(open_silence)
    assert open_silence.status == "resolved"