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

"""Tests for direct RDP logoff → workstation rdp.login.success correlation."""

import uuid
from datetime import datetime, timedelta, timezone

from app.services.event_summary import event_to_summary
from app.services.host_sessions import event_session_terminated
from app.services.ingest import ingest_event
from app.services.rdp_session_logoff import (
    SESSION_CLOSED_BY_LOGOFF_AT_KEY,
    SESSION_CLOSED_BY_LOGOFF_EVENT_ID_KEY,
    close_workstation_session_for_rdp_logoff,
    find_logoff_after_workstation_login,
    resolve_workstation_login_closed_by_logoff,
)
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


def test_logoff_marks_workstation_login_closed_on_ingest(db_session):
    t0 = datetime.now(timezone.utc)
    user = r"B26\papatramp"
    host = {"hostname": "BIV-PC", "os_family": "windows", "ipv4": "192.168.165.39"}
    source = {"product": "rdp-login-monitor", "product_version": "2.1.11-SAC"}

    login = _ingest(
        db_session,
        t0 + timedelta(seconds=1),
        host=host,
        source=source,
        type="rdp.login.success",
        category="auth",
        severity="info",
        title="RDP login",
        summary="4624",
        details={"user": user, "ip_address": "192.168.160.3", "logon_type": 10},
    )
    logoff = _ingest(
        db_session,
        t0 + timedelta(hours=1),
        host=host,
        source=source,
        type="rdp.session.logoff",
        category="auth",
        severity="info",
        title="RDP session logoff",
        summary="4634",
        details={
            "user": user,
            "ip_address": "192.168.160.3",
            "logon_type": 10,
            "event_id_windows": 4634,
        },
    )

    assert login.details[SESSION_CLOSED_BY_LOGOFF_AT_KEY] == logoff.occurred_at.isoformat()
    assert login.details[SESSION_CLOSED_BY_LOGOFF_EVENT_ID_KEY] == logoff.id
    assert event_session_terminated(login, db=db_session) is True
    assert event_to_summary(login, db_session).session_terminated is True


def test_user_mismatch_does_not_close_login(db_session):
    t0 = datetime.now(timezone.utc)
    host = {"hostname": "BIV-PC", "os_family": "windows", "ipv4": "192.168.165.39"}
    source = {"product": "rdp-login-monitor", "product_version": "2.1.11-SAC"}

    login = _ingest(
        db_session,
        t0,
        host=host,
        source=source,
        type="rdp.login.success",
        category="auth",
        severity="info",
        title="RDP login",
        summary="4624",
        details={"user": r"B26\Alice"},
    )
    logoff = _ingest(
        db_session,
        t0 + timedelta(hours=1),
        host=host,
        source=source,
        type="rdp.session.logoff",
        category="auth",
        severity="info",
        title="RDP session logoff",
        summary="4634",
        details={"user": r"B26\Bob", "logon_type": 10, "event_id_windows": 4634},
    )
    db_session.refresh(login)

    assert close_workstation_session_for_rdp_logoff(db_session, logoff) is None
    assert event_session_terminated(login, db=db_session) is False


def test_ip_mismatch_does_not_close_login_when_both_ips_present(db_session):
    t0 = datetime.now(timezone.utc)
    user = r"B26\papatramp"
    host = {"hostname": "BIV-PC", "os_family": "windows", "ipv4": "192.168.165.39"}
    source = {"product": "rdp-login-monitor", "product_version": "2.1.11-SAC"}

    login = _ingest(
        db_session,
        t0,
        host=host,
        source=source,
        type="rdp.login.success",
        category="auth",
        severity="info",
        title="RDP login",
        summary="4624",
        details={"user": user, "ip_address": "192.168.160.3", "logon_type": 10},
    )
    logoff = _ingest(
        db_session,
        t0 + timedelta(hours=1),
        host=host,
        source=source,
        type="rdp.session.logoff",
        category="auth",
        severity="info",
        title="RDP session logoff",
        summary="4634",
        details={"user": user, "ip_address": "192.168.160.99", "logon_type": 10, "event_id_windows": 4634},
    )
    db_session.refresh(login)

    assert close_workstation_session_for_rdp_logoff(db_session, logoff) is None
    assert event_session_terminated(login, db=db_session) is False


def test_runtime_resolve_for_historical_login_without_flag(db_session):
    from app.models import Event, Host

    t0 = datetime.now(timezone.utc)
    user = r"B26\papatramp"
    host = Host(
        hostname="BIV-PC",
        os_family="windows",
        product="rdp-login-monitor",
        ipv4="192.168.165.39",
    )
    db_session.add(host)
    db_session.commit()

    login = Event(
        event_id=str(uuid.uuid4()),
        host_id=host.id,
        occurred_at=t0 + timedelta(seconds=1),
        received_at=t0,
        category="auth",
        type="rdp.login.success",
        severity="info",
        title="RDP login",
        summary="4624",
        payload={},
        details={"user": "papatramp", "ip_address": "192.168.160.3"},
    )
    logoff = Event(
        event_id=str(uuid.uuid4()),
        host_id=host.id,
        occurred_at=t0 + timedelta(hours=2),
        received_at=t0,
        category="auth",
        type="rdp.session.logoff",
        severity="info",
        title="4634",
        summary="4634",
        payload={},
        details={"user": user, "ip_address": "192.168.160.3", "event_id_windows": 4634},
    )
    db_session.add_all([login, logoff])
    db_session.commit()

    assert resolve_workstation_login_closed_by_logoff(db_session, login) is True
    assert find_logoff_after_workstation_login(db_session, login) is not None
    assert event_to_summary(login, db_session).session_terminated is True


def test_sam_domain_user_match_on_logoff(db_session):
    t0 = datetime.now(timezone.utc)
    host = {"hostname": "BIV-PC", "os_family": "windows", "ipv4": "192.168.165.39"}
    source = {"product": "rdp-login-monitor", "product_version": "2.1.11-SAC"}

    login = _ingest(
        db_session,
        t0,
        host=host,
        source=source,
        type="rdp.login.success",
        category="auth",
        severity="info",
        title="RDP login",
        summary="4624",
        details={"user": r"B26\papatramp", "logon_type": 10},
    )
    _ingest(
        db_session,
        t0 + timedelta(minutes=30),
        host=host,
        source=source,
        type="rdp.session.logoff",
        category="auth",
        severity="info",
        title="RDP session logoff",
        summary="4647",
        details={"user": "papatramp", "logon_type": 10, "event_id_windows": 4647},
    )
    db_session.refresh(login)

    assert event_session_terminated(login, db=db_session) is True