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

"""Tests for RDG 302→303 session flap rule."""

import uuid
from datetime import datetime, timedelta, timezone

import pytest

from app.config import get_settings
from app.models import Problem
from app.services.ingest import ingest_event
from app.services.problems import maybe_create_problem
from app.services.problem_rules import RULE_RDG_SESSION_FLAP
from app.services.rdg_session_flap import (
    evaluate_rdg_session_flap,
    event_has_rdg_flap,
    find_rdg_success_before_end,
    resolve_rdg_flap_summary,
)
from app.services.event_summary import event_to_summary
from tests.test_ingest import VALID_EVENT


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


def _ingest(db, occurred_at: datetime, **overrides):
    payload = _payload(**overrides)
    payload["occurred_at"] = occurred_at.isoformat()
    event, _ = ingest_event(db, payload)
    db.flush()
    return event


@pytest.fixture
def rdg_settings(monkeypatch):
    monkeypatch.setenv("SAC_RDG_FLAP_WINDOW_MIN_SEC", "1")
    monkeypatch.setenv("SAC_RDG_FLAP_WINDOW_MAX_SEC", "10")
    monkeypatch.setenv("SAC_RDG_FLAP_DEDUP_SEC", "30")
    get_settings.cache_clear()
    yield
    get_settings.cache_clear()


def test_rdg_flap_detects_302_then_303(db_session, rdg_settings):
    t0 = datetime.now(timezone.utc)
    user = "B26\\test.user"
    details = {"user": user, "internal_ip": "192.168.163.32"}

    _ingest(
        db_session,
        t0,
        type="rdg.connection.success",
        category="auth",
        severity="info",
        title="RD Gateway event 302",
        summary="302",
        details=details,
    )
    end = _ingest(
        db_session,
        t0 + timedelta(seconds=4),
        type="rdg.connection.disconnected",
        category="auth",
        severity="info",
        title="RD Gateway event 303",
        summary="303",
        details=details,
    )

    match = evaluate_rdg_session_flap(db_session, end)
    assert match is not None
    assert match.rule_id == RULE_RDG_SESSION_FLAP
    assert event_has_rdg_flap(end) is True

    problem, created = maybe_create_problem(db_session, end)
    assert created is True
    assert problem.rule_id == RULE_RDG_SESSION_FLAP


def test_rdg_flap_ignores_gap_over_10_sec(db_session, rdg_settings):
    t0 = datetime.now(timezone.utc)
    user = "B26\\slow.user"
    details = {"user": user}

    _ingest(
        db_session,
        t0,
        type="rdg.connection.success",
        category="auth",
        severity="info",
        title="302",
        summary="302",
        details=details,
    )
    end = _ingest(
        db_session,
        t0 + timedelta(seconds=15),
        type="rdg.connection.disconnected",
        category="auth",
        severity="info",
        title="303",
        summary="303",
        details=details,
    )

    assert find_rdg_success_before_end(db_session, end) is None
    assert evaluate_rdg_session_flap(db_session, end) is None


def test_rdg_flap_dedup_within_30_sec(db_session, rdg_settings):
    t0 = datetime.now(timezone.utc)
    user = "B26\\dup.user"
    details = {"user": user}

    def pair(at: datetime):
        _ingest(
            db_session,
            at,
            type="rdg.connection.success",
            category="auth",
            severity="info",
            title="302",
            summary="302",
            details=details,
        )
        return _ingest(
            db_session,
            at + timedelta(seconds=5),
            type="rdg.connection.disconnected",
            category="auth",
            severity="info",
            title="303",
            summary="303",
            details=details,
        )

    end1 = pair(t0)
    maybe_create_problem(db_session, end1)
    db_session.flush()

    end2 = pair(t0 + timedelta(seconds=10))
    maybe_create_problem(db_session, end2)
    db_session.flush()

    from sqlalchemy import func, select

    open_count = db_session.scalar(
        select(func.count()).select_from(Problem).where(
            Problem.rule_id == RULE_RDG_SESSION_FLAP,
            Problem.status == "open",
        )
    )
    assert open_count == 1


def test_resolve_rdg_flap_summary_for_302_and_303(db_session, rdg_settings):
    t0 = datetime.now(timezone.utc)
    user = "B26\\pair.user"
    details = {"user": user, "internal_ip": "192.168.163.48"}

    start = _ingest(
        db_session,
        t0,
        type="rdg.connection.success",
        category="auth",
        severity="warning",
        title="RD Gateway event 302",
        summary="302",
        details=details,
    )
    end = _ingest(
        db_session,
        t0 + timedelta(seconds=5),
        type="rdg.connection.disconnected",
        category="auth",
        severity="info",
        title="RD Gateway event 303",
        summary="303",
        details=details,
    )
    evaluate_rdg_session_flap(db_session, end)
    db_session.flush()

    end_flap, end_pair, end_qwinsta = resolve_rdg_flap_summary(db_session, end)
    start_flap, start_pair, start_qwinsta = resolve_rdg_flap_summary(db_session, start)

    assert end_flap is True
    assert end_pair == start.id
    assert end_qwinsta == end.id

    assert start_flap is True
    assert start_pair == end.id
    assert start_qwinsta == end.id

    end_summary = event_to_summary(end, db_session)
    start_summary = event_to_summary(start, db_session)
    assert end_summary.rdg_flap is True
    assert start_summary.rdg_flap is True
    assert start_summary.rdg_flap_qwinsta_event_id == end.id