feat: close rdp.login.success on rdp.session.logoff (v0.5.12)

Ingest correlates direct RDP logoff with open login events on the same host and user. Terminate-session treats missing qwinsta as already logged off.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/app/services/event_actor_user.py

@@ -15,6 +15,7 @@ ACTOR_USER_EVENT_TYPES: frozenset[str] = frozenset(
         "session.logind.failed",
         "rdp.login.success",
         "rdp.login.failed",
+        "rdp.session.logoff",
         "rdp.shadow.control.started",
         "rdp.shadow.control.stopped",
         "rdp.shadow.control.permission",

backend/app/services/host_sessions.py

@@ -84,6 +84,10 @@ def _event_login_user(event: Event) -> str:
 
 
 def event_session_terminated(event: Event, db: Session | None = None) -> bool:
+    from app.services.rdp_session_logoff import (
+        event_closed_by_logoff,
+        resolve_workstation_login_closed_by_logoff,
+    )
     from app.services.rdg_workstation_session import (
         event_closed_by_rdg,
         resolve_workstation_login_closed,
@@ -97,7 +101,11 @@ def event_session_terminated(event: Event, db: Session | None = None) -> bool:
         return True
     if event_closed_by_rdg(event):
         return True
+    if event_closed_by_logoff(event):
+        return True
     if db is not None and event.type == "rdp.login.success":
+        if resolve_workstation_login_closed_by_logoff(db, event):
+            return True
         return resolve_workstation_login_closed(db, event)
     return False
 
@@ -376,6 +384,31 @@ def terminate_linux_session(
     return last
 
 
+def _normalize_sam_account(user: str) -> str:
+    text = (user or "").strip()
+    if "\\" in text:
+        return text.split("\\")[-1].strip().casefold()
+    if "@" in text:
+        return text.split("@")[0].strip().casefold()
+    return text.casefold()
+
+
+def windows_user_matches_session(login_user: str, session_user: str) -> bool:
+    login = _normalize_sam_account(login_user)
+    session = _normalize_sam_account(session_user)
+    return bool(login and session and login == session)
+
+
+def filter_windows_sessions_for_user(
+    sessions: list[HostSessionRow],
+    login_user: str,
+) -> list[HostSessionRow]:
+    user = (login_user or "").strip()
+    if not user:
+        return sessions
+    return [s for s in sessions if windows_user_matches_session(user, s.user)]
+
+
 def parse_qwinsta_sessions(stdout: str, *, filter_user: str | None = None) -> list[HostSessionRow]:
     rows: list[HostSessionRow] = []
     norm_filter = (filter_user or "").strip().lower()
@@ -475,11 +508,17 @@ def terminate_session_for_event(
             sessions, qwinsta = list_windows_sessions(host, win_cfg)
             if not qwinsta or not qwinsta.ok:
                 raise ValueError(qwinsta.message if qwinsta else "qwinsta failed")
-            matched = [s for s in sessions if user.lower() in s.user.lower()] if user else sessions
+            matched = filter_windows_sessions_for_user(sessions, user) if user else sessions
             if len(matched) == 1:
                 sid = matched[0].session_id
             elif not matched:
-                raise ValueError("No matching Windows session for user")
+                # Пользователь уже вышел из RDP — qwinsta пуст, событие входа в SAC ещё «открыто».
+                return WinRmCmdResult(
+                    ok=True,
+                    message="На хосте нет активной сессии пользователя (уже вышел из RDP)",
+                    target=qwinsta.target,
+                    stdout=qwinsta.stdout,
+                )
             else:
                 raise ValueError("Multiple sessions; specify session_id")
         result = terminate_windows_session(host, win_cfg, sid)

backend/app/services/ingest.py

@@ -9,6 +9,7 @@ from app.services.agent_update import process_agent_update_ingest
 from app.services.daily_report_format import normalize_daily_report_details
 from app.services.event_severity_overrides import apply_severity_override
 from app.services.host_inventory import INVENTORY_EVENT_TYPE, process_inventory_ingest
+from app.services.rdp_session_logoff import close_workstation_session_for_rdp_logoff
 from app.services.rdg_workstation_session import close_workstation_session_for_rdg_end
 
 DAILY_REPORT_TYPES = frozenset({"report.daily.ssh", "report.daily.rdp"})
@@ -128,4 +129,5 @@ def ingest_event(db: Session, payload: dict) -> tuple[Event, bool]:
 
     process_agent_update_ingest(db, host, payload.get("type", ""), details)
     close_workstation_session_for_rdg_end(db, event)
+    close_workstation_session_for_rdp_logoff(db, event)
     return event, True

backend/app/services/rdg_workstation_session.py

@@ -56,13 +56,17 @@ def mark_login_closed_by_rdg(login_event: Event, *, rdg_end_event: Event) -> Non
 
 
 def _login_already_closed(login_event: Event) -> bool:
+    from app.services.rdp_session_logoff import event_closed_by_logoff
+
     details = _details_dict(login_event)
     if details.get("session_terminated") is True:
         return True
     at = details.get(SESSION_TERMINATED_AT_KEY)
     if at is not None and str(at).strip() != "":
         return True
-    return event_closed_by_rdg(login_event)
+    if event_closed_by_rdg(login_event):
+        return True
+    return event_closed_by_logoff(login_event)
 
 
 def find_workstation_login_for_rdg_end(db: Session, rdg_end_event: Event) -> Event | None:

backend/app/services/rdp_session_logoff.py

@@ -0,0 +1,148 @@
+"""Correlate direct RDP logoff (Security 4634/4647) with workstation rdp.login.success."""
+
+from __future__ import annotations
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+from sqlalchemy.orm.attributes import flag_modified
+
+from app.models import Event
+from app.services.host_sessions import (
+    SESSION_TERMINATED_AT_KEY,
+    _details_dict,
+    _event_login_user,
+)
+from app.services.rdg_workstation_session import (
+    WORKSTATION_LOGIN_TYPE,
+    event_closed_by_rdg,
+    users_match_rdg,
+)
+
+SESSION_CLOSED_BY_LOGOFF_AT_KEY = "session_closed_by_logoff_at"
+SESSION_CLOSED_BY_LOGOFF_EVENT_ID_KEY = "session_closed_by_logoff_event_id"
+
+LOGOFF_EVENT_TYPE = "rdp.session.logoff"
+LOGOFF_EVENT_TYPES = frozenset({LOGOFF_EVENT_TYPE})
+
+
+def event_closed_by_logoff(event: Event) -> bool:
+    details = _details_dict(event)
+    at = details.get(SESSION_CLOSED_BY_LOGOFF_AT_KEY)
+    return at is not None and str(at).strip() != ""
+
+
+def mark_login_closed_by_logoff(login_event: Event, *, logoff_event: Event) -> None:
+    details = dict(_details_dict(login_event))
+    details[SESSION_CLOSED_BY_LOGOFF_AT_KEY] = logoff_event.occurred_at.isoformat()
+    details[SESSION_CLOSED_BY_LOGOFF_EVENT_ID_KEY] = logoff_event.id
+    login_event.details = details
+    flag_modified(login_event, "details")
+
+
+def _login_already_closed(login_event: Event) -> bool:
+    details = _details_dict(login_event)
+    if details.get("session_terminated") is True:
+        return True
+    at = details.get(SESSION_TERMINATED_AT_KEY)
+    if at is not None and str(at).strip() != "":
+        return True
+    if event_closed_by_rdg(login_event):
+        return True
+    return event_closed_by_logoff(login_event)
+
+
+def find_workstation_login_for_logoff(db: Session, logoff_event: Event) -> Event | None:
+    if logoff_event.type not in LOGOFF_EVENT_TYPES:
+        return None
+
+    logoff_user = _event_login_user(logoff_event)
+    if not logoff_user:
+        return None
+    logoff_at = logoff_event.occurred_at
+    host_id = logoff_event.host_id
+    if host_id is None:
+        return None
+
+    candidates = db.scalars(
+        select(Event)
+        .where(
+            Event.host_id == host_id,
+            Event.type == WORKSTATION_LOGIN_TYPE,
+            Event.occurred_at <= logoff_at,
+            Event.id != logoff_event.id,
+        )
+        .order_by(Event.occurred_at.desc())
+    ).all()
+
+    logoff_details = _details_dict(logoff_event)
+    logoff_ip = str(logoff_details.get("ip_address") or "").strip()
+
+    for login in candidates:
+        if _login_already_closed(login):
+            continue
+        login_user = _event_login_user(login)
+        if not users_match_rdg(login_user, logoff_user):
+            continue
+        if logoff_ip and logoff_ip not in ("", "-"):
+            login_ip = str(_details_dict(login).get("ip_address") or "").strip()
+            if login_ip and login_ip not in ("", "-") and login_ip != logoff_ip:
+                continue
+        return login
+    return None
+
+
+def find_logoff_after_workstation_login(db: Session, login_event: Event) -> Event | None:
+    """Runtime lookup for historical logins without persisted close flag."""
+    if login_event.type != WORKSTATION_LOGIN_TYPE:
+        return None
+    if _login_already_closed(login_event):
+        return None
+
+    host_id = login_event.host_id
+    if host_id is None:
+        return None
+
+    login_user = _event_login_user(login_event)
+    if not login_user:
+        return None
+    login_at = login_event.occurred_at
+    login_ip = str(_details_dict(login_event).get("ip_address") or "").strip()
+
+    candidates = db.scalars(
+        select(Event)
+        .where(
+            Event.host_id == host_id,
+            Event.type == LOGOFF_EVENT_TYPE,
+            Event.occurred_at >= login_at,
+            Event.id != login_event.id,
+        )
+        .order_by(Event.occurred_at.asc())
+    ).all()
+
+    for logoff in candidates:
+        if not users_match_rdg(_event_login_user(logoff), login_user):
+            continue
+        logoff_ip = str(_details_dict(logoff).get("ip_address") or "").strip()
+        if login_ip and login_ip not in ("", "-") and logoff_ip and logoff_ip not in ("", "-"):
+            if login_ip != logoff_ip:
+                continue
+        return logoff
+    return None
+
+
+def resolve_workstation_login_closed_by_logoff(db: Session, login_event: Event) -> bool:
+    if event_closed_by_logoff(login_event):
+        return True
+    return find_logoff_after_workstation_login(db, login_event) is not None
+
+
+def close_workstation_session_for_rdp_logoff(db: Session, logoff_event: Event) -> Event | None:
+    """On direct RDP logoff, mark matching open workstation login as session-closed."""
+    if logoff_event.type not in LOGOFF_EVENT_TYPES:
+        return None
+
+    login = find_workstation_login_for_logoff(db, logoff_event)
+    if login is None:
+        return None
+    mark_login_closed_by_logoff(login, logoff_event=logoff_event)
+    return login