"""RD Gateway event labels (access path, UI title/summary).""" from __future__ import annotations from dataclasses import dataclass from sqlalchemy.orm import Session from app.config import get_settings from app.models import Event from app.services.rdg_client_host import find_windows_host_by_ipv4 from app.services.rdg_session_flap import event_internal_ip RDG_TYPES = frozenset( { "rdg.connection.success", "rdg.connection.disconnected", "rdg.connection.failed", } ) ACCESS_PATH_HAPROXY = "Haproxy-RDG-Comp" ACCESS_PATH_DIRECT = "RDG-Comp" @dataclass(frozen=True) class RdgDisplayInfo: title: str summary: str access_path: str | None internal_ip: str | None qwinsta_enabled: bool def _details_dict(event: Event) -> dict: raw = event.details return raw if isinstance(raw, dict) else {} def _event_external_ip(event: Event) -> str: details = _details_dict(event) for key in ("external_ip", "source_ip", "ip_address"): val = details.get(key) if val is not None and str(val).strip(): return str(val).strip() return "" def _parse_haproxy_ips() -> frozenset[str]: raw = (get_settings().sac_rdg_haproxy_external_ips or "").strip() if not raw: return frozenset() parts = [p.strip() for p in raw.replace(";", ",").split(",")] return frozenset(p for p in parts if p) def classify_rdg_access_path(external_ip: str) -> str | None: ip = (external_ip or "").strip() if not ip or ip in ("-", "N/A"): return None if ip in _parse_haproxy_ips(): return ACCESS_PATH_HAPROXY return ACCESS_PATH_DIRECT def _windows_event_label(event: Event) -> str: details = _details_dict(event) win_id = details.get("event_id_windows") if win_id is None: return "" text = str(win_id).strip() return f"Win{text}" if text else "" def _action_label(event: Event) -> str: if event.type == "rdg.connection.success": return "подключение" if event.type == "rdg.connection.disconnected": return "отключение" return "ошибка" def build_rdg_display(event: Event, db: Session | None = None) -> RdgDisplayInfo | None: if event.type not in RDG_TYPES: return None details = _details_dict(event) internal_ip = event_internal_ip(event) external_ip = _event_external_ip(event) access_path = classify_rdg_access_path(external_ip) user = str(details.get("user") or "").strip() gateway = event.host.hostname if event.host else "—" client_label = internal_ip or "—" if db and internal_ip: client_host = find_windows_host_by_ipv4(db, internal_ip) if client_host is not None: client_label = f"{client_host.hostname} ({internal_ip})" path_note = f" ({access_path})" if access_path else "" win_note = _windows_event_label(event) action = _action_label(event) title = f"RDS {action} → {client_label}{path_note}" if win_note: title = f"{title} · {win_note}" summary_parts: list[str] = [] if user: summary_parts.append(user) if external_ip: summary_parts.append(f"внешний {external_ip}") summary_parts.append(f"шлюз {gateway}") if win_note: summary_parts.append(win_note) summary = " · ".join(summary_parts) qwinsta_enabled = bool(internal_ip) return RdgDisplayInfo( title=title, summary=summary, access_path=access_path, internal_ip=internal_ip or None, qwinsta_enabled=qwinsta_enabled, ) def event_supports_rdg_client_qwinsta(event: Event) -> bool: if event.type not in RDG_TYPES: return False return bool(event_internal_ip(event))