feat: notification cooldown dedup for events and problems (notif-31)

Store last notify time by dedup_key/fingerprint, gate dispatch before channels; config SAC_NOTIFY_*_COOLDOWN_SEC, migration 008. Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-29 16:24:01 +10:00
parent 1a70980920
commit 4167687dec
12 changed files with 309 additions and 14 deletions
@@ -0,0 +1,31 @@
+"""notification_cooldown for SAC notify dedup (F-NOT-03)
+
+Revision ID: 008
+Revises: 007
+"""
+
+from typing import Sequence, Union
+
+import sqlalchemy as sa
+from alembic import op
+
+revision: str = "008"
+down_revision: Union[str, None] = "007"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "notification_cooldown",
+        sa.Column("cooldown_key", sa.String(length=512), nullable=False),
+        sa.Column("kind", sa.String(length=16), nullable=False),
+        sa.Column("last_notified_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("cooldown_key"),
+    )
+    op.create_index("ix_notification_cooldown_kind", "notification_cooldown", ["kind"])
+
+
+def downgrade() -> None:
+    op.drop_index("ix_notification_cooldown_kind", table_name="notification_cooldown")
+    op.drop_table("notification_cooldown")
@@ -70,6 +70,11 @@ class Settings(BaseSettings):
    notify_min_severity: str = "warning"
    notify_channels: str = "telegram,webhook,email"

+    # F-NOT-03: cooldown исходящих оповещений (dedup_key / fingerprint)
+    sac_notify_cooldown_enabled: bool = True
+    sac_notify_event_cooldown_sec: int = 90
+    sac_notify_problem_cooldown_sec: int = 300
+
    # Порог «живости» агента
    sac_heartbeat_stale_minutes: int = 780

@@ -2,7 +2,17 @@ from app.models.api_key import ApiKey
 from app.models.event import Event
 from app.models.host import Host
 from app.models.notification_channel import NotificationChannel
+from app.models.notification_cooldown import NotificationCooldown
 from app.models.notification_policy import NotificationPolicy
 from app.models.problem import Problem, ProblemEvent

-__all__ = ["ApiKey", "Event", "Host", "NotificationChannel", "NotificationPolicy", "Problem", "ProblemEvent"]
+__all__ = [
+    "ApiKey",
+    "Event",
+    "Host",
+    "NotificationChannel",
+    "NotificationCooldown",
+    "NotificationPolicy",
+    "Problem",
+    "ProblemEvent",
+]
@@ -0,0 +1,16 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, String
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class NotificationCooldown(Base):
+    """Последняя отправка оповещения по ключу (dedup_key / fingerprint)."""
+
+    __tablename__ = "notification_cooldown"
+
+    cooldown_key: Mapped[str] = mapped_column(String(512), primary_key=True)
+    kind: Mapped[str] = mapped_column(String(16))  # event | problem
+    last_notified_at: Mapped[datetime] = mapped_column(DateTime(timezone=True))
@@ -0,0 +1,100 @@
+"""Cooldown / dedup for outbound notifications (F-NOT-03)."""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timezone
+from typing import Any
+
+from sqlalchemy.orm import Session
+
+from app.config import get_settings
+from app.models import Event, Problem
+from app.models.notification_cooldown import NotificationCooldown
+
+logger = logging.getLogger(__name__)
+
+COOLDOWN_KIND_EVENT = "event"
+COOLDOWN_KIND_PROBLEM = "problem"
+
+# Не душим служебные проверки ingest
+_EVENT_COOLDOWN_EXEMPT_TYPES = frozenset({"agent.test"})
+
+
+def _as_utc(dt: datetime) -> datetime:
+    if dt.tzinfo is None:
+        return dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)
+
+
+def build_event_cooldown_key(event: Event) -> str:
+    if event.dedup_key and str(event.dedup_key).strip():
+        return f"event:{event.dedup_key.strip()}"
+
+    details: dict[str, Any] = event.details if isinstance(event.details, dict) else {}
+    host = event.host.hostname if event.host else "unknown"
+    ip = str(details.get("source_ip") or details.get("ip_address") or details.get("ip") or "").strip()
+    user = str(details.get("user") or details.get("username") or "").strip()
+    return f"event:{event.type}|{host}|{ip}|{user}"
+
+
+def build_problem_cooldown_key(problem: Problem) -> str:
+    fp = (problem.fingerprint or "").strip() or f"id:{problem.id}"
+    return f"problem:{fp}"
+
+
+def _cooldown_seconds_for_kind(kind: str) -> int:
+    settings = get_settings()
+    if not settings.sac_notify_cooldown_enabled:
+        return 0
+    if kind == COOLDOWN_KIND_PROBLEM:
+        return max(0, int(settings.sac_notify_problem_cooldown_sec))
+    return max(0, int(settings.sac_notify_event_cooldown_sec))
+
+
+def allow_notify(db: Session | None, *, key: str, kind: str) -> bool:
+    """True = можно слать сейчас; при True обновляет last_notified_at."""
+    if kind == COOLDOWN_KIND_EVENT:
+        cooldown_sec = _cooldown_seconds_for_kind(COOLDOWN_KIND_EVENT)
+    else:
+        cooldown_sec = _cooldown_seconds_for_kind(COOLDOWN_KIND_PROBLEM)
+
+    if cooldown_sec <= 0:
+        return True
+    if db is None:
+        return True
+
+    now = datetime.now(timezone.utc)
+    row = db.get(NotificationCooldown, key)
+    if row is not None:
+        last = _as_utc(row.last_notified_at)
+        elapsed = (now - last).total_seconds()
+        if elapsed >= 0 and elapsed < cooldown_sec:
+            logger.info(
+                "notify cooldown skip kind=%s key=%s elapsed=%.0fs need=%ss",
+                kind,
+                key[:120],
+                elapsed,
+                cooldown_sec,
+            )
+            return False
+
+    if row is None:
+        db.add(NotificationCooldown(cooldown_key=key, kind=kind, last_notified_at=now))
+    else:
+        row.kind = kind
+        row.last_notified_at = now
+    db.flush()
+    return True
+
+
+def should_notify_event(event: Event, db: Session | None) -> bool:
+    if event.type in _EVENT_COOLDOWN_EXEMPT_TYPES:
+        return True
+    key = build_event_cooldown_key(event)
+    return allow_notify(db, key=key, kind=COOLDOWN_KIND_EVENT)
+
+
+def should_notify_problem(problem: Problem, db: Session | None) -> bool:
+    key = build_problem_cooldown_key(problem)
+    return allow_notify(db, key=key, kind=COOLDOWN_KIND_PROBLEM)
@@ -1,18 +1,19 @@
 """Dispatch ingest notifications per global policy (severity → channels)."""

