feat: always send RDP/SSH login success to Telegram via SAC

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-05-31 18:47:39 +10:00
parent 4baa2b426f
commit fbef1e2f4b
3 changed files with 74 additions and 2 deletions
@@ -89,6 +89,62 @@ def test_notify_event_skipped_by_cooldown():
    mock_tg.notify_event.assert_not_called()


+def test_notify_auth_login_bypasses_min_severity():
+    event = Event(
+        event_id="00000000-0000-4000-8000-000000000601",
+        host_id=1,
+        occurred_at=datetime(2026, 5, 31, 18, 36, tzinfo=timezone.utc),
+        category="auth",
+        type="rdp.login.success",
+        severity="info",
+        title="RDP login event 4624",
+        summary="papatramp from 192.168.160.3",
+        details={"telegram_via": "sac"},
+        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=True):
+            with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
+                notify_dispatch.notify_auth_login(event)
+    mock_tg.notify_event.assert_called_once()
+
+
+def test_notify_auth_login_skips_telegram_when_via_agent():
+    event = Event(
+        event_id="00000000-0000-4000-8000-000000000602",
+        host_id=1,
+        occurred_at=datetime(2026, 5, 31, 18, 36, tzinfo=timezone.utc),
+        category="auth",
+        type="ssh.login.success",
+        severity="info",
+        title="SSH login",
+        summary="user from 10.0.0.1",
+        details={"telegram_via": "agent"},
+        payload={},
+    )
+    policy = NotificationPolicyConfig(
+        min_severity="warning",
+        use_telegram=True,
+        use_webhook=True,
+        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=True):
+            with patch.object(notify_dispatch, "telegram_notify") as mock_tg:
+                with patch.object(notify_dispatch, "webhook_notify") as mock_wh:
+                    notify_dispatch.notify_auth_login(event)
+    mock_tg.notify_event.assert_not_called()
+    mock_wh.notify_event.assert_called_once()
+
+
 def test_notify_lifecycle_bypasses_min_severity():
    event = Event(
        event_id="00000000-0000-4000-8000-000000000501",