feat(multiplayer): Phase 3 task 3.4 - input validation, rate limiting, disconnect policy

MatchSim._recv_input now validates before decoding (§3.1 steps 2-3):
per-peer rolling-1s rate limiting (packet count AND byte budget, dropping
over-budget packets and disconnecting after 3 consecutive over-budget
seconds), and framing validation (redundancy count and payload size
checked against NetCodec's own layout before unpack_input ever runs,
disconnecting after 20 malformed packets). Framing has to be validated
explicitly rather than relying on decode failure: StreamPeerBuffer
silently zero-fills past EOF instead of erroring, a finding from Phase
2's adversarial review.

networked_match.gd's _on_input_received now rejects any seq claiming to
be more than 20 ticks ahead of the current server tick (§3.1 step 4) and
counts (rather than silently ignoring) input from a peer with no slot,
for observability.

Verified with two new permanent regression tests (networked_match_smoke.gd
--role=client-abuse-malformed / client-abuse-flood) that call
MatchSim._recv_input directly with garbage bytes and a legitimate-but-
too-frequent flood, respectively, bypassing the honest client encoder
entirely - the same thing a hostile custom client sending raw ENet
packets would look like. Both confirm real disconnection, not just that
the server tolerates the abuse.

Two bugs surfaced by getting these tests to actually pass cleanly: a
GDScript lambda-capture-by-value mistake in the tests themselves (a
plain `var disconnected := false` mutated inside a signal-handler lambda
never became visible to the enclosing function - fixed by capturing a
single-element Array instead, which is captured by reference); and a
narrow real race where NetworkManager's own ping/pong reply could target
a peer that a concurrent abuse-triggered disconnect had just removed
from the same poll() batch, now guarded. (Passing disconnect_peer's
`force` parameter as an attempted fix for a related one-off benign error
was tried and reverted - it made Godot's own peer-list bookkeeping
inconsistent, producing hundreds of errors instead of one; verified
empirically rather than assumed.)

Full regression suite, including the net-sim-latency milestone gate,
re-run clean.

Game/scripts/match_sim.gd

@@ -24,6 +24,38 @@ signal input_received(peer_id: int, decoded: Dictionary)    # decoded: see NetCo
 signal snapshot_received(decoded: Dictionary)                # decoded: see NetCodec.unpack_snapshot
 signal score_update_received(score: Dictionary)
 
+# Input validation (multiplayer-todo.md §3.1 steps 2-3, task 3.4). Deliberately
+# lives here rather than in NetworkedMatch: framing/rate abuse is a protocol-
+# level concern independent of any particular match's roster/slot state, and
+# this autoload already owns the RPC that receives the raw bytes.
+#
+# 60Hz * 1.5 + 20, per §3.1 step 2's own numbers.
+const RATE_LIMIT_PACKETS_PER_SEC := 110
+# "Same for a byte budget" (§3.1 step 2) — the worst-case legitimate packet
+# is a full-redundancy input (INPUT_HEADER_SIZE + MAX_REDUNDANCY entries,
+# the "40 B input" §2.3 sizes to), so the byte budget is just the packet
+# budget scaled by that worst-case size — no separate constant to keep in
+# sync by hand.
+const RATE_LIMIT_BYTES_PER_SEC := RATE_LIMIT_PACKETS_PER_SEC * (NetCodec.INPUT_HEADER_SIZE + NetCodec.MAX_REDUNDANCY * NetCodec.INPUT_ENTRY_SIZE)
+const RATE_LIMIT_WINDOW_MS := 1000
+const RATE_LIMIT_OVER_BUDGET_SECONDS_TO_DISCONNECT := 3
+const MALFORMED_LIMIT_TO_DISCONNECT := 20
+
+
+class _PeerInputState:
+	var window_start_ms := 0
+	var packets_this_window := 0
+	var bytes_this_window := 0
+	var over_budget_seconds := 0
+	var malformed_count := 0
+
+
+var _peer_input_state: Dictionary = {}   # peer_id -> _PeerInputState, server only
+
+
+func _ready() -> void:
+	NetworkManager.client_disconnected.connect(func(peer_id: int) -> void: _peer_input_state.erase(peer_id))
+
 # Server only: the last match_config actually sent, so a client whose own
 # scene load (and therefore its match_config_received listener) finishes
 # AFTER the server already broadcast can still get it — a one-shot
