51 lines
1.5 KiB
Python
51 lines
1.5 KiB
Python
"""Resolve RDG client workstation (session host) from event internal_ip."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from sqlalchemy import or_, select
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models import Event, Host
|
|
from app.services.rdg_session_flap import event_internal_ip
|
|
from app.services.winrm_connect import is_windows_host
|
|
|
|
|
|
class ClientWorkstationNotFoundError(Exception):
|
|
def __init__(self, internal_ip: str) -> None:
|
|
self.internal_ip = internal_ip
|
|
super().__init__(
|
|
f"Client workstation not found in SAC hosts for internal_ip={internal_ip}. "
|
|
"Ensure the PC is registered with matching ipv4."
|
|
)
|
|
|
|
|
|
def find_windows_host_by_ipv4(db: Session, ipv4: str) -> Host | None:
|
|
ip = (ipv4 or "").strip()
|
|
if not ip:
|
|
return None
|
|
rows = db.scalars(
|
|
select(Host)
|
|
.where(
|
|
Host.ipv4 == ip,
|
|
or_(
|
|
Host.os_family.ilike("windows"),
|
|
Host.product == "rdp-login-monitor",
|
|
),
|
|
)
|
|
.order_by(Host.last_seen_at.desc())
|
|
).all()
|
|
for host in rows:
|
|
if is_windows_host(host):
|
|
return host
|
|
return None
|
|
|
|
|
|
def resolve_client_workstation(db: Session, event: Event) -> Host:
|
|
internal_ip = event_internal_ip(event)
|
|
if not internal_ip:
|
|
raise ClientWorkstationNotFoundError("")
|
|
host = find_windows_host_by_ipv4(db, internal_ip)
|
|
if host is None:
|
|
raise ClientWorkstationNotFoundError(internal_ip)
|
|
return host
|