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 npc_authority import NpcAuthorityManager, ScopeKey, scope_from_transform from player_state import normalize_player_state_packet 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 MAX_NORMAL_MOVEMENT_SPEED = 2500.0 MOVEMENT_GRACE_DISTANCE = 512.0 MAX_MOVEMENT_VALIDATION_ELAPSED_SECONDS = 5.0 MOVEMENT_TRANSITION_TYPES = frozenset({ "teleport", "cell_change", "worldspace_change", "load", "spawn", "fast_travel", }) ALLOWED_MOVEMENT_TYPES = MOVEMENT_TRANSITION_TYPES | {"normal"} _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: 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._last_npc_state_by_scope: dict[ScopeKey, dict[str, Any]] = {} self._last_player_state_by_player_id: dict[int, dict[str, Any]] = {} self._npc_authority = NpcAuthorityManager() 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, "playerStatePacketsReceived": 0, "playerStatePacketsBroadcast": 0, "movementPacketsRejected": 0, "movementCorrectionsSent": 0, "worldStatePacketsReceived": 0, "worldStatePacketsBroadcast": 0, "npcStatePacketsReceived": 0, "npcStatePacketsBroadcast": 0, "npcAuthorityChanges": 0, "npcAuthorityRejects": 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, "npcAuthorityScopes": len(self._npc_authority.assignments()), } ) 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._last_npc_state_by_scope.clear() self._last_player_state_by_player_id.clear() self._npc_authority.clear() 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_player_states_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", "movement-correction-v1", "npc-authority-epoch-v1", "player-state-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 if client.protocol_version >= PROTOCOL_VERSION: for reliable_field in ("equippedItems", "appearance", "actionEvents", "characterName"): normalized.pop(reliable_field, None) accepted_monotonic = time.monotonic() movement_valid, movement_reason = self._validate_transform_movement(client, normalized, accepted_monotonic) if not movement_valid: with self._lock: self._stats["movementPacketsRejected"] += 1 self._reject_packet(client, movement_reason) self._send_position_correction(client, movement_reason) return False previous_scope = scope_from_transform(client.last_transform) normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() client.record_transform(normalized, accepted_monotonic) with self._lock: self._stats["transformPacketsReceived"] += 1 self._broadcast_transform(client, normalized) self._reconcile_npc_authority() current_scope = scope_from_transform(normalized) if current_scope is not None and current_scope != previous_scope: self._send_npc_authority_for_client(client, current_scope) return True if packet_type == "playerState": if client.protocol_version < PROTOCOL_VERSION: self._reject_packet(client, "playerState requires Protocol V2", warning=False) return False normalized = normalize_player_state_packet(packet, _normalize_action_events) if normalized is None: self._reject_packet(client, "Malformed playerState") return False normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() with self._lock: self._last_player_state_by_player_id[client.player_id] = dict(normalized) self._stats["playerStatePacketsReceived"] += 1 self._broadcast_player_state(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": normalized = self._normalize_npc_state_packet(packet) if normalized is None: self._reject_packet(client, "Malformed npcState") return False if client.protocol_version >= PROTOCOL_VERSION: scope = self._npc_scope_from_packet(normalized) epoch = normalized.get("authorityEpoch") if scope is None or not _is_int(epoch, 1): with self._lock: self._stats["npcAuthorityRejects"] += 1 self._reject_packet(client, "Protocol V2 npcState missing valid authority scope/epoch") return False if not self._npc_authority.authorize(client.player_id, scope, epoch): with self._lock: self._stats["npcAuthorityRejects"] += 1 self._reject_packet(client, "Stale or unauthorized npcState authority epoch") return False if any( npc.get("cellId") != scope.cell_id or npc.get("worldspaceId", "") != scope.worldspace_id for npc in normalized.get("npcs", []) ): with self._lock: self._stats["npcAuthorityRejects"] += 1 self._reject_packet(client, "npcState contains NPCs outside declared authority scope") return False normalized["authorityCellId"] = scope.cell_id normalized["authorityWorldspaceId"] = scope.worldspace_id with self._lock: self._last_npc_state_by_scope[scope] = dict(normalized) else: if not self._is_world_host(client): self._reject_packet(client, "Legacy npcState from non-authority client", warning=False) return False with self._lock: self._last_npc_state = dict(normalized) normalized["playerId"] = client.player_id normalized["serverTime"] = time.time() normalized["fullReplace"] = True with self._lock: 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 movement_type = packet.get("movementType", "normal") if not isinstance(movement_type, str) or movement_type not in ALLOWED_MOVEMENT_TYPES: 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 "" normalized["movementType"] = movement_type 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 _validate_transform_movement(self, client: ClientSession, packet: dict[str, Any], now_monotonic: float) -> tuple[bool, str]: previous, previous_monotonic = client.get_last_transform_anchor() if previous is None or previous_monotonic is None: return True, "first transform" movement_type = packet.get("movementType", "normal") if movement_type in MOVEMENT_TRANSITION_TYPES: return True, f"explicit {movement_type} transition" previous_cell = previous.get("cellId", "") current_cell = packet.get("cellId", "") previous_world = previous.get("worldspaceId", "") current_world = packet.get("worldspaceId", "") if previous_cell != current_cell or previous_world != current_world: return False, "scope changed without an explicit movement transition" elapsed = now_monotonic - previous_monotonic if not math.isfinite(elapsed) or elapsed < 0.0: elapsed = 0.0 elapsed = min(elapsed, MAX_MOVEMENT_VALIDATION_ELAPSED_SECONDS) dx = packet["x"] - previous["x"] dy = packet["y"] - previous["y"] dz = packet["z"] - previous["z"] distance = math.sqrt((dx * dx) + (dy * dy) + (dz * dz)) allowed_distance = MOVEMENT_GRACE_DISTANCE + (MAX_NORMAL_MOVEMENT_SPEED * elapsed) if not math.isfinite(distance) or distance > allowed_distance: return False, f"normal movement exceeded server envelope: distance={distance:.1f}, allowed={allowed_distance:.1f}, elapsed={elapsed:.3f}s" return True, "normal movement accepted" def _send_position_correction(self, client: ClientSession, reason: str) -> None: previous, _previous_monotonic = client.get_last_transform_anchor() if previous is None: return packet: dict[str, Any] = { "type": "positionCorrection", "reason": reason[:160], "x": previous["x"], "y": previous["y"], "z": previous["z"], "angleZ": previous["angleZ"], "cellId": previous["cellId"], "worldspaceId": previous.get("worldspaceId", ""), "serverTime": time.time(), } try: self._send_packet(client, packet) except (OSError, ValueError): self._disconnect_client(client) return with self._lock: self._stats["movementCorrectionsSent"] += 1 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 if "authorityEpoch" in packet: if not _is_int(packet["authorityEpoch"], 1): return None normalized["authorityEpoch"] = int(packet["authorityEpoch"]) if "authorityCellId" in packet: if not _is_hex_form_id(packet["authorityCellId"], allow_zero=False): return None normalized["authorityCellId"] = _normalize_hex_form_id(packet["authorityCellId"]) if "authorityWorldspaceId" in packet: world = packet["authorityWorldspaceId"] if not _is_hex_form_id(world, allow_empty=True, allow_zero=True): return None normalized["authorityWorldspaceId"] = _normalize_hex_form_id(world) if world else "" return normalized def _npc_scope_from_packet(self, packet: dict[str, Any]) -> ScopeKey | None: cell = packet.get("authorityCellId") world = packet.get("authorityWorldspaceId", "") if not _is_hex_form_id(cell, allow_zero=False): return None if not _is_hex_form_id(world, allow_empty=True, allow_zero=True): return None return ScopeKey(_normalize_hex_form_id(cell), _normalize_hex_form_id(world) if world else "") 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 _authority_packet(self, assignment_player_id: int, epoch: int, scope: ScopeKey) -> dict[str, Any]: return { "type": "npcAuthority", "authorityPlayerId": assignment_player_id, "authorityEpoch": epoch, "authorityCellId": scope.cell_id, "authorityWorldspaceId": scope.worldspace_id, "serverTime": time.time(), } def _reconcile_npc_authority(self) -> None: with self._lock: players = [] for client in self._active_clients_locked(): if client.protocol_version < PROTOCOL_VERSION: continue scope = scope_from_transform(client.last_transform) if scope is not None: players.append((client.player_id, scope)) changes = self._npc_authority.reconcile(players) for change in changes: self._last_npc_state_by_scope.pop(change.scope, None) for change in changes: packet = self._authority_packet(change.player_id, change.epoch, change.scope) successful, failed = self._broadcast_to_active(packet, v2_only=True) with self._lock: self._stats["npcAuthorityChanges"] += 1 for client in failed: self._disconnect_client(client) self._log( f"NPC authority scope {change.scope.cell_id}/{change.scope.worldspace_id or ''}: " f"{change.previous_player_id} -> {change.player_id}, epoch {change.epoch} " f"(notified {successful} client(s))." ) def _send_npc_authority_for_client(self, client: ClientSession, scope: ScopeKey) -> None: if client.protocol_version < PROTOCOL_VERSION: return assignment = self._npc_authority.get(scope) if assignment is None: return try: self._send_packet(client, self._authority_packet(assignment.player_id, assignment.epoch, scope)) except (OSError, ValueError): self._disconnect_client(client) 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 _send_existing_player_states_to_client(self, new_client: ClientSession) -> None: if new_client.protocol_version < PROTOCOL_VERSION: return with self._lock: snapshots = [ dict(packet) for player_id, packet in self._last_player_state_by_player_id.items() if player_id != new_client.player_id ] successful = 0 for packet in snapshots: 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["playerStatePacketsBroadcast"] += 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_player_state(self, sender: ClientSession, packet: dict[str, Any]) -> None: successful, failed = self._broadcast_to_active(packet, exclude=sender, v2_only=True) with self._lock: self._stats["playerStatePacketsBroadcast"] += successful for client in failed: self._disconnect_client(client) 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: if client.protocol_version >= PROTOCOL_VERSION: scope = scope_from_transform(client.last_transform) with self._lock: snapshot = dict(self._last_npc_state_by_scope[scope]) if scope in self._last_npc_state_by_scope else None else: 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, v2_only: bool = False, ) -> 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) and (not v2_only or client.protocol_version >= PROTOCOL_VERSION) ] 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: self._last_player_state_by_player_id.pop(client.player_id, 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) self._reconcile_npc_authority() 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