@@ -82,10 +114,63 @@ func _recv_input(bytes: PackedByteArray) -> void:
 	if not multiplayer.is_server():
 		return
 	var peer_id := multiplayer.get_remote_sender_id()
+
+	var state: _PeerInputState = _peer_input_state.get(peer_id)
+	if state == null:
+		state = _PeerInputState.new()
+		_peer_input_state[peer_id] = state
+
+	# Rolling 1s window (§3.1 step 2). Rolled over lazily on the first
+	# packet past the window boundary, not on a timer — this RPC only ever
+	# runs when a packet actually arrives, so there's nothing to roll over
+	# when nothing is arriving anyway.
+	var now_ms := Time.get_ticks_msec()
+	if now_ms - state.window_start_ms >= RATE_LIMIT_WINDOW_MS:
+		var was_over_budget := state.packets_this_window > RATE_LIMIT_PACKETS_PER_SEC or state.bytes_this_window > RATE_LIMIT_BYTES_PER_SEC
+		state.over_budget_seconds = (state.over_budget_seconds + 1) if was_over_budget else 0
+		state.window_start_ms = now_ms
+		state.packets_this_window = 0
+		state.bytes_this_window = 0
+		if state.over_budget_seconds >= RATE_LIMIT_OVER_BUDGET_SECONDS_TO_DISCONNECT:
+			_disconnect_abusive_peer(peer_id, "input rate limit exceeded for %d consecutive seconds" % state.over_budget_seconds)
+			return
+
+	state.packets_this_window += 1
+	state.bytes_this_window += bytes.size()
+	if state.packets_this_window > RATE_LIMIT_PACKETS_PER_SEC or state.bytes_this_window > RATE_LIMIT_BYTES_PER_SEC:
+		return   # over budget for the current window — drop, counted above at the next window roll
+
+	# Framing (§3.1 step 3), validated before decoding — unpack_input can't
+	# be trusted to catch this itself: StreamPeerBuffer silently zero-fills
+	# past EOF rather than erroring (found during Phase 2's adversarial
+	# review's hostile-client stress test), so a too-short or size-mismatched
+	# payload would otherwise decode "successfully" into garbage actions
+	# instead of being rejected.
+	if bytes.size() < NetCodec.INPUT_HEADER_SIZE:
+		_count_malformed(peer_id, state)
+		return
+	var count: int = bytes[5]   # type_version(1) + seq(4) precede count — see pack_input's own layout
+	if count == 0 or count > NetCodec.MAX_REDUNDANCY or bytes.size() != NetCodec.INPUT_HEADER_SIZE + count * NetCodec.INPUT_ENTRY_SIZE:
+		_count_malformed(peer_id, state)
+		return
+
 	var decoded := NetCodec.unpack_input(bytes)
 	input_received.emit(peer_id, decoded)
 
 
+func _count_malformed(peer_id: int, state: _PeerInputState) -> void:
+	state.malformed_count += 1
+	if state.malformed_count >= MALFORMED_LIMIT_TO_DISCONNECT:
+		_disconnect_abusive_peer(peer_id, "too many malformed input packets (%d)" % state.malformed_count)
+
+
+func _disconnect_abusive_peer(peer_id: int, reason: String) -> void:
+	push_warning("MatchSim: disconnecting peer %d for abuse: %s" % [peer_id, reason])
+	_peer_input_state.erase(peer_id)
+	if multiplayer.multiplayer_peer is ENetMultiplayerPeer:
+		multiplayer.multiplayer_peer.disconnect_peer(peer_id)
+
+
 @rpc("authority", "call_remote", "unreliable_ordered", 2)
 func _snapshot(bytes: PackedByteArray) -> void:
 	var decoded := NetCodec.unpack_snapshot(bytes)

Game/scripts/network_manager.gd

@@ -174,6 +174,16 @@ func _ping(client_send_ms: int) -> void:
 	# RTT sample and the offset estimate instead of adding to them.
 	var server_now := Time.get_ticks_msec()
 	var sender_id := multiplayer.get_remote_sender_id()
