feat: problem rules v1 brute-force, privilege spike, host silence (d1-3)

- Threshold rules on ingest; heartbeat auto-resolves host silence

- Config thresholds in sac-api.env; unit tests

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/config.py

@@ -56,6 +56,14 @@ class Settings(BaseSettings):
     # Окно корреляции Problems: host + type + rule в одном open Problem
     sac_problem_correlation_window_minutes: int = 60
 
+    # rule:brute_force_burst — ssh.login.failed / rdp.login.failed
+    sac_brute_force_window_minutes: int = 15
+    sac_brute_force_threshold: int = 30
+
+    # rule:privilege_spike — privilege.sudo.command
+    sac_privilege_spike_window_minutes: int = 10
+    sac_privilege_spike_threshold: int = 10
+
 
 @lru_cache
 def get_settings() -> Settings:

backend/app/services/problem_rules.py

@@ -0,0 +1,199 @@
+"""Problem rule evaluators (v1): brute-force, privilege spike, host silence."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.config import get_settings
+from app.models import Event, Host
+from app.services.host_health import HEARTBEAT_TYPE, agent_status
+
+BRUTE_FAILED_TYPES = frozenset({"ssh.login.failed", "rdp.login.failed"})
+PRIVILEGE_SUDO_TYPE = "privilege.sudo.command"
+
+RULE_BRUTE_FORCE = "rule:brute_force_burst"
+RULE_PRIVILEGE_SPIKE = "rule:privilege_spike"
+RULE_HOST_SILENCE = "rule:host_silence"
+
+
+@dataclass(frozen=True)
+class RuleMatch:
+    rule_id: str
+    correlation_type: str
+    title: str
+    summary: str
+    severity: str
+    fingerprint_suffix: str = ""
+
+
+def _now() -> datetime:
+    return datetime.now(timezone.utc)
+
+
+def _event_source_ip(event: Event) -> str:
+    details = event.details if isinstance(event.details, dict) else {}
+    ip = details.get("source_ip") or details.get("ip")
+    return str(ip) if ip else "unknown"
+
+
+def count_events_in_window(
+    db: Session,
+    *,
+    host_id: int,
+    event_type: str,
+    since: datetime,
+    source_ip: str | None = None,
+) -> int:
+    rows = db.scalars(
+        select(Event).where(
+            Event.host_id == host_id,
+            Event.type == event_type,
+            Event.occurred_at >= since,
+        )
+    ).all()
+    if source_ip is None:
+        return len(rows)
+    return sum(1 for e in rows if _event_source_ip(e) == source_ip)
+
+
+def last_heartbeat_at(db: Session, host_id: int) -> datetime | None:
+    return db.scalar(
+        select(func.max(Event.received_at)).where(
+            Event.host_id == host_id,
+            Event.type == HEARTBEAT_TYPE,
+        )
+    )
+
+
+def evaluate_brute_force_burst(db: Session, event: Event) -> RuleMatch | None:
+    if event.type not in BRUTE_FAILED_TYPES:
+        return None
+    settings = get_settings()
+    source_ip = _event_source_ip(event)
+    since = _now() - timedelta(minutes=settings.sac_brute_force_window_minutes)
+    count = count_events_in_window(
+        db,
+        host_id=event.host_id,
+        event_type=event.type,
+        since=since,
+        source_ip=source_ip,
+    )
+    if count < settings.sac_brute_force_threshold:
+        return None
+    return RuleMatch(
+        rule_id=RULE_BRUTE_FORCE,
+        correlation_type=event.type,
+        fingerprint_suffix=f"ip{source_ip}",
+        title=f"Brute-force burst: {count} failed logins",
+        summary=(
+            f"{count} событий {event.type} за {settings.sac_brute_force_window_minutes} мин "
+            f"(IP {source_ip}, порог {settings.sac_brute_force_threshold})"
+        ),
+        severity="high",
+    )
+
+
+def evaluate_privilege_spike(db: Session, event: Event) -> RuleMatch | None:
+    if event.type != PRIVILEGE_SUDO_TYPE:
+        return None
+    settings = get_settings()
+    since = _now() - timedelta(minutes=settings.sac_privilege_spike_window_minutes)
+    count = count_events_in_window(
+        db,
+        host_id=event.host_id,
+        event_type=event.type,
+        since=since,
+    )
+    if count < settings.sac_privilege_spike_threshold:
+        return None
+    return RuleMatch(
+        rule_id=RULE_PRIVILEGE_SPIKE,
+        correlation_type=event.type,
+        title=f"Privilege spike: {count} sudo commands",
+        summary=(
+            f"{count} событий {event.type} за {settings.sac_privilege_spike_window_minutes} мин "
+            f"(порог {settings.sac_privilege_spike_threshold})"
+        ),
+        severity="warning" if count < settings.sac_privilege_spike_threshold * 2 else "high",
+    )
+
+
+def evaluate_host_silence(db: Session, event: Event) -> RuleMatch | None:
+    if event.type == HEARTBEAT_TYPE:
+        return None
+    settings = get_settings()
+    hb = last_heartbeat_at(db, event.host_id)
+    if hb is None:
+        return None
+    if agent_status(hb, stale_minutes=settings.sac_heartbeat_stale_minutes, now=_now()) != "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)
+    return RuleMatch(
+        rule_id=RULE_HOST_SILENCE,
+        correlation_type="host.silence",
+        title=f"Host silence: {hostname}",
+        summary=(
+            f"Нет свежего agent.heartbeat более {settings.sac_heartbeat_stale_minutes} мин "
+            f"(последний ~{age_min} мин назад)"
+        ),
+        severity="high",
+    )
+
+
+def evaluate_immediate_high(event: Event) -> RuleMatch | None:
+    """Одиночные high/critical и известные типы (ban, mass bruteforce)."""
+    problem_types = frozenset({"ssh.ip.banned", "ssh.bruteforce.mass"})
+    if event.severity not in frozenset({"high", "critical"}) and event.type not in problem_types:
+        return None
+    rule_id = "high_severity" if event.severity in frozenset({"high", "critical"}) else f"type:{event.type}"
+    return RuleMatch(
+        rule_id=rule_id,
+        correlation_type=event.type,
+        title=event.title,
+        summary=event.summary,
+        severity=event.severity,
+    )
+
+
+def resolve_host_silence_problems(db: Session, host_id: int) -> int:
+    """Закрыть open/ack host silence при свежем heartbeat."""
+    from app.models import Problem
+
+    problems = db.scalars(
+        select(Problem).where(
+            Problem.host_id == host_id,
+            Problem.rule_id == RULE_HOST_SILENCE,
+            Problem.status.in_(("open", "acknowledged")),
+        )
+    ).all()
+    for problem in problems:
+        problem.status = "resolved"
+    if problems:
+        db.flush()
+    return len(problems)
+
+
+def build_fingerprint(host_id: int, correlation_type: str, rule_id: str, suffix: str = "") -> str:
+    base = f"h{host_id}"
+    if suffix:
+        base = f"{base}:{suffix}"
+    return f"{base}:t{correlation_type}:r{rule_id}"
+
+
+def pick_rule_match(db: Session, event: Event) -> RuleMatch | None:
+    """Первое сработавшее правило (приоритет: burst → spike → silence → immediate)."""
+    for evaluator in (
+        evaluate_brute_force_burst,
+        evaluate_privilege_spike,
+        evaluate_host_silence,
+    ):
+        match = evaluator(db, event)
+        if match is not None:
+            return match
+    return evaluate_immediate_high(event)

backend/app/services/problems.py

@@ -1,36 +1,24 @@
-"""Auto-create Problems from ingested events (MVP rules + correlation)."""
+"""Auto-create Problems from ingested events (rules + correlation)."""
 
-from datetime import datetime, timedelta, timezone
+from datetime import datetime, timezone
 
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
 from app.config import get_settings
 from app.models import Event, Problem, ProblemEvent
-
-PROBLEM_TYPES = frozenset(
-    {
-        "ssh.ip.banned",
-        "ssh.bruteforce.mass",
-    }
+from app.services.host_health import HEARTBEAT_TYPE
+from app.services.problem_rules import (
+    RuleMatch,
+    build_fingerprint,
+    pick_rule_match,
+    resolve_host_silence_problems,
 )
 
-HIGH_SEVERITIES = frozenset({"high", "critical"})
-
-
-def problem_rule_id(event: Event) -> str | None:
-    if event.severity in HIGH_SEVERITIES:
-        return "high_severity"
-    if event.type in PROBLEM_TYPES:
-        return f"type:{event.type}"
-    return None
-
-
-def problem_fingerprint(host_id: int, event_type: str, rule_id: str) -> str:
-    return f"h{host_id}:t{event_type}:r{rule_id}"
-
 
 def _correlation_cutoff(now: datetime) -> datetime:
+    from datetime import timedelta
+
     minutes = get_settings().sac_problem_correlation_window_minutes
     return now - timedelta(minutes=minutes)
 
@@ -48,17 +36,20 @@ def _append_event(db: Session, problem: Problem, event: Event) -> None:
     problem.event_count = (problem.event_count or 0) + 1
     problem.last_seen_at = event.occurred_at
     problem.updated_at = datetime.now(timezone.utc)
-    if event.severity in HIGH_SEVERITIES and problem.severity not in HIGH_SEVERITIES:
+    if event.severity in {"high", "critical"} and problem.severity not in {"high", "critical"}:
         problem.severity = event.severity
     db.flush()
 
 
-def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, bool]:
-    rule_id = problem_rule_id(event)
-    if rule_id is None:
-        return None, False
-
-    fingerprint = problem_fingerprint(event.host_id, event.type, rule_id)
+def open_or_append_problem(
+    db: Session, event: Event, match: RuleMatch
+) -> tuple[Problem, bool]:
+    fingerprint = build_fingerprint(
+        event.host_id,
+        match.correlation_type,
+        match.rule_id,
+        match.fingerprint_suffix,
+    )
     now = datetime.now(timezone.utc)
     cutoff = _correlation_cutoff(now)
 
@@ -75,15 +66,18 @@ def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, boo
     )
     if open_problem:
         _append_event(db, open_problem, event)
+        if match.severity in {"high", "critical"} and open_problem.severity not in {"high", "critical"}:
+            open_problem.severity = match.severity
+            open_problem.summary = match.summary
         return open_problem, False
 
     problem = Problem(
         host_id=event.host_id,
-        title=event.title,
-        summary=event.summary,
-        severity=event.severity,
+        title=match.title,
+        summary=match.summary,
+        severity=match.severity,
         status="open",
-        rule_id=rule_id,
+        rule_id=match.rule_id,
         fingerprint=fingerprint,
         event_count=1,
         last_seen_at=event.occurred_at,
@@ -93,3 +87,15 @@ def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, boo
     db.add(ProblemEvent(problem_id=problem.id, event_id=event.id))
     db.flush()
     return problem, True
+
+
+def maybe_create_problem(db: Session, event: Event) -> tuple[Problem | None, bool]:
+    if event.type == HEARTBEAT_TYPE:
+        resolve_host_silence_problems(db, event.host_id)
+        return None, False
+
+    match = pick_rule_match(db, event)
+    if match is None:
+        return None, False
+
+    return open_or_append_problem(db, event, match)

backend/tests/test_problem_rules.py

@@ -0,0 +1,184 @@
+"""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"