from __future__ import annotations import json import math import socket import sys import threading import time from collections.abc import Callable from typing import Any from ban_store import BanStore from client_session import ClientSession from lan_discovery import DISCOVERY_PORT, LanDiscoveryResponder from world_state_presets import ( normalize_fw_console_arg, relay_weather_form_id, ) HOST = "0.0.0.0" PORT = 7777 ACCEPT_TIMEOUT_SECONDS = 0.5 DEFAULT_SERVER_NAME = "Commonwealth Online Server" DEFAULT_MAX_PLAYERS = 16 DEFAULT_LOG_VERBOSITY = "info" SESSION_ENDED_BANNED = "banned" SESSION_ENDED_KICKED = "kicked" _LOG_LEVELS = { "debug": 10, "info": 20, "warning": 30, "error": 40, } def get_lan_addresses() -> list[str]: """Return likely non-loopback IPv4 addresses for this machine.""" addresses: list[str] = [] try: with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe: probe.connect(("8.8.8.8", 80)) primary = probe.getsockname()[0] if primary and not primary.startswith("127."): addresses.append(primary) except OSError: pass try: hostname = socket.gethostname() for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET): candidate = info[4][0] if candidate and not candidate.startswith("127.") and candidate not in addresses: addresses.append(candidate) except OSError: pass return addresses class FalloutTogetherServer: def __init__( self, host: str = HOST, port: int = PORT, server_name: str = DEFAULT_SERVER_NAME, server_description: str = "", max_players: int = DEFAULT_MAX_PLAYERS, bans_path: str | None = None, log_verbosity: str = DEFAULT_LOG_VERBOSITY, ) -> None: self.host = host self.port = port self.server_name = server_name or DEFAULT_SERVER_NAME self.server_description = server_description or "" self.max_players = max_players if max_players >= 1 else DEFAULT_MAX_PLAYERS verbosity = str(log_verbosity or DEFAULT_LOG_VERBOSITY).strip().lower() self.log_verbosity = verbosity if verbosity in _LOG_LEVELS else DEFAULT_LOG_VERBOSITY self._ban_store = BanStore(bans_path) self._lock = threading.RLock() self._log_lock = threading.RLock() self._clients: dict[socket.socket, ClientSession] = {} self._client_threads: set[threading.Thread] = set() self._log_listeners: list[Callable[[str], None]] = [] self._server_socket: socket.socket | None = None self._accept_thread: threading.Thread | None = None self._discovery: LanDiscoveryResponder | None = None self._running = False self._started_at: float | None = None self._next_player_id = 1 self._world_state_host_player_id: int | None = None self._server_world_state: dict[str, str] = {} self._last_npc_state: dict[str, Any] | None = None self._stats: dict[str, int] = { "clientsConnected": 0, "clientsDisconnected": 0, "packetsReceived": 0, "packetsSent": 0, "packetsBroadcast": 0, "transformPacketsReceived": 0, "transformPacketsBroadcast": 0, "worldStatePacketsReceived": 0, "worldStatePacketsBroadcast": 0, "npcStatePacketsReceived": 0, "npcStatePacketsBroadcast": 0, "combatHitsReceived": 0, "combatHitsRouted": 0, "worldStateHostPacketsBroadcast": 0, "serverWorldStatePacketsBroadcast": 0, "sessionEndedPacketsSent": 0, "bannedConnectionsRejected": 0, "disconnectPacketsBroadcast": 0, } def start(self) -> None: with self._lock: if self._running: return self._prepare_server_socket() self._start_discovery() self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True) self._accept_thread.start() def serve_forever(self) -> None: with self._lock: if self._running: raise RuntimeError("Server is already running.") self._prepare_server_socket() self._start_discovery() self._accept_loop() def stop(self) -> None: clients: list[ClientSession] server_socket: socket.socket | None client_threads: list[threading.Thread] with self._lock: if not self._running and self._server_socket is None and self._discovery is None: return self._running = False server_socket = self._server_socket self._server_socket = None clients = list(self._clients.values()) client_threads = [ thread for thread in self._client_threads if thread is not threading.current_thread() ] discovery = self._discovery self._discovery = None if discovery is not None: discovery.stop() if server_socket is not None: try: server_socket.close() except OSError: pass for client in clients: self._close_client_socket(client) accept_thread = self._accept_thread if accept_thread is not None and accept_thread is not threading.current_thread(): accept_thread.join(timeout=1.0) self._accept_thread = None for thread in client_threads: thread.join(timeout=1.0) with self._lock: self._client_threads.clear() def is_running(self) -> bool: with self._lock: return self._running def get_clients(self) -> list[dict[str, Any]]: with self._lock: clients = list(self._clients.values()) return [client.to_snapshot() for client in clients] def get_stats(self) -> dict[str, Any]: with self._lock: connected_clients = len(self._clients) started_at = self._started_at stats = dict(self._stats) stats.update( { "host": self.host, "port": self.port, "serverName": self.server_name, "serverDescription": self.server_description, "maxPlayers": self.max_players, "lanAddresses": get_lan_addresses() if self._running and self.host == "0.0.0.0" else [], "isRunning": self._running, "startedAt": started_at, "uptimeSeconds": time.time() - started_at if started_at is not None else 0.0, "connectedClients": connected_clients, "nextPlayerId": self._next_player_id, } ) return stats def get_server_world_state(self) -> dict[str, str]: with self._lock: return dict(self._server_world_state) def set_server_time(self, hhmm: str) -> bool: text = str(hhmm).strip() if not text.isdigit() or len(text) > 4: self._log(f"Rejected invalid server time HHmm value: {hhmm!r}") return False normalized = text.zfill(4) hours = int(normalized[:2]) minutes = int(normalized[2:]) if hours > 23 or minutes > 59: self._log(f"Rejected out-of-range server time HHmm value: {hhmm!r}") return False with self._lock: self._server_world_state["timeHHmm"] = normalized self._broadcast_server_world_state() return True def set_server_weather(self, fw_console_arg: str) -> bool: try: normalized_fw_arg = normalize_fw_console_arg(fw_console_arg) except ValueError: self._log(f"Rejected invalid server weather console id: {fw_console_arg!r}") return False with self._lock: self._server_world_state["weatherConsoleArg"] = normalized_fw_arg self._server_world_state["weatherFormId"] = relay_weather_form_id(normalized_fw_arg) self._broadcast_server_world_state() return True def list_bans(self) -> list[dict[str, Any]]: return [ { "ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at, } for entry in self._ban_store.list_bans() ] def unban_ip(self, ip: str) -> bool: normalized = str(ip).strip() removed = self._ban_store.unban_ip(normalized) if removed: self._log(f"Unbanned IP {normalized}") return removed def ban_ip(self, ip: str, reason: str = "") -> dict[str, Any]: normalized = str(ip).strip() if not normalized: raise ValueError("IP address cannot be empty.") entry = self._ban_store.ban_ip(normalized, reason=reason) self._log( f"Banned IP {entry.ip}" + (f" (reason: {entry.reason})" if entry.reason else "") ) ended = self._end_sessions_for_ip(normalized, code=SESSION_ENDED_BANNED, reason=entry.reason) return { "ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at, "sessionsEnded": ended, } def ban_player(self, player_id: int, reason: str = "") -> dict[str, Any]: client = self._find_client_by_player_id(player_id) if client is None: raise KeyError(f"No connected player with id {player_id}") return self.ban_ip(client.address[0], reason=reason) def kick_player(self, player_id: int, reason: str = "") -> dict[str, Any]: client = self._find_client_by_player_id(player_id) if client is None: raise KeyError(f"No connected player with id {player_id}") ip = client.address[0] self._log( f"Kicking player {player_id} ({client.label})" + (f" (reason: {reason})" if reason else "") ) self._end_client_session(client, code=SESSION_ENDED_KICKED, reason=reason) return { "playerId": player_id, "ip": ip, "code": SESSION_ENDED_KICKED, "reason": str(reason or ""), } def add_log_listener(self, callback: Callable[[str], None]) -> None: with self._log_lock: if callback not in self._log_listeners: self._log_listeners.append(callback) def remove_log_listener(self, callback: Callable[[str], None]) -> None: with self._log_lock: if callback in self._log_listeners: self._log_listeners.remove(callback) def _prepare_server_socket(self) -> None: server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: if hasattr(socket, "SO_EXCLUSIVEADDRUSE"): server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1) else: server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_socket.bind((self.host, self.port)) server_socket.listen() server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS) except OSError as error: server_socket.close() raise OSError( f"Could not bind the server to {self.host}:{self.port}. " f"The port may already be in use or unavailable. ({error})" ) from error self._server_socket = server_socket self._running = True self._started_at = time.time() # Each server run is a fresh session for player/host assignment. # Without this reset, restarts can leave stale host IDs and player IDs # > 1, which breaks weather authority handoff to the active host. self._next_player_id = 1 self._world_state_host_player_id = None self._last_npc_state = None self._log(f"Commonwealth Online server listening on {self.host}:{self.port}") if self.host == "0.0.0.0": self._log( f"Bind address 0.0.0.0 means all interfaces; clients should not join 0.0.0.0." ) self._log(f"Local clients can connect at 127.0.0.1:{self.port}") lan_addresses = get_lan_addresses() for address in lan_addresses: self._log(f"LAN clients can connect at {address}:{self.port}") if not lan_addresses: self._log("Could not detect a LAN IPv4 address for this machine.") if sys.platform == "win32": self._log( "Forward this port on your router and allow it through Windows Firewall " "for connections from outside your network." ) else: self._log( "Allow TCP " f"{self.port} through your host firewall. Router port forwarding is only " "needed for connections from outside your LAN." ) else: self._log(f"Clients can connect at {self.host}:{self.port}") self._log("Waiting for newline-separated JSON transform packets...") def _start_discovery(self) -> None: discovery = LanDiscoveryResponder(self) try: discovery.start() except OSError as error: self._log( f"LAN discovery unavailable on UDP {DISCOVERY_PORT}: {error}. " "Direct TCP connections still work.", level="warning", ) self._discovery = None return self._discovery = discovery self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}") def _accept_loop(self) -> None: try: while self.is_running(): with self._lock: server_socket = self._server_socket if server_socket is None: break try: connection, address = server_socket.accept() except socket.timeout: # Wake periodically so Ctrl+C is handled promptly in Windows terminals. continue except OSError: if self.is_running(): self._log("Server socket closed unexpectedly.") break thread = threading.Thread( target=self._handle_client, args=(connection, address), daemon=True, ) with self._lock: self._client_threads.add(thread) thread.start() finally: with self._lock: server_socket = self._server_socket self._running = False self._server_socket = None if server_socket is not None: try: server_socket.close() except OSError: pass def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession: with self._lock: # The server owns player IDs so early clients do not need to coordinate # identity with each other or know anything about remote players yet. client = ClientSession( connection=connection, address=address, player_id=self._next_player_id, connected_at=time.time(), ) self._next_player_id += 1 self._clients[connection] = client self._stats["clientsConnected"] += 1 if self._world_state_host_player_id is None: self._world_state_host_player_id = client.player_id return client def _remove_client(self, client: ClientSession) -> bool: with self._lock: removed = self._clients.pop(client.connection, None) is not None if removed: self._stats["clientsDisconnected"] += 1 return removed def _handle_client(self, connection: socket.socket, address: tuple[str, int]) -> None: peer_ip = address[0] ban_entry = self._ban_store.get_ban(peer_ip) if ban_entry is not None: with self._lock: self._stats["bannedConnectionsRejected"] += 1 self._log(f"Rejected banned IP {peer_ip}:{address[1]}") self._send_session_ended_raw( connection, code=SESSION_ENDED_BANNED, reason=ban_entry.reason, ) try: connection.close() except OSError: pass return client = self._assign_client(connection, address) self._log( f"TCP accept: {client.label} (player {client.player_id})", level="debug", ) try: with connection: buffer = b"" try: with self._lock: world_state_host_player_id = self._world_state_host_player_id welcome_packet: dict[str, Any] = { "type": "welcome", "playerId": client.player_id, "serverTime": time.time(), "serverName": self.server_name, } if self.server_description: welcome_packet["serverDescription"] = self.server_description if world_state_host_player_id is not None: welcome_packet["worldStateHostPlayerId"] = world_state_host_player_id self._send_packet(client, welcome_packet) # Delay transform/npc snapshots until the client sends its first # packet so TCP ping probes (connect+close) do not pull world state. while self.is_running(): chunk = connection.recv(4096) if not chunk: break buffer += chunk while b"\n" in buffer: line_bytes, buffer = buffer.split(b"\n", 1) line = line_bytes.decode("utf-8", errors="replace").strip() self._handle_line(client, line) except ConnectionResetError: if client.packets_received > 0: self._log( f"Client disconnected unexpectedly: {client.label} " f"(player {client.player_id})" ) else: self._log( f"Short-lived connection closed: {client.label} " f"(player {client.player_id})", level="debug", ) except OSError as error: if self.is_running(): self._log( f"Client connection error: {client.label} " f"(player {client.player_id}): {error}" ) finally: self._disconnect_client(client) finally: with self._lock: self._client_threads.discard(threading.current_thread()) def _handle_line(self, client: ClientSession, line: str) -> None: if not line: return received_at = time.time() packets_received = client.record_received(received_at) if packets_received == 1: self._log(f"Client connected: {client.label} (player {client.player_id})") self._send_existing_transforms_to_client(client) self._send_existing_npc_state_to_client(client) with self._lock: self._stats["packetsReceived"] += 1 try: packet = json.loads(line) except json.JSONDecodeError as error: self._log(f"Invalid JSON from {client.label}: {error}: {line}") return if not isinstance(packet, dict): self._log(f"Invalid packet from {client.label}: expected JSON object: {packet}") return if packet.get("type") == "transform": packet["playerId"] = client.player_id packet["serverTime"] = time.time() client.record_transform(packet) with self._lock: self._stats["transformPacketsReceived"] += 1 if packet.get("type") == "worldState": with self._lock: world_state_host_player_id = self._world_state_host_player_id if world_state_host_player_id is None or client.player_id != world_state_host_player_id: self._log( f"Ignoring worldState from non-host {client.label} " f"(player {client.player_id}, host is player {world_state_host_player_id})." ) return packet["playerId"] = client.player_id packet["serverTime"] = time.time() with self._lock: self._stats["worldStatePacketsReceived"] += 1 if packet.get("type") == "npcState": with self._lock: world_state_host_player_id = self._world_state_host_player_id if world_state_host_player_id is None or client.player_id != world_state_host_player_id: self._log( f"Ignoring npcState from non-host {client.label} " f"(player {client.player_id}, host is player {world_state_host_player_id})." ) return npcs = packet.get("npcs") if not isinstance(npcs, list) or len(npcs) > 16 or any(not isinstance(npc, dict) for npc in npcs): self._log(f"Ignoring malformed npcState from host player {client.player_id}.") return packet["playerId"] = client.player_id packet["serverTime"] = time.time() packet["fullReplace"] = True with self._lock: self._last_npc_state = dict(packet) self._stats["npcStatePacketsReceived"] += 1 if packet.get("type") == "combatHit": packet["playerId"] = client.player_id packet["serverTime"] = time.time() with self._lock: self._stats["combatHitsReceived"] += 1 self._print_packet(client, packet) if packet.get("type") == "transform": self._broadcast_transform(client, packet) elif packet.get("type") == "worldState": self._broadcast_world_state(client, packet) elif packet.get("type") == "npcState": self._broadcast_npc_state(client, packet) elif packet.get("type") == "combatHit": self._route_combat_hit(client, packet) def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None: encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n" client.connection.sendall(encoded) client.record_sent(broadcast=broadcast) with self._lock: self._stats["packetsSent"] += 1 if broadcast: self._stats["packetsBroadcast"] += 1 def _send_existing_transforms_to_client(self, new_client: ClientSession) -> None: with self._lock: snapshots = [ dict(client.last_transform) for client in self._clients.values() if client.connection != new_client.connection and client.last_transform is not None ] if not snapshots: return successful_sends = 0 failed = False for packet in snapshots: packet["serverTime"] = time.time() try: self._send_packet(new_client, packet, broadcast=True) successful_sends += 1 except OSError: failed = True break with self._lock: self._stats["transformPacketsBroadcast"] += successful_sends if successful_sends: self._log( f"Sent {successful_sends} existing transform snapshot(s) to newly connected player {new_client.player_id}" ) if failed: self._disconnect_client(new_client) def _send_existing_npc_state_to_client(self, new_client: ClientSession) -> None: with self._lock: snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None if snapshot is None or snapshot.get("playerId") == new_client.player_id: return snapshot["serverTime"] = time.time() try: self._send_packet(new_client, snapshot, broadcast=True) with self._lock: self._stats["npcStatePacketsBroadcast"] += 1 self._log( f"Sent existing npcState snapshot with {len(snapshot.get('npcs', []))} " f"enemy/enemies to newly connected player {new_client.player_id}" ) except OSError: self._disconnect_client(new_client) def _disconnect_client(self, client: ClientSession) -> None: with self._lock: was_world_state_host = client.player_id == self._world_state_host_player_id never_sent_packet = client.packets_received == 0 if not self._remove_client(client): return self._close_client_socket(client) if never_sent_packet: # TCP connect probes (browser/pause ping) never send gameplay packets. # Avoid broadcasting a peer disconnect for those ghost sessions. self._log( f"Ignored probe/short-lived client: {client.label} (player {client.player_id})", level="debug", ) if was_world_state_host: self._reassign_world_state_host() return self._log(f"Client disconnected: {client.label} (player {client.player_id})") self._broadcast_disconnect(client) if was_world_state_host: self._reassign_world_state_host() def _find_client_by_player_id(self, player_id: int) -> ClientSession | None: with self._lock: for client in self._clients.values(): if client.player_id == player_id: return client return None def _clients_for_ip(self, ip: str) -> list[ClientSession]: normalized = str(ip).strip() with self._lock: return [client for client in self._clients.values() if client.address[0] == normalized] def _build_session_ended_packet(self, code: str, reason: str = "") -> dict[str, Any]: packet: dict[str, Any] = { "type": "sessionEnded", "code": code, "serverTime": time.time(), } reason_text = str(reason or "") if reason_text: packet["reason"] = reason_text else: packet["reason"] = "" return packet def _send_session_ended_raw( self, connection: socket.socket, *, code: str, reason: str = "", ) -> None: packet = self._build_session_ended_packet(code, reason) encoded = json.dumps(packet, separators=(",", ":")).encode("utf-8") + b"\n" try: connection.sendall(encoded) with self._lock: self._stats["sessionEndedPacketsSent"] += 1 self._stats["packetsSent"] += 1 try: connection.shutdown(socket.SHUT_WR) except OSError: pass except OSError: pass def _end_client_session( self, client: ClientSession, *, code: str, reason: str = "", ) -> None: packet = self._build_session_ended_packet(code, reason) try: self._send_packet(client, packet) with self._lock: self._stats["sessionEndedPacketsSent"] += 1 try: client.connection.shutdown(socket.SHUT_WR) except OSError: pass except OSError: pass self._disconnect_client(client) def _end_sessions_for_ip(self, ip: str, *, code: str, reason: str = "") -> int: clients = self._clients_for_ip(ip) for client in clients: self._end_client_session(client, code=code, reason=reason) return len(clients) def _close_client_socket(self, client: ClientSession) -> None: try: client.connection.close() except OSError: pass def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None: with self._lock: recipients = [client for client in self._clients.values() if client.connection != sender.connection] failed_recipients: list[ClientSession] = [] successful_sends = 0 for recipient in recipients: try: # Transforms go to every other client only. The sender already knows # its own movement; echoing it back would create duplicate local state. self._send_packet(recipient, packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) with self._lock: self._stats["transformPacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) self._log( f"Broadcast transform from player {sender.player_id} to {successful_sends} other client(s)", level="debug", ) def _broadcast_world_state(self, sender: ClientSession, packet: dict[str, Any]) -> None: with self._lock: recipients = [client for client in self._clients.values() if client.connection != sender.connection] failed_recipients: list[ClientSession] = [] successful_sends = 0 for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) with self._lock: self._stats["worldStatePacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) game_hour = packet.get("gameHour", "?") game_days_passed = packet.get("gameDaysPassed", "?") weather_form_id = packet.get("weatherFormId", "") weather_detail = f", weatherFormId={weather_form_id}" if weather_form_id else "" self._log( f"Broadcast worldState from player {sender.player_id} to {successful_sends} other client(s): " f"gameHour={game_hour}, gameDaysPassed={game_days_passed}{weather_detail}", level="debug", ) def _broadcast_npc_state(self, sender: ClientSession, packet: dict[str, Any]) -> None: with self._lock: recipients = [client for client in self._clients.values() if client.connection != sender.connection] failed_recipients: list[ClientSession] = [] successful_sends = 0 for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) with self._lock: self._stats["npcStatePacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) self._log( f"Broadcast npcState from player {sender.player_id} to {successful_sends} " f"other client(s): enemies={len(packet.get('npcs', []))}", level="debug", ) def _broadcast_server_world_state(self) -> None: with self._lock: if not self._running: self._log("Cannot broadcast server world state while the server is stopped.") return recipients = list(self._clients.values()) snapshot = dict(self._server_world_state) time_hhmm = snapshot.get("timeHHmm") weather_console_arg = snapshot.get("weatherConsoleArg") weather_form_id = snapshot.get("weatherFormId") if not time_hhmm and not weather_console_arg: self._log("Server world state broadcast skipped: no time or weather set.") return server_time = time.time() failed_recipients: list[ClientSession] = [] successful_sends = 0 if time_hhmm: time_packet: dict[str, Any] = { "type": "serverWorldState", "timeHHmm": time_hhmm, "serverTime": server_time, } for recipient in recipients: try: self._send_packet(recipient, time_packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) self._log( f"Broadcast serverWorldState time to {len(recipients)} client(s): timeHHmm={time_hhmm}" ) if weather_console_arg: weather_packet: dict[str, Any] = { "type": "serverWorldState", "weatherConsoleArg": weather_console_arg, "serverTime": server_time, } if weather_form_id: weather_packet["weatherFormId"] = weather_form_id for recipient in recipients: try: self._send_packet(recipient, weather_packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) self._log( "Broadcast serverWorldState weather command to " f"{len(recipients)} client(s): fw {weather_console_arg}" ) with self._lock: self._stats["serverWorldStatePacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) def _reassign_world_state_host(self) -> None: with self._lock: if not self._clients: self._world_state_host_player_id = None self._last_npc_state = None self._log("World-state host cleared; no clients remain.") return new_host_player_id = min(client.player_id for client in self._clients.values()) self._world_state_host_player_id = new_host_player_id self._last_npc_state = None recipients = list(self._clients.values()) packet = { "type": "worldStateHost", "worldStateHostPlayerId": new_host_player_id, "serverTime": time.time(), } failed_recipients: list[ClientSession] = [] successful_sends = 0 for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) with self._lock: self._stats["worldStateHostPacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) self._log( f"Reassigned world-state host to player {new_host_player_id} " f"and notified {successful_sends} client(s)." ) def _broadcast_disconnect(self, disconnected_client: ClientSession) -> None: packet = { "type": "disconnect", "playerId": disconnected_client.player_id, "serverTime": time.time(), } with self._lock: recipients = list(self._clients.values()) failed_recipients: list[ClientSession] = [] successful_sends = 0 for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful_sends += 1 except OSError: failed_recipients.append(recipient) with self._lock: self._stats["disconnectPacketsBroadcast"] += successful_sends for recipient in failed_recipients: self._disconnect_client(recipient) self._log( f"Broadcast disconnect for player {disconnected_client.player_id} " f"to {successful_sends} other client(s)" ) def _route_combat_hit(self, sender: ClientSession, packet: dict[str, Any]) -> None: """Route a targeted combatHit packet to the victim player only.""" target_player_id = packet.get("targetPlayerId") sequence = packet.get("sequence") damage_value = packet.get("damage") # Coerce numeric JSON values: some clients/serializers emit floats for integers. if isinstance(target_player_id, float) and target_player_id.is_integer(): target_player_id = int(target_player_id) packet["targetPlayerId"] = target_player_id if isinstance(sequence, float) and sequence.is_integer(): sequence = int(sequence) packet["sequence"] = sequence if ( not isinstance(target_player_id, int) or isinstance(target_player_id, bool) or not isinstance(sequence, int) or isinstance(sequence, bool) or not isinstance(damage_value, (int, float)) or isinstance(damage_value, bool) ): self._log(f"Malformed combatHit packet from player {sender.player_id}: {packet}") return damage = float(damage_value) if ( not 0 < target_player_id <= 0xFFFFFFFF or not 0 < sequence <= 0xFFFFFFFF or not math.isfinite(damage) or not 0.0 < damage <= 10000.0 ): self._log( f"Invalid combatHit values from player {sender.player_id}: " f"targetPlayerId={target_player_id}, sequence={sequence}, damage={damage}" ) return if target_player_id == sender.player_id: self._log(f"Ignoring self-targeted combatHit from player {sender.player_id}") return weapon_form_id = packet.get("weaponFormId") if weapon_form_id is not None: if ( not isinstance(weapon_form_id, str) or not 1 <= len(weapon_form_id) <= 8 or any(character not in "0123456789abcdefABCDEF" for character in weapon_form_id) ): self._log( f"Invalid combatHit weaponFormId from player {sender.player_id}: " f"{weapon_form_id!r}" ) return packet["weaponFormId"] = weapon_form_id.upper().zfill(8) with self._lock: recipient = next( (client for client in self._clients.values() if client.player_id == target_player_id), None, ) if recipient is None: self._log( f"Target player {target_player_id} not found for combatHit from player {sender.player_id}" ) return try: self._send_packet(recipient, packet, broadcast=False) with self._lock: self._stats["combatHitsRouted"] += 1 self._log( f"Routed combatHit from player {sender.player_id} to player {target_player_id}: " f"damage={damage:.1f}" ) except OSError: self._disconnect_client(recipient) def _print_packet(self, client: ClientSession, packet: dict[str, Any]) -> None: packet_type = packet.get("type") if packet_type == "worldState": game_hour = packet.get("gameHour", "?") game_days_passed = packet.get("gameDaysPassed", "?") weather_form_id = packet.get("weatherFormId", "") extra_fields = [] for field_name in ("playerId", "clientTime", "serverTime"): if field_name in packet: extra_fields.append(f"{field_name}={packet[field_name]}") extra_details = "" if extra_fields: extra_details = ", " + ", ".join(extra_fields) weather_detail = f", weatherFormId={weather_form_id}" if weather_form_id else "" self._log( f"WorldState from {client.label}: gameHour={game_hour}, " f"gameDaysPassed={game_days_passed}{weather_detail}{extra_details}", level="debug", ) return if packet_type != "transform": self._log(f"Packet from {client.label}: {packet}", level="debug") return try: x = float(packet["x"]) y = float(packet["y"]) z = float(packet["z"]) angle_z = float(packet["angleZ"]) except (KeyError, TypeError, ValueError): self._log(f"Malformed transform packet from {client.label}: {packet}", level="warning") return movement_type = packet.get("movementType", "normal") extra_fields = [] for field_name in ("playerId", "cellId", "worldspaceId", "clientTime", "serverTime"): if field_name in packet: extra_fields.append(f"{field_name}={packet[field_name]}") extra_details = "" if extra_fields: extra_details = ", " + ", ".join(extra_fields) self._log( "Transform from " f"{client.label}: x={x:.2f}, y={y:.2f}, z={z:.2f}, angleZ={angle_z:.2f}, " f"movementType={movement_type}{extra_details}", level="debug", ) def _should_log(self, level: str) -> bool: configured = _LOG_LEVELS.get(self.log_verbosity, _LOG_LEVELS[DEFAULT_LOG_VERBOSITY]) message_level = _LOG_LEVELS.get(level, _LOG_LEVELS["info"]) return message_level >= configured def _log(self, message: str, *, level: str = "info") -> None: if not self._should_log(level): return with self._log_lock: listeners = list(self._log_listeners) for listener in listeners: try: # Prefer level-aware callbacks; fall back for older listeners. try: listener(message, level=level) # type: ignore[call-arg] except TypeError: listener(message) except Exception: # Future UI listeners must not be able to break server networking. pass