+	# A single poll() call can process several queued RPCs from the same
+	# peer in one batch — an earlier one in that same batch (e.g. task 3.4's
+	# abuse-triggered disconnect_peer(..., now=true), which removes the
+	# peer immediately rather than waiting for an acknowledged disconnect)
+	# can leave this ping's sender no longer a valid peer by the time its
+	# own turn in the batch comes up. NetSim's inactive/passthrough path
+	# (the common case — no CLI flags) dispatches immediately with no
+	# validation of its own, so check here rather than relying on it.
+	if sender_id not in multiplayer.get_peers():
+		return
 	NetSim.send(func() -> void: _pong.rpc_id(sender_id, client_send_ms, server_now), sender_id)
 
 

Game/scripts/networked_match.gd

@@ -91,6 +91,12 @@ var _input_history: Array[ShipAction] = []
 var _last_received_snapshot_tick := 0     # client only: echoed back as ack_snapshot_tick
 var _input_lead_controller := InputLeadController.new()   # client only (§3.3)
 var _last_known_input_buffer_depth := -1  # client only: -1 = no snapshot with this field yet
+# §3.1 step 4. Not 120: InputLeadController.LEAD_MAX is 12, so anything
+# claiming to be further ahead of the current server tick than this is
+# broken or hostile, not just an honest client running a legitimately fast
+# lead.
+const MAX_SEQ_LEAD_TICKS := 20
+var _unknown_sender_input_count := 0      # server only, observability (§3.1 step 1)
 var _reset_gen := 0                       # server only: bumped on every kickoff/goal reset so the client hard-snaps instead of interpolating across the teleport
 # Server only. _on_goal_scored's reset_ball()/reset_ships() only QUEUE
 # teleports (task 0.15's queue_teleport — applied on each body's next
@@ -207,9 +213,26 @@ func _start_server() -> void:
 func _on_input_received(peer_id: int, decoded: Dictionary) -> void:
 	for slot in _slots:
 		if slot.peer_id == peer_id:
-			slot.jitter_buffer.ingest(decoded["seq"], decoded["actions"])
+			var seq: int = decoded["seq"]
+			# §3.1 step 4. Not 120: input_lead is clamped to
+			# InputLeadController.LEAD_MAX (12), so anything claiming to be
+			# further ahead than this is broken or hostile, not just a fast
+			# lead. This is also why InputJitterBuffer's ring can be fixed-
+			# size — a client can never make the server allocate — but
+			# rejecting the packet here still keeps garbage-far-future seq
+			# values out of the ring entirely rather than letting them
+			# silently overwrite a near-future slot some honest, in-range
+			# packet is about to need.
+			if seq > Engine.get_physics_frames() + MAX_SEQ_LEAD_TICKS:
+				return
+			slot.jitter_buffer.ingest(seq, decoded["actions"])
 			slot.last_client_send_ms = decoded["client_send_ms"]
 			return
+	# A connected-but-not-yet-slotted peer (or one whose slot somehow
+	# vanished) sending input — harmless (the packet is simply dropped,
+	# same as always), but worth counting for observability (§3.1 step 1)
+	# rather than silently discarding with no trace at all.
+	_unknown_sender_input_count += 1
 
 
 func _on_goal_registered(conceding_team: int) -> void:

Game/tests/networked_match_smoke.gd

@@ -39,6 +39,22 @@ func _ready() -> void:
 				return
 			print("SMOKE: joining ...")
 			MatchNet.welcomed.connect(_on_client_welcomed)
+		"client-abuse-malformed", "client-abuse-flood":
+			# task 3.4's disconnect-abusive-peer paths: joins normally (so
+			# it's a real connected peer, exactly like a hostile custom
+			# client would be — the validation doesn't get to assume
+			# anything about who's on the other end of an authenticated
+			# connection), then deliberately abuses MatchSim._recv_input
+			# directly rather than going through networked_match.gd's own
+			# honest encoder.
+			MatchNet.local_player_name = "Abuser"
+			var err := NetworkManager.join("127.0.0.1", PORT)
+			if err != OK:
+				print("SMOKE FAIL: join() failed: %s" % error_string(err))
+				get_tree().quit(1)
+				return
+			print("SMOKE: joining to abuse (%s) ..." % _role)
+			MatchNet.welcomed.connect(_on_abuser_welcomed)
 		_:
 			print("SMOKE FAIL: missing or unrecognised --role=")
 			get_tree().quit(1)
@@ -69,3 +85,13 @@ func _on_client_welcomed() -> void:
 	var hooks := preload("res://tests/networked_match_test_hooks.gd").new()
 	get_tree().root.add_child.call_deferred(hooks)
 	hooks.run_client_check.call_deferred(SETTLE_SECONDS, DRIVE_SECONDS)
