From d310d77428ede6c4b73f0fca5341e71488d6550b Mon Sep 17 00:00:00 2001 From: Nomads_Reach <144523850+NomadsReach@users.noreply.github.com> Date: Sat, 15 Aug 2026 21:56:33 -0400 Subject: [PATCH] Enforce scoped NPC authority epochs for Protocol V2 --- server/server_core.py | 201 +++++++++++++++++++++++++++++++----------- 1 file changed, 148 insertions(+), 53 deletions(-) diff --git a/server/server_core.py b/server/server_core.py index dffb802..f37a6a5 100644 --- a/server/server_core.py +++ b/server/server_core.py @@ -13,6 +13,7 @@ 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 world_state_presets import normalize_fw_console_arg, relay_weather_form_id HOST = "0.0.0.0" @@ -152,12 +153,6 @@ def _scope_from_state(state: dict[str, Any] | None) -> tuple[str, str, float, fl 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: @@ -226,6 +221,8 @@ class FalloutTogetherServer: self._world_state_host_player_id: int | None = None self._server_world_state: dict[str, str] = {} self._last_npc_state: dict[str, Any] | None = None + self._last_npc_state_by_scope: dict[ScopeKey, dict[str, Any]] = {} + self._npc_authority = NpcAuthorityManager() self._connect_attempts: dict[str, deque[float]] = {} self._stats: dict[str, int] = { @@ -246,6 +243,8 @@ class FalloutTogetherServer: "worldStatePacketsBroadcast": 0, "npcStatePacketsReceived": 0, "npcStatePacketsBroadcast": 0, + "npcAuthorityChanges": 0, + "npcAuthorityRejects": 0, "combatHitsReceived": 0, "combatHitsRouted": 0, "worldStateHostPacketsBroadcast": 0, @@ -335,6 +334,7 @@ class FalloutTogetherServer: "pendingConnections": pending, "nextPlayerId": self._next_player_id, "protocolVersion": PROTOCOL_VERSION, + "npcAuthorityScopes": len(self._npc_authority.assignments()), } ) return stats @@ -442,6 +442,8 @@ class FalloutTogetherServer: 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._npc_authority.clear() self._connect_attempts.clear() self._log(f"Commonwealth Online server listening on {self.host}:{self.port}") @@ -603,7 +605,14 @@ class FalloutTogetherServer: "serverName": self.server_name, "serverDescription": self.server_description, "protocolVersion": PROTOCOL_VERSION, - "capabilities": ["interest-v1", "hello-v2", "bounded-framing", "rate-limit-v1", "movement-correction-v1"], + "capabilities": [ + "interest-v1", + "hello-v2", + "bounded-framing", + "rate-limit-v1", + "movement-correction-v1", + "npc-authority-epoch-v1", + ], }, ) buffer = b"" @@ -720,23 +729,24 @@ class FalloutTogetherServer: self._reject_packet(client, "Malformed transform") return False accepted_monotonic = time.monotonic() - movement_valid, movement_reason = self._validate_transform_movement( - client, - normalized, - accepted_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 == "worldState": @@ -755,18 +765,47 @@ class FalloutTogetherServer: 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 + + 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._last_npc_state = dict(normalized) self._stats["npcStatePacketsReceived"] += 1 self._broadcast_npc_state(client, normalized) return True @@ -794,11 +833,7 @@ class FalloutTogetherServer: 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}.", - ) + 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.") @@ -854,42 +889,30 @@ class FalloutTogetherServer: 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]: + 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 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: @@ -959,8 +982,30 @@ class FalloutTogetherServer: 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") @@ -988,6 +1033,52 @@ class FalloutTogetherServer: with self._lock: return self._world_state_host_player_id == client.player_id + def _authority_packet(self, assignment_player_id: int, epoch: int, scope: ScopeKey) -> dict[str, Any]: + return { + "type": "npcAuthority", + "authorityPlayerId": assignment_player_id, + "authorityEpoch": epoch, + "authorityCellId": scope.cell_id, + "authorityWorldspaceId": scope.worldspace_id, + "serverTime": time.time(), + } + + def _reconcile_npc_authority(self) -> None: + with self._lock: + players = [] + for client in self._active_clients_locked(): + if client.protocol_version < PROTOCOL_VERSION: + continue + scope = scope_from_transform(client.last_transform) + if scope is not None: + players.append((client.player_id, scope)) + changes = self._npc_authority.reconcile(players) + for change in changes: + self._last_npc_state_by_scope.pop(change.scope, None) + for change in changes: + packet = self._authority_packet(change.player_id, change.epoch, change.scope) + successful, failed = self._broadcast_to_active(packet, v2_only=True) + with self._lock: + self._stats["npcAuthorityChanges"] += 1 + for client in failed: + self._disconnect_client(client) + self._log( + f"NPC authority scope {change.scope.cell_id}/{change.scope.worldspace_id or ''}: " + f"{change.previous_player_id} -> {change.player_id}, epoch {change.epoch} " + f"(notified {successful} client(s))." + ) + + def _send_npc_authority_for_client(self, client: ClientSession, scope: ScopeKey) -> None: + if client.protocol_version < PROTOCOL_VERSION: + return + assignment = self._npc_authority.get(scope) + if assignment is None: + return + try: + self._send_packet(client, self._authority_packet(assignment.player_id, assignment.epoch, scope)) + except (OSError, ValueError): + self._disconnect_client(client) + def _send_packet(self, client: ClientSession, packet: dict[str, Any], *, broadcast: bool = False) -> None: encoded = json.dumps(packet, separators=(",", ":"), allow_nan=False).encode("utf-8") + b"\n" if len(encoded) > MAX_PACKET_CHARS: @@ -1001,10 +1092,7 @@ class FalloutTogetherServer: 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 - ] + 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): @@ -1073,8 +1161,13 @@ class FalloutTogetherServer: 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 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) @@ -1113,11 +1206,7 @@ class FalloutTogetherServer: 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(), - } + 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) @@ -1135,11 +1224,19 @@ class FalloutTogetherServer: 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]]: + 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 + 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] = [] @@ -1199,6 +1296,7 @@ class FalloutTogetherServer: 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() @@ -1253,10 +1351,7 @@ class FalloutTogetherServer: 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", - ) + 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])