from __future__ import annotations import json import math import socket import sys import threading import time from collections import deque 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" SESSION_ENDED_FULL = "server_full" SESSION_ENDED_PROTOCOL = "protocol_mismatch" SESSION_ENDED_RATE_LIMITED = "rate_limited" SESSION_ENDED_PACKET_TOO_LARGE = "packet_too_large" PROTOCOL_VERSION = 2 LEGACY_PROTOCOL_VERSION = 1 MAX_PACKET_CHARS = 64 * 1024 CLIENT_IDLE_TIMEOUT_SECONDS = 60.0 CLIENT_HANDSHAKE_TIMEOUT_SECONDS = 10.0 MAX_ABS_COORDINATE = 10_000_000.0 MAX_MOVEMENT_SPEED = 100_000.0 MAX_ACTION_EVENTS = 16 MAX_NPCS_PER_PACKET = 64 MAX_PACKETS_PER_SECOND = 120 MAX_CONNECT_ATTEMPTS = 8 CONNECT_ATTEMPT_WINDOW_SECONDS = 10.0 EXTERIOR_INTEREST_RADIUS = 8192.0 _LOG_LEVELS = { "debug": 10, "info": 20, "warning": 30, "error": 40, } def _bounded_repr(value: Any, max_chars: int = 256) -> str: text = repr(value) return text if len(text) <= max_chars else text[:max_chars] + "..." def _is_int(value: Any, minimum: int = 0, maximum: int = 0xFFFFFFFF) -> bool: return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum def _is_finite_number(value: Any, minimum: float, maximum: float) -> bool: return ( isinstance(value, (int, float)) and not isinstance(value, bool) and math.isfinite(float(value)) and minimum <= float(value) <= maximum ) def _is_hex_form_id(value: Any, *, allow_empty: bool = False, allow_zero: bool = True) -> bool: if not isinstance(value, str): return False if allow_empty and value == "": return True if not 1 <= len(value) <= 8: return False if any(character not in "0123456789abcdefABCDEF" for character in value): return False parsed = int(value, 16) return allow_zero or parsed != 0 def _normalize_hex_form_id(value: str) -> str: return value.upper().zfill(8) def _normalize_action_events(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] allowed = { (1, "meleeattackStart"), (2, "meleeattackStart"), (3, "fireSingle"), } normalized: list[dict[str, Any]] = [] for item in value[:MAX_ACTION_EVENTS]: if not isinstance(item, dict): continue sequence = item.get("sequence") action_type = item.get("type") event_name = item.get("eventName") if not _is_int(sequence, 1) or not _is_int(action_type, 1, 3) or not isinstance(event_name, str): continue if (action_type, event_name) not in allowed: continue clean: dict[str, Any] = { "sequence": sequence, "type": action_type, "eventName": event_name, } for name in ("actorStateFlags1", "actorStateFlags2"): if _is_int(item.get(name)): clean[name] = item[name] normalized.append(clean) return normalized def _scope_from_state(state: dict[str, Any] | None) -> tuple[str, str, float, float] | None: if not isinstance(state, dict): return None cell_id = state.get("cellId") worldspace_id = state.get("worldspaceId", "") x = state.get("x") y = state.get("y") if not _is_hex_form_id(cell_id, allow_zero=False): return None if not _is_hex_form_id(worldspace_id, allow_empty=True, allow_zero=True): return None if not _is_finite_number(x, -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE): return None if not _is_finite_number(y, -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE): return None return ( _normalize_hex_form_id(cell_id), _normalize_hex_form_id(worldspace_id) if worldspace_id else "", float(x), float(y), ) def states_share_interest(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool: """Return True when two states should see each other. Missing scope data intentionally falls back to True for compatibility with older clients. Interiors are matched by exact cell. Exteriors can also see nearby peers in the same worldspace so cell-border crossings do not make players pop in and out abruptly. """ a_scope = _scope_from_state(a) b_scope = _scope_from_state(b) if a_scope is None or b_scope is None: return True a_cell, a_world, ax, ay = a_scope b_cell, b_world, bx, by = b_scope if a_cell == b_cell: return True if not a_world or a_world != b_world: return False return math.hypot(ax - bx, ay - by) <= EXTERIOR_INTEREST_RADIUS def get_lan_addresses() -> list[str]: 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[..., 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._connect_attempts: dict[str, deque[float]] = {} self._stats: dict[str, int] = { "clientsConnected": 0, "clientsDisconnected": 0, "pendingConnectionsRejected": 0, "packetsReceived": 0, "packetsSent": 0, "packetsBroadcast": 0, "packetsRejected": 0, "rateLimitedPackets": 0, "transformPacketsReceived": 0, "transformPacketsBroadcast": 0, "transformPacketsInterestFiltered": 0, "worldStatePacketsReceived": 0, "worldStatePacketsBroadcast": 0, "npcStatePacketsReceived": 0, "npcStatePacketsBroadcast": 0, "combatHitsReceived": 0, "combatHitsRouted": 0, "worldStateHostPacketsBroadcast": 0, "serverWorldStatePacketsBroadcast": 0, "sessionEndedPacketsSent": 0, "bannedConnectionsRejected": 0, "disconnectPacketsBroadcast": 0, "protocolV2Connections": 0, "legacyConnections": 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: with self._lock: self._running = False server_socket = self._server_socket self._server_socket = None clients = list(self._clients.values()) client_threads = [t for t in self._client_threads if t 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 _active_clients_locked(self) -> list[ClientSession]: return [client for client in self._clients.values() if client.gameplay_active] def get_clients(self) -> list[dict[str, Any]]: with self._lock: clients = self._active_clients_locked() return [client.to_snapshot() for client in clients] def get_stats(self) -> dict[str, Any]: with self._lock: active_clients = self._active_clients_locked() pending = len(self._clients) - len(active_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": len(active_clients), "pendingConnections": pending, "nextPlayerId": self._next_player_id, "protocolVersion": PROTOCOL_VERSION, } ) 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) ended = self._end_sessions_for_ip(normalized, code=SESSION_ENDED_BANNED, reason=entry.reason) self._log(f"Banned IP {entry.ip}" + (f" (reason: {entry.reason})" if entry.reason else "")) 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._end_client_session(client, code=SESSION_ENDED_KICKED, reason=reason) self._log(f"Kicked player {player_id} ({client.label})" + (f" (reason: {reason})" if reason else "")) return {"playerId": player_id, "ip": ip, "code": SESSION_ENDED_KICKED, "reason": str(reason or "")} def add_log_listener(self, callback: Callable[..., 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[..., 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}. The port may already be in use or unavailable. ({error})" ) from error if self.port == 0: self.port = int(server_socket.getsockname()[1]) self._server_socket = server_socket self._running = True self._started_at = time.time() self._next_player_id = 1 self._world_state_host_player_id = None self._last_npc_state = None self._connect_attempts.clear() self._log(f"Commonwealth Online server listening on {self.host}:{self.port}") self._log(f"Protocol v{PROTOCOL_VERSION} negotiation enabled; legacy clients remain temporarily compatible.") if self.host == "0.0.0.0": self._log("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}") for address in get_lan_addresses(): self._log(f"LAN clients can connect at {address}:{self.port}") if sys.platform == "win32": self._log(f"Allow TCP {self.port} through Windows Firewall and forward it only when hosting externally.") else: self._log(f"Allow TCP {self.port} through the host firewall; router forwarding is only needed externally.") else: self._log(f"Clients can connect at {self.host}:{self.port}") 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 connections still work.", level="warning") self._discovery = None return self._discovery = discovery self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}") def _allow_connect_attempt(self, ip: str) -> bool: now = time.monotonic() cutoff = now - CONNECT_ATTEMPT_WINDOW_SECONDS with self._lock: attempts = self._connect_attempts.setdefault(ip, deque()) while attempts and attempts[0] < cutoff: attempts.popleft() attempts.append(now) return len(attempts) <= MAX_CONNECT_ATTEMPTS 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: continue except OSError: if self.is_running(): self._log("Server socket closed unexpectedly.", level="warning") break if not self._allow_connect_attempt(address[0]): with self._lock: self._stats["pendingConnectionsRejected"] += 1 self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many connection attempts.") self._close_raw_socket(connection) continue 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 | None: with self._lock: max_pending = max(16, self.max_players * 2) pending_count = sum(1 for client in self._clients.values() if not client.gameplay_active) if pending_count >= max_pending: return None 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 return client def _activate_client(self, client: ClientSession, protocol_version: int) -> bool: with self._lock: if client.gameplay_active: return True active_count = len(self._active_clients_locked()) if active_count >= self.max_players: return False if not client.mark_gameplay_active(protocol_version): return True self._stats["clientsConnected"] += 1 if protocol_version >= PROTOCOL_VERSION: self._stats["protocolV2Connections"] += 1 else: self._stats["legacyConnections"] += 1 if self._world_state_host_player_id is None: self._world_state_host_player_id = client.player_id became_host = True else: became_host = False self._log(f"Client connected: {client.label} (player {client.player_id}, protocol {protocol_version})") self._send_packet( client, { "type": "sessionReady", "playerId": client.player_id, "protocolVersion": protocol_version, "serverProtocolVersion": PROTOCOL_VERSION, "worldStateHostPlayerId": self._world_state_host_player_id, "serverTime": time.time(), }, ) self._send_existing_transforms_to_client(client) self._send_existing_npc_state_to_client(client) if became_host: self._broadcast_world_state_host_assignment(client.player_id) return True 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._send_session_ended_raw(connection, code=SESSION_ENDED_BANNED, reason=ban_entry.reason) self._close_raw_socket(connection) return client = self._assign_client(connection, address) if client is None: self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many pending connections.") self._close_raw_socket(connection) return connection.settimeout(1.0) connected_mono = time.monotonic() self._init_rate_state(client, connected_mono) self._log(f"TCP accept: {client.label} (provisional player {client.player_id})", level="debug") try: with connection: self._send_packet( client, { "type": "welcome", "playerId": client.player_id, "serverTime": time.time(), "serverName": self.server_name, "serverDescription": self.server_description, "protocolVersion": PROTOCOL_VERSION, "capabilities": ["interest-v1", "hello-v2", "bounded-framing", "rate-limit-v1"], }, ) buffer = b"" while self.is_running(): now_mono = time.monotonic() if not client.gameplay_active and now_mono - connected_mono > CLIENT_HANDSHAKE_TIMEOUT_SECONDS: self._log(f"Handshake timeout: {client.label}", level="warning") break if client.gameplay_active and client.last_packet_at is not None and time.time() - client.last_packet_at > CLIENT_IDLE_TIMEOUT_SECONDS: self._log(f"Idle timeout: player {client.player_id}", level="warning") break try: chunk = connection.recv(4096) except socket.timeout: continue if not chunk: break buffer += chunk if len(buffer) > MAX_PACKET_CHARS and b"\n" not in buffer: self._end_client_session(client, code=SESSION_ENDED_PACKET_TOO_LARGE, reason="Packet exceeded maximum line size.") break while b"\n" in buffer: line_bytes, buffer = buffer.split(b"\n", 1) if len(line_bytes) > MAX_PACKET_CHARS: self._end_client_session(client, code=SESSION_ENDED_PACKET_TOO_LARGE, reason="Packet exceeded maximum line size.") return try: line = line_bytes.decode("utf-8", errors="strict").strip() except UnicodeDecodeError: self._reject_packet(client, "Packet is not valid UTF-8") continue if not self._handle_line(client, line): if client.connection.fileno() < 0: return except (ConnectionResetError, BrokenPipeError): pass except OSError as error: if self.is_running(): self._log(f"Client connection error: {client.label}: {error}", level="debug") finally: self._disconnect_client(client) with self._lock: self._client_threads.discard(threading.current_thread()) def _init_rate_state(self, client: ClientSession, now: float) -> None: client._rate_window_start = now client._rate_window_count = 0 client._rate_violations = 0 client._rate_window_blocked = False client._last_combat_sequence = 0 def _allow_packet(self, client: ClientSession) -> bool: now = time.monotonic() start = getattr(client, "_rate_window_start", now) if now - start >= 1.0: previous_count = getattr(client, "_rate_window_count", 0) violations = getattr(client, "_rate_violations", 0) if previous_count <= MAX_PACKETS_PER_SECOND: violations = max(0, violations - 1) client._rate_violations = violations client._rate_window_start = now client._rate_window_count = 0 client._rate_window_blocked = False client._rate_window_count += 1 if client._rate_window_count <= MAX_PACKETS_PER_SECOND: return True with self._lock: self._stats["rateLimitedPackets"] += 1 if not client._rate_window_blocked: client._rate_window_blocked = True client._rate_violations += 1 self._log(f"Packet-rate limit exceeded by player {client.player_id} ({client._rate_violations}/3 windows)", level="warning") if client._rate_violations >= 3: self._end_client_session(client, code=SESSION_ENDED_RATE_LIMITED, reason="Sustained packet-rate limit exceeded.") return False def _handle_line(self, client: ClientSession, line: str) -> bool: if not line: return True if not self._allow_packet(client): return False client.record_received(time.time()) with self._lock: self._stats["packetsReceived"] += 1 try: packet = json.loads(line, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value))) except (json.JSONDecodeError, ValueError) as error: self._reject_packet(client, f"Invalid JSON: {error}") return False if not isinstance(packet, dict): self._reject_packet(client, "Packet must be a JSON object") return False packet_type = packet.get("type") if packet_type == "hello": return self._handle_hello(client, packet) if packet_type == "keepAlive": return True if not client.gameplay_active: if packet_type not in {"transform", "worldState", "npcState", "combatHit"}: self._reject_packet(client, "Gameplay packet received before session activation") return False if not self._activate_client(client, LEGACY_PROTOCOL_VERSION): self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.") return False self._log(f"Legacy client player {client.player_id} activated without hello; client upgrade recommended.", level="warning") if packet_type == "transform": normalized = self._normalize_transform_packet(packet) if normalized is None: self._reject_packet(client, "Malformed transform") return False normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() client.record_transform(normalized) with self._lock: self._stats["transformPacketsReceived"] += 1 self._broadcast_transform(client, normalized) return True if packet_type == "worldState": if not self._is_world_host(client): self._reject_packet(client, "worldState from non-authority client", warning=False) return False normalized = self._normalize_world_state_packet(packet) if normalized is None: self._reject_packet(client, "Malformed worldState") return False normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() with self._lock: self._stats["worldStatePacketsReceived"] += 1 self._broadcast_world_state(client, normalized) return True if packet_type == "npcState": if not self._is_world_host(client): self._reject_packet(client, "npcState from non-authority client", warning=False) return False normalized = self._normalize_npc_state_packet(packet) if normalized is None: self._reject_packet(client, "Malformed npcState") return False normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() normalized["fullReplace"] = True with self._lock: self._last_npc_state = dict(normalized) self._stats["npcStatePacketsReceived"] += 1 self._broadcast_npc_state(client, normalized) return True if packet_type == "combatHit": normalized = self._normalize_combat_hit_packet(packet) if normalized is None: self._reject_packet(client, "Malformed combatHit") return False sequence = normalized["sequence"] if sequence <= getattr(client, "_last_combat_sequence", 0): self._reject_packet(client, "Duplicate or out-of-order combat sequence", warning=False) return False client._last_combat_sequence = sequence normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() with self._lock: self._stats["combatHitsReceived"] += 1 self._route_combat_hit(client, normalized) return True self._reject_packet(client, f"Unknown packet type: {_bounded_repr(packet_type)}") return False def _handle_hello(self, client: ClientSession, packet: dict[str, Any]) -> bool: version = packet.get("protocolVersion") if not _is_int(version, 1, 0xFFFF) or version != PROTOCOL_VERSION: self._end_client_session( client, code=SESSION_ENDED_PROTOCOL, reason=f"Server requires protocol {PROTOCOL_VERSION}.", ) return False if not self._activate_client(client, version): self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.") return False return True def _normalize_transform_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None: for field_name in ("x", "y", "z", "angleZ"): if not _is_finite_number(packet.get(field_name), -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE): return None if not _is_hex_form_id(packet.get("cellId"), allow_zero=False): return None worldspace_id = packet.get("worldspaceId", "") if not _is_hex_form_id(worldspace_id, allow_empty=True, allow_zero=True): return None normalized = dict(packet) normalized["x"] = float(packet["x"]) normalized["y"] = float(packet["y"]) normalized["z"] = float(packet["z"]) normalized["angleZ"] = float(packet["angleZ"]) normalized["cellId"] = _normalize_hex_form_id(packet["cellId"]) normalized["worldspaceId"] = _normalize_hex_form_id(worldspace_id) if worldspace_id else "" movement_type = packet.get("movementType", "normal") normalized["movementType"] = str(movement_type)[:32] if isinstance(movement_type, str) else "normal" for field_name in ("movementSpeed", "animationGraphSpeed"): if field_name in packet: value = packet[field_name] if not _is_finite_number(value, -1.0 if field_name == "animationGraphSpeed" else 0.0, MAX_MOVEMENT_SPEED): normalized.pop(field_name, None) else: normalized[field_name] = float(value) for field_name, minimum, maximum in ( ("animationDirection", -360.0, 360.0), ("aimPitch", -180.0, 180.0), ("turnDelta", -10000.0, 10000.0), ): if field_name in packet: if not _is_finite_number(packet[field_name], minimum, maximum): normalized.pop(field_name, None) else: normalized[field_name] = float(packet[field_name]) for field_name in ("isMoving", "isSprinting", "isSneaking", "isJumping", "isCrouching", "weaponDrawn"): if field_name in packet and not isinstance(packet[field_name], bool): normalized.pop(field_name, None) for field_name in ("actorStateFlags1", "actorStateFlags2"): if field_name in packet and not _is_int(packet[field_name]): normalized.pop(field_name, None) if "actionEvents" in packet: normalized["actionEvents"] = _normalize_action_events(packet["actionEvents"]) return normalized def _normalize_world_state_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None: normalized = dict(packet) if "gameHour" in packet: if not _is_finite_number(packet["gameHour"], 0.0, 24.0): return None normalized["gameHour"] = float(packet["gameHour"]) if "gameDaysPassed" in packet: if not _is_finite_number(packet["gameDaysPassed"], 0.0, 10_000_000.0): return None normalized["gameDaysPassed"] = float(packet["gameDaysPassed"]) weather = packet.get("weatherFormId") if weather is not None: if not _is_hex_form_id(weather, allow_empty=True, allow_zero=True): return None normalized["weatherFormId"] = _normalize_hex_form_id(weather) if weather else "" return normalized def _normalize_npc_state_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None: npcs = packet.get("npcs") if not isinstance(npcs, list) or len(npcs) > MAX_NPCS_PER_PACKET: return None clean_npcs: list[dict[str, Any]] = [] for npc in npcs: if not isinstance(npc, dict): return None if not _is_hex_form_id(npc.get("sourceFormId"), allow_zero=False): return None if not _is_hex_form_id(npc.get("cellId"), allow_zero=False): return None worldspace = npc.get("worldspaceId", "") if not _is_hex_form_id(worldspace, allow_empty=True, allow_zero=True): return None for field_name in ("x", "y", "z", "angleZ"): if not _is_finite_number(npc.get(field_name), -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE): return None clean = dict(npc) clean["sourceFormId"] = _normalize_hex_form_id(npc["sourceFormId"]) clean["cellId"] = _normalize_hex_form_id(npc["cellId"]) clean["worldspaceId"] = _normalize_hex_form_id(worldspace) if worldspace else "" for field_name in ("x", "y", "z", "angleZ"): clean[field_name] = float(npc[field_name]) clean_npcs.append(clean) normalized = dict(packet) normalized["npcs"] = clean_npcs return normalized def _normalize_combat_hit_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None: target = packet.get("targetPlayerId") sequence = packet.get("sequence") damage = packet.get("damage") if isinstance(target, float) and target.is_integer(): target = int(target) if isinstance(sequence, float) and sequence.is_integer(): sequence = int(sequence) if not _is_int(target, 1) or not _is_int(sequence, 1): return None if not _is_finite_number(damage, 0.000001, 10000.0): return None weapon = packet.get("weaponFormId") if weapon is not None and not _is_hex_form_id(weapon, allow_zero=True): return None normalized = dict(packet) normalized["targetPlayerId"] = target normalized["sequence"] = sequence normalized["damage"] = float(damage) if isinstance(weapon, str): normalized["weaponFormId"] = _normalize_hex_form_id(weapon) return normalized def _is_world_host(self, client: ClientSession) -> bool: with self._lock: return self._world_state_host_player_id == client.player_id def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None: encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n" if len(encoded) > MAX_PACKET_CHARS: raise ValueError("Outbound packet exceeds maximum line size") client.send_bytes(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: peers = [ client for client in self._active_clients_locked() if client.connection != new_client.connection and client.last_transform is not None ] successful = 0 for peer in peers: if not states_share_interest(peer.last_transform, new_client.last_transform): continue packet = dict(peer.last_transform or {}) packet["serverTime"] = time.time() try: self._send_packet(new_client, packet, broadcast=True) successful += 1 except (OSError, ValueError): self._disconnect_client(new_client) break with self._lock: self._stats["transformPacketsBroadcast"] += successful def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None: with self._lock: recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection] successful = 0 filtered = 0 failed: list[ClientSession] = [] for recipient in recipients: if not states_share_interest(packet, recipient.last_transform): filtered += 1 continue try: self._send_packet(recipient, packet, broadcast=True) successful += 1 except (OSError, ValueError): failed.append(recipient) with self._lock: self._stats["transformPacketsBroadcast"] += successful self._stats["transformPacketsInterestFiltered"] += filtered for recipient in failed: self._disconnect_client(recipient) def _broadcast_world_state(self, sender: ClientSession, packet: dict[str, Any]) -> None: successful, failed = self._broadcast_to_active(packet, exclude=sender) with self._lock: self._stats["worldStatePacketsBroadcast"] += successful for client in failed: self._disconnect_client(client) def _npc_packet_for_recipient(self, packet: dict[str, Any], recipient: ClientSession) -> dict[str, Any]: if recipient.last_transform is None: return dict(packet) clean = dict(packet) clean["npcs"] = [npc for npc in packet.get("npcs", []) if states_share_interest(npc, recipient.last_transform)] return clean def _broadcast_npc_state(self, sender: ClientSession, packet: dict[str, Any]) -> None: with self._lock: recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection] successful = 0 failed: list[ClientSession] = [] for recipient in recipients: scoped_packet = self._npc_packet_for_recipient(packet, recipient) try: self._send_packet(recipient, scoped_packet, broadcast=True) successful += 1 except (OSError, ValueError): failed.append(recipient) with self._lock: self._stats["npcStatePacketsBroadcast"] += successful for client in failed: self._disconnect_client(client) def _send_existing_npc_state_to_client(self, 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") == client.player_id: return scoped = self._npc_packet_for_recipient(snapshot, client) scoped["serverTime"] = time.time() try: self._send_packet(client, scoped, broadcast=True) with self._lock: self._stats["npcStatePacketsBroadcast"] += 1 except (OSError, ValueError): self._disconnect_client(client) def _route_combat_hit(self, sender: ClientSession, packet: dict[str, Any]) -> None: target_player_id = packet["targetPlayerId"] if target_player_id == sender.player_id: self._reject_packet(sender, "Self-targeted combatHit", warning=False) return recipient = self._find_client_by_player_id(target_player_id) if recipient is None: self._reject_packet(sender, f"Combat target {target_player_id} is not connected", warning=False) return if not states_share_interest(sender.last_transform, recipient.last_transform): self._reject_packet(sender, f"Combat target {target_player_id} is outside interest scope", warning=True) return try: self._send_packet(recipient, packet) with self._lock: self._stats["combatHitsRouted"] += 1 except (OSError, ValueError): self._disconnect_client(recipient) def _broadcast_server_world_state(self) -> None: with self._lock: recipients = self._active_clients_locked() snapshot = dict(self._server_world_state) packets: list[dict[str, Any]] = [] if snapshot.get("timeHHmm"): packets.append({"type": "serverWorldState", "timeHHmm": snapshot["timeHHmm"], "serverTime": time.time()}) if snapshot.get("weatherConsoleArg"): weather_packet: dict[str, Any] = { "type": "serverWorldState", "weatherConsoleArg": snapshot["weatherConsoleArg"], "serverTime": time.time(), } if snapshot.get("weatherFormId"): weather_packet["weatherFormId"] = snapshot["weatherFormId"] packets.append(weather_packet) successful = 0 failed: list[ClientSession] = [] for packet in packets: for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful += 1 except (OSError, ValueError): failed.append(recipient) with self._lock: self._stats["serverWorldStatePacketsBroadcast"] += successful for client in {failed_client.connection: failed_client for failed_client in failed}.values(): self._disconnect_client(client) def _broadcast_to_active(self, packet: dict[str, Any], *, exclude: ClientSession | None = None) -> tuple[int, list[ClientSession]]: with self._lock: recipients = [ client for client in self._active_clients_locked() if exclude is None or client.connection != exclude.connection ] successful = 0 failed: list[ClientSession] = [] for recipient in recipients: try: self._send_packet(recipient, packet, broadcast=True) successful += 1 except (OSError, ValueError): failed.append(recipient) return successful, failed def _broadcast_world_state_host_assignment(self, player_id: int) -> None: packet = {"type": "worldStateHost", "worldStateHostPlayerId": player_id, "serverTime": time.time()} successful, failed = self._broadcast_to_active(packet) with self._lock: self._stats["worldStateHostPacketsBroadcast"] += successful for client in failed: self._disconnect_client(client) def _reassign_world_state_host(self) -> None: with self._lock: active = self._active_clients_locked() if not active: self._world_state_host_player_id = None self._last_npc_state = None return new_host = min(client.player_id for client in active) self._world_state_host_player_id = new_host self._last_npc_state = None self._broadcast_world_state_host_assignment(new_host) self._log(f"Reassigned world-state host to player {new_host}.") def _broadcast_disconnect(self, client: ClientSession) -> None: packet = {"type": "disconnect", "playerId": client.player_id, "serverTime": time.time()} successful, failed = self._broadcast_to_active(packet) with self._lock: self._stats["disconnectPacketsBroadcast"] += successful for recipient in failed: self._disconnect_client(recipient) def _remove_client(self, client: ClientSession) -> bool: with self._lock: removed = self._clients.pop(client.connection, None) is not None if removed and client.gameplay_active: self._stats["clientsDisconnected"] += 1 return removed def _disconnect_client(self, client: ClientSession) -> None: with self._lock: was_active = client.gameplay_active was_host = was_active and client.player_id == self._world_state_host_player_id if not self._remove_client(client): return self._close_client_socket(client) if not was_active: self._log(f"Closed pending/probe connection: {client.label}", level="debug") return self._log(f"Client disconnected: {client.label} (player {client.player_id})") self._broadcast_disconnect(client) if was_host: self._reassign_world_state_host() def _find_client_by_player_id(self, player_id: int) -> ClientSession | None: with self._lock: return next((client for client in self._active_clients_locked() if client.player_id == player_id), 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]: return {"type": "sessionEnded", "code": code, "reason": str(reason or ""), "serverTime": time.time()} 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=(",", ":"), allow_nan=False).encode("utf-8") + b"\n" try: connection.settimeout(1.0) connection.sendall(encoded) with self._lock: self._stats["sessionEndedPacketsSent"] += 1 self._stats["packetsSent"] += 1 except OSError: pass def _end_client_session(self, client: ClientSession, *, code: str, reason: str = "") -> None: try: self._send_packet(client, self._build_session_ended_packet(code, reason)) with self._lock: self._stats["sessionEndedPacketsSent"] += 1 except (OSError, ValueError): 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_raw_socket(self, connection: socket.socket) -> None: try: connection.close() except OSError: pass def _close_client_socket(self, client: ClientSession) -> None: self._close_raw_socket(client.connection) def _reject_packet(self, client: ClientSession, reason: str, *, warning: bool = True) -> None: with self._lock: self._stats["packetsRejected"] += 1 self._log( f"Rejected packet from {client.label} (player {client.player_id}): {reason}", level="warning" if warning else "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) if not listeners: return for listener in listeners: try: try: listener(message, level=level) except TypeError: listener(message) except Exception: pass