Enforce strict item params validation and stripping on server
This commit is contained in:
@@ -161,6 +161,64 @@ async def test_radio_media_fields_update_validate(monkeypatch: pytest.MonkeyPatc
|
||||
assert "emitrange must be between 5 and 20" in send_payloads[-1].message.lower()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_item_update_strips_unknown_params(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||
ws = _fake_ws()
|
||||
client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
|
||||
server.clients[ws] = client
|
||||
item = server.item_service.default_item(client, "radio_station")
|
||||
server.item_service.add_item(item)
|
||||
|
||||
send_payloads: list[object] = []
|
||||
|
||||
async def fake_send(websocket: ServerConnection, packet: object) -> None:
|
||||
send_payloads.append(packet)
|
||||
|
||||
async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
|
||||
return
|
||||
|
||||
monkeypatch.setattr(server, "_send", fake_send)
|
||||
monkeypatch.setattr(server, "_broadcast", fake_broadcast)
|
||||
|
||||
await server._handle_message(
|
||||
client,
|
||||
json.dumps({"type": "item_update", "itemId": item.id, "params": {"mediaVolume": 25, "hackedFlag": True}}),
|
||||
)
|
||||
assert send_payloads[-1].ok is True
|
||||
assert item.params.get("mediaVolume") == 25
|
||||
assert "hackedFlag" not in item.params
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_item_use_revalidates_updated_params(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||
ws = _fake_ws()
|
||||
client = ClientConnection(websocket=ws, id="u1", nickname="tester", x=5, y=6)
|
||||
server.clients[ws] = client
|
||||
item = server.item_service.default_item(client, "widget")
|
||||
item.params["hackedFlag"] = True
|
||||
server.item_service.add_item(item)
|
||||
|
||||
send_payloads: list[object] = []
|
||||
|
||||
async def fake_send(websocket: ServerConnection, packet: object) -> None:
|
||||
send_payloads.append(packet)
|
||||
|
||||
async def fake_broadcast(packet: object, exclude: ServerConnection | None = None) -> None:
|
||||
return
|
||||
|
||||
monkeypatch.setattr(server, "_send", fake_send)
|
||||
monkeypatch.setattr(server, "_broadcast", fake_broadcast)
|
||||
monkeypatch.setattr(server.item_service, "now_ms", lambda: 40_000)
|
||||
|
||||
await server._handle_message(client, json.dumps({"type": "item_use", "itemId": item.id}))
|
||||
|
||||
assert send_payloads[-1].ok is True
|
||||
assert item.params.get("enabled") is False
|
||||
assert "hackedFlag" not in item.params
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_clock_use_reports_time_without_use_sound_packet(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
server = SignalingServer("127.0.0.1", 8765, None, None)
|
||||
|
||||
Reference in New Issue
Block a user