diff --git a/.github/mutations/plan.json b/.github/mutations/plan.json index 7a281d7..d399a7a 100644 --- a/.github/mutations/plan.json +++ b/.github/mutations/plan.json @@ -152,5 +152,19 @@ "find": " self.coordinator_context = getattr(description, \"ssc_path\", None)", "replace": " self.coordinator_context = None", "expect": "tests/test_integration_e2e.py::test_every_optional_path_has_an_entity_asking_for_it" + }, + { + "name": "a late answer with another xid is taken as the answer", + "path": "custom_components/neumann_kh/ssc_client.py", + "find": " if _is_late(reply_xid, xid):", + "replace": " if False:", + "expect": "tests/test_ssc_client.py::test_a_line_with_another_xid_is_dropped" + }, + { + "name": "a request over the token limit still carries an xid", + "path": "custom_components/neumann_kh/ssc_client.py", + "find": " if \"osc\" in payload or json_tokens(payload) + _XID_TOKENS > _MAX_REQUEST_TOKENS:", + "replace": " if \"osc\" in payload:", + "expect": "tests/test_ssc_client.py::test_a_request_over_the_token_limit_goes_without_an_xid" } ] diff --git a/README.de.md b/README.de.md index 2eebe33..4a5a6fd 100644 --- a/README.de.md +++ b/README.de.md @@ -321,15 +321,15 @@ Entities sind „nicht verfügbar“, bis eine wieder geschlossen wird. - `ui/input_select` bleibt nur lesend. Der schreibbare Pfad ist `audio/in/interface`, bestätigt auf KH 120 II und KH 750 DSP und dort standardmäßig aktiviert. -- Antwortet ein Lautsprecher auf eine einzelne Abfrage mit mehr als einer - Zeile, kann in einem engen Fall eine Folgezeile der einen Antwort als die - nächste ausgegeben werden. Die Integration sendet keine Transaktionskennung - (`osc/xid`, die beide Testgeräte zurückgeben), eine Folgezeile für den gerade - abgefragten Pfad ist von einer frischen Antwort darauf also nicht zu - unterscheiden. Bereits anliegende Zeilen werden - vor der nächsten Anfrage verworfen; betroffen ist nur eine, die genau - dazwischen eintrifft. Beide Testgeräte antworten mit exakt einer Zeile, das - betrifft also andere Firmware, nicht die eingesetzte Hardware. +- Jede Wert-Anfrage trägt eine Transaktionskennung (`osc/xid`), die beide + Testgeräte in derselben Zeile zurückgeben, auch bei Fehlerantworten; eine + Zeile mit fremder Kennung ist eine verspätete Antwort auf eine frühere Anfrage + und wird verworfen. Ohne Kennung gehen nur EQ-Schreibvorgänge mit 20 Bändern, + weil die Kennung sie über die Größengrenze der Lautsprecher brächte, und die + Schema- und Limits-Abfrage der Geräte-Discovery; ihre Antworten werden über + den Pfad zugeordnet, wo in einem engen Fall eine Folgezeile der vorigen + Antwort als die nächste durchgehen könnte. Beide Testgeräte antworten mit + exakt einer Zeile. - Der Steuerungsmodus (`ui/control_mode`) wird mit dem Schalter CONTROL auf der Rückseite eingestellt (KH 750 DSP: CONTROL MODE) und lässt sich über das Netzwerk nicht ändern, deshalb ist er ein Sensor. Auf `LOCAL` antwortet der diff --git a/README.md b/README.md index 233ebd6..fc8fe37 100644 --- a/README.md +++ b/README.md @@ -308,14 +308,14 @@ unavailable until one is closed. - `ui/input_select` stays read-only. The writable path is `audio/in/interface`, confirmed on both the KH 120 II and the KH 750 DSP and enabled by default there. -- A speaker that answers a single query with more than one line can, in a - narrow case, have a trailing line of one answer returned as the next one. - The integration sends no transaction id (`osc/xid`, which both test - speakers echo), so a trailing line for the path just requested cannot be - told apart from a fresh reply to it. Lines already waiting on the socket - are discarded before the next request; only one that - lands in between is affected. Both test devices answer with exactly one - line, so this concerns other firmware rather than the hardware in use. +- Every value request carries a transaction id (`osc/xid`), which both test + speakers echo in the same line, error replies included; a line with another + id is a late answer to an earlier request and is dropped. Only EQ writes of + 20 bands go without one, because the id would push them past the speakers' + request size limit, and the schema and limits walk of the device + discovery; their replies are matched by path, where a trailing line of the + previous answer could in a narrow case pass for the next one. Both test + devices answer with exactly one line. - Control mode (`ui/control_mode`) is set with the CONTROL switch on the back (KH 750 DSP: CONTROL MODE) and cannot be changed over the network, so it is a sensor. On `LOCAL` the speaker keeps answering, so its values stay diff --git a/custom_components/neumann_kh/_util.py b/custom_components/neumann_kh/_util.py index b5e9159..4b8308a 100644 --- a/custom_components/neumann_kh/_util.py +++ b/custom_components/neumann_kh/_util.py @@ -37,6 +37,19 @@ def build_nested(path: tuple[str, ...], value: Any) -> dict[str, Any]: return root +def json_tokens(value: Any) -> int: + """JSON tokens in a value: every object, array, key and scalar counts once. + + The way a Neumann KH 120 II counts a request against its 29-token limit + (measured: a set of a 20-entry list four levels deep is exactly 29). + """ + if isinstance(value, dict): + return 1 + sum(1 + json_tokens(child) for child in value.values()) + if isinstance(value, list): + return 1 + sum(json_tokens(child) for child in value) + return 1 + + def nesting_exceeds(value: Any, limit: int) -> bool: """Whether a parsed JSON value nests deeper than `limit`. diff --git a/custom_components/neumann_kh/ssc_client.py b/custom_components/neumann_kh/ssc_client.py index d7bf61f..73af396 100644 --- a/custom_components/neumann_kh/ssc_client.py +++ b/custom_components/neumann_kh/ssc_client.py @@ -12,13 +12,14 @@ import asyncio import contextlib +import itertools import json import logging import os import socket from typing import Any -from ._util import build_nested, deep_merge, extract, nesting_exceeds +from ._util import build_nested, deep_merge, extract, json_tokens, nesting_exceeds from .const import DEFAULT_QUERY_SETTLE, DEFAULT_TIMEOUT _LOGGER = logging.getLogger(__name__) @@ -67,6 +68,10 @@ def _os_reason(err: OSError) -> str: _MAX_LINE_BYTES = 1_048_576 # 1 MiB # Real replies nest a handful of levels (path plus osc/error); far past that is junk. _MAX_NESTING = 64 +# A KH 120 II refuses a request of more than 29 JSON tokens with 413, and +# {"osc":{"xid":n}} adds 4: a 20-band EQ set fits only without it. +_MAX_REQUEST_TOKENS = 29 +_XID_TOKENS = 4 # Upper bounds for a single response. Every incoming line refreshes the settle # window, so a device that keeps talking would otherwise hold the read open @@ -161,6 +166,45 @@ def _error_details(osc_error: Any) -> tuple[int | None, str]: return None, "" +def _reply_xid(parsed: dict[str, Any]) -> Any: + osc = parsed.get("osc") + return osc.get("xid") if isinstance(osc, dict) else None + + +def _is_late(reply_xid: Any, xid: int | None) -> bool: + """A line carrying another request's xid answers an earlier request.""" + return reply_xid is not None and reply_xid != xid + + +def _drop_xid(merged: dict[str, Any]) -> None: + """Callers walk the reply tree; the transaction id is the client's own business.""" + osc = merged.get("osc") + if isinstance(osc, dict): + osc.pop("xid", None) + if not osc: + del merged["osc"] + + +def _answer_complete( + merged: dict[str, Any], expect_path: tuple[str, ...] | None, reply_xid: Any +) -> bool: + """The answer ends with an error or the requested value. + + The own xid alone ends it only where no path was asked for: a firmware + that sends the xid before the value would otherwise leave the value line + behind, to be taken as the answer to the next request. + """ + if extract(merged, ("osc", "error")) is not None: + return True + if expect_path is None: + return reply_xid is not None + if reply_xid is not None: + # Matched by xid, so a JSON null there is the answer, not a gap. + parent = extract(merged, expect_path[:-1]) + return isinstance(parent, dict) and expect_path[-1] in parent + return extract(merged, expect_path) is not None + + def _parse_reply(line: bytes) -> dict[str, Any] | None: """A reply line as an address tree, or None if the line is unusable.""" try: @@ -202,6 +246,11 @@ def __init__( # Tells the poll loop a user action is waiting (see request(priority= # True)); the poll then pauses before its next path. self._priority_waiting = asyncio.Event() + self._xids = itertools.count(1) + # Whether the last answer on this connection was matched by path only. + # Only then can a trailing line of it pass for the next answer, so + # only then is the stale drain worth its wait. + self._matched_by_path = False # Logical path -> path on this device (see use_path_aliases). self._path_aliases: dict[tuple[str, ...], tuple[str, ...]] = {} @@ -255,6 +304,7 @@ async def _ensure_connected(self) -> bool: def _drop_connection(self) -> None: """Drops the current connection; the next access reconnects.""" + self._matched_by_path = False if self._writer is not None: self._writer.close() self._writer = None @@ -350,7 +400,7 @@ async def _discard_stale_lines(self) -> None: self._drop_connection() async def _read_lines_until_settled( - self, expect_path: tuple[str, ...] | None = None + self, expect_path: tuple[str, ...] | None = None, xid: int | None = None ) -> dict[str, Any]: """Read response lines and merge them into one result dict. @@ -363,24 +413,19 @@ async def _read_lines_until_settled( The settle window still applies while the answer is incomplete, so a firmware that splits a response across lines is still handled. - Known limitation, accepted deliberately. Returning early can leave a - trailing line of that answer in flight. _discard_stale_lines() catches - one that is already queued when the next request goes out, but not one - that lands after the drain and before the real reply - and since the - client sends no transaction id (the firmware would echo osc/xid), a - trailing line for the requested path is indistinguishable from a fresh - answer to it. - - There is no cheap fix: waiting out the settle window every time is - what the early return exists to avoid, reconnecting per request costs - more, and learning from what the drain found does not help because in - exactly this case the drain sees nothing. Measured against both test - devices a leaf query is answered with exactly one line, so this is a - risk for other firmware rather than for the hardware in use. + A line carrying another request's osc/xid is a late answer and is + dropped; the line with this request's xid is the whole answer (the + device echoes it in the same line, error replies included). A request + without an xid - osc methods, a set too long to carry one - is matched + by path. That keeps the known limitation for those few: a trailing + line of the previous answer that lands after the stale drain is + indistinguishable from a fresh answer to the same path. """ if self._reader is None: raise SSCConnectionError(f"No active connection to {mask_host(self._host)}") merged: dict[str, Any] = {} + # Until a line echoes the xid, this answer counts as matched by path. + self._matched_by_path = True received = False limit_hit = False lines = 0 @@ -451,11 +496,14 @@ async def _read_lines_until_settled( ) continue + reply_xid = _reply_xid(parsed) + if _is_late(reply_xid, xid): + _LOGGER.debug("Dropped a late answer to an earlier request from %s", self._host) + continue received = True deep_merge(merged, parsed) - if extract(merged, ("osc", "error")) is not None: - break - if expect_path is not None and extract(merged, expect_path) is not None: + if _answer_complete(merged, expect_path, reply_xid): + self._matched_by_path = reply_xid is None break # Both bounds are evaluated at the top of the next pass, which is @@ -505,11 +553,13 @@ async def _locked_request( # await, so both can be cancelled. Left outside, a cancellation # hitting that window would keep a connection the caller believes # was dropped - the invariant the v1.15.0 hardening relies on. - # Nothing can be stale on a socket opened for this request. - if not await self._ensure_connected(): + xid = self._next_xid(payload) + # Nothing can be stale on a socket opened for this request, and + # a late line of an answer matched by xid carries its own id. + if not await self._ensure_connected() and self._matched_by_path: await self._discard_stale_lines() - await self._send_raw(payload) - merged = await self._read_lines_until_settled(expect_path) + await self._send_raw(payload if xid is None else {"osc": {"xid": xid}, **payload}) + merged = await self._read_lines_until_settled(expect_path, xid) except SSCConnectionError, SSCTimeoutError: # Drop the connection so the next attempt reconnects self._drop_connection() @@ -522,6 +572,7 @@ async def _locked_request( self._drop_connection() raise + _drop_xid(merged) osc_error = extract(merged, ("osc", "error")) if osc_error: code, description = _error_details(osc_error) @@ -533,6 +584,16 @@ async def _locked_request( return merged + def _next_xid(self, payload: dict[str, Any]) -> int | None: + """An id for a value request, or None where the device would not take one. + + osc methods were never measured with an xid, and a request the xid + would push past the token limit is refused (413) - both go without. + """ + if "osc" in payload or json_tokens(payload) + _XID_TOKENS > _MAX_REQUEST_TOKENS: + return None + return next(self._xids) + def use_path_aliases(self, aliases: dict[tuple[str, ...], tuple[str, ...]]) -> None: """Send these logical paths to a different path on this device. diff --git a/tests/README.md b/tests/README.md index 1c9b229..5ade7dd 100644 --- a/tests/README.md +++ b/tests/README.md @@ -100,9 +100,9 @@ headroom for the next person to spend. The measure is Cognitive Complexity (Campbell / SonarSource), implemented in `complexity.py` and calibrated against SonarQube Cloud: at calibration Sonar -reported 35 for `_read_lines_until_settled`, and so did this. (Today it is 33, -after the reply parsing moved out into `_parse_reply`.) `COMPLEXITY_LIMIT` is 15, -SonarSource's default. +reported 35 for `_read_lines_until_settled`, and so did this. It is lower +today, since parts moved into helpers; `test_budgets.py` holds the current +value. `COMPLEXITY_LIMIT` is 15, SonarSource's default. ## Adding a guard diff --git a/tests/complexity.py b/tests/complexity.py index 7842dc0..0c63749 100644 --- a/tests/complexity.py +++ b/tests/complexity.py @@ -3,7 +3,8 @@ A measuring stick for the budget guard next door, not a general-purpose implementation. Calibrated against SonarQube Cloud during the complexity round of 1.18.1: Sonar reported 35 for `_read_lines_until_settled`, and this -scorer returned 35 for it then (33 since `_parse_reply` was split out). +scorer returned 35 for it then (lower since parts moved into helpers; +test_budgets.py holds the current value). Three rules: diff --git a/tests/recorded_replies.py b/tests/recorded_replies.py index c913bc5..7fe7ecd 100644 --- a/tests/recorded_replies.py +++ b/tests/recorded_replies.py @@ -129,6 +129,13 @@ '{"osc":{"error":[{"audio":{"out":{"mute":[401,{"desc":"authorisation needed"}]}}}]}}', ) +# On LOCAL a read-only leaf keeps its 405: writability is checked before the +# switch (KH 120 II, 2026-11-01, sent with an xid; the xid is left out here). +LOCAL_MODE_READ_ONLY_SET_REPLY = ( + {"ui": {"control_mode": "LOCAL"}}, + '{"osc":{"error":[{"ui":{"control_mode":[405,{"desc":"method not allowed"}]}}]}}', +) + # osc/schema and osc/limits - answered by both models in exactly the form the # integration sends (an older note said 400; it is not). SCHEMA_ROOT_REPLY = ( @@ -164,6 +171,42 @@ ), ) +# Error replies to requests with an xid, both models alike (2026-11-01): the +# xid sits next to "error" in the same osc object, never in a second one. +XID_ERROR_REPLIES = ( + ( + {"osc": {"xid": 101}, "audio": {"out": {"dimm": None}}}, + '{"osc":{"error":[{"audio":{"out":{"dimm":[404,{"desc":"address not found"}]}}}],"xid":101}}', + ), + ( + {"osc": {"xid": 102}, "ui": {"control_mode": "NETWORK"}}, + '{"osc":{"error":[{"ui":{"control_mode":[405,{"desc":"method not allowed"}]}}],"xid":102}}', + ), + ( + {"osc": {"xid": 103}, "audio": {"out": {"mute": "x"}}}, + '{"osc":{"error":[{"audio":{"out":{"mute":[406,{"desc":"not acceptable"}]}}}],"xid":103}}', + ), + ( + {"osc": {"xid": 104}, "audio": {"out": {"level": None, "dimm": None}}}, + ( + '{"audio":{"out":{"level":100.0}},' + '"osc":{"error":[{"audio":{"out":[404,{"desc":"address not found"}]}}],"xid":104}}' + ), + ), + ( + {"osc": {"xid": 105, "nonexistent": None}}, + '{"osc":{"error":[{"osc":[404,{"desc":"address not found"}]}],"xid":105}}', + ), +) + +# More than 29 JSON tokens, both models (KH 120 II 2026-10-20, KH 750 +# 2026-11-01): refused before anything is read, so the reply carries no xid. +# Here a get bundle of 13 unknown leaves (31 tokens); 12 (29 tokens) answer 404. +REQUEST_TOO_LONG_REPLY = ( + {"osc": {"xid": 1}, "audio": {"out": {f"x{n}": None for n in range(13)}}}, + '{"osc":{"error":[413,{"desc":"request too long"}]}}', +) + # Protocol methods of the osc container and the transaction id, both models. OSC_METHOD_REPLIES = ( ({"osc": {"version": None}}, '{"osc":{"version":"1.1"}}'), diff --git a/tests/test_budgets.py b/tests/test_budgets.py index 2d815c7..3680854 100644 --- a/tests/test_budgets.py +++ b/tests/test_budgets.py @@ -42,7 +42,9 @@ # Raised to 556 for _parse_reply: the nesting limit on faulty replies, taken # out of the read loop so its complexity went down instead of up. # Raised to 559 to skip the stale drain on a connection opened just now. - "ssc_client.py": 559, + # Raised to 608 for matching replies by osc/xid (token limit, late lines). + # Raised to 620 so a split xid answer waits for its value (re-audit R1). + "ssc_client.py": 620, } # SonarSource's default. Everything above needs a written entry. @@ -51,7 +53,7 @@ # The one function this project keeps having trouble with. Kept whole by # owner decision (ledger A8, 2026-09-02) rather than split under pressure: # every previous attempt moved the branching instead of removing it. - "ssc_client.py::SSCClient._read_lines_until_settled": 33, + "ssc_client.py::SSCClient._read_lines_until_settled": 32, # Writes, counts and reports per path, with a deadline between paths and # three ways out that all have to keep what already landed (the third: # LOCAL mode, where every further write would be refused too). diff --git a/tests/test_simulator_matches_device.py b/tests/test_simulator_matches_device.py index 7304a4d..0fc9af3 100644 --- a/tests/test_simulator_matches_device.py +++ b/tests/test_simulator_matches_device.py @@ -20,15 +20,18 @@ KH_750_SWITCH_SET_REPLIES, LIMITS_LEVEL_REQUEST, LIMITS_LEVEL_TYPE, + LOCAL_MODE_READ_ONLY_SET_REPLY, LOCAL_MODE_SET_REPLY, MALFORMED_OSC_REPLIES, NOT_FOUND_REPLIES, OSC_METHOD_REPLIES, + REQUEST_TOO_LONG_REPLY, SCHEMA_AUDIO_REPLIES, SCHEMA_AUDIO_REQUEST, SCHEMA_ROOT_REPLY, SET_REPLIES, SWITCH_SET_REPLIES, + XID_ERROR_REPLIES, ) from tools.ssc_simulator import MODEL_KH_120_II, MODEL_KH_750, SSCSimulator @@ -91,6 +94,15 @@ def test_a_simulator_in_local_mode_answers_like_the_device(model): assert _get(simulator, "ui", "control_mode") == "LOCAL" +@pytest.mark.parametrize("model", MODELS) +def test_a_simulator_in_local_mode_refuses_a_read_only_leaf_with_405(model): + """The simulator guessed 401 first; the device checks writability before the switch.""" + simulator = SSCSimulator(model, local_mode=True) + request_, line = LOCAL_MODE_READ_ONLY_SET_REPLY + + assert simulator.handle(request_) == json.loads(line) + + @pytest.mark.parametrize("model", MODELS) def test_the_simulator_describes_its_tree_like_the_device(model): simulator = SSCSimulator(model) @@ -127,6 +139,34 @@ def test_the_simulator_answers_osc_methods_like_the_device(model, request_, line assert SSCSimulator(model).handle(request_) == json.loads(line) +@pytest.mark.parametrize("model", MODELS) +def test_the_simulator_refuses_a_request_over_29_tokens_like_the_device(model): + request_, line = REQUEST_TOO_LONG_REPLY + assert SSCSimulator(model).handle(request_) == json.loads(line) + + +def test_the_simulator_takes_29_tokens(): + """Counter-check: a 20-band EQ set of a KH 120 II is exactly 29 tokens and goes through.""" + simulator = SSCSimulator(MODEL_KH_120_II) + gain = _get(simulator, "audio", "out", "eq3", "gain") + assert len(gain) == 20 + assert "osc" not in simulator.handle({"audio": {"out": {"eq3": {"gain": gain}}}}) + + +@pytest.mark.parametrize("model", MODELS) +@pytest.mark.parametrize( + ("request_", "line"), + XID_ERROR_REPLIES, + ids=["404", "405", "406", "partial-bundle", "unknown-osc-method"], +) +def test_the_simulator_echoes_the_xid_on_errors_like_the_device(model, request_, line): + answer = SSCSimulator(model).handle(request_) + # The level differs per model; the shape and the xid are what is recorded. + if "audio" in answer and "level" in answer["audio"].get("out", {}): + answer["audio"]["out"]["level"] = 100.0 + assert answer == json.loads(line) + + def _get(simulator: SSCSimulator, *path: str): request: object = None for key in reversed(path): diff --git a/tests/test_ssc_client.py b/tests/test_ssc_client.py index 5edb386..44afb36 100644 --- a/tests/test_ssc_client.py +++ b/tests/test_ssc_client.py @@ -366,8 +366,10 @@ async def _hanging_drain(self) -> None: in_drain.set() await asyncio.sleep(60) # hold the request inside the drain window + # Answered without the xid: only an answer matched by path is drained + # after, and a fresh connection skips the drain, so a first request opens it. + server.responder = lambda req: [{"device": {"name": "x"}}] client = _client(server.port) - # A fresh connection skips the drain, so open it with a first request. await client.get(("device", "name")) monkeypatch.setattr(SSCClient, "_discard_stale_lines", _hanging_drain) try: @@ -966,6 +968,11 @@ def _respond(request): return _respond +def _without_xid(request: dict) -> dict: + """The value part of a request; the transaction id is tested on its own.""" + return {key: value for key, value in request.items() if key != "osc"} + + async def test_an_aliased_get_asks_the_device_path(server): received: list = [] server.responder = _recording(received, lambda req: {"audio": {"out5": {"level": 87.0}}}) @@ -973,7 +980,7 @@ async def test_an_aliased_get_asks_the_device_path(server): try: client.use_path_aliases({("audio", "out", "level"): ("audio", "out5", "level")}) assert await client.get(("audio", "out", "level")) == 87.0 - assert received == [{"audio": {"out5": {"level": None}}}] + assert [_without_xid(r) for r in received] == [{"audio": {"out5": {"level": None}}}] finally: await client.close() @@ -985,7 +992,7 @@ async def test_an_aliased_set_writes_the_device_path(server): try: client.use_path_aliases({("audio", "out", "mute"): ("audio", "out5", "mute")}) assert await client.set(("audio", "out", "mute"), True) is True - assert received == [{"audio": {"out5": {"mute": True}}}] + assert [_without_xid(r) for r in received] == [{"audio": {"out5": {"mute": True}}}] finally: await client.close() @@ -997,7 +1004,7 @@ async def test_an_unaliased_path_is_sent_unchanged(server): try: client.use_path_aliases({("audio", "out", "level"): ("audio", "out5", "level")}) await client.get(("audio", "out", "delay")) - assert received == [{"audio": {"out": {"delay": None}}}] + assert [_without_xid(r) for r in received] == [{"audio": {"out": {"delay": None}}}] finally: await client.close() @@ -1017,6 +1024,8 @@ async def _counting_drain(self) -> None: await real_drain(self) monkeypatch.setattr(SSCClient, "_discard_stale_lines", _counting_drain) + # Without the xid echoed, so the reused connection has something to drain. + server.responder = lambda req: [{"device": {"name": "x"}}] client = _client(server.port) try: await client.get(("device", "name")) @@ -1025,3 +1034,189 @@ async def _counting_drain(self) -> None: assert drains == 1, "the reused connection was not drained" finally: await client.close() + + +# --- osc/xid ----------------------------------------------------------------- +# +# Every value request carries a transaction id, which the device echoes in the +# same line - error replies included (KH 120 II and KH 750, 2026-11-01). A line +# with another id is a stale answer to an earlier request. + + +def _xid_of(request: dict) -> object: + return request.get("osc", {}).get("xid") + + +async def test_a_value_request_carries_an_xid(server): + sent: list[dict] = [] + + def _record(request): + sent.append(request) + return [request] + + server.responder = _record + client = _client(server.port) + try: + await client.get(("device", "name")) + await client.get(("device", "name")) + finally: + await client.close() + + xids = [_xid_of(request) for request in sent] + assert None not in xids, sent + assert xids[0] != xids[1], "two requests shared an xid" + + +async def test_a_line_with_another_xid_is_dropped(server): + """A late answer to an earlier request names the same path - only its xid tells. + + Without the xid it was taken as the answer: the documented limitation of + the early return, now closed for every request that carries an id. + """ + + def _stale_then_fresh(request): + stale = {"device": {"name": "old"}, "osc": {"xid": "earlier"}} + return [stale, request | {"device": {"name": "new"}}] + + server.responder = _stale_then_fresh + client = _client(server.port) + try: + assert await client.get(("device", "name")) == "new" + finally: + await client.close() + + +async def test_a_stale_error_with_another_xid_does_not_fail_the_request(server): + def _stale_error_then_fresh(request): + error = [{"device": {"name": [404, {"desc": "address not found"}]}}] + stale = {"osc": {"error": error, "xid": "earlier"}} + return [stale, request | {"device": {"name": "ok"}}] + + server.responder = _stale_error_then_fresh + client = _client(server.port) + try: + assert await client.get(("device", "name")) == "ok" + finally: + await client.close() + + +async def test_an_error_with_the_own_xid_is_the_answer(server): + def _refuse(request): + error = [{"ui": {"control_mode": [405, {"desc": "method not allowed"}]}}] + return [{"osc": {"error": error, "xid": _xid_of(request)}}] + + server.responder = _refuse + client = _client(server.port) + try: + with pytest.raises(SSCDeviceError) as err: + await client.set(("ui", "control_mode"), "NETWORK") + finally: + await client.close() + assert err.value.code == 405 + + +async def test_the_xid_does_not_reach_the_caller(server): + """Callers walk the reply tree; an osc/xid there is noise they never asked for.""" + client = _client(server.port) + try: + reply = await client.request({"device": {"name": None}}, expect_path=("device", "name")) + finally: + await client.close() + assert "osc" not in reply, reply + + +async def test_a_request_over_the_token_limit_goes_without_an_xid(server): + """A KH 120 II refuses more than 29 JSON tokens with 413; the xid costs 4. + + A set of a 20-band EQ leaf is exactly 29 tokens without it. + """ + sent: list[dict] = [] + + def _record(request): + sent.append(request) + return [request] + + server.responder = _record + client = _client(server.port) + try: + await client.set(("audio", "out", "eq3", "gain"), [0.0] * 20) + await client.set(("audio", "out", "eq2", "gain"), [0.0] * 10) + finally: + await client.close() + + assert _xid_of(sent[0]) is None, "the 20-band set would be refused with 413" + assert _xid_of(sent[1]) is not None + + +async def test_an_answer_matched_by_xid_needs_no_stale_drain(server, monkeypatch): + """The drain guards replies matched by path; an xid reply cannot be mistaken.""" + drains = 0 + real_drain = SSCClient._discard_stale_lines + + async def _counting_drain(self) -> None: + nonlocal drains + drains += 1 + await real_drain(self) + + monkeypatch.setattr(SSCClient, "_discard_stale_lines", _counting_drain) + client = _client(server.port) + try: + await client.get(("device", "name")) + await client.get(("device", "name")) + finally: + await client.close() + + assert drains == 0 + + +async def test_an_xid_line_without_the_value_does_not_end_the_answer(socket_enabled): + """A firmware that splits an answer may send the xid before the value. + + Ending the read on the xid line returned None, and the value line - which + carries no xid - was then taken as the answer to the next request. + """ + + async def _handle(reader, writer): + while line := await reader.readline(): + xid = json.loads(line)["osc"]["xid"] + writer.write(json.dumps({"osc": {"xid": xid}}).encode() + b"\r\n") + await writer.drain() + await asyncio.sleep(_SETTLE / 5) # within the settle window + writer.write(json.dumps({"device": {"name": f"v{xid}"}}).encode() + b"\r\n") + await writer.drain() + + raw_server, handled = await _serve(_handle) + port = raw_server.sockets[0].getsockname()[1] + client = _client(port) + try: + first = await client.get(("device", "name")) + second = await client.get(("device", "name")) + finally: + await client.close() + await _shutdown(raw_server, handled) + assert (first, second) == ("v1", "v2") + + +async def test_an_answer_ended_by_the_settle_window_is_drained_after(server, monkeypatch): + """No xid and no requested path: the read ends on the settle window. + + Such an answer is matched by nothing, so a trailing line of it could pass + for the next answer - the drain has to run before the next request. + """ + drains = 0 + real_drain = SSCClient._discard_stale_lines + + async def _counting_drain(self) -> None: + nonlocal drains + drains += 1 + await real_drain(self) + + monkeypatch.setattr(SSCClient, "_discard_stale_lines", _counting_drain) + server.responder = lambda req: [{"other": {"value": 1}}] + client = _client(server.port) + try: + await client.get(("device", "name")) + await client.get(("device", "name")) + finally: + await client.close() + assert drains == 1 diff --git a/tools/README.md b/tools/README.md index f529859..b778dc0 100644 --- a/tools/README.md +++ b/tools/README.md @@ -53,9 +53,11 @@ read no error code from a real device passed every test. `--disable-schema` rejects them. `osc/version`, `osc/ping`, `osc/feature/pattern|subscription` and the `osc/xid` echo answer as recorded. - LOCAL control mode: `SSCSimulator(local_mode=True)` (tests only, no command - line option) answers every set with `401` at the field, as recorded for a - writable leaf. Which code a read-only field gets first is not recorded; - the simulator assumes `401`. + line option) answers every set of a writable field with `401` at the + field; a read-only field keeps its `405`, as recorded. +- A request of more than 29 JSON tokens (every object, array, key and scalar) + gets a flat `413` "request too long" without an xid, measured on both + models; a 20-band EQ set is exactly 29. - Not simulated: address patterns (`*`, `?`) and subscriptions. - Malformed JSON lines are ignored, matching the client's behaviour. diff --git a/tools/ssc_simulator.py b/tools/ssc_simulator.py index 2267f9c..5970877 100644 --- a/tools/ssc_simulator.py +++ b/tools/ssc_simulator.py @@ -49,6 +49,7 @@ _ERR_NOT_FOUND = 404 _ERR_NOT_ALLOWED = 405 _ERR_NOT_ACCEPTABLE = 406 +_ERR_REQUEST_TOO_LONG = 413 _ERR_PARAMETER_NOT_FOUND = 454 _ERR_DESCRIPTIONS = { @@ -57,6 +58,7 @@ _ERR_NOT_FOUND: "address not found", _ERR_NOT_ALLOWED: "method not allowed", _ERR_NOT_ACCEPTABLE: "not acceptable", + _ERR_REQUEST_TOO_LONG: "request too long", _ERR_PARAMETER_NOT_FOUND: "parameter address not found", } @@ -452,8 +454,20 @@ def _introspection_path(value: Any, method: str) -> tuple[tuple[str, ...] | None return path, None +# Every object, array, key and scalar counts; a 20-band EQ set is exactly 29. +_MAX_REQUEST_TOKENS = 29 + + +def _json_tokens(value: Any) -> int: + if isinstance(value, dict): + return 1 + sum(1 + _json_tokens(child) for child in value.values()) + if isinstance(value, list): + return 1 + sum(_json_tokens(child) for child in value) + return 1 + + def _error(code: int) -> dict[str, Any]: - """A flat SSC error, for the cases whose real reply was never recorded (400).""" + """A flat SSC error: the recorded 413, and the never recorded cases (400).""" return {"osc": {"error": [code, {"desc": _ERR_DESCRIPTIONS[code]}]}} @@ -532,12 +546,12 @@ def _answer_leaf( return None, _error_entry(node, _ERR_NOT_FOUND) if value is None: return _build_nested(path, current), None - # Recorded for a writable leaf only; which code a read-only leaf gets - # in LOCAL mode is not, and 401 first is the guess. - if self._local_mode(): - return None, _error_entry(path, _ERR_AUTHORISATION_NEEDED) + # Recorded on LOCAL: a read-only leaf keeps its 405, only a writable + # one gets 401 - the device checks writability before the switch. if path in self.read_only: return None, _error_entry(path, _ERR_NOT_ALLOWED) + if self._local_mode(): + return None, _error_entry(path, _ERR_AUTHORISATION_NEEDED) if not _same_kind(current, value): return None, _error_entry(path, _ERR_NOT_ACCEPTABLE) self._apply_set(path, value) @@ -594,6 +608,10 @@ def _osc_limits(self, value: Any) -> dict[str, Any]: def handle(self, request: dict[str, Any]) -> dict[str, Any]: """Answer a single decoded SSC request.""" + # Recorded on both models: refused before anything is read, so a + # request over the limit gets a flat 413 and no xid. + if _json_tokens(request) > _MAX_REQUEST_TOKENS: + return _error(_ERR_REQUEST_TOO_LONG) # osc/schema and osc/limits are optional per the specification; both # tested firmwares answer them. Without them, the reply is unrecorded. if "osc" in request: @@ -606,6 +624,13 @@ def handle(self, request: dict[str, Any]) -> dict[str, Any]: if "schema" in osc: return self._osc_schema(osc["schema"]) return self._osc_limits(osc["limits"]) + unknown = set(osc) - set(_OSC_TREE) + if unknown and "xid" in osc: + # Recorded with an xid: an unknown method is a 404 at the osc + # node (two keys, so the bundle rule moves it up), xid echoed. + return { + "osc": {"error": [_error_entry(("osc",), _ERR_NOT_FOUND)], "xid": osc["xid"]} + } methods = _osc_method_answer(osc) if methods is None: return _error(_ERR_BAD_REQUEST)