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

"""Tests for host session list/parse helpers."""

from types import SimpleNamespace

from app.services.host_sessions import (
    HostSessionRow,
    _event_login_user,
    event_session_terminated,
    event_supports_session_terminate,
    filter_logind_session_rows,
    filter_windows_sessions_for_user,
    mark_event_session_terminated,
    parse_loginctl_sessions,
    parse_loginctl_sessions_json,
    parse_qwinsta_sessions,
    terminate_session_for_event,
)
from app.services.linux_admin_settings import LinuxAdminConfig
from app.services.win_admin_settings import WinAdminConfig
from app.services.winrm_connect import WinRmCmdResult


def test_parse_loginctl_sessions():
    stdout = "c1 1000 alice seat0 pts/0 active -\n 2 1001 bob - tty2 active -\n"
    rows = parse_loginctl_sessions(stdout)
    assert len(rows) == 2
    assert rows[0].session_id == "c1"
    assert rows[0].user == "alice"
    assert rows[0].tty == "pts/0"


def test_parse_loginctl_sessions_prefers_pts_over_ephemeral_duplicate():
    stdout = """21166 1000 papatramp - pts/0 active no -
21169 1000 papatramp - -     active no -
"""
    rows = parse_loginctl_sessions(stdout)
    assert len(rows) == 1
    assert rows[0].session_id == "21166"
    assert rows[0].tty == "pts/0"


def test_parse_loginctl_sessions_json():
    stdout = """[
  {"session":"21166","uid":1000,"user":"papatramp","seat":null,"tty":"pts/0","state":"active","idle":false,"since":null},
  {"session":"21169","uid":1000,"user":"papatramp","seat":null,"tty":null,"state":"active","idle":false,"since":null}
]"""
    rows = parse_loginctl_sessions_json(stdout)
    assert len(rows) == 1
    assert rows[0].session_id == "21166"
    assert rows[0].tty == "pts/0"


def test_filter_logind_session_rows_keeps_distinct_users():
    from app.services.host_sessions import HostSessionRow

    rows = filter_logind_session_rows(
        [
            HostSessionRow(session_id="1", user="alice", tty="pts/0", state="active"),
            HostSessionRow(session_id="2", user="bob", tty=None, state="active"),
        ]
    )
    assert len(rows) == 2


def test_parse_qwinsta_sessions_filters_user():
    stdout = """SESSIONNAME       USERNAME                 ID  STATE
 console           Administrator             1  Active
 rdp-tcp#0         B26\\alice                 2  Active
"""
    all_rows = parse_qwinsta_sessions(stdout)
    assert len(all_rows) == 2
    filtered = parse_qwinsta_sessions(stdout, filter_user="alice")
    assert len(filtered) == 1
    assert filtered[0].session_id == "2"


def test_event_supports_session_terminate_types():
    class HostStub:
        os_family = "linux"
        product = "ssh-monitor"

    class EventStub:
        def __init__(self, event_type: str):
            self.type = event_type
            self.host = HostStub()

    assert event_supports_session_terminate(EventStub("ssh.login.success")) is True


def test_event_login_user_without_orm_actor_user_attr():
    event = SimpleNamespace(
        type="rdp.login.success",
        details={"user": "papatramp"},
    )
    assert _event_login_user(event) == "papatramp"


def test_filter_windows_sessions_for_user_matches_domain():
    rows = [
        HostSessionRow(session_id="2", user="B26\\papatramp", state="Active"),
        HostSessionRow(session_id="3", user="B26\\alice", state="Active"),
    ]
    matched = filter_windows_sessions_for_user(rows, "papatramp")
    assert len(matched) == 1
    assert matched[0].session_id == "2"


def test_terminate_session_for_event_windows_already_logged_off(monkeypatch):
    host = SimpleNamespace(
        os_family="windows",
        product="rdp-login-monitor",
        hostname="BIV-PC",
        ipv4="192.168.165.39",
        display_name=None,
        inventory={},
    )
    event = SimpleNamespace(
        type="rdp.login.success",
        details={"user": "papatramp"},
        host=host,
    )

    def fake_list(_host, _cfg):
        return [], WinRmCmdResult(ok=True, message="ok", target="BIV-PC", stdout="SESSIONNAME USERNAME ID STATE")

    monkeypatch.setattr(
        "app.services.host_sessions.list_windows_sessions",
        fake_list,
    )

    result = terminate_session_for_event(
        event,
        linux_cfg=LinuxAdminConfig(user="", password="", source="test"),
        win_cfg=WinAdminConfig(user="B26\\admin", password="x", source="test"),
    )
    assert result.ok is True
    assert "уже вышел" in result.message


def test_terminate_session_for_event_windows_no_actor_user_attr(monkeypatch):
    host = SimpleNamespace(
        os_family="windows",
        product="rdp-login-monitor",
        hostname="srv01",
        ipv4="10.0.0.1",
        display_name=None,
        inventory={},
    )
    event = SimpleNamespace(
        type="rdp.login.success",
        details={"user": "papatramp"},
        host=host,
    )
    rows = [HostSessionRow(session_id="2", user="B26\\papatramp", state="Active")]

    def fake_list(_host, _cfg):
        return rows, WinRmCmdResult(ok=True, message="ok", target="srv01")

    def fake_term(_host, _cfg, sid):
        assert sid == "2"
        return WinRmCmdResult(ok=True, message="logged off", target="srv01")

    monkeypatch.setattr(
        "app.services.host_sessions.list_windows_sessions",
        fake_list,
    )
    monkeypatch.setattr(
        "app.services.host_sessions.terminate_windows_session",
        fake_term,
    )

    result = terminate_session_for_event(
        event,
        linux_cfg=LinuxAdminConfig(user="", password="", source="test"),
        win_cfg=WinAdminConfig(user="B26\\admin", password="x", source="test"),
    )
    assert result.ok is True


def test_event_session_terminated_flag():
    event = SimpleNamespace(
        type="rdp.login.success",
        details={"session_terminated_at": "2026-06-24T10:00:00+00:00"},
    )
    assert event_session_terminated(event) is True

    fresh = SimpleNamespace(type="rdp.login.success", details={"user": "alice"})
    assert event_session_terminated(fresh) is False

    mark_event_session_terminated(fresh, by_username="admin")
    assert event_session_terminated(fresh) is True
    assert fresh.details["session_terminated_by"] == "admin"