From 7a531d62ffd3be8814f0c62dcdcf5e3936f88e21 Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Tue, 18 Aug 2026 11:54:24 -0700 Subject: [PATCH] feat(protocol): validate GET and POST request options --- README.md | 18 ++ smartthings_local/protocol/coap.py | 10 +- smartthings_local/protocol/dtls_session.py | 114 +++++++- tests/test_public_api_contract.py | 9 + tests/test_request_options.py | 293 +++++++++++++++++++++ 5 files changed, 433 insertions(+), 11 deletions(-) create mode 100644 tests/test_request_options.py diff --git a/README.md b/README.md index 3d88677..7ddbfd1 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,24 @@ sess.subscribe(["operational", "state", "vs", "0"], # OBSERVE sess.close() ``` +`get()` and `post()` accept repeated URI-query strings. They also accept +ordered `(number, bytes)` options for reviewed OCF extensions while retaining +ownership of path, query, content-format, Accept, Observe, and blockwise +options: + +```python +code, body = sess.get( + ["oic", "res"], + query=("rt=oic.r.doxm",), + extra_options=((2049, b"\x08\x00"),), +) +``` + +Path and query text is UTF-8 encoded, and every value is size bounded before +anything is sent. Additional options remain arbitrary bytes, must already be +ordered by option number, and may repeat a number when the option is +repeatable. + `connect()` uses a 12-second monotonic DTLS handshake deadline by default. A caller that needs a shorter bounded attempt can pass a positive finite value without changing later reader timeouts. OpenSSL's DTLS timer schedules flight diff --git a/smartthings_local/protocol/coap.py b/smartthings_local/protocol/coap.py index b384fc0..5d0a7dc 100644 --- a/smartthings_local/protocol/coap.py +++ b/smartthings_local/protocol/coap.py @@ -18,7 +18,9 @@ CONTENT_FORMAT = 12 ACCEPT = 17 BLOCK2 = 23 +BLOCK1 = 27 SIZE2 = 28 +SIZE1 = 60 # CoAP message types TYPE_CON = 0 @@ -249,12 +251,13 @@ def _option_bytes(value, *, name): def build_get_request( mtype, mid, token, path_segs, query=(), *, accept=CF_CBOR, - block_number=None, block_szx=BLOCK_SZX): - """Build a GET with optional Uri-Query, Accept, and Block2 options. + block_number=None, block_szx=BLOCK_SZX, extra_options=()): + """Build a GET with optional query, Block2, and extension options. ``block_number=None`` omits Block2 for the initial request. Continuation requests pass the accumulator's ``expected_number`` and ``szx``. Path and - query values may be either text or already encoded bytes. + query values may be either text or already encoded bytes. Extension + options are expected to have been validated by the session. """ options = [ (URI_PATH, _option_bytes(segment, name='path segment')) @@ -278,6 +281,7 @@ def build_get_request( or not 0 <= block_szx <= BLOCK_SZX): raise ValueError('block_szx must be between 0 and 6') options.append((BLOCK2, block_value(block_number, 0, block_szx))) + options.extend(extra_options) return build_coap(mtype, METHOD_GET, mid, token, options) diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index d362100..ccb30d5 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -54,6 +54,7 @@ ) from .coap import ( ACCEPT, + BLOCK1, BLOCK2, BLOCK2_COMPLETE, CF_CBOR, @@ -67,8 +68,11 @@ RESPONSE_EMPTY_ACK, RESPONSE_MESSAGE, RESPONSE_RESET, + SIZE1, + SIZE2, TYPE_CON, URI_PATH, + URI_QUERY, Block2Accumulator, block_fields, build_coap, @@ -178,6 +182,75 @@ def _validate_handshake_timeout(timeout, default): return value +_MAX_REQUEST_OPTION_BYTES = 1024 +_MAX_REQUEST_OPTION_COUNT = 32 +_MAX_REQUEST_OPTION_NUMBER = 65535 +_MANAGED_REQUEST_OPTIONS = frozenset(( + URI_PATH, URI_QUERY, OBSERVE, CONTENT_FORMAT, ACCEPT, + BLOCK2, BLOCK1, SIZE2, SIZE1, +)) + + +def _validated_text_options(values, *, name, allow_empty): + """Return bounded UTF-8 option values without echoing caller data.""" + if isinstance(values, (str, bytes, bytearray, memoryview)): + raise TypeError(f'{name} must be an iterable of strings') + try: + iterator = iter(values) + except TypeError: + raise TypeError(f'{name} must be an iterable of strings') from None + result = [] + for value in iterator: + if len(result) >= _MAX_REQUEST_OPTION_COUNT: + raise ValueError(f'{name} must contain at most 32 values') + if not isinstance(value, str): + raise TypeError(f'{name} values must be strings') + try: + encoded = value.encode('utf-8') + except UnicodeEncodeError: + raise ValueError(f'{name} values must be valid UTF-8') from None + if (not allow_empty and not encoded) or \ + len(encoded) > _MAX_REQUEST_OPTION_BYTES: + raise ValueError(f'{name} values must be non-empty and bounded') + result.append(value) + return tuple(result) + + +def _validated_extra_options(extra_options): + """Return bounded, ordered options not owned by the request methods.""" + if isinstance(extra_options, (str, bytes, bytearray, memoryview)): + raise TypeError('extra_options must contain (number, bytes) tuples') + try: + iterator = iter(extra_options) + except TypeError: + raise TypeError( + 'extra_options must contain (number, bytes) tuples') from None + result = [] + previous = -1 + for item in iterator: + if len(result) >= _MAX_REQUEST_OPTION_COUNT: + raise ValueError('extra_options must contain at most 32 values') + if not isinstance(item, tuple) or len(item) != 2: + raise TypeError( + 'extra_options must contain (number, bytes) tuples') + number, value = item + if isinstance(number, bool) or not isinstance(number, int): + raise TypeError('extra option numbers must be integers') + if not 1 <= number <= _MAX_REQUEST_OPTION_NUMBER: + raise ValueError('extra option numbers must be bounded') + if number < previous: + raise ValueError('extra options must be ordered by number') + if number in _MANAGED_REQUEST_OPTIONS: + raise ValueError('extra option is managed by the CoAP transport') + if not isinstance(value, bytes): + raise TypeError('extra option values must be bytes') + if len(value) > _MAX_REQUEST_OPTION_BYTES: + raise ValueError('extra option values must be bounded') + result.append((number, value)) + previous = number + return tuple(result) + + class ConnectCancellation: """One-way, socket-backed cancellation signal for ``connect()``. @@ -958,7 +1031,7 @@ def _refetch_one(self, href, seq): # ---- request primitives ------------------------------------------ - def get(self, path_segs, query=(), timeout=10.0): + def get(self, path_segs, query=(), timeout=10.0, *, extra_options=()): """Token-stable Block2 GET. Returns (code, payload_bytes). Reuses one CoAP token across every block of a multi-block @@ -966,11 +1039,17 @@ def get(self, path_segs, query=(), timeout=10.0): token, and dropping a fresh token on block 1+ silently drops the request.""" self._check_live() + path_segs = _validated_text_options( + path_segs, name='path_segs', allow_empty=False) + query = _validated_text_options( + query, name='query', allow_empty=False) + extra_options = _validated_extra_options(extra_options) code, blob, _blocks, _tok = self._blockwise_get( - path_segs, query, timeout) + path_segs, query, timeout, extra_options=extra_options) return code, blob - def _blockwise_get(self, path_segs, query=(), timeout=10.0): + def _blockwise_get( + self, path_segs, query=(), timeout=10.0, *, extra_options=()): """Shared token-stable Block2 reassembly (RFC 7959 §2.4). Returns (code, payload, block_count, token). The last two are @@ -989,19 +1068,22 @@ def _blockwise_get(self, path_segs, query=(), timeout=10.0): when the server supplies them. None of the tested appliances emit option 4, so on those this is inert.""" try: - return self._blockwise_get_once(path_segs, query, timeout) + return self._blockwise_get_once( + path_segs, query, timeout, extra_options) except _EtagChanged: logger.debug("GET %s /%s: ETag changed mid-transfer, restarting", self.host, '/'.join(path_segs)) try: - return self._blockwise_get_once(path_segs, query, timeout) + return self._blockwise_get_once( + path_segs, query, timeout, extra_options) except _EtagChanged: logger.debug( "GET %s /%s: representation kept changing mid-transfer", self.host, '/'.join(path_segs)) raise BlockwiseError() from None - def _blockwise_get_once(self, path_segs, query, timeout): + def _blockwise_get_once( + self, path_segs, query, timeout, extra_options): """One attempt at a full Block2 transfer. Raises _EtagChanged if the server's representation changed while we were reassembling.""" tok = self._next_tok() @@ -1018,6 +1100,7 @@ def _blockwise_get_once(self, path_segs, query, timeout): num, accumulator.szx, deadline, + extra_options, ) prior_blocks = accumulator.blocks_received @@ -1045,7 +1128,9 @@ def _blockwise_get_once(self, path_segs, query, timeout): raise BlockwiseError() - def _exchange_block(self, tok, path_segs, query, num, szx, deadline): + def _exchange_block( + self, tok, path_segs, query, num, szx, deadline, + extra_options): """Send one block request under `tok` and return its response message, retransmitting up to _BLOCK_MAX_ATTEMPTS times. @@ -1070,6 +1155,7 @@ def _exchange_block(self, tok, path_segs, query, num, szx, deadline): query, block_number=num if num > 0 else None, block_szx=szx, + extra_options=extra_options, ) try: for attempt in range(_BLOCK_MAX_ATTEMPTS): @@ -1167,7 +1253,9 @@ def _block_num_matches(message, num, szx): response_offset = response_num << (response_szx + 4) return response_offset == requested_offset - def post(self, path_segs, body_cbor, timeout=8.0): + def post( + self, path_segs, body_cbor, timeout=8.0, *, query=(), + extra_options=()): """Single-frame POST with a CBOR-encoded body. Returns (code, payload_bytes). body_cbor must already be encoded. @@ -1181,10 +1269,20 @@ def post(self, path_segs, body_cbor, timeout=8.0): retry cannot offer that, since it mints a fresh MID and token. Defaults to one attempt — see _WRITE_ACK_TIMEOUT.""" self._check_live() + path_segs = _validated_text_options( + path_segs, name='path_segs', allow_empty=False) + query = _validated_text_options( + query, name='query', allow_empty=False) + extra_options = _validated_extra_options(extra_options) + if not isinstance(body_cbor, bytes): + raise TypeError('body_cbor must be bytes') tok = self._next_tok() opts = [(URI_PATH, s.encode()) for s in path_segs] + for q in query: + opts.append((URI_QUERY, q.encode())) opts.append((CONTENT_FORMAT, CF_CBOR)) opts.append((ACCEPT, CF_CBOR)) + opts.extend(extra_options) ev = threading.Event() container = {} mid, exchange = self._register_pending_request(tok, ev, container) diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index e08148f..9f805f1 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -206,6 +206,11 @@ def test_dtls_session_keeps_current_consumer_methods(): "timeout", ], ) + get_extra_options = inspect.signature(DtlsCoapSession.get).parameters[ + "extra_options" + ] + assert get_extra_options.kind is inspect.Parameter.KEYWORD_ONLY + assert get_extra_options.default == () _assert_compatible_signature( DtlsCoapSession.post, [ @@ -215,6 +220,10 @@ def test_dtls_session_keeps_current_consumer_methods(): "timeout", ], ) + post_parameters = inspect.signature(DtlsCoapSession.post).parameters + for name in ("query", "extra_options"): + assert post_parameters[name].kind is inspect.Parameter.KEYWORD_ONLY + assert post_parameters[name].default == () _assert_compatible_signature( DtlsCoapSession.subscribe, ["self", "path_segs"], diff --git a/tests/test_request_options.py b/tests/test_request_options.py new file mode 100644 index 0000000..486049a --- /dev/null +++ b/tests/test_request_options.py @@ -0,0 +1,293 @@ +"""Validated URI query and additional CoAP request option tests.""" + +from __future__ import annotations + +import pytest + +import smartthings_local.protocol.dtls_session as dtls_session +from smartthings_local.protocol.coap import ( + ACCEPT, + BLOCK1, + BLOCK2, + CONTENT_FORMAT, + METHOD_GET, + METHOD_POST, + OBSERVE, + SIZE1, + SIZE2, + TYPE_ACK, + URI_PATH, + URI_QUERY, + block_value, + build_coap, + parse_coap, +) +from smartthings_local.protocol.dtls_session import DtlsCoapSession + +_ROUTING_OPTION = (65524, b"\xc0") +_VERSION_OPTION = (2049, b"\x08\x00") + + +class _NullAuth: + def configure_context(self, _context): + return None + + +def _session(responder, **session_kwargs): + session = DtlsCoapSession( + "device.example", + 5684, + auth=_NullAuth(), + rate_limit_rps=1_000_000, + **session_kwargs, + ) + session.conn = object() + requests = [] + + def send(datagram): + request = parse_coap(datagram) + requests.append(request) + response = responder(request, len(requests) - 1) + if response is not None: + session._dispatch_coap(response) + + session._send_dgram = send + return session, requests + + +def _response(request, payload=b"ok", *, options=()): + _mtype, _code, mid, token, _options, _payload = request + return build_coap(TYPE_ACK, 0x45, mid, token, options, payload) + + +def test_get_keeps_query_and_extra_options_on_every_block(): + def respond(request, request_number): + if request_number == 0: + return _response( + request, + b"a" * 1024, + options=((BLOCK2, block_value(0, 1, 6)),), + ) + return _response( + request, + b"done", + options=((BLOCK2, block_value(1, 0, 6)),), + ) + + session, requests = _session(respond) + code, payload = session.get( + ["oic", "res"], + query=("rt=oic.r.doxm", "if=oic.if.baseline"), + extra_options=(_VERSION_OPTION, _ROUTING_OPTION), + ) + + assert code == 0x45 + assert payload == b"a" * 1024 + b"done" + assert len(requests) == 2 + assert all(request[1] == METHOD_GET for request in requests) + for request in requests: + options = request[4] + assert [value for number, value in options if number == URI_PATH] == [ + b"oic", + b"res", + ] + assert [value for number, value in options if number == URI_QUERY] == [ + b"rt=oic.r.doxm", + b"if=oic.if.baseline", + ] + assert _VERSION_OPTION in options + assert _ROUTING_OPTION in options + assert not [value for number, value in requests[0][4] if number == BLOCK2] + assert [value for number, value in requests[1][4] if number == BLOCK2] == [ + block_value(1, 0, 6) + ] + + +def test_get_retransmission_keeps_query_and_extension_wire_bytes(monkeypatch): + monkeypatch.setattr(dtls_session, "_BLOCK_ACK_TIMEOUT", 0.001) + + def respond(request, request_number): + return _response(request) if request_number == 1 else None + + session, requests = _session(respond) + datagrams = [] + send = session._send_dgram + + def record_and_send(datagram): + datagrams.append(datagram) + send(datagram) + + session._send_dgram = record_and_send + + assert session.get( + ["oic", "res"], + timeout=1.0, + query=("if=oic.if.baseline",), + extra_options=(_VERSION_OPTION, _ROUTING_OPTION), + ) == (0x45, b"ok") + + assert len(requests) == 2 + assert datagrams[0] == datagrams[1] + assert _VERSION_OPTION in requests[1][4] + assert _ROUTING_OPTION in requests[1][4] + + +def test_post_encodes_repeated_queries_and_ordered_extra_options(): + session, requests = _session(lambda request, _number: _response(request)) + + assert session.post( + ["mode", "vs", "0"], + b"payload", + query=("if=oic.if.a", "x.example=value"), + extra_options=(_VERSION_OPTION, _ROUTING_OPTION), + ) == (0x45, b"ok") + + assert len(requests) == 1 + _mtype, code, _mid, _token, options, payload = requests[0] + assert code == METHOD_POST + assert payload == b"payload" + assert [value for number, value in options if number == URI_QUERY] == [ + b"if=oic.if.a", + b"x.example=value", + ] + assert _VERSION_OPTION in options + assert _ROUTING_OPTION in options + + +def test_post_retransmission_keeps_query_and_extension_wire_bytes(monkeypatch): + monkeypatch.setattr(dtls_session, "_WRITE_ACK_TIMEOUT", 0.001) + + def respond(request, request_number): + return _response(request) if request_number == 1 else None + + session, requests = _session(respond, write_max_attempts=2) + datagrams = [] + send = session._send_dgram + + def record_and_send(datagram): + datagrams.append(datagram) + send(datagram) + + session._send_dgram = record_and_send + + assert session.post( + ["mode", "vs", "0"], + b"payload", + timeout=1.0, + query=("if=oic.if.a",), + extra_options=(_VERSION_OPTION, _ROUTING_OPTION), + ) == (0x45, b"ok") + + assert len(requests) == 2 + assert datagrams[0] == datagrams[1] + assert [value for number, value in requests[1][4] if number == URI_QUERY] == [ + b"if=oic.if.a" + ] + assert _VERSION_OPTION in requests[1][4] + assert _ROUTING_OPTION in requests[1][4] + + +@pytest.mark.parametrize( + ("operation", "error_type"), + ( + (lambda session: session.get("oic"), TypeError), + (lambda session: session.get([1]), TypeError), + (lambda session: session.get(["x" * 1025]), ValueError), + (lambda session: session.get(["\ud800"]), ValueError), + (lambda session: session.get(["oic"], query="x=y"), TypeError), + (lambda session: session.get(["oic"], query=(1,)), TypeError), + (lambda session: session.get(["oic"], query=("",)), ValueError), + (lambda session: session.get(["oic"], query=("x" * 1025,)), ValueError), + ( + lambda session: session.get(["oic"], query=("x=y",) * 33), + ValueError, + ), + (lambda session: session.post(["oic"], bytearray(b"x")), TypeError), + ( + lambda session: session.get(["oic"], extra_options=(("2049", b"x"),)), + TypeError, + ), + ( + lambda session: session.get( + ["oic"], extra_options=((2049, bytearray(b"x")),) + ), + TypeError, + ), + ( + lambda session: session.get(["oic"], extra_options=((2049, b"x" * 1025),)), + ValueError, + ), + ( + lambda session: session.get( + ["oic"], extra_options=((65524, b"x"), (2049, b"y")) + ), + ValueError, + ), + ( + lambda session: session.get(["oic"], extra_options=((0, b"x"),)), + ValueError, + ), + ( + lambda session: session.get(["oic"], extra_options=((65536, b"x"),)), + ValueError, + ), + ( + lambda session: session.get( + ["oic"], extra_options=tuple((2049, b"x") for _ in range(33)) + ), + ValueError, + ), + ), +) +def test_invalid_request_options_fail_before_send(operation, error_type): + session, requests = _session(lambda request, _number: _response(request)) + + with pytest.raises(error_type): + operation(session) + + assert requests == [] + + +@pytest.mark.parametrize( + "option_number", + ( + URI_PATH, + URI_QUERY, + OBSERVE, + CONTENT_FORMAT, + ACCEPT, + BLOCK2, + BLOCK1, + SIZE2, + SIZE1, + ), +) +def test_transport_managed_options_cannot_be_overridden(option_number): + session, requests = _session(lambda request, _number: _response(request)) + + with pytest.raises(ValueError, match="managed"): + session.get(["oic"], extra_options=((option_number, b"x"),)) + + assert requests == [] + + +def test_repeated_additional_option_numbers_preserve_order(): + session, requests = _session(lambda request, _number: _response(request)) + repeated = ((2049, b"a"), (2049, b"b")) + + session.get(["oic"], extra_options=repeated) + + assert [item for item in requests[0][4] if item[0] == 2049] == list(repeated) + + +def test_validation_errors_do_not_echo_option_values(): + session, _requests = _session(lambda request, _number: _response(request)) + private_value = b"credential-value" + + with pytest.raises(ValueError) as error: + session.get( + ["oic"], + extra_options=((65524, private_value), (2049, b"out-of-order")), + ) + + assert private_value.decode() not in str(error.value)