281 lines
7.7 KiB
Python
281 lines
7.7 KiB
Python
"""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"
|
|
assert open_silence.resolved_by == "auto"
|
|
|
|
|
|
def test_host_silence_suppressed_after_manual_resolve_on_ingest(db_session, rule_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()
|
|
|
|
trigger = _ingest(
|
|
db_session,
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="ok",
|
|
summary="login",
|
|
)
|
|
problem, created = maybe_create_problem(db_session, trigger)
|
|
assert created is True
|
|
problem.status = "resolved"
|
|
problem.resolved_by = "manual"
|
|
problem.updated_at = datetime.now(timezone.utc)
|
|
db_session.flush()
|
|
|
|
trigger2 = _ingest(
|
|
db_session,
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="ok2",
|
|
summary="login2",
|
|
)
|
|
problem2, created2 = maybe_create_problem(db_session, trigger2)
|
|
assert problem2 is None
|
|
assert created2 is False
|
|
|
|
open_silence = db_session.scalar(
|
|
select(Problem).where(
|
|
Problem.rule_id == RULE_HOST_SILENCE,
|
|
Problem.status == "open",
|
|
)
|
|
)
|
|
assert open_silence is None
|
|
|
|
|
|
def test_host_silence_reopens_after_auto_resolve_when_still_stale(db_session, rule_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()
|
|
|
|
trigger = _ingest(
|
|
db_session,
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="ok",
|
|
summary="login",
|
|
)
|
|
problem, created = maybe_create_problem(db_session, trigger)
|
|
assert created is True
|
|
|
|
from app.services.problem_rules import resolve_host_silence_problems
|
|
|
|
resolve_host_silence_problems(db_session, hb.host_id)
|
|
db_session.refresh(problem)
|
|
assert problem.status == "resolved"
|
|
assert problem.resolved_by == "auto"
|
|
|
|
trigger2 = _ingest(
|
|
db_session,
|
|
type="ssh.login.success",
|
|
severity="info",
|
|
title="ok2",
|
|
summary="login2",
|
|
)
|
|
problem2, created2 = maybe_create_problem(db_session, trigger2)
|
|
assert created2 is True
|
|
assert problem2 is not None
|
|
assert problem2.status == "open"
|
|
assert problem2.id != problem.id
|