085fd714a5
Extracts _connect_and_auth helper from query_status, adds execute_command using the trailing-marker pattern for multi-packet reassembly, and covers all paths (happy path, multi-packet, empty reply, auth failure, timeout, input validation, marker drain) with 10 new tests. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
303 lines
9.8 KiB
Python
303 lines
9.8 KiB
Python
"""Source RCON client tests against an in-process TCP fixture.
|
|
|
|
The handshake quirk we verified live: after a SERVERDATA_AUTH (type=3) the
|
|
server sends a SERVERDATA_RESPONSE_VALUE (type=0) FIRST and THEN the
|
|
SERVERDATA_AUTH_RESPONSE (type=2). The auth response's req_id == -1 means
|
|
bad password. The client must consume both packets before sending the
|
|
command.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import socket
|
|
import struct
|
|
import threading
|
|
from contextlib import contextmanager
|
|
from typing import Iterator
|
|
|
|
import pytest
|
|
|
|
from l4d2web.services.rcon import (
|
|
RconAuthError,
|
|
RconError,
|
|
execute_command,
|
|
query_status,
|
|
)
|
|
|
|
# req_id constants (must match rcon.py)
|
|
_EXEC_REQ_ID = 2
|
|
_MARKER_REQ_ID = 0xC0DE
|
|
|
|
|
|
def _pack(req_id: int, ptype: int, body: str) -> bytes:
|
|
body_bytes = body.encode("utf-8") + b"\x00\x00"
|
|
size = 4 + 4 + len(body_bytes)
|
|
return struct.pack("<iii", size, req_id, ptype) + body_bytes
|
|
|
|
|
|
def _unpack_one(conn: socket.socket) -> tuple[int, int, str]:
|
|
raw_size = conn.recv(4)
|
|
size = struct.unpack("<i", raw_size)[0]
|
|
payload = b""
|
|
while len(payload) < size:
|
|
payload += conn.recv(size - len(payload))
|
|
req_id, ptype = struct.unpack("<ii", payload[:8])
|
|
body = payload[8:].rstrip(b"\x00").decode("utf-8", errors="replace")
|
|
return req_id, ptype, body
|
|
|
|
|
|
@contextmanager
|
|
def fake_rcon_server(handler) -> Iterator[int]:
|
|
"""Start a TCP server on an ephemeral port; handler(conn) runs in a thread.
|
|
|
|
Handler exceptions propagate at context exit so a buggy handler surfaces
|
|
as a real test failure instead of degrading into a client-side timeout.
|
|
"""
|
|
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
|
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
|
server.bind(("127.0.0.1", 0))
|
|
server.listen(1)
|
|
port = server.getsockname()[1]
|
|
server.settimeout(3.0)
|
|
|
|
handler_error: list[BaseException] = []
|
|
|
|
def serve() -> None:
|
|
try:
|
|
conn, _ = server.accept()
|
|
try:
|
|
handler(conn)
|
|
finally:
|
|
conn.close()
|
|
except BaseException as exc:
|
|
handler_error.append(exc)
|
|
|
|
t = threading.Thread(target=serve, daemon=True)
|
|
t.start()
|
|
try:
|
|
yield port
|
|
finally:
|
|
server.close()
|
|
t.join(timeout=1.0)
|
|
if handler_error:
|
|
raise AssertionError("fake_rcon_server handler raised") from handler_error[0]
|
|
|
|
|
|
def _do_auth_handshake(conn: socket.socket) -> None:
|
|
"""Receive AUTH packet and respond with the standard two-packet handshake."""
|
|
req_id, ptype, body = _unpack_one(conn)
|
|
assert ptype == 3 # SERVERDATA_AUTH
|
|
conn.sendall(_pack(req_id, 0, "")) # leading empty type-0
|
|
conn.sendall(_pack(req_id, 2, "")) # auth response — req_id != -1 means OK
|
|
|
|
|
|
def test_auth_success_then_status(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
response_body = (
|
|
"hostname: Left 4 Dead 2\n"
|
|
"version : 2.2.4.3 9309 secure (unknown)\n"
|
|
"udp/ip : 127.0.0.1:27016 [ public 1.2.3.4:27016 ]\n"
|
|
"os : Linux Dedicated\n"
|
|
"map : c1m2_streets\n"
|
|
"players : 1 humans, 0 bots (4 max) (not hibernating) (reserved 1860000e7e5e446)\n"
|
|
"\n"
|
|
"# userid name uniqueid connected ping loss state rate adr\n"
|
|
'# 2 1 "Crone" STEAM_1:0:12376499 00:21 185 20 active 30000 91.55.5.100:27005\n'
|
|
"#end\n"
|
|
)
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
req_id, ptype, body = _unpack_one(conn)
|
|
assert ptype == 3
|
|
assert body == "letmein"
|
|
conn.sendall(_pack(req_id, 0, ""))
|
|
conn.sendall(_pack(req_id, 2, ""))
|
|
cmd_id, cmd_type, cmd = _unpack_one(conn)
|
|
assert cmd_type == 2
|
|
assert cmd == "status"
|
|
conn.sendall(_pack(cmd_id, 0, response_body))
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
result = query_status("127.0.0.1", port, "letmein", timeout=2.0)
|
|
|
|
assert result.map == "c1m2_streets"
|
|
assert result.players == 1
|
|
assert result.bots == 0
|
|
assert result.max_players == 4
|
|
assert result.hibernating is False
|
|
assert len(result.roster) == 1
|
|
p = result.roster[0]
|
|
assert p.name == "Crone"
|
|
assert p.steam_id_64 == "76561197985018726" # 76561197960265728 + 0 + 12376499*2
|
|
assert p.connected_seconds == 21
|
|
assert p.ping == 185
|
|
|
|
|
|
def test_auth_failure_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def handler(conn: socket.socket) -> None:
|
|
req_id, _, _ = _unpack_one(conn)
|
|
conn.sendall(_pack(req_id, 0, ""))
|
|
conn.sendall(_pack(-1, 2, "")) # bad password sentinel
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
with pytest.raises(RconAuthError):
|
|
query_status("127.0.0.1", port, "wrong", timeout=2.0)
|
|
|
|
|
|
def test_timeout_raises(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
def handler(conn: socket.socket) -> None:
|
|
import time
|
|
time.sleep(3.0)
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
with pytest.raises(RconError):
|
|
query_status("127.0.0.1", port, "x", timeout=0.3)
|
|
|
|
|
|
def test_parse_duration_handles_hours() -> None:
|
|
from l4d2web.services.rcon import _parse_duration
|
|
|
|
assert _parse_duration("00:21") == 21
|
|
assert _parse_duration("01:23:45") == 5025
|
|
assert _parse_duration("12:00") == 720
|
|
|
|
|
|
def test_parse_duration_rejects_malformed_as_rcon_error() -> None:
|
|
from l4d2web.services.rcon import _parse_duration
|
|
|
|
for bad in ["", ":", "abc", "1:", ":5", "1:2:3:4"]:
|
|
with pytest.raises(RconError):
|
|
_parse_duration(bad)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# execute_command tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def test_execute_command_single_packet_reply() -> None:
|
|
"""Handler sends one body packet then the marker echo; client returns body."""
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
_do_auth_handshake(conn)
|
|
|
|
# Receive EXECCOMMAND
|
|
exec_id, exec_type, cmd = _unpack_one(conn)
|
|
assert exec_type == 2 # SERVERDATA_EXECCOMMAND
|
|
assert cmd == "echo hello"
|
|
|
|
# Receive the marker packet
|
|
marker_id, marker_type, marker_body = _unpack_one(conn)
|
|
assert marker_type == 0 # SERVERDATA_RESPONSE_VALUE
|
|
assert marker_body == ""
|
|
|
|
# Send reply body, then marker echo
|
|
conn.sendall(_pack(exec_id, 0, "hello"))
|
|
conn.sendall(_pack(marker_id, 0, ""))
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
result = execute_command("127.0.0.1", port, "letmein", "echo hello", timeout=2.0)
|
|
|
|
assert result == "hello"
|
|
|
|
|
|
def test_execute_command_multi_packet_reply() -> None:
|
|
"""Multi-packet replies are concatenated in order."""
|
|
chunk1 = "A" * 3000
|
|
chunk2 = "B" * 3000
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
_do_auth_handshake(conn)
|
|
|
|
exec_id, _, _ = _unpack_one(conn)
|
|
marker_id, _, _ = _unpack_one(conn)
|
|
|
|
conn.sendall(_pack(exec_id, 0, chunk1))
|
|
conn.sendall(_pack(exec_id, 0, chunk2))
|
|
conn.sendall(_pack(marker_id, 0, ""))
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
result = execute_command("127.0.0.1", port, "letmein", "cvarlist", timeout=2.0)
|
|
|
|
assert result == chunk1 + chunk2
|
|
|
|
|
|
def test_execute_command_empty_reply() -> None:
|
|
"""Server echoes only the marker (e.g. for `say`); result is empty string."""
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
_do_auth_handshake(conn)
|
|
|
|
_unpack_one(conn) # exec packet
|
|
marker_id, _, _ = _unpack_one(conn)
|
|
|
|
conn.sendall(_pack(marker_id, 0, ""))
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
result = execute_command("127.0.0.1", port, "letmein", "say hi", timeout=2.0)
|
|
|
|
assert result == ""
|
|
|
|
|
|
def test_execute_command_bad_password() -> None:
|
|
"""Bad password on auth raises RconAuthError."""
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
req_id, _, _ = _unpack_one(conn)
|
|
conn.sendall(_pack(req_id, 0, ""))
|
|
conn.sendall(_pack(-1, 2, "")) # bad password sentinel
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
with pytest.raises(RconAuthError):
|
|
execute_command("127.0.0.1", port, "wrong", "status", timeout=2.0)
|
|
|
|
|
|
def test_execute_command_timeout() -> None:
|
|
"""Server hangs; client raises RconError."""
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
import time
|
|
time.sleep(3.0)
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
with pytest.raises(RconError):
|
|
execute_command("127.0.0.1", port, "x", "status", timeout=0.3)
|
|
|
|
|
|
def test_execute_command_rejects_empty_command() -> None:
|
|
with pytest.raises(ValueError):
|
|
execute_command("127.0.0.1", 27015, "pw", "")
|
|
|
|
|
|
def test_execute_command_rejects_whitespace_only_command() -> None:
|
|
with pytest.raises(ValueError):
|
|
execute_command("127.0.0.1", 27015, "pw", " ")
|
|
|
|
|
|
def test_execute_command_rejects_null_byte_in_command() -> None:
|
|
with pytest.raises(ValueError):
|
|
execute_command("127.0.0.1", 27015, "pw", "echo\x00hello")
|
|
|
|
|
|
def test_execute_command_rejects_oversized_command() -> None:
|
|
with pytest.raises(ValueError):
|
|
execute_command("127.0.0.1", 27015, "pw", "x" * 1001)
|
|
|
|
|
|
def test_execute_command_drains_marker_after_response() -> None:
|
|
"""Client stops reading exactly after the marker — does not block on a 5th packet."""
|
|
|
|
def handler(conn: socket.socket) -> None:
|
|
_do_auth_handshake(conn)
|
|
|
|
exec_id, _, _ = _unpack_one(conn)
|
|
marker_id, _, _ = _unpack_one(conn)
|
|
|
|
# Send one body packet then the marker — nothing else.
|
|
conn.sendall(_pack(exec_id, 0, "done"))
|
|
conn.sendall(_pack(marker_id, 0, ""))
|
|
# Handler returns immediately; if client reads past the marker it would
|
|
# block (no more data) and hit a timeout — the 2 s timeout guards us.
|
|
|
|
with fake_rcon_server(handler) as port:
|
|
result = execute_command("127.0.0.1", port, "letmein", "status", timeout=2.0)
|
|
|
|
assert result == "done"
|