Commonwealth-Online/Commonwealth-Online-Server
3
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases 2 Wiki Activity
Files
94be46979e6aedc4f7104c30e6823373e5533886
Commonwealth-Online-Server/server/server_core.py
T

1281 lines
54 KiB
Python
Raw Blame History

from __future__ import annotations
import json
import math
import socket
import sys
import threading
import time
from collections import deque
from collections.abc import Callable
from typing import Any
from ban_store import BanStore
from client_session import ClientSession
from lan_discovery import DISCOVERY_PORT, LanDiscoveryResponder
from world_state_presets import normalize_fw_console_arg, relay_weather_form_id
HOST = "0.0.0.0"
PORT = 7777
ACCEPT_TIMEOUT_SECONDS = 0.5
DEFAULT_SERVER_NAME = "Commonwealth Online Server"
DEFAULT_MAX_PLAYERS = 16
DEFAULT_LOG_VERBOSITY = "info"
SESSION_ENDED_BANNED = "banned"
SESSION_ENDED_KICKED = "kicked"
SESSION_ENDED_FULL = "server_full"
SESSION_ENDED_PROTOCOL = "protocol_mismatch"
SESSION_ENDED_RATE_LIMITED = "rate_limited"
SESSION_ENDED_PACKET_TOO_LARGE = "packet_too_large"
PROTOCOL_VERSION = 2
LEGACY_PROTOCOL_VERSION = 1
MAX_PACKET_CHARS = 64 * 1024
CLIENT_IDLE_TIMEOUT_SECONDS = 60.0
CLIENT_HANDSHAKE_TIMEOUT_SECONDS = 10.0
MAX_ABS_COORDINATE = 10_000_000.0
MAX_MOVEMENT_SPEED = 100_000.0
MAX_ACTION_EVENTS = 16
MAX_NPCS_PER_PACKET = 64
MAX_PACKETS_PER_SECOND = 120
MAX_CONNECT_ATTEMPTS = 8
CONNECT_ATTEMPT_WINDOW_SECONDS = 10.0
EXTERIOR_INTEREST_RADIUS = 8192.0
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:
"""Return True when two states should see each other.
Missing scope data intentionally falls back to True for compatibility with older clients.
Interiors are matched by exact cell. Exteriors can also see nearby peers in the same
worldspace so cell-border crossings do not make players pop in and out abruptly.
"""
a_scope = _scope_from_state(a)
b_scope = _scope_from_state(b)
if a_scope is None or b_scope is None:
return True
a_cell, a_world, ax, ay = a_scope
b_cell, b_world, bx, by = b_scope
if a_cell == b_cell:
return True
if not a_world or a_world != b_world:
return False
return math.hypot(ax - bx, ay - by) <= EXTERIOR_INTEREST_RADIUS
def get_lan_addresses() -> list[str]:
addresses: list[str] = []
try:
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
probe.connect(("8.8.8.8", 80))
primary = probe.getsockname()[0]
if primary and not primary.startswith("127."):
addresses.append(primary)
except OSError:
pass
try:
hostname = socket.gethostname()
for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET):
candidate = info[4][0]
if candidate and not candidate.startswith("127.") and candidate not in addresses:
addresses.append(candidate)
except OSError:
pass
return addresses
class FalloutTogetherServer:
def __init__(
self,
host: str = HOST,
port: int = PORT,
server_name: str = DEFAULT_SERVER_NAME,
server_description: str = "",
max_players: int = DEFAULT_MAX_PLAYERS,
bans_path: str | None = None,
log_verbosity: str = DEFAULT_LOG_VERBOSITY,
) -> None:
self.host = host
self.port = port
self.server_name = server_name or DEFAULT_SERVER_NAME
self.server_description = server_description or ""
self.max_players = max_players if max_players >= 1 else DEFAULT_MAX_PLAYERS
verbosity = str(log_verbosity or DEFAULT_LOG_VERBOSITY).strip().lower()
self.log_verbosity = verbosity if verbosity in _LOG_LEVELS else DEFAULT_LOG_VERBOSITY
self._ban_store = BanStore(bans_path)
self._lock = threading.RLock()
self._log_lock = threading.RLock()
self._clients: dict[socket.socket, ClientSession] = {}
self._client_threads: set[threading.Thread] = set()
self._log_listeners: list[Callable[..., None]] = []
self._server_socket: socket.socket | None = None
self._accept_thread: threading.Thread | None = None
self._discovery: LanDiscoveryResponder | None = None
self._running = False
self._started_at: float | None = None
self._next_player_id = 1
self._world_state_host_player_id: int | None = None
self._server_world_state: dict[str, str] = {}
self._last_npc_state: dict[str, Any] | None = None
self._connect_attempts: dict[str, deque[float]] = {}
self._stats: dict[str, int] = {
"clientsConnected": 0,
"clientsDisconnected": 0,
"pendingConnectionsRejected": 0,
"packetsReceived": 0,
"packetsSent": 0,
"packetsBroadcast": 0,
"packetsRejected": 0,
"rateLimitedPackets": 0,
"transformPacketsReceived": 0,
"transformPacketsBroadcast": 0,
"transformPacketsInterestFiltered": 0,
"movementPacketsRejected": 0,
"movementCorrectionsSent": 0,
"worldStatePacketsReceived": 0,
"worldStatePacketsBroadcast": 0,
"npcStatePacketsReceived": 0,
"npcStatePacketsBroadcast": 0,
"combatHitsReceived": 0,
"combatHitsRouted": 0,
"worldStateHostPacketsBroadcast": 0,
"serverWorldStatePacketsBroadcast": 0,
"sessionEndedPacketsSent": 0,
"bannedConnectionsRejected": 0,
"disconnectPacketsBroadcast": 0,
"protocolV2Connections": 0,
"legacyConnections": 0,
}
def start(self) -> None:
with self._lock:
if self._running:
return
self._prepare_server_socket()
self._start_discovery()
self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True)
self._accept_thread.start()
def serve_forever(self) -> None:
with self._lock:
if self._running:
raise RuntimeError("Server is already running.")
self._prepare_server_socket()
self._start_discovery()
self._accept_loop()
def stop(self) -> None:
with self._lock:
self._running = False
server_socket = self._server_socket
self._server_socket = None
clients = list(self._clients.values())
client_threads = [t for t in self._client_threads if t is not threading.current_thread()]
discovery = self._discovery
self._discovery = None
if discovery is not None:
discovery.stop()
if server_socket is not None:
try:
server_socket.close()
except OSError:
pass
for client in clients:
self._close_client_socket(client)
accept_thread = self._accept_thread
if accept_thread is not None and accept_thread is not threading.current_thread():
accept_thread.join(timeout=1.0)
self._accept_thread = None
for thread in client_threads:
thread.join(timeout=1.0)
with self._lock:
self._client_threads.clear()
def is_running(self) -> bool:
with self._lock:
return self._running
def _active_clients_locked(self) -> list[ClientSession]:
return [client for client in self._clients.values() if client.gameplay_active]
def get_clients(self) -> list[dict[str, Any]]:
with self._lock:
clients = self._active_clients_locked()
return [client.to_snapshot() for client in clients]
def get_stats(self) -> dict[str, Any]:
with self._lock:
active_clients = self._active_clients_locked()
pending = len(self._clients) - len(active_clients)
started_at = self._started_at
stats = dict(self._stats)
stats.update(
{
"host": self.host,
"port": self.port,
"serverName": self.server_name,
"serverDescription": self.server_description,
"maxPlayers": self.max_players,
"lanAddresses": get_lan_addresses() if self._running and self.host == "0.0.0.0" else [],
"isRunning": self._running,
"startedAt": started_at,
"uptimeSeconds": time.time() - started_at if started_at is not None else 0.0,
"connectedClients": len(active_clients),
"pendingConnections": pending,
"nextPlayerId": self._next_player_id,
"protocolVersion": PROTOCOL_VERSION,
}
)
return stats
def get_server_world_state(self) -> dict[str, str]:
with self._lock:
return dict(self._server_world_state)
def set_server_time(self, hhmm: str) -> bool:
text = str(hhmm).strip()
if not text.isdigit() or len(text) > 4:
self._log(f"Rejected invalid server time HHmm value: {hhmm!r}")
return False
normalized = text.zfill(4)
hours = int(normalized[:2])
minutes = int(normalized[2:])
if hours > 23 or minutes > 59:
self._log(f"Rejected out-of-range server time HHmm value: {hhmm!r}")
return False
with self._lock:
self._server_world_state["timeHHmm"] = normalized
self._broadcast_server_world_state()
return True
def set_server_weather(self, fw_console_arg: str) -> bool:
try:
normalized_fw_arg = normalize_fw_console_arg(fw_console_arg)
except ValueError:
self._log(f"Rejected invalid server weather console id: {fw_console_arg!r}")
return False
with self._lock:
self._server_world_state["weatherConsoleArg"] = normalized_fw_arg
self._server_world_state["weatherFormId"] = relay_weather_form_id(normalized_fw_arg)
self._broadcast_server_world_state()
return True
def list_bans(self) -> list[dict[str, Any]]:
return [
{"ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at}
for entry in self._ban_store.list_bans()
]
def unban_ip(self, ip: str) -> bool:
normalized = str(ip).strip()
removed = self._ban_store.unban_ip(normalized)
if removed:
self._log(f"Unbanned IP {normalized}")
return removed
def ban_ip(self, ip: str, reason: str = "") -> dict[str, Any]:
normalized = str(ip).strip()
if not normalized:
raise ValueError("IP address cannot be empty.")
entry = self._ban_store.ban_ip(normalized, reason=reason)
ended = self._end_sessions_for_ip(normalized, code=SESSION_ENDED_BANNED, reason=entry.reason)
self._log(f"Banned IP {entry.ip}" + (f" (reason: {entry.reason})" if entry.reason else ""))
return {"ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at, "sessionsEnded": ended}
def ban_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
client = self._find_client_by_player_id(player_id)
if client is None:
raise KeyError(f"No connected player with id {player_id}")
return self.ban_ip(client.address[0], reason=reason)
def kick_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
client = self._find_client_by_player_id(player_id)
if client is None:
raise KeyError(f"No connected player with id {player_id}")
ip = client.address[0]
self._end_client_session(client, code=SESSION_ENDED_KICKED, reason=reason)
self._log(f"Kicked player {player_id} ({client.label})" + (f" (reason: {reason})" if reason else ""))
return {"playerId": player_id, "ip": ip, "code": SESSION_ENDED_KICKED, "reason": str(reason or "")}
def add_log_listener(self, callback: Callable[..., None]) -> None:
with self._log_lock:
if callback not in self._log_listeners:
self._log_listeners.append(callback)
def remove_log_listener(self, callback: Callable[..., None]) -> None:
with self._log_lock:
if callback in self._log_listeners:
self._log_listeners.remove(callback)
def _prepare_server_socket(self) -> None:
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
try:
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
else:
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind((self.host, self.port))
server_socket.listen()
server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
except OSError as error:
server_socket.close()
raise OSError(
f"Could not bind the server to {self.host}:{self.port}. The port may already be in use or unavailable. ({error})"
) from error
if self.port == 0:
self.port = int(server_socket.getsockname()[1])
self._server_socket = server_socket
self._running = True
self._started_at = time.time()
self._next_player_id = 1
self._world_state_host_player_id = None
self._last_npc_state = None
self._connect_attempts.clear()
self._log(f"Commonwealth Online server listening on {self.host}:{self.port}")
self._log(f"Protocol v{PROTOCOL_VERSION} negotiation enabled; legacy clients remain temporarily compatible.")
if self.host == "0.0.0.0":
self._log("Bind address 0.0.0.0 means all interfaces; clients should not join 0.0.0.0.")
self._log(f"Local clients can connect at 127.0.0.1:{self.port}")
for address in get_lan_addresses():
self._log(f"LAN clients can connect at {address}:{self.port}")
if sys.platform == "win32":
self._log(f"Allow TCP {self.port} through Windows Firewall and forward it only when hosting externally.")
else:
self._log(f"Allow TCP {self.port} through the host firewall; router forwarding is only needed externally.")
else:
self._log(f"Clients can connect at {self.host}:{self.port}")
def _start_discovery(self) -> None:
discovery = LanDiscoveryResponder(self)
try:
discovery.start()
except OSError as error:
self._log(f"LAN discovery unavailable on UDP {DISCOVERY_PORT}: {error}. Direct connections still work.", level="warning")
self._discovery = None
return
self._discovery = discovery
self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}")
def _allow_connect_attempt(self, ip: str) -> bool:
now = time.monotonic()
cutoff = now - CONNECT_ATTEMPT_WINDOW_SECONDS
with self._lock:
attempts = self._connect_attempts.setdefault(ip, deque())
while attempts and attempts[0] < cutoff:
attempts.popleft()
attempts.append(now)
return len(attempts) <= MAX_CONNECT_ATTEMPTS
def _accept_loop(self) -> None:
try:
while self.is_running():
with self._lock:
server_socket = self._server_socket
if server_socket is None:
break
try:
connection, address = server_socket.accept()
except socket.timeout:
continue
except OSError:
if self.is_running():
self._log("Server socket closed unexpectedly.", level="warning")
break
if not self._allow_connect_attempt(address[0]):
with self._lock:
self._stats["pendingConnectionsRejected"] += 1
self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many connection attempts.")
self._close_raw_socket(connection)
continue
thread = threading.Thread(target=self._handle_client, args=(connection, address), daemon=True)
with self._lock:
self._client_threads.add(thread)
thread.start()
finally:
with self._lock:
server_socket = self._server_socket
self._running = False
self._server_socket = None
if server_socket is not None:
try:
server_socket.close()
except OSError:
pass
def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession | None:
with self._lock:
max_pending = max(16, self.max_players * 2)
pending_count = sum(1 for client in self._clients.values() if not client.gameplay_active)
if pending_count >= max_pending:
return None
client = ClientSession(
connection=connection,
address=address,
player_id=self._next_player_id,
connected_at=time.time(),
)
self._next_player_id += 1
self._clients[connection] = client
return client
def _activate_client(self, client: ClientSession, protocol_version: int) -> bool:
with self._lock:
if client.gameplay_active:
return True
active_count = len(self._active_clients_locked())
if active_count >= self.max_players:
return False
if not client.mark_gameplay_active(protocol_version):
return True
self._stats["clientsConnected"] += 1
if protocol_version >= PROTOCOL_VERSION:
self._stats["protocolV2Connections"] += 1
else:
self._stats["legacyConnections"] += 1
if self._world_state_host_player_id is None:
self._world_state_host_player_id = client.player_id
became_host = True
else:
became_host = False
self._log(f"Client connected: {client.label} (player {client.player_id}, protocol {protocol_version})")
self._send_packet(
client,
{
"type": "sessionReady",
"playerId": client.player_id,
"protocolVersion": protocol_version,
"serverProtocolVersion": PROTOCOL_VERSION,
"worldStateHostPlayerId": self._world_state_host_player_id,
"serverTime": time.time(),
},
)
self._send_existing_transforms_to_client(client)
self._send_existing_npc_state_to_client(client)
if became_host:
self._broadcast_world_state_host_assignment(client.player_id)
return True
def _handle_client(self, connection: socket.socket, address: tuple[str, int]) -> None:
peer_ip = address[0]
ban_entry = self._ban_store.get_ban(peer_ip)
if ban_entry is not None:
with self._lock:
self._stats["bannedConnectionsRejected"] += 1
self._send_session_ended_raw(connection, code=SESSION_ENDED_BANNED, reason=ban_entry.reason)
self._close_raw_socket(connection)
return
client = self._assign_client(connection, address)
if client is None:
self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many pending connections.")
self._close_raw_socket(connection)
return
connection.settimeout(1.0)
connected_mono = time.monotonic()
self._init_rate_state(client, connected_mono)
self._log(f"TCP accept: {client.label} (provisional player {client.player_id})", level="debug")
try:
with connection:
self._send_packet(
client,
{
"type": "welcome",
"playerId": client.player_id,
"serverTime": time.time(),
"serverName": self.server_name,
"serverDescription": self.server_description,
"protocolVersion": PROTOCOL_VERSION,
"capabilities": ["interest-v1", "hello-v2", "bounded-framing", "rate-limit-v1", "movement-correction-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
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
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)
return True
if packet_type == "worldState":
if not self._is_world_host(client):
self._reject_packet(client, "worldState from non-authority client", warning=False)
return False
normalized = self._normalize_world_state_packet(packet)
if normalized is None:
self._reject_packet(client, "Malformed worldState")
return False
normalized["playerId"] = client.player_id
normalized["serverTime"] = time.time()
with self._lock:
self._stats["worldStatePacketsReceived"] += 1
self._broadcast_world_state(client, normalized)
return True
if packet_type == "npcState":
if not self._is_world_host(client):
self._reject_packet(client, "npcState from non-authority client", warning=False)
return False
normalized = self._normalize_npc_state_packet(packet)
if normalized is None:
self._reject_packet(client, "Malformed npcState")
return False
normalized["playerId"] = client.player_id
normalized["serverTime"] = time.time()
normalized["fullReplace"] = True
with self._lock:
self._last_npc_state = dict(normalized)
self._stats["npcStatePacketsReceived"] += 1
self._broadcast_npc_state(client, normalized)
return True
if packet_type == "combatHit":
normalized = self._normalize_combat_hit_packet(packet)
if normalized is None:
self._reject_packet(client, "Malformed combatHit")
return False
sequence = normalized["sequence"]
if sequence <= getattr(client, "_last_combat_sequence", 0):
self._reject_packet(client, "Duplicate or out-of-order combat sequence", warning=False)
return False
client._last_combat_sequence = sequence
normalized["playerId"] = client.player_id
normalized["serverTime"] = time.time()
with self._lock:
self._stats["combatHitsReceived"] += 1
self._route_combat_hit(client, normalized)
return True
self._reject_packet(client, f"Unknown packet type: {_bounded_repr(packet_type)}")
return False
def _handle_hello(self, client: ClientSession, packet: dict[str, Any]) -> bool:
version = packet.get("protocolVersion")
if not _is_int(version, 1, 0xFFFF) or version != PROTOCOL_VERSION:
self._end_client_session(
client,
code=SESSION_ENDED_PROTOCOL,
reason=f"Server requires protocol {PROTOCOL_VERSION}.",
)
return False
if not self._activate_client(client, version):
self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.")
return False
return True
def _normalize_transform_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
for field_name in ("x", "y", "z", "angleZ"):
if not _is_finite_number(packet.get(field_name), -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE):
return None
if not _is_hex_form_id(packet.get("cellId"), allow_zero=False):
return None
worldspace_id = packet.get("worldspaceId", "")
if not _is_hex_form_id(worldspace_id, allow_empty=True, allow_zero=True):
return None
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
return normalized
def _normalize_combat_hit_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
target = packet.get("targetPlayerId")
sequence = packet.get("sequence")
damage = packet.get("damage")
if isinstance(target, float) and target.is_integer():
target = int(target)
if isinstance(sequence, float) and sequence.is_integer():
sequence = int(sequence)
if not _is_int(target, 1) or not _is_int(sequence, 1):
return None
if not _is_finite_number(damage, 0.000001, 10000.0):
return None
weapon = packet.get("weaponFormId")
if weapon is not None and not _is_hex_form_id(weapon, allow_zero=True):
return None
normalized = dict(packet)
normalized["targetPlayerId"] = target
normalized["sequence"] = sequence
normalized["damage"] = float(damage)
if isinstance(weapon, str):
normalized["weaponFormId"] = _normalize_hex_form_id(weapon)
return normalized
def _is_world_host(self, client: ClientSession) -> bool:
with self._lock:
return self._world_state_host_player_id == client.player_id
def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None:
encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
if len(encoded) > MAX_PACKET_CHARS:
raise ValueError("Outbound packet exceeds maximum line size")
client.send_bytes(encoded)
client.record_sent(broadcast=broadcast)
with self._lock:
self._stats["packetsSent"] += 1
if broadcast:
self._stats["packetsBroadcast"] += 1
def _send_existing_transforms_to_client(self, new_client: ClientSession) -> None:
with self._lock:
peers = [
client for client in self._active_clients_locked()
if client.connection != new_client.connection and client.last_transform is not None
]
successful = 0
for peer in peers:
if not states_share_interest(peer.last_transform, new_client.last_transform):
continue
packet = dict(peer.last_transform or {})
packet["serverTime"] = time.time()
try:
self._send_packet(new_client, packet, broadcast=True)
successful += 1
except (OSError, ValueError):
self._disconnect_client(new_client)
break
with self._lock:
self._stats["transformPacketsBroadcast"] += successful
def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None:
with self._lock:
recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection]
successful = 0
filtered = 0
failed: list[ClientSession] = []
for recipient in recipients:
if not states_share_interest(packet, recipient.last_transform):
filtered += 1
continue
try:
self._send_packet(recipient, packet, broadcast=True)
successful += 1
except (OSError, ValueError):
failed.append(recipient)
with self._lock:
self._stats["transformPacketsBroadcast"] += successful
self._stats["transformPacketsInterestFiltered"] += filtered
for recipient in failed:
self._disconnect_client(recipient)
def _broadcast_world_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
successful, failed = self._broadcast_to_active(packet, exclude=sender)
with self._lock:
self._stats["worldStatePacketsBroadcast"] += successful
for client in failed:
self._disconnect_client(client)
def _npc_packet_for_recipient(self, packet: dict[str, Any], recipient: ClientSession) -> dict[str, Any]:
if recipient.last_transform is None:
return dict(packet)
clean = dict(packet)
clean["npcs"] = [npc for npc in packet.get("npcs", []) if states_share_interest(npc, recipient.last_transform)]
return clean
def _broadcast_npc_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
with self._lock:
recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection]
successful = 0
failed: list[ClientSession] = []
for recipient in recipients:
scoped_packet = self._npc_packet_for_recipient(packet, recipient)
try:
self._send_packet(recipient, scoped_packet, broadcast=True)
successful += 1
except (OSError, ValueError):
failed.append(recipient)
with self._lock:
self._stats["npcStatePacketsBroadcast"] += successful
for client in failed:
self._disconnect_client(client)
def _send_existing_npc_state_to_client(self, client: ClientSession) -> None:
with self._lock:
snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None
if snapshot is None or snapshot.get("playerId") == client.player_id:
return
scoped = self._npc_packet_for_recipient(snapshot, client)
scoped["serverTime"] = time.time()
try:
self._send_packet(client, scoped, broadcast=True)
with self._lock:
self._stats["npcStatePacketsBroadcast"] += 1
except (OSError, ValueError):
self._disconnect_client(client)
def _route_combat_hit(self, sender: ClientSession, packet: dict[str, Any]) -> None:
target_player_id = packet["targetPlayerId"]
if target_player_id == sender.player_id:
self._reject_packet(sender, "Self-targeted combatHit", warning=False)
return
recipient = self._find_client_by_player_id(target_player_id)
if recipient is None:
self._reject_packet(sender, f"Combat target {target_player_id} is not connected", warning=False)
return
if not states_share_interest(sender.last_transform, recipient.last_transform):
self._reject_packet(sender, f"Combat target {target_player_id} is outside interest scope", warning=True)
return
try:
self._send_packet(recipient, packet)
with self._lock:
self._stats["combatHitsRouted"] += 1
except (OSError, ValueError):
self._disconnect_client(recipient)
def _broadcast_server_world_state(self) -> None:
with self._lock:
recipients = self._active_clients_locked()
snapshot = dict(self._server_world_state)
packets: list[dict[str, Any]] = []
if snapshot.get("timeHHmm"):
packets.append({"type": "serverWorldState", "timeHHmm": snapshot["timeHHmm"], "serverTime": time.time()})
if snapshot.get("weatherConsoleArg"):
weather_packet: dict[str, Any] = {
"type": "serverWorldState",
"weatherConsoleArg": snapshot["weatherConsoleArg"],
"serverTime": time.time(),
}
if snapshot.get("weatherFormId"):
weather_packet["weatherFormId"] = snapshot["weatherFormId"]
packets.append(weather_packet)
successful = 0
failed: list[ClientSession] = []
for packet in packets:
for recipient in recipients:
try:
self._send_packet(recipient, packet, broadcast=True)
successful += 1
except (OSError, ValueError):
failed.append(recipient)
with self._lock:
self._stats["serverWorldStatePacketsBroadcast"] += successful
for client in {failed_client.connection: failed_client for failed_client in failed}.values():
self._disconnect_client(client)
def _broadcast_to_active(self, packet: dict[str, Any], *, exclude: ClientSession | None = None) -> tuple[int, list[ClientSession]]:
with self._lock:
recipients = [
client for client in self._active_clients_locked()
if exclude is None or client.connection != exclude.connection
]
successful = 0
failed: list[ClientSession] = []
for recipient in recipients:
try:
self._send_packet(recipient, packet, broadcast=True)
successful += 1
except (OSError, ValueError):
failed.append(recipient)
return successful, failed
def _broadcast_world_state_host_assignment(self, player_id: int) -> None:
packet = {"type": "worldStateHost", "worldStateHostPlayerId": player_id, "serverTime": time.time()}
successful, failed = self._broadcast_to_active(packet)
with self._lock:
self._stats["worldStateHostPacketsBroadcast"] += successful
for client in failed:
self._disconnect_client(client)
def _reassign_world_state_host(self) -> None:
with self._lock:
active = self._active_clients_locked()
if not active:
self._world_state_host_player_id = None
self._last_npc_state = None
return
new_host = min(client.player_id for client in active)
self._world_state_host_player_id = new_host
self._last_npc_state = None
self._broadcast_world_state_host_assignment(new_host)
self._log(f"Reassigned world-state host to player {new_host}.")
def _broadcast_disconnect(self, client: ClientSession) -> None:
packet = {"type": "disconnect", "playerId": client.player_id, "serverTime": time.time()}
successful, failed = self._broadcast_to_active(packet)
with self._lock:
self._stats["disconnectPacketsBroadcast"] += successful
for recipient in failed:
self._disconnect_client(recipient)
def _remove_client(self, client: ClientSession) -> bool:
with self._lock:
removed = self._clients.pop(client.connection, None) is not None
if removed and client.gameplay_active:
self._stats["clientsDisconnected"] += 1
return removed
def _disconnect_client(self, client: ClientSession) -> None:
with self._lock:
was_active = client.gameplay_active
was_host = was_active and client.player_id == self._world_state_host_player_id
if not self._remove_client(client):
return
self._close_client_socket(client)
if not was_active:
self._log(f"Closed pending/probe connection: {client.label}", level="debug")
return
self._log(f"Client disconnected: {client.label} (player {client.player_id})")
self._broadcast_disconnect(client)
if was_host:
self._reassign_world_state_host()
def _find_client_by_player_id(self, player_id: int) -> ClientSession | None:
with self._lock:
return next((client for client in self._active_clients_locked() if client.player_id == player_id), None)
def _clients_for_ip(self, ip: str) -> list[ClientSession]:
normalized = str(ip).strip()
with self._lock:
return [client for client in self._clients.values() if client.address[0] == normalized]
def _build_session_ended_packet(self, code: str, reason: str = "") -> dict[str, Any]:
return {"type": "sessionEnded", "code": code, "reason": str(reason or ""), "serverTime": time.time()}
def _send_session_ended_raw(self, connection: socket.socket, *, code: str, reason: str = "") -> None:
packet = self._build_session_ended_packet(code, reason)
encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
try:
connection.settimeout(1.0)
connection.sendall(encoded)
with self._lock:
self._stats["sessionEndedPacketsSent"] += 1
self._stats["packetsSent"] += 1
except OSError:
pass
def _end_client_session(self, client: ClientSession, *, code: str, reason: str = "") -> None:
try:
self._send_packet(client, self._build_session_ended_packet(code, reason))
with self._lock:
self._stats["sessionEndedPacketsSent"] += 1
except (OSError, ValueError):
pass
self._disconnect_client(client)
def _end_sessions_for_ip(self, ip: str, *, code: str, reason: str = "") -> int:
clients = self._clients_for_ip(ip)
for client in clients:
self._end_client_session(client, code=code, reason=reason)
return len(clients)
def _close_raw_socket(self, connection: socket.socket) -> None:
try:
connection.close()
except OSError:
pass
def _close_client_socket(self, client: ClientSession) -> None:
self._close_raw_socket(client.connection)
def _reject_packet(self, client: ClientSession, reason: str, *, warning: bool = True) -> None:
with self._lock:
self._stats["packetsRejected"] += 1
self._log(
f"Rejected packet from {client.label} (player {client.player_id}): {reason}",
level="warning" if warning else "debug",
)
def _should_log(self, level: str) -> bool:
configured = _LOG_LEVELS.get(self.log_verbosity, _LOG_LEVELS[DEFAULT_LOG_VERBOSITY])
message_level = _LOG_LEVELS.get(level, _LOG_LEVELS["info"])
return message_level >= configured
def _log(self, message: str, *, level: str = "info") -> None:
if not self._should_log(level):
return
with self._log_lock:
listeners = list(self._log_listeners)
if not listeners:
return
for listener in listeners:
try:
try:
listener(message, level=level)
except TypeError:
listener(message)
except Exception:
pass
Reference in New Issue View Git Blame Copy Permalink