server/gns_transport.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import ctypes
+import ipaddress
 import os
 import sys
 from dataclasses import dataclass
@@ -57,6 +58,12 @@ class GnsEvent:
     debug: str = ""
 
 
+@dataclass(frozen=True)
+class RemoteEndpoint:
+    host: str
+    port: int
+
+
 def _default_library_names() -> tuple[str, ...]:
     if sys.platform == "win32":
         return ("commonwealth_online_gns_bridge.dll",)
@@ -103,6 +110,7 @@ class _NativeApi:
         self.poll = library.co_gns_server_poll
         self.send = library.co_gns_server_send
         self.disconnect = library.co_gns_server_disconnect
+        self.remote_ipv4 = library.co_gns_server_remote_ipv4
 
         self.create.argtypes = [
             ctypes.c_char_p,
@@ -140,6 +148,13 @@ class _NativeApi:
             ctypes.c_char_p,
         ]
         self.disconnect.restype = ctypes.c_int
+        self.remote_ipv4.argtypes = [
+            ctypes.c_void_p,
+            ctypes.c_uint32,
+            ctypes.POINTER(ctypes.c_uint32),
+            ctypes.POINTER(ctypes.c_uint16),
+        ]
+        self.remote_ipv4.restype = ctypes.c_int
 
 
 class GnsServerTransport:
@@ -197,6 +212,24 @@ class GnsServerTransport:
         if self._closed or not self._handle.value:
             raise GnsTransportError("GNS transport is closed")
 
+    def remote_endpoint(self, connection_id: int) -> RemoteEndpoint | None:
+        self._require_open()
+        if not isinstance(connection_id, int) or isinstance(connection_id, bool) or connection_id <= 0:
+            raise ValueError("connection_id must be a positive integer")
+        ipv4 = ctypes.c_uint32()
+        port = ctypes.c_uint16()
+        found = int(
+            self._api.remote_ipv4(
+                self._handle,
+                connection_id,
+                ctypes.byref(ipv4),
+                ctypes.byref(port),
+            )
+        )
+        if found != 1:
+            return None
+        return RemoteEndpoint(str(ipaddress.IPv4Address(ipv4.value)), int(port.value))
+
     def poll(self) -> GnsEvent | None:
         self._require_open()
         event = _CEvent()