diff --git a/server/server_core.py b/server/server_core.py index 5c964f5..3b4a2fc 100644 --- a/server/server_core.py +++ b/server/server_core.py @@ -6,17 +6,14 @@ 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, -) - +from world_state_presets import normalize_fw_console_arg, relay_weather_form_id HOST = "0.0.0.0" PORT = 7777 @@ -26,6 +23,24 @@ 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, @@ -35,10 +50,117 @@ _LOG_LEVELS = { } -def get_lan_addresses() -> list[str]: - """Return likely non-loopback IPv4 addresses for this machine.""" - addresses: list[str] = [] +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)) @@ -47,7 +169,6 @@ def get_lan_addresses() -> list[str]: addresses.append(primary) except OSError: pass - try: hostname = socket.gethostname() for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET): @@ -56,7 +177,6 @@ def get_lan_addresses() -> list[str]: addresses.append(candidate) except OSError: pass - return addresses @@ -84,7 +204,7 @@ class FalloutTogetherServer: 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._log_listeners: list[Callable[..., None]] = [] self._server_socket: socket.socket | None = None self._accept_thread: threading.Thread | None = None self._discovery: LanDiscoveryResponder | None = None @@ -94,15 +214,20 @@ class FalloutTogetherServer: 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, @@ -114,13 +239,14 @@ class FalloutTogetherServer: "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) @@ -130,50 +256,35 @@ class FalloutTogetherServer: 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() - ] + 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() @@ -181,15 +292,18 @@ class FalloutTogetherServer: 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 = list(self._clients.values()) - + clients = self._active_clients_locked() return [client.to_snapshot() for client in clients] def get_stats(self) -> dict[str, Any]: with self._lock: - connected_clients = len(self._clients) + active_clients = self._active_clients_locked() + pending = len(self._clients) - len(active_clients) started_at = self._started_at stats = dict(self._stats) stats.update( @@ -203,11 +317,12 @@ class FalloutTogetherServer: "isRunning": self._running, "startedAt": started_at, "uptimeSeconds": time.time() - started_at if started_at is not None else 0.0, - "connectedClients": connected_clients, + "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]: @@ -219,17 +334,14 @@ class FalloutTogetherServer: 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 @@ -239,21 +351,15 @@ class FalloutTogetherServer: 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, - } + {"ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at} for entry in self._ban_store.list_bans() ] @@ -268,19 +374,10 @@ class FalloutTogetherServer: 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, - } + 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) @@ -292,26 +389,17 @@ class FalloutTogetherServer: 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 ""), - } + 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[[str], None]) -> None: + 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[[str], None]) -> None: + def remove_log_listener(self, callback: Callable[..., None]) -> None: with self._log_lock: if callback in self._log_listeners: self._log_listeners.remove(callback) @@ -329,86 +417,78 @@ class FalloutTogetherServer: 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})" + 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() - # 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._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( - f"Bind address 0.0.0.0 means all interfaces; clients should not join 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}") - lan_addresses = get_lan_addresses() - for address in lan_addresses: + for address in get_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." - ) + self._log(f"Allow TCP {self.port} through Windows Firewall and forward it only when hosting externally.") else: - self._log( - "Allow TCP " - f"{self.port} through your host firewall. Router port forwarding is only " - "needed for connections from outside your LAN." - ) + 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}") - 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._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: - # Wake periodically so Ctrl+C is handled promptly in Windows terminals. continue except OSError: if self.is_running(): - self._log("Server socket closed unexpectedly.") + self._log("Server socket closed unexpectedly.", level="warning") break - thread = threading.Thread( - target=self._handle_client, - args=(connection, address), - daemon=True, - ) + 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() @@ -417,17 +497,18 @@ class FalloutTogetherServer: 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: + def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession | None: 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. + 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, @@ -436,18 +517,45 @@ class FalloutTogetherServer: ) 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: + def _activate_client(self, client: ClientSession, protocol_version: int) -> bool: with self._lock: - removed = self._clients.pop(client.connection, None) is not None - if removed: - self._stats["clientsDisconnected"] += 1 - return removed + 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] @@ -455,173 +563,348 @@ class FalloutTogetherServer: 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 + 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) - self._log( - f"TCP accept: {client.label} (player {client.player_id})", - level="debug", - ) + 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: - buffer = b"" - - try: - with self._lock: - world_state_host_player_id = self._world_state_host_player_id - - welcome_packet: dict[str, Any] = { + self._send_packet( + client, + { "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(): + "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) - 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) + 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 _handle_line(self, client: ClientSession, line: str) -> None: + 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 - - 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) + 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) - except json.JSONDecodeError as error: - self._log(f"Invalid JSON from {client.label}: {error}: {line}") - return - + 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._log(f"Invalid packet from {client.label}: expected JSON object: {packet}") - return + self._reject_packet(client, "Packet must be a JSON object") + return False - if packet.get("type") == "transform": - packet["playerId"] = client.player_id - packet["serverTime"] = time.time() - client.record_transform(packet) + 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.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() + 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.get("type") == "npcState": + 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: - 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._last_npc_state = dict(normalized) self._stats["npcStatePacketsReceived"] += 1 + self._broadcast_npc_state(client, normalized) + return True - if packet.get("type") == "combatHit": - packet["playerId"] = client.player_id - packet["serverTime"] = time.time() + 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._print_packet(client, packet) + self._reject_packet(client, f"Unknown packet type: {_bounded_repr(packet_type)}") + return False - 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 _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=(",", ":")).encode("utf-8") + b"\n" - client.connection.sendall(encoded) + 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: @@ -629,89 +912,210 @@ class FalloutTogetherServer: 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() + peers = [ + client for client in self._active_clients_locked() 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: + 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_sends += 1 - except OSError: - failed = True + successful += 1 + except (OSError, ValueError): + self._disconnect_client(new_client) break - with self._lock: - self._stats["transformPacketsBroadcast"] += successful_sends + self._stats["transformPacketsBroadcast"] += successful - if successful_sends: - self._log( - f"Sent {successful_sends} existing transform snapshot(s) to newly connected player {new_client.player_id}" - ) + 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) - if failed: - self._disconnect_client(new_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 _send_existing_npc_state_to_client(self, new_client: ClientSession) -> None: + 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") == new_client.player_id: + if snapshot is None or snapshot.get("playerId") == client.player_id: return - - snapshot["serverTime"] = time.time() + scoped = self._npc_packet_for_recipient(snapshot, client) + scoped["serverTime"] = time.time() try: - self._send_packet(new_client, snapshot, broadcast=True) + self._send_packet(client, scoped, 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) + 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_world_state_host = client.player_id == self._world_state_host_player_id - never_sent_packet = client.packets_received == 0 - + 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 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() + 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_world_state_host: + if was_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 + 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() @@ -719,56 +1123,26 @@ class FalloutTogetherServer: 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 + 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: + 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" + 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 - 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) + def _end_client_session(self, client: ClientSession, *, code: str, reason: str = "") -> None: try: - self._send_packet(client, packet) + self._send_packet(client, self._build_session_ended_packet(code, reason)) with self._lock: self._stats["sessionEndedPacketsSent"] += 1 - try: - client.connection.shutdown(socket.SHUT_WR) - except OSError: - pass - except OSError: + except (OSError, ValueError): pass self._disconnect_client(client) @@ -778,353 +1152,21 @@ class FalloutTogetherServer: self._end_client_session(client, code=code, reason=reason) return len(clients) - def _close_client_socket(self, client: ClientSession) -> None: + def _close_raw_socket(self, connection: socket.socket) -> None: try: - client.connection.close() + connection.close() except OSError: pass - def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None: + 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: - 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._stats["packetsRejected"] += 1 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", + f"Rejected packet from {client.label} (player {client.player_id}): {reason}", + level="warning" if warning else "debug", ) def _should_log(self, level: str) -> bool: @@ -1135,17 +1177,15 @@ class FalloutTogetherServer: 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: - # Prefer level-aware callbacks; fall back for older listeners. try: - listener(message, level=level) # type: ignore[call-arg] + listener(message, level=level) except TypeError: listener(message) except Exception: - # Future UI listeners must not be able to break server networking. pass