1462 lines
64 KiB
Python
1462 lines
64 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import math
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import time
|
|
from collections import deque
|
|
from collections.abc import Callable
|
|
from typing import Any
|
|
|
|
from ban_store import BanStore
|
|
from client_session import ClientSession
|
|
from lan_discovery import DISCOVERY_PORT, LanDiscoveryResponder
|
|
from npc_authority import NpcAuthorityManager, ScopeKey, scope_from_transform
|
|
from player_state import normalize_player_state_packet
|
|
from world_state_presets import normalize_fw_console_arg, relay_weather_form_id
|
|
|
|
HOST = "0.0.0.0"
|
|
PORT = 7777
|
|
ACCEPT_TIMEOUT_SECONDS = 0.5
|
|
DEFAULT_SERVER_NAME = "Commonwealth Online Server"
|
|
DEFAULT_MAX_PLAYERS = 16
|
|
DEFAULT_LOG_VERBOSITY = "info"
|
|
SESSION_ENDED_BANNED = "banned"
|
|
SESSION_ENDED_KICKED = "kicked"
|
|
SESSION_ENDED_FULL = "server_full"
|
|
SESSION_ENDED_PROTOCOL = "protocol_mismatch"
|
|
SESSION_ENDED_RATE_LIMITED = "rate_limited"
|
|
SESSION_ENDED_PACKET_TOO_LARGE = "packet_too_large"
|
|
|
|
PROTOCOL_VERSION = 2
|
|
LEGACY_PROTOCOL_VERSION = 1
|
|
MAX_PACKET_CHARS = 64 * 1024
|
|
CLIENT_IDLE_TIMEOUT_SECONDS = 60.0
|
|
CLIENT_HANDSHAKE_TIMEOUT_SECONDS = 10.0
|
|
MAX_ABS_COORDINATE = 10_000_000.0
|
|
MAX_MOVEMENT_SPEED = 100_000.0
|
|
MAX_ACTION_EVENTS = 16
|
|
MAX_NPCS_PER_PACKET = 64
|
|
MAX_PACKETS_PER_SECOND = 120
|
|
MAX_CONNECT_ATTEMPTS = 8
|
|
CONNECT_ATTEMPT_WINDOW_SECONDS = 10.0
|
|
EXTERIOR_INTEREST_RADIUS = 8192.0
|
|
MAX_NORMAL_MOVEMENT_SPEED = 2500.0
|
|
MOVEMENT_GRACE_DISTANCE = 512.0
|
|
MAX_MOVEMENT_VALIDATION_ELAPSED_SECONDS = 5.0
|
|
MOVEMENT_TRANSITION_TYPES = frozenset({
|
|
"teleport",
|
|
"cell_change",
|
|
"worldspace_change",
|
|
"load",
|
|
"spawn",
|
|
"fast_travel",
|
|
})
|
|
ALLOWED_MOVEMENT_TYPES = MOVEMENT_TRANSITION_TYPES | {"normal"}
|
|
|
|
_LOG_LEVELS = {
|
|
"debug": 10,
|
|
"info": 20,
|
|
"warning": 30,
|
|
"error": 40,
|
|
}
|
|
|
|
|
|
def _bounded_repr(value: Any, max_chars: int = 256) -> str:
|
|
text = repr(value)
|
|
return text if len(text) <= max_chars else text[:max_chars] + "..."
|
|
|
|
|
|
def _is_int(value: Any, minimum: int = 0, maximum: int = 0xFFFFFFFF) -> bool:
|
|
return isinstance(value, int) and not isinstance(value, bool) and minimum <= value <= maximum
|
|
|
|
|
|
def _is_finite_number(value: Any, minimum: float, maximum: float) -> bool:
|
|
return (
|
|
isinstance(value, (int, float))
|
|
and not isinstance(value, bool)
|
|
and math.isfinite(float(value))
|
|
and minimum <= float(value) <= maximum
|
|
)
|
|
|
|
|
|
def _is_hex_form_id(value: Any, *, allow_empty: bool = False, allow_zero: bool = True) -> bool:
|
|
if not isinstance(value, str):
|
|
return False
|
|
if allow_empty and value == "":
|
|
return True
|
|
if not 1 <= len(value) <= 8:
|
|
return False
|
|
if any(character not in "0123456789abcdefABCDEF" for character in value):
|
|
return False
|
|
parsed = int(value, 16)
|
|
return allow_zero or parsed != 0
|
|
|
|
|
|
def _normalize_hex_form_id(value: str) -> str:
|
|
return value.upper().zfill(8)
|
|
|
|
|
|
def _normalize_action_events(value: Any) -> list[dict[str, Any]]:
|
|
if not isinstance(value, list):
|
|
return []
|
|
allowed = {
|
|
(1, "meleeattackStart"),
|
|
(2, "meleeattackStart"),
|
|
(3, "fireSingle"),
|
|
}
|
|
normalized: list[dict[str, Any]] = []
|
|
for item in value[:MAX_ACTION_EVENTS]:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
sequence = item.get("sequence")
|
|
action_type = item.get("type")
|
|
event_name = item.get("eventName")
|
|
if not _is_int(sequence, 1) or not _is_int(action_type, 1, 3) or not isinstance(event_name, str):
|
|
continue
|
|
if (action_type, event_name) not in allowed:
|
|
continue
|
|
clean: dict[str, Any] = {
|
|
"sequence": sequence,
|
|
"type": action_type,
|
|
"eventName": event_name,
|
|
}
|
|
for name in ("actorStateFlags1", "actorStateFlags2"):
|
|
if _is_int(item.get(name)):
|
|
clean[name] = item[name]
|
|
normalized.append(clean)
|
|
return normalized
|
|
|
|
|
|
def _scope_from_state(state: dict[str, Any] | None) -> tuple[str, str, float, float] | None:
|
|
if not isinstance(state, dict):
|
|
return None
|
|
cell_id = state.get("cellId")
|
|
worldspace_id = state.get("worldspaceId", "")
|
|
x = state.get("x")
|
|
y = state.get("y")
|
|
if not _is_hex_form_id(cell_id, allow_zero=False):
|
|
return None
|
|
if not _is_hex_form_id(worldspace_id, allow_empty=True, allow_zero=True):
|
|
return None
|
|
if not _is_finite_number(x, -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE):
|
|
return None
|
|
if not _is_finite_number(y, -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE):
|
|
return None
|
|
return (
|
|
_normalize_hex_form_id(cell_id),
|
|
_normalize_hex_form_id(worldspace_id) if worldspace_id else "",
|
|
float(x),
|
|
float(y),
|
|
)
|
|
|
|
|
|
def states_share_interest(a: dict[str, Any] | None, b: dict[str, Any] | None) -> bool:
|
|
a_scope = _scope_from_state(a)
|
|
b_scope = _scope_from_state(b)
|
|
if a_scope is None or b_scope is None:
|
|
return True
|
|
a_cell, a_world, ax, ay = a_scope
|
|
b_cell, b_world, bx, by = b_scope
|
|
if a_cell == b_cell:
|
|
return True
|
|
if not a_world or a_world != b_world:
|
|
return False
|
|
return math.hypot(ax - bx, ay - by) <= EXTERIOR_INTEREST_RADIUS
|
|
|
|
|
|
def get_lan_addresses() -> list[str]:
|
|
addresses: list[str] = []
|
|
try:
|
|
with socket.socket(socket.AF_INET, socket.SOCK_DGRAM) as probe:
|
|
probe.connect(("8.8.8.8", 80))
|
|
primary = probe.getsockname()[0]
|
|
if primary and not primary.startswith("127."):
|
|
addresses.append(primary)
|
|
except OSError:
|
|
pass
|
|
try:
|
|
hostname = socket.gethostname()
|
|
for info in socket.getaddrinfo(hostname, None, family=socket.AF_INET):
|
|
candidate = info[4][0]
|
|
if candidate and not candidate.startswith("127.") and candidate not in addresses:
|
|
addresses.append(candidate)
|
|
except OSError:
|
|
pass
|
|
return addresses
|
|
|
|
|
|
class FalloutTogetherServer:
|
|
def __init__(
|
|
self,
|
|
host: str = HOST,
|
|
port: int = PORT,
|
|
server_name: str = DEFAULT_SERVER_NAME,
|
|
server_description: str = "",
|
|
max_players: int = DEFAULT_MAX_PLAYERS,
|
|
bans_path: str | None = None,
|
|
log_verbosity: str = DEFAULT_LOG_VERBOSITY,
|
|
) -> None:
|
|
self.host = host
|
|
self.port = port
|
|
self.server_name = server_name or DEFAULT_SERVER_NAME
|
|
self.server_description = server_description or ""
|
|
self.max_players = max_players if max_players >= 1 else DEFAULT_MAX_PLAYERS
|
|
verbosity = str(log_verbosity or DEFAULT_LOG_VERBOSITY).strip().lower()
|
|
self.log_verbosity = verbosity if verbosity in _LOG_LEVELS else DEFAULT_LOG_VERBOSITY
|
|
self._ban_store = BanStore(bans_path)
|
|
|
|
self._lock = threading.RLock()
|
|
self._log_lock = threading.RLock()
|
|
self._clients: dict[socket.socket, ClientSession] = {}
|
|
self._client_threads: set[threading.Thread] = set()
|
|
self._log_listeners: list[Callable[..., None]] = []
|
|
self._server_socket: socket.socket | None = None
|
|
self._accept_thread: threading.Thread | None = None
|
|
self._discovery: LanDiscoveryResponder | None = None
|
|
self._running = False
|
|
self._started_at: float | None = None
|
|
self._next_player_id = 1
|
|
self._world_state_host_player_id: int | None = None
|
|
self._server_world_state: dict[str, str] = {}
|
|
self._last_npc_state: dict[str, Any] | None = None
|
|
self._last_npc_state_by_scope: dict[ScopeKey, dict[str, Any]] = {}
|
|
self._last_player_state_by_player_id: dict[int, dict[str, Any]] = {}
|
|
self._npc_authority = NpcAuthorityManager()
|
|
self._connect_attempts: dict[str, deque[float]] = {}
|
|
|
|
self._stats: dict[str, int] = {
|
|
"clientsConnected": 0,
|
|
"clientsDisconnected": 0,
|
|
"pendingConnectionsRejected": 0,
|
|
"packetsReceived": 0,
|
|
"packetsSent": 0,
|
|
"packetsBroadcast": 0,
|
|
"packetsRejected": 0,
|
|
"rateLimitedPackets": 0,
|
|
"transformPacketsReceived": 0,
|
|
"transformPacketsBroadcast": 0,
|
|
"transformPacketsInterestFiltered": 0,
|
|
"playerStatePacketsReceived": 0,
|
|
"playerStatePacketsBroadcast": 0,
|
|
"movementPacketsRejected": 0,
|
|
"movementCorrectionsSent": 0,
|
|
"worldStatePacketsReceived": 0,
|
|
"worldStatePacketsBroadcast": 0,
|
|
"npcStatePacketsReceived": 0,
|
|
"npcStatePacketsBroadcast": 0,
|
|
"npcAuthorityChanges": 0,
|
|
"npcAuthorityRejects": 0,
|
|
"combatHitsReceived": 0,
|
|
"combatHitsRouted": 0,
|
|
"worldStateHostPacketsBroadcast": 0,
|
|
"serverWorldStatePacketsBroadcast": 0,
|
|
"sessionEndedPacketsSent": 0,
|
|
"bannedConnectionsRejected": 0,
|
|
"disconnectPacketsBroadcast": 0,
|
|
"protocolV2Connections": 0,
|
|
"legacyConnections": 0,
|
|
}
|
|
|
|
def start(self) -> None:
|
|
with self._lock:
|
|
if self._running:
|
|
return
|
|
self._prepare_server_socket()
|
|
self._start_discovery()
|
|
self._accept_thread = threading.Thread(target=self._accept_loop, daemon=True)
|
|
self._accept_thread.start()
|
|
|
|
def serve_forever(self) -> None:
|
|
with self._lock:
|
|
if self._running:
|
|
raise RuntimeError("Server is already running.")
|
|
self._prepare_server_socket()
|
|
self._start_discovery()
|
|
self._accept_loop()
|
|
|
|
def stop(self) -> None:
|
|
with self._lock:
|
|
self._running = False
|
|
server_socket = self._server_socket
|
|
self._server_socket = None
|
|
clients = list(self._clients.values())
|
|
client_threads = [t for t in self._client_threads if t is not threading.current_thread()]
|
|
|
|
discovery = self._discovery
|
|
self._discovery = None
|
|
if discovery is not None:
|
|
discovery.stop()
|
|
if server_socket is not None:
|
|
try:
|
|
server_socket.close()
|
|
except OSError:
|
|
pass
|
|
for client in clients:
|
|
self._close_client_socket(client)
|
|
accept_thread = self._accept_thread
|
|
if accept_thread is not None and accept_thread is not threading.current_thread():
|
|
accept_thread.join(timeout=1.0)
|
|
self._accept_thread = None
|
|
for thread in client_threads:
|
|
thread.join(timeout=1.0)
|
|
with self._lock:
|
|
self._client_threads.clear()
|
|
|
|
def is_running(self) -> bool:
|
|
with self._lock:
|
|
return self._running
|
|
|
|
def _active_clients_locked(self) -> list[ClientSession]:
|
|
return [client for client in self._clients.values() if client.gameplay_active]
|
|
|
|
def get_clients(self) -> list[dict[str, Any]]:
|
|
with self._lock:
|
|
clients = self._active_clients_locked()
|
|
return [client.to_snapshot() for client in clients]
|
|
|
|
def get_stats(self) -> dict[str, Any]:
|
|
with self._lock:
|
|
active_clients = self._active_clients_locked()
|
|
pending = len(self._clients) - len(active_clients)
|
|
started_at = self._started_at
|
|
stats = dict(self._stats)
|
|
stats.update(
|
|
{
|
|
"host": self.host,
|
|
"port": self.port,
|
|
"serverName": self.server_name,
|
|
"serverDescription": self.server_description,
|
|
"maxPlayers": self.max_players,
|
|
"lanAddresses": get_lan_addresses() if self._running and self.host == "0.0.0.0" else [],
|
|
"isRunning": self._running,
|
|
"startedAt": started_at,
|
|
"uptimeSeconds": time.time() - started_at if started_at is not None else 0.0,
|
|
"connectedClients": len(active_clients),
|
|
"pendingConnections": pending,
|
|
"nextPlayerId": self._next_player_id,
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"npcAuthorityScopes": len(self._npc_authority.assignments()),
|
|
}
|
|
)
|
|
return stats
|
|
|
|
def get_server_world_state(self) -> dict[str, str]:
|
|
with self._lock:
|
|
return dict(self._server_world_state)
|
|
|
|
def set_server_time(self, hhmm: str) -> bool:
|
|
text = str(hhmm).strip()
|
|
if not text.isdigit() or len(text) > 4:
|
|
self._log(f"Rejected invalid server time HHmm value: {hhmm!r}")
|
|
return False
|
|
normalized = text.zfill(4)
|
|
hours = int(normalized[:2])
|
|
minutes = int(normalized[2:])
|
|
if hours > 23 or minutes > 59:
|
|
self._log(f"Rejected out-of-range server time HHmm value: {hhmm!r}")
|
|
return False
|
|
with self._lock:
|
|
self._server_world_state["timeHHmm"] = normalized
|
|
self._broadcast_server_world_state()
|
|
return True
|
|
|
|
def set_server_weather(self, fw_console_arg: str) -> bool:
|
|
try:
|
|
normalized_fw_arg = normalize_fw_console_arg(fw_console_arg)
|
|
except ValueError:
|
|
self._log(f"Rejected invalid server weather console id: {fw_console_arg!r}")
|
|
return False
|
|
with self._lock:
|
|
self._server_world_state["weatherConsoleArg"] = normalized_fw_arg
|
|
self._server_world_state["weatherFormId"] = relay_weather_form_id(normalized_fw_arg)
|
|
self._broadcast_server_world_state()
|
|
return True
|
|
|
|
def list_bans(self) -> list[dict[str, Any]]:
|
|
return [
|
|
{"ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at}
|
|
for entry in self._ban_store.list_bans()
|
|
]
|
|
|
|
def unban_ip(self, ip: str) -> bool:
|
|
normalized = str(ip).strip()
|
|
removed = self._ban_store.unban_ip(normalized)
|
|
if removed:
|
|
self._log(f"Unbanned IP {normalized}")
|
|
return removed
|
|
|
|
def ban_ip(self, ip: str, reason: str = "") -> dict[str, Any]:
|
|
normalized = str(ip).strip()
|
|
if not normalized:
|
|
raise ValueError("IP address cannot be empty.")
|
|
entry = self._ban_store.ban_ip(normalized, reason=reason)
|
|
ended = self._end_sessions_for_ip(normalized, code=SESSION_ENDED_BANNED, reason=entry.reason)
|
|
self._log(f"Banned IP {entry.ip}" + (f" (reason: {entry.reason})" if entry.reason else ""))
|
|
return {"ip": entry.ip, "reason": entry.reason, "bannedAt": entry.banned_at, "sessionsEnded": ended}
|
|
|
|
def ban_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
|
|
client = self._find_client_by_player_id(player_id)
|
|
if client is None:
|
|
raise KeyError(f"No connected player with id {player_id}")
|
|
return self.ban_ip(client.address[0], reason=reason)
|
|
|
|
def kick_player(self, player_id: int, reason: str = "") -> dict[str, Any]:
|
|
client = self._find_client_by_player_id(player_id)
|
|
if client is None:
|
|
raise KeyError(f"No connected player with id {player_id}")
|
|
ip = client.address[0]
|
|
self._end_client_session(client, code=SESSION_ENDED_KICKED, reason=reason)
|
|
self._log(f"Kicked player {player_id} ({client.label})" + (f" (reason: {reason})" if reason else ""))
|
|
return {"playerId": player_id, "ip": ip, "code": SESSION_ENDED_KICKED, "reason": str(reason or "")}
|
|
|
|
def add_log_listener(self, callback: Callable[..., None]) -> None:
|
|
with self._log_lock:
|
|
if callback not in self._log_listeners:
|
|
self._log_listeners.append(callback)
|
|
|
|
def remove_log_listener(self, callback: Callable[..., None]) -> None:
|
|
with self._log_lock:
|
|
if callback in self._log_listeners:
|
|
self._log_listeners.remove(callback)
|
|
|
|
def _prepare_server_socket(self) -> None:
|
|
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
try:
|
|
if hasattr(socket, "SO_EXCLUSIVEADDRUSE"):
|
|
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_EXCLUSIVEADDRUSE, 1)
|
|
else:
|
|
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server_socket.bind((self.host, self.port))
|
|
server_socket.listen()
|
|
server_socket.settimeout(ACCEPT_TIMEOUT_SECONDS)
|
|
except OSError as error:
|
|
server_socket.close()
|
|
raise OSError(
|
|
f"Could not bind the server to {self.host}:{self.port}. The port may already be in use or unavailable. ({error})"
|
|
) from error
|
|
|
|
if self.port == 0:
|
|
self.port = int(server_socket.getsockname()[1])
|
|
self._server_socket = server_socket
|
|
self._running = True
|
|
self._started_at = time.time()
|
|
self._next_player_id = 1
|
|
self._world_state_host_player_id = None
|
|
self._last_npc_state = None
|
|
self._last_npc_state_by_scope.clear()
|
|
self._last_player_state_by_player_id.clear()
|
|
self._npc_authority.clear()
|
|
self._connect_attempts.clear()
|
|
|
|
self._log(f"Commonwealth Online server listening on {self.host}:{self.port}")
|
|
self._log(f"Protocol v{PROTOCOL_VERSION} negotiation enabled; legacy clients remain temporarily compatible.")
|
|
if self.host == "0.0.0.0":
|
|
self._log("Bind address 0.0.0.0 means all interfaces; clients should not join 0.0.0.0.")
|
|
self._log(f"Local clients can connect at 127.0.0.1:{self.port}")
|
|
for address in get_lan_addresses():
|
|
self._log(f"LAN clients can connect at {address}:{self.port}")
|
|
if sys.platform == "win32":
|
|
self._log(f"Allow TCP {self.port} through Windows Firewall and forward it only when hosting externally.")
|
|
else:
|
|
self._log(f"Allow TCP {self.port} through the host firewall; router forwarding is only needed externally.")
|
|
else:
|
|
self._log(f"Clients can connect at {self.host}:{self.port}")
|
|
|
|
def _start_discovery(self) -> None:
|
|
discovery = LanDiscoveryResponder(self)
|
|
try:
|
|
discovery.start()
|
|
except OSError as error:
|
|
self._log(f"LAN discovery unavailable on UDP {DISCOVERY_PORT}: {error}. Direct connections still work.", level="warning")
|
|
self._discovery = None
|
|
return
|
|
self._discovery = discovery
|
|
self._log(f"LAN discovery listening on UDP port {DISCOVERY_PORT}")
|
|
|
|
def _allow_connect_attempt(self, ip: str) -> bool:
|
|
now = time.monotonic()
|
|
cutoff = now - CONNECT_ATTEMPT_WINDOW_SECONDS
|
|
with self._lock:
|
|
attempts = self._connect_attempts.setdefault(ip, deque())
|
|
while attempts and attempts[0] < cutoff:
|
|
attempts.popleft()
|
|
attempts.append(now)
|
|
return len(attempts) <= MAX_CONNECT_ATTEMPTS
|
|
|
|
def _accept_loop(self) -> None:
|
|
try:
|
|
while self.is_running():
|
|
with self._lock:
|
|
server_socket = self._server_socket
|
|
if server_socket is None:
|
|
break
|
|
try:
|
|
connection, address = server_socket.accept()
|
|
except socket.timeout:
|
|
continue
|
|
except OSError:
|
|
if self.is_running():
|
|
self._log("Server socket closed unexpectedly.", level="warning")
|
|
break
|
|
|
|
if not self._allow_connect_attempt(address[0]):
|
|
with self._lock:
|
|
self._stats["pendingConnectionsRejected"] += 1
|
|
self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many connection attempts.")
|
|
self._close_raw_socket(connection)
|
|
continue
|
|
|
|
thread = threading.Thread(target=self._handle_client, args=(connection, address), daemon=True)
|
|
with self._lock:
|
|
self._client_threads.add(thread)
|
|
thread.start()
|
|
finally:
|
|
with self._lock:
|
|
server_socket = self._server_socket
|
|
self._running = False
|
|
self._server_socket = None
|
|
if server_socket is not None:
|
|
try:
|
|
server_socket.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def _assign_client(self, connection: socket.socket, address: tuple[str, int]) -> ClientSession | None:
|
|
with self._lock:
|
|
max_pending = max(16, self.max_players * 2)
|
|
pending_count = sum(1 for client in self._clients.values() if not client.gameplay_active)
|
|
if pending_count >= max_pending:
|
|
return None
|
|
client = ClientSession(
|
|
connection=connection,
|
|
address=address,
|
|
player_id=self._next_player_id,
|
|
connected_at=time.time(),
|
|
)
|
|
self._next_player_id += 1
|
|
self._clients[connection] = client
|
|
return client
|
|
|
|
def _activate_client(self, client: ClientSession, protocol_version: int) -> bool:
|
|
with self._lock:
|
|
if client.gameplay_active:
|
|
return True
|
|
active_count = len(self._active_clients_locked())
|
|
if active_count >= self.max_players:
|
|
return False
|
|
if not client.mark_gameplay_active(protocol_version):
|
|
return True
|
|
self._stats["clientsConnected"] += 1
|
|
if protocol_version >= PROTOCOL_VERSION:
|
|
self._stats["protocolV2Connections"] += 1
|
|
else:
|
|
self._stats["legacyConnections"] += 1
|
|
if self._world_state_host_player_id is None:
|
|
self._world_state_host_player_id = client.player_id
|
|
became_host = True
|
|
else:
|
|
became_host = False
|
|
|
|
self._log(f"Client connected: {client.label} (player {client.player_id}, protocol {protocol_version})")
|
|
self._send_packet(
|
|
client,
|
|
{
|
|
"type": "sessionReady",
|
|
"playerId": client.player_id,
|
|
"protocolVersion": protocol_version,
|
|
"serverProtocolVersion": PROTOCOL_VERSION,
|
|
"worldStateHostPlayerId": self._world_state_host_player_id,
|
|
"serverTime": time.time(),
|
|
},
|
|
)
|
|
self._send_existing_transforms_to_client(client)
|
|
self._send_existing_player_states_to_client(client)
|
|
self._send_existing_npc_state_to_client(client)
|
|
if became_host:
|
|
self._broadcast_world_state_host_assignment(client.player_id)
|
|
return True
|
|
|
|
def _handle_client(self, connection: socket.socket, address: tuple[str, int]) -> None:
|
|
peer_ip = address[0]
|
|
ban_entry = self._ban_store.get_ban(peer_ip)
|
|
if ban_entry is not None:
|
|
with self._lock:
|
|
self._stats["bannedConnectionsRejected"] += 1
|
|
self._send_session_ended_raw(connection, code=SESSION_ENDED_BANNED, reason=ban_entry.reason)
|
|
self._close_raw_socket(connection)
|
|
return
|
|
|
|
client = self._assign_client(connection, address)
|
|
if client is None:
|
|
self._send_session_ended_raw(connection, code=SESSION_ENDED_RATE_LIMITED, reason="Too many pending connections.")
|
|
self._close_raw_socket(connection)
|
|
return
|
|
|
|
connection.settimeout(1.0)
|
|
connected_mono = time.monotonic()
|
|
self._init_rate_state(client, connected_mono)
|
|
self._log(f"TCP accept: {client.label} (provisional player {client.player_id})", level="debug")
|
|
|
|
try:
|
|
with connection:
|
|
self._send_packet(
|
|
client,
|
|
{
|
|
"type": "welcome",
|
|
"playerId": client.player_id,
|
|
"serverTime": time.time(),
|
|
"serverName": self.server_name,
|
|
"serverDescription": self.server_description,
|
|
"protocolVersion": PROTOCOL_VERSION,
|
|
"capabilities": [
|
|
"interest-v1",
|
|
"hello-v2",
|
|
"bounded-framing",
|
|
"rate-limit-v1",
|
|
"movement-correction-v1",
|
|
"npc-authority-epoch-v1",
|
|
"player-state-v1",
|
|
],
|
|
},
|
|
)
|
|
buffer = b""
|
|
while self.is_running():
|
|
now_mono = time.monotonic()
|
|
if not client.gameplay_active and now_mono - connected_mono > CLIENT_HANDSHAKE_TIMEOUT_SECONDS:
|
|
self._log(f"Handshake timeout: {client.label}", level="warning")
|
|
break
|
|
if client.gameplay_active and client.last_packet_at is not None and time.time() - client.last_packet_at > CLIENT_IDLE_TIMEOUT_SECONDS:
|
|
self._log(f"Idle timeout: player {client.player_id}", level="warning")
|
|
break
|
|
try:
|
|
chunk = connection.recv(4096)
|
|
except socket.timeout:
|
|
continue
|
|
if not chunk:
|
|
break
|
|
buffer += chunk
|
|
if len(buffer) > MAX_PACKET_CHARS and b"\n" not in buffer:
|
|
self._end_client_session(client, code=SESSION_ENDED_PACKET_TOO_LARGE, reason="Packet exceeded maximum line size.")
|
|
break
|
|
while b"\n" in buffer:
|
|
line_bytes, buffer = buffer.split(b"\n", 1)
|
|
if len(line_bytes) > MAX_PACKET_CHARS:
|
|
self._end_client_session(client, code=SESSION_ENDED_PACKET_TOO_LARGE, reason="Packet exceeded maximum line size.")
|
|
return
|
|
try:
|
|
line = line_bytes.decode("utf-8", errors="strict").strip()
|
|
except UnicodeDecodeError:
|
|
self._reject_packet(client, "Packet is not valid UTF-8")
|
|
continue
|
|
if not self._handle_line(client, line):
|
|
if client.connection.fileno() < 0:
|
|
return
|
|
except (ConnectionResetError, BrokenPipeError):
|
|
pass
|
|
except OSError as error:
|
|
if self.is_running():
|
|
self._log(f"Client connection error: {client.label}: {error}", level="debug")
|
|
finally:
|
|
self._disconnect_client(client)
|
|
with self._lock:
|
|
self._client_threads.discard(threading.current_thread())
|
|
|
|
def _init_rate_state(self, client: ClientSession, now: float) -> None:
|
|
client._rate_window_start = now
|
|
client._rate_window_count = 0
|
|
client._rate_violations = 0
|
|
client._rate_window_blocked = False
|
|
client._last_combat_sequence = 0
|
|
|
|
def _allow_packet(self, client: ClientSession) -> bool:
|
|
now = time.monotonic()
|
|
start = getattr(client, "_rate_window_start", now)
|
|
if now - start >= 1.0:
|
|
previous_count = getattr(client, "_rate_window_count", 0)
|
|
violations = getattr(client, "_rate_violations", 0)
|
|
if previous_count <= MAX_PACKETS_PER_SECOND:
|
|
violations = max(0, violations - 1)
|
|
client._rate_violations = violations
|
|
client._rate_window_start = now
|
|
client._rate_window_count = 0
|
|
client._rate_window_blocked = False
|
|
client._rate_window_count += 1
|
|
if client._rate_window_count <= MAX_PACKETS_PER_SECOND:
|
|
return True
|
|
with self._lock:
|
|
self._stats["rateLimitedPackets"] += 1
|
|
if not client._rate_window_blocked:
|
|
client._rate_window_blocked = True
|
|
client._rate_violations += 1
|
|
self._log(f"Packet-rate limit exceeded by player {client.player_id} ({client._rate_violations}/3 windows)", level="warning")
|
|
if client._rate_violations >= 3:
|
|
self._end_client_session(client, code=SESSION_ENDED_RATE_LIMITED, reason="Sustained packet-rate limit exceeded.")
|
|
return False
|
|
|
|
def _handle_line(self, client: ClientSession, line: str) -> bool:
|
|
if not line:
|
|
return True
|
|
if not self._allow_packet(client):
|
|
return False
|
|
|
|
client.record_received(time.time())
|
|
with self._lock:
|
|
self._stats["packetsReceived"] += 1
|
|
|
|
try:
|
|
packet = json.loads(line, parse_constant=lambda value: (_ for _ in ()).throw(ValueError(value)))
|
|
except (json.JSONDecodeError, ValueError) as error:
|
|
self._reject_packet(client, f"Invalid JSON: {error}")
|
|
return False
|
|
if not isinstance(packet, dict):
|
|
self._reject_packet(client, "Packet must be a JSON object")
|
|
return False
|
|
|
|
packet_type = packet.get("type")
|
|
if packet_type == "hello":
|
|
return self._handle_hello(client, packet)
|
|
if packet_type == "keepAlive":
|
|
return True
|
|
|
|
if not client.gameplay_active:
|
|
if packet_type not in {"transform", "worldState", "npcState", "combatHit"}:
|
|
self._reject_packet(client, "Gameplay packet received before session activation")
|
|
return False
|
|
if not self._activate_client(client, LEGACY_PROTOCOL_VERSION):
|
|
self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.")
|
|
return False
|
|
self._log(f"Legacy client player {client.player_id} activated without hello; client upgrade recommended.", level="warning")
|
|
|
|
if packet_type == "transform":
|
|
normalized = self._normalize_transform_packet(packet)
|
|
if normalized is None:
|
|
self._reject_packet(client, "Malformed transform")
|
|
return False
|
|
if client.protocol_version >= PROTOCOL_VERSION:
|
|
for reliable_field in ("equippedItems", "appearance", "actionEvents", "characterName"):
|
|
normalized.pop(reliable_field, None)
|
|
accepted_monotonic = time.monotonic()
|
|
movement_valid, movement_reason = self._validate_transform_movement(client, normalized, accepted_monotonic)
|
|
if not movement_valid:
|
|
with self._lock:
|
|
self._stats["movementPacketsRejected"] += 1
|
|
self._reject_packet(client, movement_reason)
|
|
self._send_position_correction(client, movement_reason)
|
|
return False
|
|
previous_scope = scope_from_transform(client.last_transform)
|
|
normalized["playerId"] = client.player_id
|
|
normalized["serverTime"] = time.time()
|
|
client.record_transform(normalized, accepted_monotonic)
|
|
with self._lock:
|
|
self._stats["transformPacketsReceived"] += 1
|
|
self._broadcast_transform(client, normalized)
|
|
self._reconcile_npc_authority()
|
|
current_scope = scope_from_transform(normalized)
|
|
if current_scope is not None and current_scope != previous_scope:
|
|
self._send_npc_authority_for_client(client, current_scope)
|
|
return True
|
|
|
|
if packet_type == "playerState":
|
|
if client.protocol_version < PROTOCOL_VERSION:
|
|
self._reject_packet(client, "playerState requires Protocol V2", warning=False)
|
|
return False
|
|
normalized = normalize_player_state_packet(packet, _normalize_action_events)
|
|
if normalized is None:
|
|
self._reject_packet(client, "Malformed playerState")
|
|
return False
|
|
normalized["playerId"] = client.player_id
|
|
normalized["serverTime"] = time.time()
|
|
durable_fields = ("equippedItems", "appearance", "characterName")
|
|
with self._lock:
|
|
if any(field in normalized for field in durable_fields):
|
|
cached = dict(self._last_player_state_by_player_id.get(client.player_id, {}))
|
|
cached.update({"type": "playerState", "playerId": client.player_id, "serverTime": normalized["serverTime"]})
|
|
for field in durable_fields:
|
|
if field in normalized:
|
|
cached[field] = normalized[field]
|
|
cached.pop("actionEvents", None)
|
|
self._last_player_state_by_player_id[client.player_id] = cached
|
|
self._stats["playerStatePacketsReceived"] += 1
|
|
self._broadcast_player_state(client, normalized)
|
|
return True
|
|
|
|
if packet_type == "worldState":
|
|
if not self._is_world_host(client):
|
|
self._reject_packet(client, "worldState from non-authority client", warning=False)
|
|
return False
|
|
normalized = self._normalize_world_state_packet(packet)
|
|
if normalized is None:
|
|
self._reject_packet(client, "Malformed worldState")
|
|
return False
|
|
normalized["playerId"] = client.player_id
|
|
normalized["serverTime"] = time.time()
|
|
with self._lock:
|
|
self._stats["worldStatePacketsReceived"] += 1
|
|
self._broadcast_world_state(client, normalized)
|
|
return True
|
|
|
|
if packet_type == "npcState":
|
|
normalized = self._normalize_npc_state_packet(packet)
|
|
if normalized is None:
|
|
self._reject_packet(client, "Malformed npcState")
|
|
return False
|
|
|
|
if client.protocol_version >= PROTOCOL_VERSION:
|
|
scope = self._npc_scope_from_packet(normalized)
|
|
epoch = normalized.get("authorityEpoch")
|
|
if scope is None or not _is_int(epoch, 1):
|
|
with self._lock:
|
|
self._stats["npcAuthorityRejects"] += 1
|
|
self._reject_packet(client, "Protocol V2 npcState missing valid authority scope/epoch")
|
|
return False
|
|
if not self._npc_authority.authorize(client.player_id, scope, epoch):
|
|
with self._lock:
|
|
self._stats["npcAuthorityRejects"] += 1
|
|
self._reject_packet(client, "Stale or unauthorized npcState authority epoch")
|
|
return False
|
|
if any(
|
|
npc.get("cellId") != scope.cell_id or npc.get("worldspaceId", "") != scope.worldspace_id
|
|
for npc in normalized.get("npcs", [])
|
|
):
|
|
with self._lock:
|
|
self._stats["npcAuthorityRejects"] += 1
|
|
self._reject_packet(client, "npcState contains NPCs outside declared authority scope")
|
|
return False
|
|
normalized["authorityCellId"] = scope.cell_id
|
|
normalized["authorityWorldspaceId"] = scope.worldspace_id
|
|
with self._lock:
|
|
self._last_npc_state_by_scope[scope] = dict(normalized)
|
|
else:
|
|
if not self._is_world_host(client):
|
|
self._reject_packet(client, "Legacy npcState from non-authority client", warning=False)
|
|
return False
|
|
with self._lock:
|
|
self._last_npc_state = dict(normalized)
|
|
|
|
normalized["playerId"] = client.player_id
|
|
normalized["serverTime"] = time.time()
|
|
normalized["fullReplace"] = True
|
|
with self._lock:
|
|
self._stats["npcStatePacketsReceived"] += 1
|
|
self._broadcast_npc_state(client, normalized)
|
|
return True
|
|
|
|
if packet_type == "combatHit":
|
|
normalized = self._normalize_combat_hit_packet(packet)
|
|
if normalized is None:
|
|
self._reject_packet(client, "Malformed combatHit")
|
|
return False
|
|
sequence = normalized["sequence"]
|
|
if sequence <= getattr(client, "_last_combat_sequence", 0):
|
|
self._reject_packet(client, "Duplicate or out-of-order combat sequence", warning=False)
|
|
return False
|
|
client._last_combat_sequence = sequence
|
|
normalized["playerId"] = client.player_id
|
|
normalized["serverTime"] = time.time()
|
|
with self._lock:
|
|
self._stats["combatHitsReceived"] += 1
|
|
self._route_combat_hit(client, normalized)
|
|
return True
|
|
|
|
self._reject_packet(client, f"Unknown packet type: {_bounded_repr(packet_type)}")
|
|
return False
|
|
|
|
def _handle_hello(self, client: ClientSession, packet: dict[str, Any]) -> bool:
|
|
version = packet.get("protocolVersion")
|
|
if not _is_int(version, 1, 0xFFFF) or version != PROTOCOL_VERSION:
|
|
self._end_client_session(client, code=SESSION_ENDED_PROTOCOL, reason=f"Server requires protocol {PROTOCOL_VERSION}.")
|
|
return False
|
|
if not self._activate_client(client, version):
|
|
self._end_client_session(client, code=SESSION_ENDED_FULL, reason="Server is full.")
|
|
return False
|
|
return True
|
|
|
|
def _normalize_transform_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
|
|
for field_name in ("x", "y", "z", "angleZ"):
|
|
if not _is_finite_number(packet.get(field_name), -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE):
|
|
return None
|
|
if not _is_hex_form_id(packet.get("cellId"), allow_zero=False):
|
|
return None
|
|
worldspace_id = packet.get("worldspaceId", "")
|
|
if not _is_hex_form_id(worldspace_id, allow_empty=True, allow_zero=True):
|
|
return None
|
|
|
|
movement_type = packet.get("movementType", "normal")
|
|
if not isinstance(movement_type, str) or movement_type not in ALLOWED_MOVEMENT_TYPES:
|
|
return None
|
|
|
|
normalized = dict(packet)
|
|
normalized["x"] = float(packet["x"])
|
|
normalized["y"] = float(packet["y"])
|
|
normalized["z"] = float(packet["z"])
|
|
normalized["angleZ"] = float(packet["angleZ"])
|
|
normalized["cellId"] = _normalize_hex_form_id(packet["cellId"])
|
|
normalized["worldspaceId"] = _normalize_hex_form_id(worldspace_id) if worldspace_id else ""
|
|
normalized["movementType"] = movement_type
|
|
for field_name in ("movementSpeed", "animationGraphSpeed"):
|
|
if field_name in packet:
|
|
value = packet[field_name]
|
|
if not _is_finite_number(value, -1.0 if field_name == "animationGraphSpeed" else 0.0, MAX_MOVEMENT_SPEED):
|
|
normalized.pop(field_name, None)
|
|
else:
|
|
normalized[field_name] = float(value)
|
|
for field_name, minimum, maximum in (
|
|
("animationDirection", -360.0, 360.0),
|
|
("aimPitch", -180.0, 180.0),
|
|
("turnDelta", -10000.0, 10000.0),
|
|
):
|
|
if field_name in packet:
|
|
if not _is_finite_number(packet[field_name], minimum, maximum):
|
|
normalized.pop(field_name, None)
|
|
else:
|
|
normalized[field_name] = float(packet[field_name])
|
|
for field_name in ("isMoving", "isSprinting", "isSneaking", "isJumping", "isCrouching", "weaponDrawn"):
|
|
if field_name in packet and not isinstance(packet[field_name], bool):
|
|
normalized.pop(field_name, None)
|
|
for field_name in ("actorStateFlags1", "actorStateFlags2"):
|
|
if field_name in packet and not _is_int(packet[field_name]):
|
|
normalized.pop(field_name, None)
|
|
if "actionEvents" in packet:
|
|
normalized["actionEvents"] = _normalize_action_events(packet["actionEvents"])
|
|
return normalized
|
|
|
|
def _validate_transform_movement(self, client: ClientSession, packet: dict[str, Any], now_monotonic: float) -> tuple[bool, str]:
|
|
previous, previous_monotonic = client.get_last_transform_anchor()
|
|
if previous is None or previous_monotonic is None:
|
|
return True, "first transform"
|
|
movement_type = packet.get("movementType", "normal")
|
|
if movement_type in MOVEMENT_TRANSITION_TYPES:
|
|
return True, f"explicit {movement_type} transition"
|
|
previous_cell = previous.get("cellId", "")
|
|
current_cell = packet.get("cellId", "")
|
|
previous_world = previous.get("worldspaceId", "")
|
|
current_world = packet.get("worldspaceId", "")
|
|
if previous_cell != current_cell or previous_world != current_world:
|
|
return False, "scope changed without an explicit movement transition"
|
|
elapsed = now_monotonic - previous_monotonic
|
|
if not math.isfinite(elapsed) or elapsed < 0.0:
|
|
elapsed = 0.0
|
|
elapsed = min(elapsed, MAX_MOVEMENT_VALIDATION_ELAPSED_SECONDS)
|
|
dx = packet["x"] - previous["x"]
|
|
dy = packet["y"] - previous["y"]
|
|
dz = packet["z"] - previous["z"]
|
|
distance = math.sqrt((dx * dx) + (dy * dy) + (dz * dz))
|
|
allowed_distance = MOVEMENT_GRACE_DISTANCE + (MAX_NORMAL_MOVEMENT_SPEED * elapsed)
|
|
if not math.isfinite(distance) or distance > allowed_distance:
|
|
return False, f"normal movement exceeded server envelope: distance={distance:.1f}, allowed={allowed_distance:.1f}, elapsed={elapsed:.3f}s"
|
|
return True, "normal movement accepted"
|
|
|
|
def _send_position_correction(self, client: ClientSession, reason: str) -> None:
|
|
previous, _previous_monotonic = client.get_last_transform_anchor()
|
|
if previous is None:
|
|
return
|
|
packet: dict[str, Any] = {
|
|
"type": "positionCorrection",
|
|
"reason": reason[:160],
|
|
"x": previous["x"],
|
|
"y": previous["y"],
|
|
"z": previous["z"],
|
|
"angleZ": previous["angleZ"],
|
|
"cellId": previous["cellId"],
|
|
"worldspaceId": previous.get("worldspaceId", ""),
|
|
"serverTime": time.time(),
|
|
}
|
|
try:
|
|
self._send_packet(client, packet)
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(client)
|
|
return
|
|
with self._lock:
|
|
self._stats["movementCorrectionsSent"] += 1
|
|
|
|
def _normalize_world_state_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
|
|
normalized = dict(packet)
|
|
if "gameHour" in packet:
|
|
if not _is_finite_number(packet["gameHour"], 0.0, 24.0):
|
|
return None
|
|
normalized["gameHour"] = float(packet["gameHour"])
|
|
if "gameDaysPassed" in packet:
|
|
if not _is_finite_number(packet["gameDaysPassed"], 0.0, 10_000_000.0):
|
|
return None
|
|
normalized["gameDaysPassed"] = float(packet["gameDaysPassed"])
|
|
weather = packet.get("weatherFormId")
|
|
if weather is not None:
|
|
if not _is_hex_form_id(weather, allow_empty=True, allow_zero=True):
|
|
return None
|
|
normalized["weatherFormId"] = _normalize_hex_form_id(weather) if weather else ""
|
|
return normalized
|
|
|
|
def _normalize_npc_state_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
|
|
npcs = packet.get("npcs")
|
|
if not isinstance(npcs, list) or len(npcs) > MAX_NPCS_PER_PACKET:
|
|
return None
|
|
clean_npcs: list[dict[str, Any]] = []
|
|
for npc in npcs:
|
|
if not isinstance(npc, dict):
|
|
return None
|
|
if not _is_hex_form_id(npc.get("sourceFormId"), allow_zero=False):
|
|
return None
|
|
if not _is_hex_form_id(npc.get("cellId"), allow_zero=False):
|
|
return None
|
|
worldspace = npc.get("worldspaceId", "")
|
|
if not _is_hex_form_id(worldspace, allow_empty=True, allow_zero=True):
|
|
return None
|
|
for field_name in ("x", "y", "z", "angleZ"):
|
|
if not _is_finite_number(npc.get(field_name), -MAX_ABS_COORDINATE, MAX_ABS_COORDINATE):
|
|
return None
|
|
clean = dict(npc)
|
|
clean["sourceFormId"] = _normalize_hex_form_id(npc["sourceFormId"])
|
|
clean["cellId"] = _normalize_hex_form_id(npc["cellId"])
|
|
clean["worldspaceId"] = _normalize_hex_form_id(worldspace) if worldspace else ""
|
|
for field_name in ("x", "y", "z", "angleZ"):
|
|
clean[field_name] = float(npc[field_name])
|
|
clean_npcs.append(clean)
|
|
normalized = dict(packet)
|
|
normalized["npcs"] = clean_npcs
|
|
if "authorityEpoch" in packet:
|
|
if not _is_int(packet["authorityEpoch"], 1):
|
|
return None
|
|
normalized["authorityEpoch"] = int(packet["authorityEpoch"])
|
|
if "authorityCellId" in packet:
|
|
if not _is_hex_form_id(packet["authorityCellId"], allow_zero=False):
|
|
return None
|
|
normalized["authorityCellId"] = _normalize_hex_form_id(packet["authorityCellId"])
|
|
if "authorityWorldspaceId" in packet:
|
|
world = packet["authorityWorldspaceId"]
|
|
if not _is_hex_form_id(world, allow_empty=True, allow_zero=True):
|
|
return None
|
|
normalized["authorityWorldspaceId"] = _normalize_hex_form_id(world) if world else ""
|
|
return normalized
|
|
|
|
def _npc_scope_from_packet(self, packet: dict[str, Any]) -> ScopeKey | None:
|
|
cell = packet.get("authorityCellId")
|
|
world = packet.get("authorityWorldspaceId", "")
|
|
if not _is_hex_form_id(cell, allow_zero=False):
|
|
return None
|
|
if not _is_hex_form_id(world, allow_empty=True, allow_zero=True):
|
|
return None
|
|
return ScopeKey(_normalize_hex_form_id(cell), _normalize_hex_form_id(world) if world else "")
|
|
|
|
def _normalize_combat_hit_packet(self, packet: dict[str, Any]) -> dict[str, Any] | None:
|
|
target = packet.get("targetPlayerId")
|
|
sequence = packet.get("sequence")
|
|
damage = packet.get("damage")
|
|
if isinstance(target, float) and target.is_integer():
|
|
target = int(target)
|
|
if isinstance(sequence, float) and sequence.is_integer():
|
|
sequence = int(sequence)
|
|
if not _is_int(target, 1) or not _is_int(sequence, 1):
|
|
return None
|
|
if not _is_finite_number(damage, 0.000001, 10000.0):
|
|
return None
|
|
weapon = packet.get("weaponFormId")
|
|
if weapon is not None and not _is_hex_form_id(weapon, allow_zero=True):
|
|
return None
|
|
normalized = dict(packet)
|
|
normalized["targetPlayerId"] = target
|
|
normalized["sequence"] = sequence
|
|
normalized["damage"] = float(damage)
|
|
if isinstance(weapon, str):
|
|
normalized["weaponFormId"] = _normalize_hex_form_id(weapon)
|
|
return normalized
|
|
|
|
def _is_world_host(self, client: ClientSession) -> bool:
|
|
with self._lock:
|
|
return self._world_state_host_player_id == client.player_id
|
|
|
|
def _authority_packet(self, assignment_player_id: int, epoch: int, scope: ScopeKey) -> dict[str, Any]:
|
|
return {
|
|
"type": "npcAuthority",
|
|
"authorityPlayerId": assignment_player_id,
|
|
"authorityEpoch": epoch,
|
|
"authorityCellId": scope.cell_id,
|
|
"authorityWorldspaceId": scope.worldspace_id,
|
|
"serverTime": time.time(),
|
|
}
|
|
|
|
def _reconcile_npc_authority(self) -> None:
|
|
with self._lock:
|
|
players = []
|
|
for client in self._active_clients_locked():
|
|
if client.protocol_version < PROTOCOL_VERSION:
|
|
continue
|
|
scope = scope_from_transform(client.last_transform)
|
|
if scope is not None:
|
|
players.append((client.player_id, scope))
|
|
changes = self._npc_authority.reconcile(players)
|
|
for change in changes:
|
|
self._last_npc_state_by_scope.pop(change.scope, None)
|
|
for change in changes:
|
|
packet = self._authority_packet(change.player_id, change.epoch, change.scope)
|
|
successful, failed = self._broadcast_to_active(packet, v2_only=True)
|
|
with self._lock:
|
|
self._stats["npcAuthorityChanges"] += 1
|
|
for client in failed:
|
|
self._disconnect_client(client)
|
|
self._log(
|
|
f"NPC authority scope {change.scope.cell_id}/{change.scope.worldspace_id or '<interior>'}: "
|
|
f"{change.previous_player_id} -> {change.player_id}, epoch {change.epoch} "
|
|
f"(notified {successful} client(s))."
|
|
)
|
|
|
|
def _send_npc_authority_for_client(self, client: ClientSession, scope: ScopeKey) -> None:
|
|
if client.protocol_version < PROTOCOL_VERSION:
|
|
return
|
|
assignment = self._npc_authority.get(scope)
|
|
if assignment is None:
|
|
return
|
|
try:
|
|
self._send_packet(client, self._authority_packet(assignment.player_id, assignment.epoch, scope))
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(client)
|
|
|
|
def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None:
|
|
encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
|
|
if len(encoded) > MAX_PACKET_CHARS:
|
|
raise ValueError("Outbound packet exceeds maximum line size")
|
|
client.send_bytes(encoded)
|
|
client.record_sent(broadcast=broadcast)
|
|
with self._lock:
|
|
self._stats["packetsSent"] += 1
|
|
if broadcast:
|
|
self._stats["packetsBroadcast"] += 1
|
|
|
|
def _send_existing_transforms_to_client(self, new_client: ClientSession) -> None:
|
|
with self._lock:
|
|
peers = [client for client in self._active_clients_locked() if client.connection != new_client.connection and client.last_transform is not None]
|
|
successful = 0
|
|
for peer in peers:
|
|
if not states_share_interest(peer.last_transform, new_client.last_transform):
|
|
continue
|
|
packet = dict(peer.last_transform or {})
|
|
packet["serverTime"] = time.time()
|
|
try:
|
|
self._send_packet(new_client, packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(new_client)
|
|
break
|
|
with self._lock:
|
|
self._stats["transformPacketsBroadcast"] += successful
|
|
|
|
def _send_existing_player_states_to_client(self, new_client: ClientSession) -> None:
|
|
if new_client.protocol_version < PROTOCOL_VERSION:
|
|
return
|
|
with self._lock:
|
|
snapshots = [
|
|
dict(packet)
|
|
for player_id, packet in self._last_player_state_by_player_id.items()
|
|
if player_id != new_client.player_id
|
|
]
|
|
successful = 0
|
|
for packet in snapshots:
|
|
packet["serverTime"] = time.time()
|
|
try:
|
|
self._send_packet(new_client, packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(new_client)
|
|
break
|
|
with self._lock:
|
|
self._stats["playerStatePacketsBroadcast"] += successful
|
|
|
|
def _broadcast_transform(self, sender: ClientSession, packet: dict[str, Any]) -> None:
|
|
with self._lock:
|
|
recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection]
|
|
successful = 0
|
|
filtered = 0
|
|
failed: list[ClientSession] = []
|
|
for recipient in recipients:
|
|
if not states_share_interest(packet, recipient.last_transform):
|
|
filtered += 1
|
|
continue
|
|
try:
|
|
self._send_packet(recipient, packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
failed.append(recipient)
|
|
with self._lock:
|
|
self._stats["transformPacketsBroadcast"] += successful
|
|
self._stats["transformPacketsInterestFiltered"] += filtered
|
|
for recipient in failed:
|
|
self._disconnect_client(recipient)
|
|
|
|
def _broadcast_player_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
|
|
with self._lock:
|
|
recipients = [
|
|
client
|
|
for client in self._active_clients_locked()
|
|
if client.connection != sender.connection and client.protocol_version >= PROTOCOL_VERSION
|
|
]
|
|
successful = 0
|
|
failed: list[ClientSession] = []
|
|
durable_fields = ("equippedItems", "appearance", "characterName")
|
|
sender_scope_known = _scope_from_state(sender.last_transform) is not None
|
|
for recipient in recipients:
|
|
relay_packet = dict(packet)
|
|
if "actionEvents" in relay_packet:
|
|
recipient_scope_known = _scope_from_state(recipient.last_transform) is not None
|
|
if not sender_scope_known or not recipient_scope_known or not states_share_interest(sender.last_transform, recipient.last_transform):
|
|
relay_packet.pop("actionEvents", None)
|
|
if "actionEvents" not in relay_packet and not any(field in relay_packet for field in durable_fields):
|
|
continue
|
|
try:
|
|
self._send_packet(recipient, relay_packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
failed.append(recipient)
|
|
with self._lock:
|
|
self._stats["playerStatePacketsBroadcast"] += successful
|
|
for client in failed:
|
|
self._disconnect_client(client)
|
|
|
|
def _broadcast_world_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
|
|
successful, failed = self._broadcast_to_active(packet, exclude=sender)
|
|
with self._lock:
|
|
self._stats["worldStatePacketsBroadcast"] += successful
|
|
for client in failed:
|
|
self._disconnect_client(client)
|
|
|
|
def _npc_packet_for_recipient(self, packet: dict[str, Any], recipient: ClientSession) -> dict[str, Any]:
|
|
if recipient.last_transform is None:
|
|
return dict(packet)
|
|
clean = dict(packet)
|
|
clean["npcs"] = [npc for npc in packet.get("npcs", []) if states_share_interest(npc, recipient.last_transform)]
|
|
return clean
|
|
|
|
def _broadcast_npc_state(self, sender: ClientSession, packet: dict[str, Any]) -> None:
|
|
with self._lock:
|
|
recipients = [client for client in self._active_clients_locked() if client.connection != sender.connection]
|
|
successful = 0
|
|
failed: list[ClientSession] = []
|
|
for recipient in recipients:
|
|
scoped_packet = self._npc_packet_for_recipient(packet, recipient)
|
|
try:
|
|
self._send_packet(recipient, scoped_packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
failed.append(recipient)
|
|
with self._lock:
|
|
self._stats["npcStatePacketsBroadcast"] += successful
|
|
for client in failed:
|
|
self._disconnect_client(client)
|
|
|
|
def _send_existing_npc_state_to_client(self, client: ClientSession) -> None:
|
|
if client.protocol_version >= PROTOCOL_VERSION:
|
|
scope = scope_from_transform(client.last_transform)
|
|
with self._lock:
|
|
snapshot = dict(self._last_npc_state_by_scope[scope]) if scope in self._last_npc_state_by_scope else None
|
|
else:
|
|
with self._lock:
|
|
snapshot = dict(self._last_npc_state) if self._last_npc_state is not None else None
|
|
if snapshot is None or snapshot.get("playerId") == client.player_id:
|
|
return
|
|
scoped = self._npc_packet_for_recipient(snapshot, client)
|
|
scoped["serverTime"] = time.time()
|
|
try:
|
|
self._send_packet(client, scoped, broadcast=True)
|
|
with self._lock:
|
|
self._stats["npcStatePacketsBroadcast"] += 1
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(client)
|
|
|
|
def _route_combat_hit(self, sender: ClientSession, packet: dict[str, Any]) -> None:
|
|
target_player_id = packet["targetPlayerId"]
|
|
if target_player_id == sender.player_id:
|
|
self._reject_packet(sender, "Self-targeted combatHit", warning=False)
|
|
return
|
|
recipient = self._find_client_by_player_id(target_player_id)
|
|
if recipient is None:
|
|
self._reject_packet(sender, f"Combat target {target_player_id} is not connected", warning=False)
|
|
return
|
|
if not states_share_interest(sender.last_transform, recipient.last_transform):
|
|
self._reject_packet(sender, f"Combat target {target_player_id} is outside interest scope", warning=True)
|
|
return
|
|
try:
|
|
self._send_packet(recipient, packet)
|
|
with self._lock:
|
|
self._stats["combatHitsRouted"] += 1
|
|
except (OSError, ValueError):
|
|
self._disconnect_client(recipient)
|
|
|
|
def _broadcast_server_world_state(self) -> None:
|
|
with self._lock:
|
|
recipients = self._active_clients_locked()
|
|
snapshot = dict(self._server_world_state)
|
|
packets: list[dict[str, Any]] = []
|
|
if snapshot.get("timeHHmm"):
|
|
packets.append({"type": "serverWorldState", "timeHHmm": snapshot["timeHHmm"], "serverTime": time.time()})
|
|
if snapshot.get("weatherConsoleArg"):
|
|
weather_packet: dict[str, Any] = {"type": "serverWorldState", "weatherConsoleArg": snapshot["weatherConsoleArg"], "serverTime": time.time()}
|
|
if snapshot.get("weatherFormId"):
|
|
weather_packet["weatherFormId"] = snapshot["weatherFormId"]
|
|
packets.append(weather_packet)
|
|
successful = 0
|
|
failed: list[ClientSession] = []
|
|
for packet in packets:
|
|
for recipient in recipients:
|
|
try:
|
|
self._send_packet(recipient, packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
failed.append(recipient)
|
|
with self._lock:
|
|
self._stats["serverWorldStatePacketsBroadcast"] += successful
|
|
for client in {failed_client.connection: failed_client for failed_client in failed}.values():
|
|
self._disconnect_client(client)
|
|
|
|
def _broadcast_to_active(
|
|
self,
|
|
packet: dict[str, Any],
|
|
*,
|
|
exclude: ClientSession | None = None,
|
|
v2_only: bool = False,
|
|
) -> tuple[int, list[ClientSession]]:
|
|
with self._lock:
|
|
recipients = [
|
|
client
|
|
for client in self._active_clients_locked()
|
|
if (exclude is None or client.connection != exclude.connection)
|
|
and (not v2_only or client.protocol_version >= PROTOCOL_VERSION)
|
|
]
|
|
successful = 0
|
|
failed: list[ClientSession] = []
|
|
for recipient in recipients:
|
|
try:
|
|
self._send_packet(recipient, packet, broadcast=True)
|
|
successful += 1
|
|
except (OSError, ValueError):
|
|
failed.append(recipient)
|
|
return successful, failed
|
|
|
|
def _broadcast_world_state_host_assignment(self, player_id: int) -> None:
|
|
packet = {"type": "worldStateHost", "worldStateHostPlayerId": player_id, "serverTime": time.time()}
|
|
successful, failed = self._broadcast_to_active(packet)
|
|
with self._lock:
|
|
self._stats["worldStateHostPacketsBroadcast"] += successful
|
|
for client in failed:
|
|
self._disconnect_client(client)
|
|
|
|
def _reassign_world_state_host(self) -> None:
|
|
with self._lock:
|
|
active = self._active_clients_locked()
|
|
if not active:
|
|
self._world_state_host_player_id = None
|
|
self._last_npc_state = None
|
|
return
|
|
new_host = min(client.player_id for client in active)
|
|
self._world_state_host_player_id = new_host
|
|
self._last_npc_state = None
|
|
self._broadcast_world_state_host_assignment(new_host)
|
|
self._log(f"Reassigned world-state host to player {new_host}.")
|
|
|
|
def _broadcast_disconnect(self, client: ClientSession) -> None:
|
|
packet = {"type": "disconnect", "playerId": client.player_id, "serverTime": time.time()}
|
|
successful, failed = self._broadcast_to_active(packet)
|
|
with self._lock:
|
|
self._stats["disconnectPacketsBroadcast"] += successful
|
|
for recipient in failed:
|
|
self._disconnect_client(recipient)
|
|
|
|
def _remove_client(self, client: ClientSession) -> bool:
|
|
with self._lock:
|
|
removed = self._clients.pop(client.connection, None) is not None
|
|
if removed:
|
|
self._last_player_state_by_player_id.pop(client.player_id, None)
|
|
if removed and client.gameplay_active:
|
|
self._stats["clientsDisconnected"] += 1
|
|
return removed
|
|
|
|
def _disconnect_client(self, client: ClientSession) -> None:
|
|
with self._lock:
|
|
was_active = client.gameplay_active
|
|
was_host = was_active and client.player_id == self._world_state_host_player_id
|
|
if not self._remove_client(client):
|
|
return
|
|
self._close_client_socket(client)
|
|
if not was_active:
|
|
self._log(f"Closed pending/probe connection: {client.label}", level="debug")
|
|
return
|
|
self._log(f"Client disconnected: {client.label} (player {client.player_id})")
|
|
self._broadcast_disconnect(client)
|
|
self._reconcile_npc_authority()
|
|
if was_host:
|
|
self._reassign_world_state_host()
|
|
|
|
def _find_client_by_player_id(self, player_id: int) -> ClientSession | None:
|
|
with self._lock:
|
|
return next((client for client in self._active_clients_locked() if client.player_id == player_id), None)
|
|
|
|
def _clients_for_ip(self, ip: str) -> list[ClientSession]:
|
|
normalized = str(ip).strip()
|
|
with self._lock:
|
|
return [client for client in self._clients.values() if client.address[0] == normalized]
|
|
|
|
def _build_session_ended_packet(self, code: str, reason: str = "") -> dict[str, Any]:
|
|
return {"type": "sessionEnded", "code": code, "reason": str(reason or ""), "serverTime": time.time()}
|
|
|
|
def _send_session_ended_raw(self, connection: socket.socket, *, code: str, reason: str = "") -> None:
|
|
packet = self._build_session_ended_packet(code, reason)
|
|
encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n"
|
|
try:
|
|
connection.settimeout(1.0)
|
|
connection.sendall(encoded)
|
|
with self._lock:
|
|
self._stats["sessionEndedPacketsSent"] += 1
|
|
self._stats["packetsSent"] += 1
|
|
except OSError:
|
|
pass
|
|
|
|
def _end_client_session(self, client: ClientSession, *, code: str, reason: str = "") -> None:
|
|
try:
|
|
self._send_packet(client, self._build_session_ended_packet(code, reason))
|
|
with self._lock:
|
|
self._stats["sessionEndedPacketsSent"] += 1
|
|
except (OSError, ValueError):
|
|
pass
|
|
self._disconnect_client(client)
|
|
|
|
def _end_sessions_for_ip(self, ip: str, *, code: str, reason: str = "") -> int:
|
|
clients = self._clients_for_ip(ip)
|
|
for client in clients:
|
|
self._end_client_session(client, code=code, reason=reason)
|
|
return len(clients)
|
|
|
|
def _close_raw_socket(self, connection: socket.socket) -> None:
|
|
try:
|
|
connection.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def _close_client_socket(self, client: ClientSession) -> None:
|
|
self._close_raw_socket(client.connection)
|
|
|
|
def _reject_packet(self, client: ClientSession, reason: str, *, warning: bool = True) -> None:
|
|
with self._lock:
|
|
self._stats["packetsRejected"] += 1
|
|
self._log(f"Rejected packet from {client.label} (player {client.player_id}): {reason}", level="warning" if warning else "debug")
|
|
|
|
def _should_log(self, level: str) -> bool:
|
|
configured = _LOG_LEVELS.get(self.log_verbosity, _LOG_LEVELS[DEFAULT_LOG_VERBOSITY])
|
|
message_level = _LOG_LEVELS.get(level, _LOG_LEVELS["info"])
|
|
return message_level >= configured
|
|
|
|
def _log(self, message: str, *, level: str = "info") -> None:
|
|
if not self._should_log(level):
|
|
return
|
|
with self._log_lock:
|
|
listeners = list(self._log_listeners)
|
|
if not listeners:
|
|
return
|
|
for listener in listeners:
|
|
try:
|
|
try:
|
|
listener(message, level=level)
|
|
except TypeError:
|
|
listener(message)
|
|
except Exception:
|
|
pass
|