+import logging
+
 from sqlalchemy.orm import Session

 from app.models import Event, Problem
 from app.services import email_notify, telegram_notify, webhook_notify
+from app.services.notification_cooldown import should_notify_event, should_notify_problem
 from app.services.notification_policy import get_effective_notification_policy
 from app.services.notification_severity import severity_meets_minimum
-from app.services.notification_settings import CHANNEL_EMAIL, CHANNEL_TELEGRAM, CHANNEL_WEBHOOK
+
+logger = logging.getLogger(__name__)


-def notify_event(event: Event, *, db: Session | None = None) -> None:
-    policy = get_effective_notification_policy(db)
-    if not severity_meets_minimum(event.severity, policy.min_severity):
-        return
+def _dispatch_event_channels(event: Event, *, db: Session | None, policy) -> None:
    if policy.use_telegram:
        telegram_notify.notify_event(event, db=db, apply_policy_gate=False)
    if policy.use_webhook:
@@ -21,13 +22,28 @@ def notify_event(event: Event, *, db: Session | None = None) -> None:
        email_notify.notify_event(event, db=db, apply_policy_gate=False)


-def notify_problem(problem: Problem, event: Event | None = None, *, db: Session | None = None) -> None:
-    policy = get_effective_notification_policy(db)
-    if not severity_meets_minimum(problem.severity, policy.min_severity):
-        return
+def _dispatch_problem_channels(problem: Problem, event: Event | None, *, db: Session | None, policy) -> None:
    if policy.use_telegram:
        telegram_notify.notify_problem(problem, event, db=db, apply_policy_gate=False)
    if policy.use_webhook:
        webhook_notify.notify_problem(problem, event, db=db, apply_policy_gate=False)
    if policy.use_email:
        email_notify.notify_problem(problem, event, db=db, apply_policy_gate=False)
