From 9f8a6bdcc83cdd015d3c589021c041e48f97f2d7 Mon Sep 17 00:00:00 2001 From: Jage9 Date: Tue, 24 Feb 2026 02:39:51 -0500 Subject: [PATCH] Enforce strict item params validation and stripping on server --- server/app/items/clock.py | 5 ++- server/app/items/dice.py | 4 +- server/app/items/helpers.py | 6 +++ server/app/items/piano.py | 4 +- server/app/items/radio.py | 20 +++++---- server/app/items/wheel.py | 4 +- server/app/items/widget.py | 18 +++++++- server/app/server.py | 6 ++- server/tests/test_item_use_cooldown.py | 58 ++++++++++++++++++++++++++ 9 files changed, 110 insertions(+), 15 deletions(-) diff --git a/server/app/items/clock.py b/server/app/items/clock.py index 8f7e4b8..8018789 100644 --- a/server/app/items/clock.py +++ b/server/app/items/clock.py @@ -6,7 +6,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem -from .helpers import parse_bool_like_or_none +from .helpers import keep_only_known_params, parse_bool_like_or_none LABEL = "clock" TOOLTIP = "It tells the time. What did you think it did?" @@ -62,6 +62,7 @@ TIME_ZONE_OPTIONS: tuple[str, ...] = ( "UTC", ) DEFAULT_PARAMS: dict = {"timeZone": DEFAULT_TIME_ZONE, "use24Hour": False} +PARAM_KEYS: tuple[str, ...] = ("timeZone", "use24Hour") PROPERTY_METADATA: dict[str, dict[str, object]] = { "title": {"valueType": "text", "tooltip": "Display name spoken and shown for this item.", "maxLength": 80}, @@ -81,7 +82,7 @@ def validate_update(_item: WorldItem, next_params: dict) -> dict: raise ValueError("use24Hour must be on/off.") next_params["timeZone"] = time_zone next_params["use24Hour"] = use_24_hour - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/items/dice.py b/server/app/items/dice.py index 0ec9df9..0b283ba 100644 --- a/server/app/items/dice.py +++ b/server/app/items/dice.py @@ -7,6 +7,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem +from .helpers import keep_only_known_params LABEL = "dice" TOOLTIP = "Great for drinking games or boredom." @@ -19,6 +20,7 @@ EMIT_RANGE = 15 DIRECTIONAL = False DEFAULT_TITLE = "Dice" DEFAULT_PARAMS: dict = {"sides": 6, "number": 2} +PARAM_KEYS: tuple[str, ...] = ("sides", "number") PROPERTY_METADATA: dict[str, dict[str, object]] = { "title": {"valueType": "text", "tooltip": "Display name spoken and shown for this item.", "maxLength": 80}, @@ -47,7 +49,7 @@ def validate_update(_item: WorldItem, next_params: dict) -> dict: raise ValueError("Dice sides and number must be between 1 and 100.") next_params["sides"] = sides next_params["number"] = number - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, _clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/items/helpers.py b/server/app/items/helpers.py index 5add62e..e2abd51 100644 --- a/server/app/items/helpers.py +++ b/server/app/items/helpers.py @@ -41,3 +41,9 @@ def toggle_bool_param(params: dict, key: str, *, default: bool = True) -> bool: current = parse_bool_like(params.get(key), default=default) return not current + +def keep_only_known_params(params: dict, allowed_keys: tuple[str, ...]) -> dict: + """Return a copy containing only explicitly allowed item param keys.""" + + allowed = set(allowed_keys) + return {key: value for key, value in params.items() if key in allowed} diff --git a/server/app/items/piano.py b/server/app/items/piano.py index fff8ebe..8f9e07d 100644 --- a/server/app/items/piano.py +++ b/server/app/items/piano.py @@ -6,6 +6,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem +from .helpers import keep_only_known_params LABEL = "piano" TOOLTIP = "Playable keyboard instrument with multiple synth voices." @@ -38,6 +39,7 @@ DEFAULT_PARAMS: dict = { "emitRange": 15, "songId": "unterlandersheimweh", } +PARAM_KEYS: tuple[str, ...] = ("instrument", "voiceMode", "octave", "attack", "decay", "release", "brightness", "emitRange", "songId") INSTRUMENT_OPTIONS: tuple[str, ...] = ( "piano", @@ -179,7 +181,7 @@ def validate_update(_item: WorldItem, next_params: dict) -> dict: if isinstance(preserved_song_id, str) and preserved_song_id.strip(): next_params["songId"] = preserved_song_id.strip() - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, _clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/items/radio.py b/server/app/items/radio.py index c18faf9..6be9c9b 100644 --- a/server/app/items/radio.py +++ b/server/app/items/radio.py @@ -6,7 +6,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem -from .helpers import toggle_bool_param +from .helpers import keep_only_known_params, toggle_bool_param LABEL = "radio" TOOLTIP = "Can play stations from the Internet. Tune multiple to the same station and they will sync up." @@ -38,6 +38,16 @@ DEFAULT_PARAMS: dict = { "facing": 0, "emitRange": 20, } +PARAM_KEYS: tuple[str, ...] = ( + "streamUrl", + "enabled", + "mediaVolume", + "mediaChannel", + "mediaEffect", + "mediaEffectValue", + "facing", + "emitRange", +) CHANNEL_OPTIONS: tuple[str, ...] = ("stereo", "mono", "left", "right") EFFECT_OPTIONS: tuple[str, ...] = ("reverb", "echo", "flanger", "high_pass", "low_pass", "off") @@ -62,6 +72,7 @@ PROPERTY_METADATA: dict[str, dict[str, object]] = { "valueType": "number", "tooltip": "Facing direction in degrees used for directional emit.", "range": {"min": 0, "max": 360, "step": 1}, + "visibleWhen": {"directional": True}, }, "emitRange": { "valueType": "number", @@ -77,7 +88,6 @@ def validate_update(item: WorldItem, next_params: dict) -> dict: stream_url = str(next_params.get("streamUrl", "")).strip() if len(stream_url) > 2048: raise ValueError("streamUrl must be 2048 characters or less.") - previous_stream_url = str(item.params.get("streamUrl", "")).strip() next_params["streamUrl"] = stream_url enabled_value = next_params.get("enabled", True) @@ -95,10 +105,6 @@ def validate_update(item: WorldItem, next_params: dict) -> dict: raise ValueError("enabled must be true/false or on/off.") else: raise ValueError("enabled must be true/false or on/off.") - if stream_url and stream_url != previous_stream_url: - enabled = True - if not stream_url: - enabled = False next_params["enabled"] = enabled try: @@ -142,7 +148,7 @@ def validate_update(item: WorldItem, next_params: dict) -> dict: if not (5 <= emit_range <= 20): raise ValueError("emitRange must be between 5 and 20.") next_params["emitRange"] = emit_range - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, _clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/items/wheel.py b/server/app/items/wheel.py index 85fdd0b..d7dc2ce 100644 --- a/server/app/items/wheel.py +++ b/server/app/items/wheel.py @@ -7,6 +7,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem +from .helpers import keep_only_known_params LABEL = "wheel" TOOLTIP = "Spin to win fabulous prizes." @@ -19,6 +20,7 @@ EMIT_RANGE = 15 DIRECTIONAL = False DEFAULT_TITLE = "wheel" DEFAULT_PARAMS: dict = {"spaces": "yes, no"} +PARAM_KEYS: tuple[str, ...] = ("spaces",) PROPERTY_METADATA: dict[str, dict[str, object]] = { "title": {"valueType": "text", "tooltip": "Display name spoken and shown for this item.", "maxLength": 80}, @@ -46,7 +48,7 @@ def validate_update(_item: WorldItem, next_params: dict) -> dict: if any(len(token) > 80 for token in spaces): raise ValueError("each space must be 80 chars or less.") next_params["spaces"] = ", ".join(spaces) - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, _clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/items/widget.py b/server/app/items/widget.py index d2a01d1..2eda0ad 100644 --- a/server/app/items/widget.py +++ b/server/app/items/widget.py @@ -6,7 +6,7 @@ from typing import Callable from ..item_types import ItemUseResult from ..models import WorldItem -from .helpers import parse_bool_like, toggle_bool_param +from .helpers import keep_only_known_params, parse_bool_like, toggle_bool_param LABEL = "widget" TOOLTIP = "A basic item. Make it a beacon or whatever you want." @@ -44,6 +44,19 @@ DEFAULT_PARAMS: dict = { "useSound": "", "emitSound": "", } +PARAM_KEYS: tuple[str, ...] = ( + "enabled", + "directional", + "facing", + "emitRange", + "emitVolume", + "emitSoundSpeed", + "emitSoundTempo", + "emitEffect", + "emitEffectValue", + "useSound", + "emitSound", +) EFFECT_OPTIONS: tuple[str, ...] = ("reverb", "echo", "flanger", "high_pass", "low_pass", "off") PROPERTY_METADATA: dict[str, dict[str, object]] = { @@ -54,6 +67,7 @@ PROPERTY_METADATA: dict[str, dict[str, object]] = { "valueType": "number", "tooltip": "Facing direction in degrees used when directional is on.", "range": {"min": 0, "max": 360, "step": 1}, + "visibleWhen": {"directional": True}, }, "emitRange": { "valueType": "number", @@ -181,7 +195,7 @@ def validate_update(item: WorldItem, next_params: dict) -> dict: raise ValueError("useSound must be 2048 characters or less.") if len(next_params["emitSound"]) > 2048: raise ValueError("emitSound must be 2048 characters or less.") - return next_params + return keep_only_known_params(next_params, PARAM_KEYS) def use_item(item: WorldItem, nickname: str, _clock_formatter: Callable[[dict], str]) -> ItemUseResult: diff --git a/server/app/server.py b/server/app/server.py index ed46fb8..4125f7a 100644 --- a/server/app/server.py +++ b/server/app/server.py @@ -1005,7 +1005,11 @@ class SignalingServer: return if use_result.updated_params is not None: - item.params = use_result.updated_params + try: + item.params = handler.validate_update(item, {**item.params, **use_result.updated_params}) + except ValueError as exc: + await self._send_item_result(client, False, "use", str(exc), item.id) + return item.updatedAt = now_ms self.item_service.save_state() await self._broadcast_item(item) diff --git a/server/tests/test_item_use_cooldown.py b/server/tests/test_item_use_cooldown.py index 1d8e9cb..87df17d 100644 --- a/server/tests/test_item_use_cooldown.py +++ b/server/tests/test_item_use_cooldown.py @@ -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)