PapaTramp
1b96a3b2c7
Use event 302/303 instead of Win302 in RDG titles. Match event types by prefix in events list (e.g. rdg.connection). Co-authored-by: Cursor <cursoragent@cursor.com>
133 lines
3.7 KiB
Python
133 lines
3.7 KiB
Python
"""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"event {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))
|