+
+
+def notify_event(event: Event, *, db: Session | None = None) -> None:
+    policy = get_effective_notification_policy(db)
+    if not severity_meets_minimum(event.severity, policy.min_severity):
+        return
+    if not should_notify_event(event, db):
+        return
+    _dispatch_event_channels(event, db=db, policy=policy)
+
+
+def notify_problem(problem: Problem, event: Event | None = None, *, db: Session | None = None) -> None:
+    policy = get_effective_notification_policy(db)
+    if not severity_meets_minimum(problem.severity, policy.min_severity):
+        return
+    if not should_notify_problem(problem, db):
+        return
+    _dispatch_problem_channels(problem, event, db=db, policy=policy)
@@ -0,0 +1,85 @@
+"""Notification cooldown / dedup (notif-31)."""
+
+from datetime import datetime, timedelta, timezone
+
+from app.config import get_settings
+from app.models import Event
+from app.models.notification_cooldown import NotificationCooldown
+from app.services.notification_cooldown import (
+    allow_notify,
+    build_event_cooldown_key,
+    build_problem_cooldown_key,
+    should_notify_event,
+)
+from app.models import Problem
+
+
+def test_build_event_cooldown_key_prefers_dedup_key():
+    event = Event(
+        event_id="00000000-0000-4000-8000-000000000601",
+        host_id=1,
+        occurred_at=datetime(2026, 5, 29, 16, 0, tzinfo=timezone.utc),
+        category="auth",
+        type="rdp.login.failed",
+        severity="warning",
+        title="t",
+        summary="s",
+        dedup_key="rdp|host|failed|10.0.0.1|admin",
+        payload={},
+    )
+    assert build_event_cooldown_key(event) == "event:rdp|host|failed|10.0.0.1|admin"
+
+
+def test_allow_notify_blocks_within_cooldown(db_session, monkeypatch):
+    monkeypatch.setenv("SAC_NOTIFY_COOLDOWN_ENABLED", "true")
+    monkeypatch.setenv("SAC_NOTIFY_EVENT_COOLDOWN_SEC", "90")
+    get_settings.cache_clear()
+
+    key = "event:test-key"
+    past = datetime.now(timezone.utc) - timedelta(seconds=30)
+    db_session.add(NotificationCooldown(cooldown_key=key, kind="event", last_notified_at=past))
+    db_session.commit()
+
+    assert allow_notify(db_session, key=key, kind="event") is False
+
+
+def test_allow_notify_allows_after_cooldown(db_session, monkeypatch):
+    monkeypatch.setenv("SAC_NOTIFY_COOLDOWN_ENABLED", "true")
+    monkeypatch.setenv("SAC_NOTIFY_EVENT_COOLDOWN_SEC", "60")
+    get_settings.cache_clear()
+
+    key = "event:old"
+    past = datetime.now(timezone.utc) - timedelta(seconds=120)
+    db_session.add(NotificationCooldown(cooldown_key=key, kind="event", last_notified_at=past))
+    db_session.commit()
+
+    assert allow_notify(db_session, key=key, kind="event") is True
+
+
+def test_should_notify_event_exempt_agent_test(db_session, monkeypatch):
+    monkeypatch.setenv("SAC_NOTIFY_EVENT_COOLDOWN_SEC", "90")
+    get_settings.cache_clear()
+
+    event = Event(
+        event_id="00000000-0000-4000-8000-000000000602",
+        host_id=1,
+        occurred_at=datetime(2026, 5, 29, 16, 0, tzinfo=timezone.utc),
+        category="agent",
+        type="agent.test",
+        severity="info",
+        title="t",
+        summary="s",
+        payload={},
+    )
+    assert should_notify_event(event, db_session) is True
+
+
+def test_build_problem_cooldown_key():
+    problem = Problem(
+        title="brute",
+        summary="x",
+        severity="high",
+        status="open",
+        fingerprint="host:1|rdp.login.failed|rule:brute",
+    )
+    assert build_problem_cooldown_key(problem).startswith("problem:")
@@ -8,7 +8,7 @@ from app.services import notify_dispatch
 from app.services.notification_policy import NotificationPolicyConfig


-def test_notify_event_skipped_below_policy_severity():
+def test_notify_event_calls_selected_channels():
    event = Event(
        event_id="00000000-0000-4000-8000-000000000401",
        host_id=1,
@@ -60,3 +60,30 @@ def test_notify_event_calls_selected_channels():
    mock_tg.notify_event.assert_called_once()
    mock_wh.notify_event.assert_called_once()
    mock_em.notify_event.assert_not_called()
+
+
+def test_notify_event_skipped_by_cooldown():
+    event = Event(
+        event_id="00000000-0000-4000-8000-000000000403",
+        host_id=1,
+        occurred_at=datetime(2026, 5, 29, 15, 0, tzinfo=timezone.utc),
+        category="auth",
+        type="rdp.login.failed",
+        severity="warning",
+        title="fail",
+        summary="send",
+        dedup_key="same-key",
+        payload={},
+    )
+    policy = NotificationPolicyConfig(
+        min_severity="warning",
+        use_telegram=True,
+        use_webhook=False,
+        use_email=False,
+        source="db",
+    )
+    with patch.object(notify_dispatch, "get_effective_notification_policy", return_value=policy):
+        with patch.object(notify_dispatch, "should_notify_event", return_value=False):
+            with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
+                notify_dispatch.notify_event(event)
+    mock_tg.notify_event.assert_not_called()