+
+
+func _on_abuser_welcomed() -> void:
+	MatchNet.welcomed.disconnect(_on_abuser_welcomed)
+	var hooks := preload("res://tests/networked_match_test_hooks.gd").new()
+	get_tree().root.add_child.call_deferred(hooks)
+	if _role == "client-abuse-malformed":
+		hooks.run_malformed_abuse_check.call_deferred()
+	else:
+		hooks.run_rate_limit_abuse_check.call_deferred()

Game/tests/networked_match_test_hooks.gd

@@ -114,3 +114,66 @@ func run_client_check(settle_seconds: float, drive_seconds: float) -> void:
 	await get_tree().create_timer(0.3).timeout
 	NetworkManager.shutdown()
 	get_tree().quit(0 if success else 1)
+
+
+# task 3.4: MatchSim._recv_input must count malformed packets and disconnect
+# after MALFORMED_LIMIT_TO_DISCONNECT (20) of them. Calls the RPC directly
+# with garbage bytes rather than going through networked_match.gd's own
+# honest encoder — this IS what a hostile custom client sending raw ENet
+# packets would look like, so bypassing the normal send path is the point,
+# not a shortcut.
+func run_malformed_abuse_check() -> void:
+	await get_tree().create_timer(1.0).timeout
+	# A single-element Array, not a plain bool: GDScript lambdas capture
+	# outer local variables BY VALUE at creation time, not by reference, so
+	# `disconnected = true` inside the lambda below would silently mutate
+	# only the lambda's own captured copy — invisible to this function's
+	# own `disconnected` if it were a plain bool. Mutating an Array's
+	# CONTENTS from inside the lambda works because the Array object
+	# itself (not a copy of it) is what got captured.
+	var disconnected := [false]
+	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)
+
+	for i in 25:
+		MatchSim._recv_input.rpc_id(1, PackedByteArray([1, 2, 3]))   # far too short to even hold a header
+		NetworkManager.poll()
+		await get_tree().physics_frame
+	await get_tree().create_timer(1.0).timeout
+	NetworkManager.poll()
+
+	print("SMOKE %s: 25 malformed packets %s" % [
+		"PASS" if disconnected[0] else "FAIL",
+		"resulted in disconnect" if disconnected[0] else "did NOT disconnect the abusive peer",
+	])
+	get_tree().quit(0 if disconnected[0] else 1)
+
+
+# task 3.4: MatchSim._recv_input must rate-limit and disconnect after
+# RATE_LIMIT_OVER_BUDGET_SECONDS_TO_DISCONNECT (3) consecutive seconds over
+# RATE_LIMIT_PACKETS_PER_SEC (110/s). Every packet here is individually
+# well-formed (a real NetCodec.pack_input payload) — only the SEND RATE is
+# abusive, confirming the rate limiter fires independently of the malformed-
+# packet counter, not as a side effect of it.
+func run_rate_limit_abuse_check() -> void:
+	await get_tree().create_timer(1.0).timeout
+	var disconnected := [false]   # see run_malformed_abuse_check's comment on why not a plain bool
+	NetworkManager.disconnected_from_server.connect(func() -> void: disconnected[0] = true)
+
+	var net_codec := preload("res://scripts/net_codec.gd")
+	var ship_action_script := preload("res://scripts/ship_action.gd")
+	var bytes: PackedByteArray = net_codec.pack_input(1, 0, Time.get_ticks_msec(), [ship_action_script.new()])
+
+	var deadline_ms := Time.get_ticks_msec() + 4000
+	while Time.get_ticks_msec() < deadline_ms and not disconnected[0]:
+		for i in 40:   # well above 110/s once summed across a frame's worth of iterations
+			MatchSim._recv_input.rpc_id(1, bytes)
+		NetworkManager.poll()
+		await get_tree().process_frame
+	await get_tree().create_timer(0.5).timeout
+	NetworkManager.poll()
+
+	print("SMOKE %s: sustained packet flood %s" % [
+		"PASS" if disconnected[0] else "FAIL",
+		"resulted in disconnect" if disconnected[0] else "did NOT disconnect the abusive peer",
+	])
+	get_tree().quit(0 if disconnected[0] else 1)