From ba82fc691b79e395f9cf1e4570975e08ddb51d53 Mon Sep 17 00:00:00 2001 From: hoon <230467962+atc722@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:49:34 +0900 Subject: [PATCH 1/3] refactor(protocol): share CoAP GET response handling --- smartthings_local/protocol/coap.py | 511 ++++++++++++++++++++- smartthings_local/protocol/dtls_session.py | 251 +++++----- tests/test_coap_wire.py | 463 ++++++++++++++++++- tests/test_dtls_session_reader_death.py | 376 ++++++++++++++- 4 files changed, 1463 insertions(+), 138 deletions(-) diff --git a/smartthings_local/protocol/coap.py b/smartthings_local/protocol/coap.py index 741829f..f716190 100644 --- a/smartthings_local/protocol/coap.py +++ b/smartthings_local/protocol/coap.py @@ -6,8 +6,9 @@ independently. """ import struct +from dataclasses import dataclass -from ..errors import MalformedMessageError +from ..errors import BlockwiseError, MalformedMessageError # CoAP option numbers (RFC 7252 + 7641 + 7959) URI_PATH = 11 @@ -40,11 +41,70 @@ # will honour and the only one the probes have validated end-to-end. BLOCK_SZX = 6 +# Shared bounds for callers that assemble untrusted Block2 responses. Thirty- +# two means exactly blocks 0..31; a response that advertises another block from +# block 31 is rejected. The payload cap applies to the fully assembled body. +MAX_BLOCK2_BLOCKS = 32 +MAX_BLOCK2_PAYLOAD_BYTES = 64 * 1024 + +# ``classify_coap_response`` outcomes. Strings keep the helper lightweight for +# transports that already use their own event/state machinery. +RESPONSE_IGNORE = 'ignore' +RESPONSE_EMPTY_ACK = 'empty_ack' +RESPONSE_RESET = 'reset' +RESPONSE_MESSAGE = 'response' + +BLOCK2_DUPLICATE = 'duplicate' +BLOCK2_CONTINUE = 'continue' +BLOCK2_COMPLETE = 'complete' + + +@dataclass(frozen=True, slots=True, repr=False) +class CoapMessage: + """One decoded CoAP datagram. + + The representation is deliberately metadata-only: tokens and payloads can + contain device data and must not leak through exception/debug reprs. + """ + + mtype: int + code: int + mid: int + token: bytes + options: tuple[tuple[int, bytes], ...] + payload: bytes + + def __repr__(self): + return ( + 'CoapMessage(' + f'mtype={self.mtype!r}, code={self.code!r}, ' + f'option_count={len(self.options)}, ' + f'payload_length={len(self.payload)})' + ) + + +@dataclass(frozen=True, slots=True, repr=False) +class CoapResponseClassification: + """Transport-independent classification of a possible response.""" + + kind: str + message: CoapMessage | None = None + acknowledgement: bytes | None = None + + def __repr__(self): + return ( + 'CoapResponseClassification(' + f'kind={self.kind!r}, has_message={self.message is not None!r}, ' + f'has_acknowledgement={self.acknowledgement is not None!r})' + ) + def _vlen(v): """Variable-length integer encoder used in option deltas + lengths.""" - if v < 13: return v, b'' - if v < 269: return 13, bytes([v - 13]) + if v < 13: + return v, b'' + if v < 269: + return 13, bytes([v - 13]) return 14, struct.pack('>H', v - 269) @@ -54,17 +114,30 @@ def encode_options(opts): prev = 0 for n, val in sorted(opts, key=lambda x: x[0]): d, dx = _vlen(n - prev) - l, lx = _vlen(len(val)) - out += bytes([(d << 4) | l]) + dx + lx + val + length, lx = _vlen(len(val)) + out += bytes([(d << 4) | length]) + dx + lx + val prev = n return out def parse_coap(data): """Decode a CoAP datagram. Returns (mtype, code, mid, token, - options, payload). options is a list of (num, value_bytes).""" + options, payload). options is a list of (num, value_bytes). + + The decoder is intentionally strict because some callers use it on + unauthenticated UDP datagrams. Truncated headers, tokens, extended option + fields, option values, and empty payload markers are classified rather + than leaking ``IndexError`` or being accepted as partial messages. + """ + if not isinstance(data, (bytes, bytearray, memoryview)): + raise MalformedMessageError() + data = bytes(data) + if len(data) < 4 or data[0] >> 6 != 1: + raise MalformedMessageError() mt = (data[0] >> 4) & 0x03 tkl = data[0] & 0x0F + if tkl > 8 or len(data) < 4 + tkl: + raise MalformedMessageError() code = data[1] mid = int.from_bytes(data[2:4], 'big') tok = data[4:4 + tkl] @@ -75,33 +148,62 @@ def parse_coap(data): while i < len(data): b = data[i] if b == 0xFF: + if i + 1 >= len(data): + raise MalformedMessageError() payload = data[i + 1:] break d_nib, l_nib = b >> 4, b & 0x0F i += 1 if d_nib == 13: - delta = 13 + data[i]; i += 1 + if i >= len(data): + raise MalformedMessageError() + delta = 13 + data[i] + i += 1 elif d_nib == 14: - delta = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2 + if i + 2 > len(data): + raise MalformedMessageError() + delta = 269 + int.from_bytes(data[i:i + 2], 'big') + i += 2 elif d_nib == 15: raise MalformedMessageError() else: delta = d_nib if l_nib == 13: - length = 13 + data[i]; i += 1 + if i >= len(data): + raise MalformedMessageError() + length = 13 + data[i] + i += 1 elif l_nib == 14: - length = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2 + if i + 2 > len(data): + raise MalformedMessageError() + length = 269 + int.from_bytes(data[i:i + 2], 'big') + i += 2 elif l_nib == 15: raise MalformedMessageError() else: length = l_nib num = prev + delta + if i + length > len(data): + raise MalformedMessageError() opts.append((num, data[i:i + length])) i += length prev = num return mt, code, mid, tok, opts, payload +def parse_coap_message(data): + """Decode ``data`` into an immutable :class:`CoapMessage`.""" + mtype, code, mid, token, options, payload = parse_coap(data) + return CoapMessage( + mtype=mtype, + code=code, + mid=mid, + token=token, + options=tuple(options), + payload=payload, + ) + + def build_coap(mtype, code, mid, token, options, payload=b''): """Build a CoAP datagram. mtype: CON/NON/ACK/RST. token: bytes (may be empty for ACK). options: list of (num, value_bytes).""" @@ -114,11 +216,18 @@ def build_coap(mtype, code, mid, token, options, payload=b''): return body +def build_empty_ack(mid): + """Build the bare ACK required for a confirmable CoAP response.""" + return build_coap(TYPE_ACK, 0, mid, b'', []) + + def block_value(num, more, szx): """Encode a CoAP Block-N option value.""" v = (num << 4) | ((more & 1) << 3) | (szx & 7) - if v <= 0xFF: return bytes([v]) - if v <= 0xFFFF: return struct.pack('>H', v) + if v <= 0xFF: + return bytes([v]) + if v <= 0xFFFF: + return struct.pack('>H', v) return struct.pack('>I', v)[1:] @@ -130,6 +239,384 @@ def block_fields(value): return v >> 4, (v >> 3) & 1, v & 0x07 +def _option_bytes(value, *, name): + if isinstance(value, str): + return value.encode() + if isinstance(value, (bytes, bytearray, memoryview)): + return bytes(value) + raise TypeError(f'{name} values must be strings or bytes') + + +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`` 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. + """ + options = [ + (URI_PATH, _option_bytes(segment, name='path segment')) + for segment in path_segs + ] + options.extend( + (URI_QUERY, _option_bytes(value, name='query')) + for value in query + ) + if accept is not None: + if not isinstance(accept, (bytes, bytearray, memoryview)): + raise TypeError('accept must be bytes or None') + options.append((ACCEPT, bytes(accept))) + if block_number is not None: + if (isinstance(block_number, bool) + or not isinstance(block_number, int) + or block_number < 0): + raise ValueError('block_number must be a non-negative integer') + if (isinstance(block_szx, bool) + or not isinstance(block_szx, int) + 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))) + return build_coap(mtype, METHOD_GET, mid, token, options) + + +def option_values(options, number): + """Return all values for one option number, preserving wire order.""" + return tuple(value for option_number, value in options + if option_number == number) + + +def decode_uint_option(options, number, *, max_length): + """Decode one optional CoAP uint option. + + Returns ``None`` when absent. Repeated options and values longer than the + caller's bound are malformed; an empty value is the canonical integer 0. + """ + values = option_values(options, number) + if len(values) > 1: + raise MalformedMessageError() + if not values: + return None + value = values[0] + if len(value) > max_length: + raise MalformedMessageError() + return int.from_bytes(value, 'big') + + +def classify_coap_response(datagram, *, token=None, request_mid=None): + """Classify a response without coupling it to a socket implementation. + + Empty ACK and RST frames correlate by ``request_mid`` because RFC 7252 + requires them to carry no token. Piggyback ACK responses correlate by + both MID (when supplied) and token. Separate CON/NON responses correlate + by token; a CON classification includes the bare ACK bytes the transport + should send to the response source. + + ``token=None`` disables token filtering and is useful to a connected + session that performs its own token dispatch. Structurally valid but + unrelated messages return ``RESPONSE_IGNORE``. Invalid ACK/RST semantics + raise :class:`MalformedMessageError`. + """ + message = parse_coap_message(datagram) + + if message.mtype in (TYPE_ACK, TYPE_RST) and message.code == 0: + if message.token or message.options or message.payload: + raise MalformedMessageError() + if request_mid is not None and message.mid != request_mid: + return CoapResponseClassification(RESPONSE_IGNORE, message) + kind = (RESPONSE_EMPTY_ACK + if message.mtype == TYPE_ACK else RESPONSE_RESET) + return CoapResponseClassification(kind, message) + + if message.mtype == TYPE_RST: + # A Reset is always empty. A non-empty/code-bearing RST is malformed, + # rather than an unrelated response that callers may silently accept. + raise MalformedMessageError() + + acknowledgement = ( + build_empty_ack(message.mid) if message.mtype == TYPE_CON else None + ) + if message.mtype not in (TYPE_CON, TYPE_NON, TYPE_ACK): + return CoapResponseClassification( + RESPONSE_IGNORE, message, acknowledgement) + if message.code == 0: + return CoapResponseClassification( + RESPONSE_IGNORE, message, acknowledgement) + if token is not None and message.token != token: + return CoapResponseClassification( + RESPONSE_IGNORE, message, acknowledgement) + if (message.mtype == TYPE_ACK and request_mid is not None + and message.mid != request_mid): + return CoapResponseClassification(RESPONSE_IGNORE, message) + return CoapResponseClassification( + RESPONSE_MESSAGE, message, acknowledgement) + + +class Block2Accumulator: + """Bounded, token-stable Block2 representation accumulator. + + At most ``max_blocks`` response blocks (32 by default) and + ``max_payload_bytes`` assembled bytes (64 KiB by default) are accepted. + Block offsets must remain contiguous. A bounded SZX downshift is accepted + because Samsung RT-OCF may return the requested payload size while + advertising a smaller size for the next request; upshifts and unaligned + transitions remain invalid. ETag and Content-Format omissions on + continuation blocks are tolerated, while conflicting values that are + present remain invalid. Size2 is only an informational RFC 7959 estimate: + it may change, may differ from the final length, and never affects + allocation or acceptance. + + ``add_response`` returns ``BLOCK2_DUPLICATE`` without changing state for a + retransmitted earlier block, ``BLOCK2_CONTINUE`` when the next block is + required, and ``BLOCK2_COMPLETE`` when ``code`` and ``payload`` are ready. + Any correlated transfer-contract violation raises ``BlockwiseError``. + """ + + def __init__( + self, token, *, max_blocks=MAX_BLOCK2_BLOCKS, + max_payload_bytes=MAX_BLOCK2_PAYLOAD_BYTES, + accepted_content_formats=None): + if not isinstance(token, (bytes, bytearray, memoryview)): + raise TypeError('token must be bytes') + token = bytes(token) + if len(token) > 8: + raise ValueError('token must contain at most 8 bytes') + if (isinstance(max_blocks, bool) or not isinstance(max_blocks, int) + or max_blocks <= 0): + raise ValueError('max_blocks must be a positive integer') + if (isinstance(max_payload_bytes, bool) + or not isinstance(max_payload_bytes, int) + or max_payload_bytes <= 0): + raise ValueError( + 'max_payload_bytes must be a positive integer') + if accepted_content_formats is None: + formats = None + else: + try: + formats = frozenset(accepted_content_formats) + except TypeError as exc: + raise TypeError( + 'accepted_content_formats must be an iterable') from exc + if any(isinstance(value, bool) or not isinstance(value, int) + or value < 0 for value in formats): + raise ValueError( + 'accepted_content_formats must contain non-negative ' + 'integers') + + self._token = token + self._max_blocks = max_blocks + self._max_payload_bytes = max_payload_bytes + self._accepted_content_formats = formats + self._expected_number = 0 + self._negotiated_szx = None + self._etag = None + self._content_format = None + self._size2 = None + self._payload = bytearray() + self._blocks_received = 0 + self._code = None + self._complete = False + + @property + def expected_number(self): + """Block number required next at the currently negotiated SZX.""" + return self._expected_number + + @property + def szx(self): + """SZX for the next request (1024-byte blocks until negotiated).""" + if self._negotiated_szx is None: + return BLOCK_SZX + return self._negotiated_szx + + @property + def complete(self): + return self._complete + + @property + def code(self): + return self._code + + @property + def payload(self): + return bytes(self._payload) + + @property + def blocks_received(self): + return self._blocks_received + + @property + def etag(self): + return self._etag + + @property + def content_format(self): + return self._content_format + + @property + def size2(self): + return self._size2 + + @staticmethod + def _single_etag(options): + values = option_values(options, ETAG) + if len(values) > 1: + raise BlockwiseError() + if not values: + return None + value = values[0] + if not 1 <= len(value) <= 8: + raise BlockwiseError() + return value + + @staticmethod + def _uint_option(options, number, *, max_length): + try: + return decode_uint_option( + options, number, max_length=max_length) + except MalformedMessageError: + raise BlockwiseError() from None + + def add_response(self, message): + if not isinstance(message, CoapMessage): + raise TypeError('message must be a CoapMessage') + if self._complete: + raise BlockwiseError() + if message.token != self._token: + raise BlockwiseError() + if message.mtype not in (TYPE_CON, TYPE_NON, TYPE_ACK): + raise BlockwiseError() + if message.code == 0: + raise BlockwiseError() + + # A non-success response terminates the logical GET immediately while + # preserving the connected-session contract of returning bytes already + # accumulated before and in the error response. + if message.code >> 5 != 2: + if len(self._payload) + len(message.payload) > \ + self._max_payload_bytes: + raise BlockwiseError() + self._code = message.code + self._payload.extend(message.payload) + self._blocks_received += 1 + self._complete = True + return BLOCK2_COMPLETE + + block_values = option_values(message.options, BLOCK2) + if len(block_values) > 1: + raise BlockwiseError() + if self._expected_number > 0 and not block_values: + raise BlockwiseError() + if block_values: + encoded = block_values[0] + if len(encoded) > 3: + raise BlockwiseError() + number, more, response_szx = block_fields(encoded) + more = bool(more) + if response_szx > BLOCK_SZX: + raise BlockwiseError() + else: + number = 0 + more = False + response_szx = None + + if self._expected_number > 0 and response_szx is None: + raise BlockwiseError() + + if response_szx is not None: + block_size = 1 << (response_szx + 4) + response_offset = number * block_size + expected_offset = len(self._payload) + if response_offset < expected_offset: + return BLOCK2_DUPLICATE + if response_offset != expected_offset: + raise BlockwiseError() + + request_szx = self.szx + if (self._negotiated_szx is not None + and response_szx != self._negotiated_szx + and response_szx > self._negotiated_szx): + raise BlockwiseError() + + # Some Samsung RT-OCF versions answer a request at the previous + # SZX-sized payload while advertising a smaller SZX for the next + # request. Compatibility mode accepts only this bounded downward + # transition. The byte offset below still has to land exactly on + # a block boundary at the newly advertised size. + downshifted = response_szx < request_szx + payload_limit_szx = ( + request_szx if downshifted else response_szx + ) + payload_limit = 1 << (payload_limit_szx + 4) + if len(message.payload) > payload_limit: + raise BlockwiseError() + next_offset = expected_offset + len(message.payload) + if more and ( + not message.payload + or next_offset % block_size + or (not downshifted + and len(message.payload) != block_size)): + raise BlockwiseError() + else: + next_offset = len(message.payload) + + etag = self._single_etag(message.options) + content_format = self._uint_option( + message.options, CONTENT_FORMAT, max_length=2) + size2 = self._uint_option(message.options, SIZE2, max_length=4) + if (self._accepted_content_formats is not None + and content_format is not None + and content_format not in self._accepted_content_formats): + raise BlockwiseError() + + if self._blocks_received == 0: + next_etag = etag + next_content_format = content_format + next_size2 = size2 + next_szx = response_szx + next_code = message.code + else: + if message.code != self._code: + raise BlockwiseError() + if (etag is not None and self._etag is not None + and etag != self._etag): + raise BlockwiseError() + if (content_format is not None + and self._content_format is not None + and content_format != self._content_format): + raise BlockwiseError() + next_etag = self._etag if etag is None else etag + next_content_format = ( + self._content_format + if content_format is None else content_format + ) + next_size2 = self._size2 if size2 is None else size2 + next_szx = response_szx + next_code = self._code + + next_length = len(self._payload) + len(message.payload) + if next_length > self._max_payload_bytes: + raise BlockwiseError() + if self._blocks_received >= self._max_blocks: + raise BlockwiseError() + if more and self._blocks_received + 1 >= self._max_blocks: + raise BlockwiseError() + + self._etag = next_etag + self._content_format = next_content_format + self._size2 = next_size2 + self._negotiated_szx = next_szx + self._code = next_code + self._payload.extend(message.payload) + self._blocks_received += 1 + if more: + self._expected_number = next_offset // (1 << (next_szx + 4)) + return BLOCK2_CONTINUE + self._complete = True + return BLOCK2_COMPLETE + + def fmt_code(c): """0x45 → '2.05', 0x84 → '4.04'. Used in log lines.""" return f"{c >> 5}.{c & 0x1F:02d}" diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index b3f5177..4b88153 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -26,6 +26,7 @@ block 0 on a fresh one-shot token by a worker thread. See #39. """ import errno +import logging import math import os import socket @@ -38,35 +39,55 @@ from ..errors import ( BlockwiseError, EndpointError, + MalformedMessageError, SessionClosedError, SessionError, SessionIdentifierError, SessionResetError, SessionTimeoutError, ) -from .coap import ( - URI_PATH, URI_QUERY, OBSERVE, ETAG, CONTENT_FORMAT, ACCEPT, BLOCK2, SIZE2, - TYPE_CON, TYPE_NON, TYPE_ACK, TYPE_RST, - METHOD_GET, METHOD_POST, CF_CBOR, - OBSERVE_REGISTER, OBSERVE_DEREGISTER, BLOCK_SZX, - encode_options, parse_coap, build_coap, block_value, block_fields, - fmt_code, - split_dtls as _split_dtls, -) +from . import auth as _auth +from . import coap as _coap from .auth import ( AuthenticationProvider, CertificateAuth, - _DTLS_CIPHERS, - _OCF_ROOT_CA, - _load_pem_chain, +) +from .coap import ( + ACCEPT, + BLOCK2, + BLOCK2_COMPLETE, + CF_CBOR, + CONTENT_FORMAT, + ETAG, + METHOD_GET, + METHOD_POST, + OBSERVE, + OBSERVE_DEREGISTER, + OBSERVE_REGISTER, + RESPONSE_EMPTY_ACK, + RESPONSE_MESSAGE, + RESPONSE_RESET, + TYPE_CON, + URI_PATH, + Block2Accumulator, + block_fields, + build_coap, + build_get_request, + classify_coap_response, + fmt_code, ) from .dtls_handshake import ( _HANDSHAKE_POLL_S, - _HandshakeCancelled, _drive_dtls_handshake, + _HandshakeCancelled, ) from .endpoint import open_connected_udp_socket -import logging + +# Private compatibility exports used by dtls_probe and existing callers. +_DTLS_CIPHERS = _auth._DTLS_CIPHERS +_OCF_ROOT_CA = _auth._OCF_ROOT_CA +_load_pem_chain = _auth._load_pem_chain +_split_dtls = _coap.split_dtls logger = logging.getLogger(__name__) @@ -715,42 +736,39 @@ def _reader_loop(self): def _dispatch_coap(self, datagram): try: - mt, code, mid, tok, ropts, payload = parse_coap(datagram) - except Exception as e: + classification = classify_coap_response(datagram) + except MalformedMessageError as e: logger.debug("malformed CoAP: %s", e) return + message = classification.message + if message is None: + return + mt = message.mtype + code = message.code + mid = message.mid + tok = message.token + ropts = message.options + payload = message.payload + if DEBUG_BRIDGE: kind = ['CON', 'NON', 'ACK', 'RST'][mt] logger.info("rx %s code=%s mid=%04x tok=%s opts=%d pl=%d", kind, fmt_code(code), mid, tok.hex() or '-', len(ropts), len(payload)) - # RFC 7252 empty messages are exactly the four-byte v1 header with - # TKL=0 and code=0. parse_coap() intentionally stays a lightweight - # general parser, so validate the raw shape before a control frame is - # allowed to mutate a MID-indexed exchange. - bare_control = ( - len(datagram) == 4 - and datagram[0] >> 6 == 1 - and datagram[0] & 0x0F == 0 - and code == 0 - ) - # ACK back any CON from the device to suppress retransmits. # RFC 7252 §4.2 — ACK is a bare frame (token len 0, code 0). - if mt == TYPE_CON: + if classification.acknowledgement is not None: try: - self._send_dgram(build_coap(TYPE_ACK, 0, mid, b'', [])) + self._send_dgram(classification.acknowledgement) except Exception as e: logger.warning("ACK send: %s", e) # Empty ACK with no options & no payload = "separate response # coming" — used by Samsung's RT-OCF for the larger reads. Stop # the retransmit timer on the client side and wait for the CON. - if mt == TYPE_ACK and code == 0: - if not bare_control: - return + if classification.kind == RESPONSE_EMPTY_ACK: with self._state_lock: exchange = self._pending_mids.get(mid) if exchange is not None: @@ -762,9 +780,7 @@ def _dispatch_coap(self, datagram): # A reset rejects the matching exchange. Like an empty ACK it has no # response token, so surface it through the request's MID registry. - if mt == TYPE_RST: - if not bare_control: - return + if classification.kind == RESPONSE_RESET: with self._state_lock: exchange = self._pending_mids.get(mid) if exchange is not None: @@ -774,6 +790,8 @@ def _dispatch_coap(self, datagram): if exchange is not None: ev.set() return + if classification.kind != RESPONSE_MESSAGE: + return # Pending one-shot? Resolve and return. with self._state_lock: @@ -785,6 +803,7 @@ def _dispatch_coap(self, datagram): container['mid'] = mid container['options'] = ropts container['payload'] = payload + container['message'] = message if rec is not None: ev.set() return @@ -970,62 +989,49 @@ def _blockwise_get_once(self, path_segs, query, timeout): """One attempt at a full Block2 transfer. Raises _EtagChanged if the server's representation changed while we were reassembling.""" tok = self._next_tok() - blob = b'' - num = 0 - blocks = 0 - last_code = None + accumulator = Block2Accumulator(tok, max_blocks=self.MAX_BLOCKS) etag = None - deadline = time.time() + timeout - szx = BLOCK_SZX # server may negotiate down; track per-transfer - while True: + deadline = time.monotonic() + timeout + while not accumulator.complete: + num = accumulator.expected_number self.pace() - self._check_live() - container = self._exchange_block( - tok, path_segs, query, num, szx, deadline) - if 'err' in container: - raise container['err'] - blocks += 1 - - code = container['code'] - payload = container['payload'] - ropts = container['options'] - last_code = code - # 4.xx / 5.xx responses don't carry Block2 continuation — - # bail with whatever we got. Caller decides if 4.xx is fatal. - if code >> 5 != 2: - return code, blob, blocks, tok + message = self._exchange_block( + tok, + path_segs, + query, + num, + accumulator.szx, + deadline, + ) + prior_blocks = accumulator.blocks_received # RFC 7959 §2.4: compare ETags across blocks, or we splice # two versions of the resource into one buffer. - block_etag = next((v for n, v in ropts if n == ETAG), None) - if num == 0: - etag = block_etag - elif etag is not None and block_etag != etag: - raise _EtagChanged() + if message.code >> 5 == 2: + block_etag = next( + (value for number, value in message.options + if number == ETAG), + None, + ) + if prior_blocks == 0: + etag = block_etag + elif etag is not None and block_etag != etag: + raise _EtagChanged() + + status = accumulator.add_response(message) + if status == BLOCK2_COMPLETE: + return ( + accumulator.code, + accumulator.payload, + accumulator.blocks_received, + tok, + ) - blob += payload - b2 = [v for n, v in ropts if n == BLOCK2] - if not b2: - break - _, more, server_szx = block_fields(b2[0]) - if not more: - break - if server_szx != szx: - # Server negotiated the block size down. Block numbers - # are indices into the new size, so the next one has to - # come off the byte offset we have actually accumulated, - # not off num + 1. - szx = server_szx - num = len(blob) >> (szx + 4) - else: - num += 1 - if num > self.MAX_BLOCKS: - raise BlockwiseError() - return last_code, blob, blocks, tok + raise BlockwiseError() def _exchange_block(self, tok, path_segs, query, num, szx, deadline): """Send one block request under `tok` and return its response - container, retransmitting up to _BLOCK_MAX_ATTEMPTS times. + message, retransmitting up to _BLOCK_MAX_ATTEMPTS times. A response whose Block2 NUM is not the one we asked for is a retransmit of an earlier block, not the next one. Concatenating @@ -1040,37 +1046,47 @@ def _exchange_block(self, tok, path_segs, query, num, szx, deadline): ev = threading.Event() container = {} mid, exchange = self._register_pending_request(tok, ev, container) - opts = [(URI_PATH, s.encode()) for s in path_segs] - for q in query: - opts.append((URI_QUERY, q.encode())) - opts.append((ACCEPT, CF_CBOR)) - if num > 0: - opts.append((BLOCK2, block_value(num, 0, szx))) - datagram = build_coap(TYPE_CON, METHOD_GET, mid, tok, opts) + datagram = build_get_request( + TYPE_CON, + mid, + tok, + path_segs, + query, + block_number=num if num > 0 else None, + block_szx=szx, + ) try: for attempt in range(_BLOCK_MAX_ATTEMPTS): # Close the reader-death registration race: after this request # is visible to reader-finally, recheck that the reader still # owns the session before sending. self._check_live() + if time.monotonic() >= deadline: + raise SessionTimeoutError() self._send_dgram(datagram) while True: with self._state_lock: - if ('err' in container - or ('code' in container - and self._block_num_matches( - container, num))): - return container - if 'code' in container: + error = container.get('err') + message = container.get('message') + if (error is None and message is not None + and self._block_num_matches( + message, num, szx)): + return message + if error is None and message is not None: logger.debug( "GET %s /%s block %d: stale block, " "still waiting", self.host, '/'.join(path_segs), num, ) - container.clear() + for key in ( + 'code', 'mtype', 'mid', 'options', + 'payload', 'message'): + container.pop(key, None) acknowledged = exchange.acknowledged ev.clear() - remaining = deadline - time.time() + if error is not None: + raise error + remaining = deadline - time.monotonic() if remaining <= 0: if acknowledged: raise SessionTimeoutError() @@ -1083,7 +1099,7 @@ def _exchange_block(self, tok, path_segs, query, num, szx, deadline): if acknowledged: raise SessionTimeoutError() break - remaining = deadline - time.time() + remaining = deadline - time.monotonic() if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1: logger.debug( "GET %s /%s block %d: timed out after %d attempt(s)", @@ -1100,18 +1116,13 @@ def _exchange_block(self, tok, path_segs, query, num, szx, deadline): raise SessionTimeoutError() def _wait_for_block(self, ev, per_wait): - """Wait for one block response, giving up early if the reader - dies underneath us. - - Only the reader thread can resolve a token, so once it is gone - the wait can never succeed. Polling in slices turns what would - be a full per-block timeout into an immediate SessionClosedError, - which is the same fail-fast contract get() gets from _check_live() - at entry — it just has to hold for every block, not only the - first.""" - deadline = time.time() + per_wait + """Wait for a block response while checking reader liveness.""" + deadline = time.monotonic() + per_wait while True: - slice_s = min(_BLOCK_LIVENESS_POLL_S, deadline - time.time()) + slice_s = min( + _BLOCK_LIVENESS_POLL_S, + deadline - time.monotonic(), + ) if slice_s <= 0: return False if ev.wait(slice_s): @@ -1119,16 +1130,18 @@ def _wait_for_block(self, ev, per_wait): self._check_live() @staticmethod - def _block_num_matches(container, num): - """True if this response carries the block we asked for. A - response with no Block2 option is the whole representation, so - it only answers block 0.""" - if container.get('code', 0) >> 5 != 2: - return True # error responses end the transfer either way - b2 = [v for n, v in container.get('options', ()) if n == BLOCK2] - if not b2: + def _block_num_matches(message, num, szx): + """Return whether ``message`` answers the requested byte offset.""" + if message.code >> 5 != 2: + return True + block2 = [value for number, value in message.options + if number == BLOCK2] + if not block2: return num == 0 - return block_fields(b2[0])[0] == num + response_num, _more, response_szx = block_fields(block2[0]) + requested_offset = num << (szx + 4) + response_offset = response_num << (response_szx + 4) + return response_offset == requested_offset def post(self, path_segs, body_cbor, timeout=8.0): """Single-frame POST with a CBOR-encoded body. Returns @@ -1142,6 +1155,8 @@ def post(self, path_segs, body_cbor, timeout=8.0): container = {} mid, exchange = self._register_pending_request(tok, ev, container) try: + # See get(): a reader can die after the entry check but before + # registration. This post-registration snapshot fails closed. self.pace() # The reader can exit between the entry liveness check and the # registration above. Recheck after registration so its teardown diff --git a/tests/test_coap_wire.py b/tests/test_coap_wire.py index baca36f..f6eba6a 100644 --- a/tests/test_coap_wire.py +++ b/tests/test_coap_wire.py @@ -1,10 +1,38 @@ import pytest -from smartthings_local.errors import MalformedMessageError +from smartthings_local.errors import BlockwiseError, MalformedMessageError from smartthings_local.protocol.coap import ( - build_coap, parse_coap, encode_options, block_value, block_fields, + ACCEPT, + BLOCK2, + BLOCK2_COMPLETE, + BLOCK2_CONTINUE, + BLOCK2_DUPLICATE, + CF_CBOR, + CONTENT_FORMAT, + ETAG, + METHOD_GET, + RESPONSE_EMPTY_ACK, + RESPONSE_IGNORE, + RESPONSE_MESSAGE, + SIZE2, + TYPE_ACK, + TYPE_CON, + TYPE_NON, + URI_PATH, + URI_QUERY, + Block2Accumulator, + CoapMessage, + block_fields, + block_value, + build_coap, + build_empty_ack, + build_get_request, + classify_coap_response, + decode_uint_option, + encode_options, fmt_code, - TYPE_CON, METHOD_GET, URI_PATH, ACCEPT, CF_CBOR, BLOCK2, + option_values, + parse_coap, ) @@ -74,3 +102,432 @@ def test_reserved_option_nibbles_raise_classified_value_error(option_header): parse_coap(datagram) assert isinstance(exc.value, ValueError) + + +@pytest.mark.parametrize( + 'datagram', + ( + b'', + b'\x40\x01\x00', + b'\x80\x01\x00\x01', # unsupported CoAP version + b'\x49\x01\x00\x01' + b'x' * 9, # reserved token length + b'\x44\x01\x00\x01abc', # truncated token + b'\x40\x01\x00\x01\xd0', # truncated extended delta + b'\x40\x01\x00\x01\xe0\x00', + b'\x40\x01\x00\x01\x0d', # truncated extended length + b'\x40\x01\x00\x01\x0e\x00', + b'\x40\x01\x00\x01\x03ab', # truncated option value + b'\x40\x01\x00\x01\xff', # empty payload marker + ), +) +def test_truncated_or_structurally_invalid_datagrams_are_classified(datagram): + with pytest.raises(MalformedMessageError): + parse_coap(datagram) + + +def test_non_bytes_coap_input_is_classified(): + with pytest.raises(MalformedMessageError): + parse_coap('not wire bytes') + + +def test_build_get_request_adds_query_and_only_requested_block2(): + initial = build_get_request( + TYPE_NON, + 0x1234, + b'token', + ('oic', b'res'), + ('rt=oic.r.doxm',), + ) + _, code, _, _, initial_options, _ = parse_coap(initial) + assert code == METHOD_GET + assert option_values(initial_options, URI_PATH) == (b'oic', b'res') + assert option_values(initial_options, URI_QUERY) == (b'rt=oic.r.doxm',) + assert option_values(initial_options, ACCEPT) == (CF_CBOR,) + assert option_values(initial_options, BLOCK2) == () + + continuation = build_get_request( + TYPE_NON, + 0x1235, + b'token', + ('oic', 'res'), + block_number=2, + block_szx=4, + ) + *_, continuation_options, _ = parse_coap(continuation) + assert option_values(continuation_options, BLOCK2) == ( + block_value(2, 0, 4), + ) + + +def test_decode_uint_option_is_optional_unique_and_bounded(): + assert decode_uint_option([], SIZE2, max_length=4) is None + assert decode_uint_option([(SIZE2, b'')], SIZE2, max_length=4) == 0 + assert decode_uint_option( + [(SIZE2, b'\x01\x00')], SIZE2, max_length=4) == 256 + with pytest.raises(MalformedMessageError): + decode_uint_option( + [(SIZE2, b'\x01'), (SIZE2, b'\x02')], + SIZE2, + max_length=4, + ) + with pytest.raises(MalformedMessageError): + decode_uint_option([(SIZE2, b'12345')], SIZE2, max_length=4) + + +def test_response_classification_handles_empty_ack_then_separate_con(): + empty_ack = classify_coap_response( + build_empty_ack(0x1234), + token=b'token', + request_mid=0x1234, + ) + assert empty_ack.kind == RESPONSE_EMPTY_ACK + assert empty_ack.acknowledgement is None + + separate = classify_coap_response( + build_coap( + TYPE_CON, + 0x45, + 0xBEEF, + b'token', + [(CONTENT_FORMAT, CF_CBOR)], + b'body', + ), + token=b'token', + request_mid=0x1234, + ) + assert separate.kind == RESPONSE_MESSAGE + assert separate.message.payload == b'body' + assert parse_coap(separate.acknowledgement) == ( + TYPE_ACK, + 0, + 0xBEEF, + b'', + [], + b'', + ) + + +def test_response_classification_accepts_matching_piggyback_ack(): + response = classify_coap_response( + build_coap(TYPE_ACK, 0x45, 0x1234, b'token', [], b'body'), + token=b'token', + request_mid=0x1234, + ) + assert response.kind == RESPONSE_MESSAGE + assert response.acknowledgement is None + + stale = classify_coap_response( + build_coap(TYPE_ACK, 0x45, 0x1235, b'token', [], b'body'), + token=b'token', + request_mid=0x1234, + ) + assert stale.kind == RESPONSE_IGNORE + + +def test_nonempty_empty_ack_is_malformed(): + with pytest.raises(MalformedMessageError): + classify_coap_response( + build_coap(TYPE_ACK, 0, 0x1234, b'x', []), + token=b'x', + request_mid=0x1234, + ) + + +def _message( + *, number=0, more=False, szx=0, token=b'token', payload=b'', + etag=b'etag', content_format=60, size2=None, code=0x45, + mtype=TYPE_NON, include_block=True): + options = [] + if include_block: + options.append((BLOCK2, block_value(number, more, szx))) + if etag is not None: + options.append((ETAG, etag)) + if content_format is not None: + options.append(( + CONTENT_FORMAT, + content_format.to_bytes(max(1, (content_format.bit_length() + 7) // 8), + 'big'), + )) + if size2 is not None: + options.append(( + SIZE2, + size2.to_bytes(max(1, (size2.bit_length() + 7) // 8), 'big'), + )) + return CoapMessage( + mtype=mtype, + code=code, + mid=number, + token=token, + options=tuple(options), + payload=payload, + ) + + +def test_block2_accumulator_is_token_stable_and_assembles_exact_body(): + accumulator = Block2Accumulator( + b'token', + accepted_content_formats={60, 10000}, + ) + first = _message( + number=0, + more=True, + payload=b'a' * 16, + size2=20, + ) + assert accumulator.add_response(first) == BLOCK2_CONTINUE + assert accumulator.expected_number == 1 + assert accumulator.szx == 0 + assert accumulator.blocks_received == 1 + + # A retransmitted earlier block is ignored without duplicating its bytes. + assert accumulator.add_response(first) == BLOCK2_DUPLICATE + assert accumulator.payload == b'a' * 16 + + final = _message(number=1, payload=b'done', size2=20) + assert accumulator.add_response(final) == BLOCK2_COMPLETE + assert accumulator.complete + assert accumulator.code == 0x45 + assert accumulator.payload == b'a' * 16 + b'done' + assert accumulator.etag == b'etag' + assert accumulator.content_format == 60 + assert accumulator.size2 == 20 + + +def test_block2_accumulator_allows_bounded_szx_downshift_by_default(): + accumulator = Block2Accumulator(b'token') + assert accumulator.add_response(_message( + number=0, + more=True, + szx=6, + payload=b'a' * 1024, + )) == BLOCK2_CONTINUE + assert accumulator.add_response(_message( + number=4, + szx=4, + payload=b'done', + )) == BLOCK2_COMPLETE + + +def test_block2_accumulator_downshift_uses_byte_offset(): + accumulator = Block2Accumulator(b'token') + + # Samsung may return the full 1024 bytes requested for block zero while + # advertising SZX=4 (256 bytes) for continuation requests. + assert accumulator.add_response(_message( + number=0, + more=True, + szx=4, + payload=b'a' * 1024, + )) == BLOCK2_CONTINUE + assert accumulator.expected_number == 4 + assert accumulator.szx == 4 + + assert accumulator.add_response(_message( + number=4, + szx=4, + payload=b'done', + )) == BLOCK2_COMPLETE + assert accumulator.payload == b'a' * 1024 + b'done' + + +def test_block2_accumulator_rejects_upshift_and_unaligned_downshift(): + upshift = Block2Accumulator(b'token') + upshift.add_response(_message( + number=0, + more=True, + szx=4, + payload=b'a' * 1024, + )) + with pytest.raises(BlockwiseError): + upshift.add_response(_message( + number=1, + szx=6, + payload=b'done', + )) + + unaligned = Block2Accumulator(b'token') + with pytest.raises(BlockwiseError): + unaligned.add_response(_message( + number=0, + more=True, + szx=4, + payload=b'a' * 300, + )) + + +@pytest.mark.parametrize( + 'second', + ( + _message(number=2, payload=b'done'), + _message(number=1, szx=1, payload=b'done'), + _message(number=1, token=b'other', payload=b'done'), + _message(number=1, etag=b'other', payload=b'done'), + _message(number=1, content_format=10000, payload=b'done'), + ), +) +def test_block2_accumulator_rejects_transfer_identity_changes(second): + accumulator = Block2Accumulator(b'token') + accumulator.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + size2=20, + )) + with pytest.raises(BlockwiseError): + accumulator.add_response(second) + + +def test_block2_metadata_omissions_preserve_legacy_peer_compatibility(): + accumulator = Block2Accumulator(b'token') + accumulator.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + etag=b'first', + content_format=60, + size2=20, + )) + assert accumulator.add_response(_message( + number=1, + payload=b'done', + etag=None, + content_format=None, + size2=None, + )) == BLOCK2_COMPLETE + assert accumulator.payload == b'a' * 16 + b'done' + assert accumulator.etag == b'first' + assert accumulator.content_format == 60 + assert accumulator.size2 == 20 + + +def test_block2_size2_is_an_informational_estimate_only(): + accumulator = Block2Accumulator( + b'token', + max_payload_bytes=32, + ) + accumulator.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + size2=1, + )) + assert accumulator.add_response(_message( + number=1, + payload=b'done', + size2=1_000_000, + )) == BLOCK2_COMPLETE + assert accumulator.payload == b'a' * 16 + b'done' + assert accumulator.size2 == 1_000_000 + + +def test_block2_continuation_requires_an_explicit_block2_option(): + accumulator = Block2Accumulator(b'token') + accumulator.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + )) + with pytest.raises(BlockwiseError): + accumulator.add_response(_message( + payload=b'done', + include_block=False, + )) + + +def test_mid_transfer_error_preserves_connected_session_contract_by_default(): + error = _message( + code=0x80, + payload=b'error', + include_block=False, + etag=None, + content_format=None, + ) + accumulator = Block2Accumulator(b'token') + accumulator.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + )) + assert accumulator.add_response(error) == BLOCK2_COMPLETE + assert accumulator.code == 0x80 + assert accumulator.payload == b'a' * 16 + b'error' + + +def test_block2_accumulator_enforces_exact_block_and_payload_bounds(): + two_blocks = Block2Accumulator( + b'token', + max_blocks=2, + max_payload_bytes=32, + ) + assert two_blocks.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + size2=32, + )) == BLOCK2_CONTINUE + assert two_blocks.add_response(_message( + number=1, + payload=b'b' * 16, + size2=32, + )) == BLOCK2_COMPLETE + + needs_third = Block2Accumulator( + b'token', + max_blocks=2, + max_payload_bytes=48, + ) + needs_third.add_response(_message( + number=0, + more=True, + payload=b'a' * 16, + size2=48, + )) + with pytest.raises(BlockwiseError): + needs_third.add_response(_message( + number=1, + more=True, + payload=b'b' * 16, + size2=48, + )) + + too_large = Block2Accumulator(b'token', max_payload_bytes=16) + with pytest.raises(BlockwiseError): + too_large.add_response(_message( + payload=b'x' * 17, + include_block=False, + etag=None, + content_format=None, + )) + + +def test_block2_accumulator_default_payload_bound_is_exactly_64_kib(): + exact = Block2Accumulator(b'token') + response = _message( + payload=b'x' * 65536, + include_block=False, + etag=None, + content_format=None, + ) + assert exact.add_response(response) == BLOCK2_COMPLETE + + oversized = Block2Accumulator(b'token') + with pytest.raises(BlockwiseError): + oversized.add_response(CoapMessage( + mtype=TYPE_NON, + code=0x80, + mid=1, + token=b'token', + options=(), + payload=b'x' * 65537, + )) + + +def test_internal_response_reprs_redact_token_payload_and_ack_bytes(): + sensitive_value = b'synthetic-sensitive-device-value' + classification = classify_coap_response( + build_coap(TYPE_CON, 0x45, 1, b'token', [], sensitive_value), + token=b'token', + ) + assert 'token' not in repr(classification.message) + assert sensitive_value.decode() not in repr(classification.message) + assert 'acknowledgement=b' not in repr(classification) diff --git a/tests/test_dtls_session_reader_death.py b/tests/test_dtls_session_reader_death.py index d341e30..4f07570 100644 --- a/tests/test_dtls_session_reader_death.py +++ b/tests/test_dtls_session_reader_death.py @@ -9,7 +9,6 @@ """ import errno import logging -import socket import threading import time @@ -17,6 +16,19 @@ from OpenSSL import SSL from smartthings_local.errors import SessionClosedError +from smartthings_local.protocol import dtls_session +from smartthings_local.protocol.coap import ( + BLOCK2, + CF_CBOR, + CONTENT_FORMAT, + TYPE_ACK, + TYPE_CON, + block_value, + build_coap, + build_empty_ack, + option_values, + parse_coap, +) from smartthings_local.protocol.dtls_session import DtlsCoapSession _LOGGER_NAME = "smartthings_local.protocol.dtls_session" @@ -36,6 +48,7 @@ class _FakeConn: def __init__(self): self._decrypted = [] + self.sent = [] def bio_write(self, datagram): self._decrypted.append(datagram) @@ -48,8 +61,8 @@ def recv(self, _n): def bio_read(self, _n): return b"" - def send(self, _datagram): - return None + def send(self, datagram): + self.sent.append(datagram) def shutdown(self): return None @@ -76,12 +89,12 @@ def recv(self, _n): step = self._steps.pop(0) if callable(step): step() - raise socket.timeout() + raise TimeoutError() if isinstance(step, BaseException): raise step return step time.sleep(0.01) - raise socket.timeout() + raise TimeoutError() def send(self, data): return len(data) @@ -191,3 +204,356 @@ def test_check_live_without_reader_matches_old_conn_guard(): sess.conn = None with pytest.raises(SessionClosedError): sess._check_live() + + +def test_dispatch_empty_ack_then_separate_con_resolves_and_acks_response(): + sess = _make_session() + token = b'token' + event = threading.Event() + container = {} + mid, exchange = sess._register_pending_request(token, event, container) + try: + sess._dispatch_coap(build_empty_ack((mid + 1) & 0xFFFF)) + assert not event.is_set() + + sess._dispatch_coap(build_empty_ack(mid)) + assert event.is_set() + assert exchange.acknowledged is True + assert 'acknowledged' not in container + assert sess.conn.sent == [] + event.clear() + + sess._dispatch_coap(build_coap( + TYPE_CON, + 0x45, + 0xBEEF, + token, + [(CONTENT_FORMAT, CF_CBOR)], + b'body', + )) + assert event.is_set() + assert container['payload'] == b'body' + assert parse_coap(sess.conn.sent[-1]) == ( + TYPE_ACK, + 0, + 0xBEEF, + b'', + [], + b'', + ) + finally: + sess._unregister_pending_request(token, mid, exchange) + + +def test_dispatch_piggyback_ack_resolves_without_sending_another_ack(): + sess = _make_session() + token = b'token' + event = threading.Event() + container = {} + sess._pending[token] = (event, container) + + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + 0x1234, + token, + [(CONTENT_FORMAT, CF_CBOR)], + b'body', + )) + assert event.is_set() + assert container['payload'] == b'body' + assert sess.conn.sent == [] + + +def test_get_uses_one_token_and_shared_block2_continuation_builder(): + sess = _make_session() + requests = [] + + def respond(datagram): + request = parse_coap(datagram) + requests.append(request) + _mtype, _code, mid, token, options, _payload = request + requested_block = option_values(options, BLOCK2) + number = 0 if not requested_block else \ + int.from_bytes(requested_block[0], 'big') >> 4 + response_payload = b'a' * 16 if number == 0 else b'done' + response_options = [ + (BLOCK2, block_value(number, number == 0, 0)), + ] + # Some RFC 7959 peers only repeat representation metadata on block 0. + # The connected session historically accepted that shape. + if number == 0: + response_options.append((CONTENT_FORMAT, CF_CBOR)) + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + response_options, + response_payload, + )) + + sess._send_dgram = respond + sess.pace = lambda: None + + code, payload = sess.get(['oic', 'res'], query=('if=oic.if.baseline',)) + assert code == 0x45 + assert payload == b'a' * 16 + b'done' + assert len(requests) == 2 + assert requests[0][3] == requests[1][3] + assert option_values(requests[0][4], BLOCK2) == () + assert option_values(requests[1][4], BLOCK2) == ( + block_value(1, 0, 0), + ) + + +def test_get_matches_a_downshifted_response_by_byte_offset(): + sess = _make_session() + requests = [] + + def respond(datagram): + request = parse_coap(datagram) + requests.append(request) + _mtype, _code, mid, token, options, _payload = request + requested = option_values(options, BLOCK2) + if not requested: + response_block = block_value(0, 1, 6) + response_payload = b'a' * 1024 + else: + assert requested == (block_value(1, 0, 6),) + # The same byte offset is NUM=4 after the server downshifts to + # 256-byte blocks. + response_block = block_value(4, 0, 4) + response_payload = b'done' + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + [(BLOCK2, response_block)], + response_payload, + )) + + sess._send_dgram = respond + sess.pace = lambda: None + + assert sess.get(['oic', 'res']) == ( + 0x45, + b'a' * 1024 + b'done', + ) + assert len(requests) == 2 + + +def test_stale_block_does_not_clear_interleaved_current_block(): + sess = _make_session() + requests = [] + + def respond(datagram): + _mtype, _code, mid, token, options, _payload = parse_coap(datagram) + requested = option_values(options, BLOCK2) + number = 0 if not requested else \ + int.from_bytes(requested[0], 'big') >> 4 + requests.append(number) + + if number == 0: + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + [(BLOCK2, block_value(0, 1, 0))], + b'a' * 16, + )) + elif number == 1: + # The reader replaces a delayed duplicate with the requested block + # before the GET thread wakes and inspects the pending slot. + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + [(BLOCK2, block_value(0, 1, 0))], + b'a' * 16, + )) + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + [(BLOCK2, block_value(1, 0, 0))], + b'done', + )) + sess._send_dgram = respond + sess.pace = lambda: None + + assert sess.get(['oic', 'res'], timeout=0.2) == ( + 0x45, + b'a' * 16 + b'done', + ) + assert requests == [0, 1] + + +def test_empty_ack_stops_get_retransmit_until_separate_response( + monkeypatch): + monkeypatch.setattr(dtls_session, '_BLOCK_ACK_TIMEOUT', 0.01) + sess = _make_session() + requests = [] + timers = [] + + def respond(datagram): + mtype, code, mid, token, _options, _payload = parse_coap(datagram) + if mtype == TYPE_ACK and code == 0: + return + requests.append(datagram) + sess._dispatch_coap(build_empty_ack(mid)) + timer = threading.Timer( + 0.04, + lambda: sess._dispatch_coap(build_coap( + TYPE_CON, + 0x45, + 0xBEEF, + token, + [], + b'body', + )), + ) + timers.append(timer) + timer.start() + + sess._send_dgram = respond + try: + assert sess.get(['oic', 'res'], timeout=0.2) == (0x45, b'body') + finally: + for timer in timers: + timer.join(1.0) + assert len(requests) == 1 + + +def test_empty_ack_survives_stale_block_until_separate_response( + monkeypatch): + monkeypatch.setattr(dtls_session, '_BLOCK_ACK_TIMEOUT', 0.01) + sess = _make_session() + requested_blocks = [] + waits = [] + pending = {} + + def respond(datagram): + mtype, code, mid, token, options, _payload = parse_coap(datagram) + if mtype == TYPE_ACK and code == 0: + return + requested = option_values(options, BLOCK2) + number = 0 if not requested else \ + int.from_bytes(requested[0], 'big') >> 4 + requested_blocks.append(number) + + if number == 0: + sess._dispatch_coap(build_coap( + TYPE_ACK, + 0x45, + mid, + token, + [(BLOCK2, block_value(0, 1, 0))], + b'a' * 16, + )) + return + + sess._dispatch_coap(build_empty_ack(mid)) + pending['token'] = token + + def wait_for_block(_event, per_wait): + waits.append(per_wait) + token = pending['token'] + if len(waits) == 1: + sess._dispatch_coap(build_coap( + TYPE_CON, + 0x45, + 0xBEEF, + token, + [(BLOCK2, block_value(0, 1, 0))], + b'a' * 16, + )) + elif len(waits) == 2: + sess._dispatch_coap(build_coap( + TYPE_CON, + 0x45, + 0xCAFE, + token, + [(BLOCK2, block_value(1, 0, 0))], + b'done', + )) + else: + pytest.fail('unexpected additional Block2 wait') + return True + + sess._send_dgram = respond + sess._wait_for_block = wait_for_block + sess.pace = lambda: None + assert sess.get(['oic', 'res'], timeout=0.2) == ( + 0x45, + b'a' * 16 + b'done', + ) + + assert requested_blocks == [0, 1] + assert len(waits) == 2 + assert all(wait > dtls_session._BLOCK_ACK_TIMEOUT for wait in waits) + assert sess._pending == {} + assert sess._pending_mids == {} + + +def test_get_preserves_mid_transfer_error_payload_contract(): + sess = _make_session() + + def respond(datagram): + _mtype, _code, mid, token, options, _payload = parse_coap(datagram) + requested_block = option_values(options, BLOCK2) + number = 0 if not requested_block else \ + int.from_bytes(requested_block[0], 'big') >> 4 + if number == 0: + code = 0x45 + response_options = [(BLOCK2, block_value(0, 1, 0))] + payload = b'a' * 16 + else: + code = 0x80 + response_options = [] + payload = b'error' + sess._dispatch_coap(build_coap( + TYPE_ACK, + code, + mid, + token, + response_options, + payload, + )) + + sess._send_dgram = respond + sess.pace = lambda: None + assert sess.get(['oic', 'res']) == (0x80, b'a' * 16 + b'error') + + +@pytest.mark.parametrize('method', ('get', 'post')) +def test_request_rechecks_reader_after_pending_registration(method): + sess = _make_session() + sess._reader_thread = object() + sess._reader_running.set() + original_check_live = sess._check_live + checks = 0 + sent = [] + + def fail_on_post_registration_snapshot(): + nonlocal checks + checks += 1 + if checks == 2: + sess._reader_running.clear() + original_check_live() + + sess._check_live = fail_on_post_registration_snapshot + sess._send_dgram = sent.append + with pytest.raises(SessionClosedError): + if method == 'get': + sess.get(['oic', 'res']) + else: + sess.post(['mode', 'vs', '0'], b'body') + + assert checks == 2 + assert sent == [] + assert sess._pending == {} From 73c1d73605306fcfbcfa57e9d5a98047959a4de3 Mon Sep 17 00:00:00 2001 From: hoon <230467962+atc722@users.noreply.github.com> Date: Sat, 15 Aug 2026 10:49:55 +0900 Subject: [PATCH 2/3] feat(protocol): discover advertised OCF secure ports --- README.md | 30 + smartthings_local/protocol/ocf_discovery.py | 690 +++++++++++++++++ tests/test_import_isolation.py | 2 + tests/test_ocf_discovery.py | 818 ++++++++++++++++++++ tests/test_public_api_contract.py | 16 + 5 files changed, 1556 insertions(+) create mode 100644 smartthings_local/protocol/ocf_discovery.py create mode 100644 tests/test_ocf_discovery.py diff --git a/README.md b/README.md index 23dabb0..59aa195 100644 --- a/README.md +++ b/README.md @@ -336,6 +336,35 @@ python -m smartthings_local.protocol.dtls_probe "$APPLIANCE_IP" 5684 49153 49154 `live` means a DTLS server answered its first flight; `dead` means silent or not DTLS. Once you have the client cert (Part 2), add the explicit `--diagnostic` flag to run the stateful diagnostic drive, which reports `completed` (cert accepted) or `rejected` with the server's fatal alert. Diagnostic mode can allocate appliance-side DTLS state and is never used by discovery or reconnect. An `unsupported_certificate` / `unknown_ca` alert means the endpoint is reachable but this certificate profile was rejected. It is not a reason to disable verification or keep retrying. The same bounded stateless API gates the bridge's reconnect loop and, when `OCF_PORT` is unset, probes both standard 5684 and ports 49152–49160. +Consumers can discover ports outside that fallback range through the public, +read-only OCF resource directory before probing them: + +```python +from smartthings_local.protocol.dtls_probe import probe_dtls_ports +from smartthings_local.protocol.ocf_discovery import discover_ocf_secure_ports + +fallback_ports = (5684, *range(49152, 49161)) +advertisement = discover_ocf_secure_ports(appliance_host) +candidates = advertisement.ports or fallback_ports +probe = probe_dtls_ports(appliance_host, candidates) +``` + +`discover_ocf_secure_ports()` first reads the public `/oic/res` directory and +uses only `coaps://` endpoints whose literal host matches the correlated +response source. If that first lookup yields no correlated response or no +usable secure endpoint, the same overall deadline also bounds a filtered +`/oic/res?rt=oic.r.doxm` fallback for Samsung's legacy secure-port policy. It +accepts Samsung's dynamic plaintext response source port while still requiring +the resolved target address and CoAP token, and assembles Block2 responses +within fixed time, block-count, and payload limits. + +Directory discovery and the DTLS probe have separate jobs: discovery can learn +a device-advertised port outside the caller's fixed fallback set, while +`probe_dtls_ports()` only checks the candidates it receives for a stateless +DTLS first-flight response. Neither step authenticates the appliance. An +advertised port therefore remains only a candidate: require a successful +stateless DTLS probe before attempting authentication. + ### Tested combinations | Appliance class | Model family | Confirmed | @@ -664,6 +693,7 @@ smartthings_local/ The installable library — `pip install sm dtls_probe.py Stateless DTLS liveness + opt-in stateful diagnostic dtls_handshake.py Shared memory-BIO handshake driver, bounded by a monotonic deadline (used by session + probe) owner_psk.py Pure manufacturer-certificate OwnerPSK derivation + ocf_discovery.py Bounded public OCF secure-port discovery ocf_root_ca.pem Samsung OCF root CA, bundled for handshake verification ocf/ OCF resource + state layer (reusable) __init__.py diff --git a/smartthings_local/protocol/ocf_discovery.py b/smartthings_local/protocol/ocf_discovery.py new file mode 100644 index 0000000..f0c4a8d --- /dev/null +++ b/smartthings_local/protocol/ocf_discovery.py @@ -0,0 +1,690 @@ +"""Bounded discovery of OCF-advertised secure UDP ports. + +Samsung appliances normally receive public CoAP discovery on UDP 5683, but +some firmware sends the response from a different source port. This module +therefore uses unconnected UDP sockets, validates the resolved target address +and CoAP token, then pins the first valid response endpoint for the remainder +of each bounded Block2 transfer. + +Discovery first reads the unfiltered ``/oic/res`` directory and accepts only +secure ``eps`` entries bound to that response source. If the representation +contains no secure endpoint, a second, separately correlated +``/oic/res?rt=oic.r.doxm`` lookup supports legacy ``p.sec``/``port`` forms. +Both lookups share one monotonic socket-I/O deadline. + +Directory discovery learns advertised candidates, including ports outside a +caller's conventional scan set. It does not prove that a DTLS service is +present: callers should pass the returned candidates to ``probe_dtls_ports``. +Neither operation authenticates a device, transfers ownership, or writes an +OCF security resource. +""" + +from __future__ import annotations + +import io +import math +import secrets +import selectors +import socket +import time +from dataclasses import dataclass +from urllib.parse import unquote, urlsplit + +import cbor2 + +from ..errors import BlockwiseError, MalformedMessageError +from .coap import ( + BLOCK2_DUPLICATE, + CF_CBOR, + RESPONSE_MESSAGE, + TYPE_CON, + TYPE_NON, + Block2Accumulator, + build_get_request, + classify_coap_response, +) +from .endpoint import ResolvedUdpEndpoint, resolve_udp_endpoints + +__all__ = [ + 'OcfSecurePortDiscoveryResult', + 'discover_ocf_secure_ports', +] + +_DISCOVERY_PORT = 5683 +_MAX_ENDPOINTS = 8 +_MAX_PORTS = 8 +_MAX_BLOCKS = 32 +_MAX_DATAGRAM_BYTES = 8192 +_MAX_PAYLOAD_BYTES = 65536 +_MAX_LINKS = 256 +_MAX_ENDPOINT_URIS_PER_LINK = 32 +_OCF_CBOR_CONTENT_FORMAT = 10000 +_CONTENT = 0x45 +_PRIMARY_QUERY = () +_FALLBACK_QUERY = (b'rt=oic.r.doxm',) +_UNSET = object() + +_TRANSFER_COMPLETE = 'complete' +_TRANSFER_ENDPOINT_UNAVAILABLE = 'endpoint_unavailable' +_TRANSFER_MALFORMED = 'malformed' +_TRANSFER_NO_RESPONSE = 'no_response' + +_PORTS_FOUND = 'ports' +_PORTS_ABSENT = 'absent' +_PORTS_MALFORMED = 'malformed' +_PORTS_UNTRUSTED = 'untrusted' + +_ENDPOINT_IGNORE = 'ignore' +_ENDPOINT_MATCH = 'match' +_ENDPOINT_UNTRUSTED = 'untrusted' + + +@dataclass(frozen=True, slots=True, repr=False) +class OcfSecurePortDiscoveryResult: + """Redacted outcome of one bounded secure-port discovery operation. + + ``attempts`` counts logical request attempts across the primary and, when + needed, fallback lookup rather than destination addresses. + ``response_received`` is true when either lookup accepted at least one + correlated response. The custom representation deliberately omits + discovered ports, addresses, and wire data. + """ + + ports: tuple[int, ...] + attempts: int + response_received: bool + error_code: str | None = None + + @property + def found(self): + """Return whether at least one validated secure port was advertised.""" + return bool(self.ports) + + def __repr__(self): + return ( + 'OcfSecurePortDiscoveryResult(' + f'found={self.found!r}, port_count={len(self.ports)}, ' + f'attempts={self.attempts}, ' + f'response_received={self.response_received!r}, ' + f'error_code={self.error_code!r})' + ) + + +@dataclass(slots=True, repr=False) +class _Route: + sock: socket.socket + endpoint: ResolvedUdpEndpoint + host_key: tuple[bytes, int] + + +@dataclass(frozen=True, slots=True, repr=False) +class _TransferResult: + status: str + payload: bytes + code: int | None + family: int | None + source_key: tuple[bytes, int] | None + attempts: int + response_received: bool + + +def _validate_options(discovery_port, timeout, retries, family): + if isinstance(discovery_port, bool) or not isinstance(discovery_port, int): + raise TypeError('discovery_port must be an integer') + if not 1 <= discovery_port <= 65535: + raise ValueError('discovery_port must be between 1 and 65535') + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)): + raise TypeError('timeout must be a number') + if not math.isfinite(timeout) or not 0 < timeout <= 30: + raise ValueError('timeout must be greater than zero and at most 30') + if isinstance(retries, bool) or not isinstance(retries, int): + raise TypeError('retries must be an integer') + if not 0 <= retries <= 4: + raise ValueError('retries must be between zero and four') + if isinstance(family, bool) or not isinstance(family, int): + raise TypeError('family must be an address-family integer') + if family not in (socket.AF_UNSPEC, socket.AF_INET, socket.AF_INET6): + raise ValueError('family must be AF_UNSPEC, AF_INET, or AF_INET6') + + +def _host_key(family, sockaddr): + """Return canonical address bytes plus an IPv6 scope ID.""" + expected_length = 2 if family == socket.AF_INET else 4 + if not isinstance(sockaddr, tuple) or len(sockaddr) != expected_length: + return None + host = sockaddr[0] + if not isinstance(host, str): + return None + if family == socket.AF_INET6: + host = host.split('%', 1)[0] + try: + packed = socket.inet_pton(family, host) + except OSError: + return None + scope_id = sockaddr[3] if family == socket.AF_INET6 else 0 + if isinstance(scope_id, bool) or not isinstance(scope_id, int): + return None + return packed, scope_id + + +def _peer_key(family, sockaddr): + host_key = _host_key(family, sockaddr) + if host_key is None: + return None + port = sockaddr[1] + if isinstance(port, bool) or not isinstance(port, int): + return None + if not 1 <= port <= 65535: + return None + return family, host_key[0], port, host_key[1] + + +def _open_routes(endpoints, selector): + routes = [] + for endpoint in endpoints[:_MAX_ENDPOINTS]: + key = _host_key(endpoint.family, endpoint.sockaddr) + if key is None: + continue + sock = None + try: + sock = socket.socket( + endpoint.family, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.bind(endpoint.bind_address(0)) + sock.setblocking(False) + route = _Route(sock, endpoint, key) + selector.register(sock, selectors.EVENT_READ, route) + routes.append(route) + except (OSError, ValueError): + if sock is not None: + try: + sock.close() + except OSError: + pass + return routes + + +def _decode_cbor(payload): + stream = io.BytesIO(payload) + try: + value = cbor2.CBORDecoder(stream).decode() + except Exception: # noqa: BLE001 - untrusted CBOR must fail closed + return _UNSET + if stream.tell() != len(payload): + return _UNSET + return value + + +def _resource_links(value): + """Return a shallow, bounded OCF link sequence or ``None``.""" + containers = value if isinstance(value, list) else [value] + if not all(isinstance(container, dict) for container in containers): + return None + + links = [] + for container in containers: + if 'links' in container: + nested = container.get('links') + if not isinstance(nested, list): + return None + candidates = nested + elif 'href' in container: + candidates = [container] + else: + candidates = [] + for link in candidates: + if not isinstance(link, dict): + return None + links.append(link) + if len(links) > _MAX_LINKS: + return None + return links + + +def _uri_scope_id(zone): + if not zone: + return None + if zone.isdecimal(): + value = int(zone, 10) + return value if value <= 0xFFFFFFFF else None + try: + return socket.if_nametoindex(zone) + except (OSError, ValueError): + return None + + +def _secure_endpoint_for_source(value, family, source_key): + """Classify one endpoint URI without resolving untrusted hostnames.""" + if not isinstance(value, str): + return _ENDPOINT_IGNORE, None + secure_hint = value.lower().startswith('coaps:') + try: + parsed = urlsplit(value) + hostname = parsed.hostname + port = parsed.port + username = parsed.username + password = parsed.password + except ValueError: + return (_ENDPOINT_UNTRUSTED if secure_hint else _ENDPOINT_IGNORE), None + + if parsed.scheme.lower() != 'coaps': + return _ENDPOINT_IGNORE, None + if (not hostname or username is not None or password is not None + or parsed.path or parsed.query or parsed.fragment + or parsed.netloc.endswith(':')): + return _ENDPOINT_UNTRUSTED, None + + hostname = unquote(hostname) + scope_id = 0 + if family == socket.AF_INET6: + if '%' in hostname: + address, zone = hostname.rsplit('%', 1) + scope_id = _uri_scope_id(zone) + if scope_id is None: + return _ENDPOINT_UNTRUSTED, None + else: + address = hostname + # A zone identifier is local to the receiver. An omitted zone in + # an advertised link-local URI is interpreted in the exact scope + # on which the correlated datagram arrived. + scope_id = source_key[1] + else: + if '%' in hostname: + return _ENDPOINT_UNTRUSTED, None + address = hostname + + try: + packed = socket.inet_pton(family, address) + except OSError: + # Do not perform DNS for a hostname supplied by an unauthenticated + # directory response. + return _ENDPOINT_UNTRUSTED, None + if (packed, scope_id) != source_key: + return _ENDPOINT_UNTRUSTED, None + + if port is None: + port = 5684 + if not 1 <= port <= 65535: + return _ENDPOINT_UNTRUSTED, None + return _ENDPOINT_MATCH, port + + +def _add_port(ports, seen, port): + if (isinstance(port, bool) or not isinstance(port, int) + or not 1 <= port <= 65535 or port in seen): + return + seen.add(port) + if len(ports) < _MAX_PORTS: + ports.append(port) + + +def _ports_from_links(links, family, source_key, *, fallback): + ports = [] + seen = set() + saw_untrusted = False + + for link in links: + if fallback: + if link.get('href') != '/oic/sec/doxm': + continue + resource_types = link.get('rt') + if isinstance(resource_types, str): + resource_types = [resource_types] + if (not isinstance(resource_types, list) + or 'oic.r.doxm' not in resource_types): + continue + + policy = link.get('p') + if isinstance(policy, dict) and policy.get('sec') is True: + _add_port(ports, seen, policy.get('port')) + + endpoints = link.get('eps') + if not isinstance(endpoints, list): + continue + for endpoint in endpoints[:_MAX_ENDPOINT_URIS_PER_LINK]: + if not isinstance(endpoint, dict): + continue + status, port = _secure_endpoint_for_source( + endpoint.get('ep'), family, source_key) + if status == _ENDPOINT_MATCH: + _add_port(ports, seen, port) + elif status == _ENDPOINT_UNTRUSTED: + saw_untrusted = True + + if ports: + return _PORTS_FOUND, tuple(ports) + if saw_untrusted: + return _PORTS_UNTRUSTED, () + return _PORTS_ABSENT, () + + +def _primary_secure_ports_from_payload(payload, family, source_key): + value = _decode_cbor(payload) + if value is _UNSET: + return _PORTS_MALFORMED, () + links = _resource_links(value) + if links is None: + return _PORTS_MALFORMED, () + return _ports_from_links(links, family, source_key, fallback=False) + + +def _fallback_secure_ports_from_payload(payload, family, source_key): + value = _decode_cbor(payload) + if value is _UNSET: + return _PORTS_MALFORMED, () + links = _resource_links(value) + if links is None: + return _PORTS_MALFORMED, () + return _ports_from_links(links, family, source_key, fallback=True) + + +def _transfer_result( + status, *, attempts, response_received, accumulator=None, route=None): + complete = accumulator is not None and accumulator.complete + return _TransferResult( + status=status, + payload=accumulator.payload if complete else b'', + code=accumulator.code if complete else None, + family=route.endpoint.family if complete and route is not None else None, + source_key=route.host_key if complete and route is not None else None, + attempts=attempts, + response_received=response_received, + ) + + +def _fetch_directory( + routes, selector, *, query, cutoff, retries, used_mids): + """Fetch one representation without retaining an address in its repr.""" + token = secrets.token_bytes(8) + accumulator = Block2Accumulator( + token, + max_blocks=_MAX_BLOCKS, + max_payload_bytes=_MAX_PAYLOAD_BYTES, + accepted_content_formats={ + int.from_bytes(CF_CBOR, 'big'), + _OCF_CBOR_CONTENT_FORMAT, + }, + ) + wait_slice = max( + 0.0, + min(1.0, (cutoff - time.monotonic()) / (retries + 1)), + ) + attempts = 0 + response_received = False + saw_malformed = False + pinned_route = None + pinned_peer = None + pinned_destination = None + + while not accumulator.complete: + accepted = False + sent_for_block = False + for block_attempt in range(retries + 1): + if time.monotonic() >= cutoff: + break + mid = secrets.randbits(16) + while mid in used_mids: + mid = (mid + 1) & 0xFFFF + used_mids.add(mid) + request = build_get_request( + TYPE_NON, + mid, + token, + (b'oic', b'res'), + query, + block_number=( + accumulator.expected_number + if accumulator.expected_number > 0 else None + ), + block_szx=accumulator.szx, + ) + send_routes = [pinned_route] if pinned_route else routes + sent = False + for route in send_routes: + try: + destination = ( + pinned_destination + if route is pinned_route and pinned_destination + else route.endpoint.sockaddr + ) + sent_length = route.sock.sendto(request, destination) + sent = sent or sent_length == len(request) + except OSError: + continue + attempts += 1 + if not sent: + continue + sent_for_block = True + + now = time.monotonic() + attempt_deadline = ( + cutoff if block_attempt == retries + else min(cutoff, now + wait_slice) + ) + while True: + remaining = attempt_deadline - time.monotonic() + if remaining <= 0: + break + try: + events = selector.select(remaining) + except (OSError, ValueError): + events = [] + if not events: + break + for key, _mask in events: + route = key.data + try: + datagram, source = route.sock.recvfrom( + _MAX_DATAGRAM_BYTES + 1) + except (BlockingIOError, OSError): + continue + source_host_key = _host_key( + route.endpoint.family, source) + peer_key = _peer_key(route.endpoint.family, source) + if source_host_key != route.host_key or peer_key is None: + continue + if pinned_peer is not None and ( + route is not pinned_route + or peer_key != pinned_peer): + continue + if len(datagram) > _MAX_DATAGRAM_BYTES: + saw_malformed = True + continue + + try: + classification = classify_coap_response( + datagram, token=token, request_mid=mid) + except MalformedMessageError: + saw_malformed = True + continue + message = classification.message + if (classification.kind != RESPONSE_MESSAGE + or message is None + or message.mtype not in (TYPE_CON, TYPE_NON)): + continue + + try: + block_status = accumulator.add_response(message) + except BlockwiseError: + saw_malformed = True + continue + + # ACK only an accepted, correlated CON. The shared + # classifier may offer an ACK for an ignored CON to + # preserve connected-session behavior. + if (message.mtype == TYPE_CON + and classification.acknowledgement is not None): + try: + route.sock.sendto( + classification.acknowledgement, source) + except OSError: + pass + + response_received = True + if pinned_peer is None: + pinned_route = route + pinned_peer = peer_key + pinned_destination = tuple(source) + if block_status == BLOCK2_DUPLICATE: + continue + accepted = True + break + if accepted: + break + if accepted: + break + + if accumulator.complete: + return _transfer_result( + _TRANSFER_COMPLETE, + attempts=attempts, + response_received=response_received, + accumulator=accumulator, + route=pinned_route, + ) + if not accepted: + if response_received or saw_malformed: + status = _TRANSFER_MALFORMED + elif not sent_for_block and attempts: + status = _TRANSFER_ENDPOINT_UNAVAILABLE + else: + status = _TRANSFER_NO_RESPONSE + return _transfer_result( + status, + attempts=attempts, + response_received=response_received, + ) + + return _transfer_result( + _TRANSFER_MALFORMED, + attempts=attempts, + response_received=response_received, + ) + + +def _result(ports, attempts, response_received, error_code=None): + return OcfSecurePortDiscoveryResult( + ports=ports, + attempts=attempts, + response_received=response_received, + error_code=error_code, + ) + + +def _extraction_for_transfer(transfer, *, fallback): + if transfer.status != _TRANSFER_COMPLETE: + return None + if transfer.code != _CONTENT: + return _PORTS_ABSENT, () + extractor = ( + _fallback_secure_ports_from_payload + if fallback else _primary_secure_ports_from_payload + ) + return extractor(transfer.payload, transfer.family, transfer.source_key) + + +def discover_ocf_secure_ports( + host, *, discovery_port=_DISCOVERY_PORT, timeout=3.0, retries=1, + family=socket.AF_UNSPEC): + """Discover secure ports advertised by a target's public OCF directory. + + Name resolution happens synchronously first. ``timeout`` then bounds both + explicit directory lookups and every Block2 continuation. An advertisement + is only a candidate; callers should prove it with + :func:`smartthings_local.protocol.dtls_probe.probe_dtls_ports` before a + DTLS handshake. + """ + _validate_options(discovery_port, timeout, retries, family) + try: + endpoints = resolve_udp_endpoints( + host, discovery_port, family=family) + except OSError: + return _result((), 0, False, 'endpoint_unavailable') + + selector = selectors.DefaultSelector() + routes = _open_routes(endpoints, selector) + if not routes: + selector.close() + return _result((), 0, False, 'endpoint_unavailable') + + started = time.monotonic() + deadline = started + float(timeout) + # Reserve half of short timeouts, capped at one second, so a filtered-only + # legacy target can still answer inside the same total deadline. + primary_cutoff = deadline - min(1.0, float(timeout) / 2) + attempts = 0 + response_received = False + used_mids = set() + + try: + primary = _fetch_directory( + routes, + selector, + query=_PRIMARY_QUERY, + cutoff=primary_cutoff, + retries=retries, + used_mids=used_mids, + ) + attempts += primary.attempts + response_received = response_received or primary.response_received + + if primary.status == _TRANSFER_ENDPOINT_UNAVAILABLE: + return _result( + (), attempts, response_received, 'endpoint_unavailable') + if primary.status == _TRANSFER_MALFORMED: + return _result( + (), attempts, response_received, 'malformed_ocf_response') + if primary.status == _TRANSFER_COMPLETE: + extraction = _extraction_for_transfer(primary, fallback=False) + status, ports = extraction + if status == _PORTS_FOUND: + return _result(ports, attempts, response_received) + if status in (_PORTS_MALFORMED, _PORTS_UNTRUSTED): + return _result( + (), attempts, response_received, + 'malformed_ocf_response') + + # A completely unanswered primary request and a valid representation + # with no usable secure eps are the only fallback conditions. The + # fallback gets a fresh token, accumulator, and peer pin and starts at + # the original public discovery routes. + fallback_result = _fetch_directory( + routes, + selector, + query=_FALLBACK_QUERY, + cutoff=deadline, + retries=retries, + used_mids=used_mids, + ) + attempts += fallback_result.attempts + response_received = ( + response_received or fallback_result.response_received) + + if fallback_result.status == _TRANSFER_ENDPOINT_UNAVAILABLE: + return _result( + (), attempts, response_received, 'endpoint_unavailable') + if fallback_result.status == _TRANSFER_MALFORMED: + return _result( + (), attempts, response_received, 'malformed_ocf_response') + if fallback_result.status == _TRANSFER_NO_RESPONSE: + return _result( + (), attempts, response_received, 'no_ocf_response') + + status, ports = _extraction_for_transfer( + fallback_result, fallback=True) + if status == _PORTS_FOUND: + return _result(ports, attempts, response_received) + if status in (_PORTS_MALFORMED, _PORTS_UNTRUSTED): + return _result( + (), attempts, response_received, 'malformed_ocf_response') + return _result((), attempts, response_received, 'no_secure_ports') + finally: + for route in routes: + try: + selector.unregister(route.sock) + except (KeyError, OSError, ValueError): + pass + try: + route.sock.close() + except OSError: + pass + selector.close() diff --git a/tests/test_import_isolation.py b/tests/test_import_isolation.py index a6d50a2..459c3d9 100644 --- a/tests/test_import_isolation.py +++ b/tests/test_import_isolation.py @@ -19,6 +19,7 @@ def test_smartthings_local_imports_without_mqtt_demo_present(tmp_path): "import smartthings_local.protocol.coap", "import smartthings_local.protocol.ocf_multicast", "import smartthings_local.protocol.dtls_session", + "import smartthings_local.protocol.ocf_discovery", "import smartthings_local.ocf.state_cache", "import smartthings_local.ocf.poll_scheduler", "import smartthings_local.ocf.keepalive", @@ -34,6 +35,7 @@ def test_smartthings_local_imports_without_mqtt_demo_present(tmp_path): cwd=str(tmp_path), env=env, capture_output=True, text=True, + check=False, ) assert result.returncode == 0, ( f"smartthings_local failed to import without mqtt_demo/ present:\n" diff --git a/tests/test_ocf_discovery.py b/tests/test_ocf_discovery.py new file mode 100644 index 0000000..5e8af51 --- /dev/null +++ b/tests/test_ocf_discovery.py @@ -0,0 +1,818 @@ +"""Public OCF secure-port discovery stays bounded and source-correlated.""" + +import socket +import threading +import time +import traceback +from dataclasses import FrozenInstanceError + +import cbor2 +import pytest + +from smartthings_local.errors import EndpointError +from smartthings_local.protocol import ocf_discovery as discovery +from smartthings_local.protocol.coap import ( + ACCEPT, + BLOCK2, + CF_CBOR, + CONTENT_FORMAT, + ETAG, + METHOD_GET, + SIZE2, + TYPE_ACK, + TYPE_CON, + TYPE_NON, + URI_PATH, + URI_QUERY, + block_value, + build_coap, + parse_coap, +) + +# Synthetic private-use UDP fixtures; none are captured appliance endpoints. + + +def _doxm_link(port=61002, *, endpoints=None): + link = { + 'href': '/oic/sec/doxm', + 'rt': ['oic.r.doxm'], + 'p': {'sec': True, 'port': port}, + } + if endpoints is not None: + link['eps'] = [{'ep': endpoint} for endpoint in endpoints] + return link + + +def _eps_link(*endpoints, href='/oic/d'): + return { + 'href': href, + 'rt': ['oic.wk.d'], + 'eps': [{'ep': endpoint} for endpoint in endpoints], + } + + +def _payload(*links, padding=''): + value = {'links': list(links)} + if padding: + value['padding'] = padding + return cbor2.dumps(value) + + +def _option_map(options): + result = {} + for number, value in options: + result.setdefault(number, []).append(value) + return result + + +def _uint_bytes(value): + length = max(1, (value.bit_length() + 7) // 8) + return value.to_bytes(length, 'big') + + +def _ipv4_key(address='192.0.2.20'): + return socket.inet_pton(socket.AF_INET, address), 0 + + +def test_primary_uses_source_bound_eps_from_all_links_and_ignores_legacy(): + payload = _payload( + _doxm_link(61002), + _eps_link( + 'coaps://192.0.2.20:61003', + 'coaps://192.0.2.20', + 'coap://192.0.2.20:61004', + 'coaps+tcp://192.0.2.20:61005', + 'coaps://192.0.2.21:61006', + ), + ) + + status, ports = discovery._primary_secure_ports_from_payload( + payload, socket.AF_INET, _ipv4_key()) + + assert status == discovery._PORTS_FOUND + assert ports == (61003, 5684) + + +def test_fallback_uses_only_doxm_eps_and_legacy_ports_and_stays_bounded(): + links = [ + {'href': '/oic/d', 'rt': ['oic.wk.d'], + 'p': {'sec': True, 'port': 49000}}, + _doxm_link( + 61000, + endpoints=[ + 'coaps://192.0.2.20:61001', + 'coaps://192.0.2.21:61999', + ], + ), + *[_doxm_link(port) for port in range(61002, 61012)], + ] + + status, ports = discovery._fallback_secure_ports_from_payload( + cbor2.dumps(links), socket.AF_INET, _ipv4_key()) + + assert status == discovery._PORTS_FOUND + assert ports == tuple(range(61000, 61008)) + + +@pytest.mark.parametrize( + 'payload', + ( + b'not-cbor', + cbor2.dumps({'links': 'not-a-list'}), + cbor2.dumps([{'links': [None]}]), + cbor2.dumps({'links': []}) + cbor2.dumps(1), + ), +) +@pytest.mark.parametrize( + 'extractor', + ( + discovery._primary_secure_ports_from_payload, + discovery._fallback_secure_ports_from_payload, + ), +) +def test_malformed_or_trailing_cbor_is_rejected(payload, extractor): + assert extractor( + payload, socket.AF_INET, _ipv4_key()) == ( + discovery._PORTS_MALFORMED, ()) + + +def test_primary_distinguishes_absence_from_untrusted_secure_eps(): + absent = _payload(_doxm_link(), _eps_link('coap://192.0.2.20:61003')) + untrusted_values = ( + 'coaps://192.0.2.21:61003', + 'coaps://appliance.invalid:61003', + 'coaps://192.0.2.20:', + 'coaps://user:secret@192.0.2.20:61003', + 'coaps://192.0.2.20:61003/path', + ) + + assert discovery._primary_secure_ports_from_payload( + absent, socket.AF_INET, _ipv4_key()) == ( + discovery._PORTS_ABSENT, ()) + for endpoint in untrusted_values: + assert discovery._primary_secure_ports_from_payload( + _payload(_eps_link(endpoint)), + socket.AF_INET, + _ipv4_key(), + ) == (discovery._PORTS_UNTRUSTED, ()) + + +def test_ipv6_eps_binding_inherits_or_exactly_matches_response_scope(): + source_key = ( + socket.inet_pton(socket.AF_INET6, '2001:db8::20'), + 7, + ) + matching = _payload(_eps_link( + 'coaps://[2001:db8::20]:62000', + 'coaps://[2001:db8::20%257]:62001', + 'coaps://[2001:db8::20%258]:62002', + )) + + status, ports = discovery._primary_secure_ports_from_payload( + matching, socket.AF_INET6, source_key) + + assert status == discovery._PORTS_FOUND + assert ports == (62000, 62001) + assert discovery._primary_secure_ports_from_payload( + _payload(_eps_link('coaps://[2001:db8::20%258]:62002')), + socket.AF_INET6, + source_key, + ) == (discovery._PORTS_UNTRUSTED, ()) + assert discovery._primary_secure_ports_from_payload( + _payload(_eps_link('coaps://[2001:db8::21]:62003')), + socket.AF_INET6, + source_key, + ) == (discovery._PORTS_UNTRUSTED, ()) + + +def test_unfiltered_dynamic_source_and_two_block_response_are_supported(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + responder.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + responder.settimeout(2.0) + assert listener.getsockname()[1] != responder.getsockname()[1] + + body = _payload( + _eps_link('coaps://127.0.0.1:61002'), + padding='x' * 300, + ) + block_size = 256 + assert block_size < len(body) <= block_size * 2 + errors = [] + + def respond(): + try: + first_request, client = listener.recvfrom(8192) + mtype, code, first_mid, token, options, request_payload = \ + parse_coap(first_request) + option_map = _option_map(options) + assert (mtype, code) == (TYPE_NON, METHOD_GET) + assert len(token) == 8 and request_payload == b'' + assert option_map[URI_PATH] == [b'oic', b'res'] + assert URI_QUERY not in option_map + assert option_map[ACCEPT] == [CF_CBOR] + assert BLOCK2 not in option_map + + common = [ + (CONTENT_FORMAT, CF_CBOR), + (ETAG, b'test'), + (SIZE2, _uint_bytes(len(body))), + ] + responder.sendto( + build_coap( + TYPE_CON, + 0x45, + 0x7001, + token, + [*common, (BLOCK2, block_value(0, 1, 4))], + body[:block_size], + ), + client, + ) + first_ack, ack_peer = responder.recvfrom(8192) + assert ack_peer == client + assert parse_coap(first_ack)[:4] == ( + TYPE_ACK, 0, 0x7001, b'') + + second_request, second_client = responder.recvfrom(8192) + mtype, code, second_mid, second_token, options, request_payload = \ + parse_coap(second_request) + option_map = _option_map(options) + assert second_client == client + assert (mtype, code) == (TYPE_NON, METHOD_GET) + assert second_token == token and second_mid != first_mid + assert request_payload == b'' and URI_QUERY not in option_map + assert option_map[BLOCK2] == [block_value(1, 0, 4)] + + responder.sendto( + build_coap( + TYPE_CON, + 0x45, + 0x7002, + token, + [*common, (BLOCK2, block_value(1, 0, 4))], + body[block_size:], + ), + client, + ) + second_ack, _ack_peer = responder.recvfrom(8192) + assert parse_coap(second_ack)[:4] == ( + TYPE_ACK, 0, 0x7002, b'') + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=1.5, + retries=1, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=3.0) + listener.close() + responder.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61002,) + assert result.response_received + assert result.error_code is None + assert result.attempts == 2 + + +def test_absent_primary_falls_back_with_fresh_token_on_original_route(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + responder.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + responder.settimeout(2.0) + errors = [] + + def respond(): + try: + primary, client = listener.recvfrom(8192) + primary_parsed = parse_coap(primary) + assert URI_QUERY not in _option_map(primary_parsed[4]) + responder.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7101, + primary_parsed[3], + [], + _payload({'href': '/oic/d', 'rt': ['oic.wk.d']}), + ), + client, + ) + + # A new logical GET starts from the original discovery socket, + # rather than the primary response's dynamic source port. + fallback, fallback_client = listener.recvfrom(8192) + fallback_parsed = parse_coap(fallback) + option_map = _option_map(fallback_parsed[4]) + assert fallback_client == client + assert option_map[URI_QUERY] == [b'rt=oic.r.doxm'] + assert fallback_parsed[3] != primary_parsed[3] + assert fallback_parsed[2] != primary_parsed[2] + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7102, + fallback_parsed[3], + [], + _payload(_doxm_link()), + ), + fallback_client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=1.5, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=3.0) + listener.close() + responder.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61002,) + assert result.attempts == 2 + assert result.response_received + + +def test_unanswered_primary_reserves_time_for_filtered_fallback(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + errors = [] + + def respond(): + try: + primary, client = listener.recvfrom(8192) + primary_parsed = parse_coap(primary) + assert URI_QUERY not in _option_map(primary_parsed[4]) + fallback, fallback_client = listener.recvfrom(8192) + fallback_parsed = parse_coap(fallback) + assert fallback_client == client + assert _option_map(fallback_parsed[4])[URI_QUERY] == [ + b'rt=oic.r.doxm'] + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7201, + fallback_parsed[3], + [], + _payload(_doxm_link(61003)), + ), + fallback_client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + started = time.monotonic() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.8, + retries=0, + family=socket.AF_INET, + ) + finally: + elapsed = time.monotonic() - started + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61003,) + assert result.attempts == 2 + assert elapsed < 1.0 + + +@pytest.mark.parametrize( + 'primary_payload', + ( + b'not-cbor', + _payload(_eps_link('coaps://127.0.0.2:61002')), + ), +) +def test_malformed_or_cross_source_primary_never_starts_fallback( + primary_payload): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(0.4) + errors = [] + saw_fallback = [] + + def respond(): + try: + request, client = listener.recvfrom(8192) + parsed = parse_coap(request) + listener.sendto( + build_coap( + TYPE_NON, 0x45, 0x7301, parsed[3], [], primary_payload), + client, + ) + try: + listener.recvfrom(8192) + except TimeoutError: + return + saw_fallback.append(True) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.6, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert saw_fallback == [] + assert result.ports == () + assert result.response_received + assert result.error_code == 'malformed_ocf_response' + + +def test_stale_primary_token_is_ignored_during_fallback(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + errors = [] + + def respond(): + try: + primary, client = listener.recvfrom(8192) + primary_token = parse_coap(primary)[3] + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7401, + primary_token, + [], + _payload({'href': '/oic/d'}), + ), + client, + ) + fallback, fallback_client = listener.recvfrom(8192) + fallback_token = parse_coap(fallback)[3] + assert fallback_token != primary_token + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7402, + primary_token, + [], + _payload(_doxm_link(61999)), + ), + fallback_client, + ) + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7403, + fallback_token, + [], + _payload(_doxm_link(61004)), + ), + fallback_client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=1.0, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61004,) + assert result.attempts == 2 + + +def test_non_request_rejects_piggyback_ack_and_uses_fallback(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + errors = [] + + def respond(): + try: + primary, client = listener.recvfrom(8192) + primary_parsed = parse_coap(primary) + listener.sendto( + build_coap( + TYPE_ACK, + 0x45, + primary_parsed[2], + primary_parsed[3], + [], + _payload(_eps_link('coaps://127.0.0.1:61999')), + ), + client, + ) + fallback, fallback_client = listener.recvfrom(8192) + fallback_parsed = parse_coap(fallback) + assert _option_map(fallback_parsed[4])[URI_QUERY] == [ + b'rt=oic.r.doxm'] + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7501, + fallback_parsed[3], + [], + _payload(_doxm_link(61005)), + ), + fallback_client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.8, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61005,) + assert result.attempts == 2 + + +def test_wrong_token_con_is_not_acknowledged(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(0.4) + errors = [] + unexpected_datagrams = [] + + def respond(): + try: + request, client = listener.recvfrom(8192) + token = parse_coap(request)[3] + listener.sendto( + build_coap( + TYPE_CON, + 0x45, + 0x7601, + b'badtoken', + [], + _payload(_eps_link('coaps://127.0.0.1:61999')), + ), + client, + ) + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7602, + token, + [], + _payload(_eps_link('coaps://127.0.0.1:61006')), + ), + client, + ) + try: + unexpected_datagrams.append(listener.recvfrom(8192)) + except TimeoutError: + pass + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.8, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert unexpected_datagrams == [] + assert result.ports == (61006,) + + +def test_partial_block_transfer_stays_pinned_and_fails_closed(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + first_responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + other_responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + first_responder.bind(('127.0.0.1', 0)) + other_responder.bind(('127.0.0.1', 0)) + listener.settimeout(1.0) + first_responder.settimeout(1.0) + body = _payload( + _eps_link('coaps://127.0.0.1:61007'), padding='x' * 10) + assert 64 < len(body) <= 128 + errors = [] + + def respond(): + try: + request, client = listener.recvfrom(8192) + token = parse_coap(request)[3] + first_responder.sendto( + build_coap( + TYPE_NON, 0x45, 0x7701, token, + [(BLOCK2, block_value(0, 1, 2))], body[:64]), + client, + ) + _request, second_client = first_responder.recvfrom(8192) + assert second_client == client + other_responder.sendto( + build_coap( + TYPE_NON, 0x45, 0x7702, token, + [(BLOCK2, block_value(1, 0, 2))], body[64:]), + client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.6, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + first_responder.close() + other_responder.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == () + assert result.response_received + assert result.error_code == 'malformed_ocf_response' + + +def test_primary_retry_keeps_token_and_changes_message_id(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + errors = [] + + def respond(): + try: + first, client = listener.recvfrom(8192) + second, second_client = listener.recvfrom(8192) + first_parsed = parse_coap(first) + second_parsed = parse_coap(second) + assert second_client == client + assert first_parsed[0] == second_parsed[0] == TYPE_NON + assert first_parsed[3] == second_parsed[3] + assert first_parsed[2] != second_parsed[2] + assert URI_QUERY not in _option_map(second_parsed[4]) + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7801, + second_parsed[3], + [], + _payload(_eps_link('coaps://127.0.0.1:61008')), + ), + client, + ) + except Exception as exc: # noqa: BLE001 - surfaced through errors below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=1.0, + retries=1, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (61008,) + assert result.attempts == 2 + + +def test_resolution_failure_and_result_repr_are_redacted(monkeypatch): + remote_host = 'private-appliance.invalid' + + def fail(host, port, *, family): + assert host == remote_host + assert port == 5683 + assert family == socket.AF_INET6 + raise EndpointError() + + monkeypatch.setattr(discovery, 'resolve_udp_endpoints', fail) + + result = discovery.discover_ocf_secure_ports( + remote_host, family=socket.AF_INET6) + rendered = repr(result) + ''.join( + traceback.format_exception(EndpointError())) + + assert result.error_code == 'endpoint_unavailable' + assert result.attempts == 0 + assert remote_host not in rendered + assert '61002' not in repr( + discovery.OcfSecurePortDiscoveryResult((61002,), 1, True)) + + +def test_result_is_immutable_and_ipv6_scope_is_part_of_source_identity(): + result = discovery.OcfSecurePortDiscoveryResult((61002,), 1, True) + + with pytest.raises(FrozenInstanceError): + result.attempts = 2 + assert discovery._host_key( + socket.AF_INET, ('192.0.2.20', 5683)) != discovery._host_key( + socket.AF_INET, ('192.0.2.21', 5683)) + assert discovery._host_key( + socket.AF_INET6, ('2001:db8::20', 5683, 0, 7)) != \ + discovery._host_key( + socket.AF_INET6, ('2001:db8::20', 5683, 0, 8)) + + +@pytest.mark.parametrize( + ('keyword', 'value', 'error_type'), + ( + ('discovery_port', 0, ValueError), + ('discovery_port', True, TypeError), + ('timeout', 0, ValueError), + ('timeout', float('nan'), ValueError), + ('timeout', True, TypeError), + ('retries', 5, ValueError), + ('retries', True, TypeError), + ('family', 9999, ValueError), + ('family', 'AF_INET', TypeError), + ), +) +def test_invalid_options_fail_before_network(keyword, value, error_type): + with pytest.raises(error_type): + discovery.discover_ocf_secure_ports( + '192.0.2.20', **{keyword: value}) diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index b070bec..e08148f 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -18,6 +18,10 @@ ConnectCancellation, DtlsCoapSession, ) +from smartthings_local.protocol.ocf_discovery import ( + OcfSecurePortDiscoveryResult, + discover_ocf_secure_ports, +) from smartthings_local.protocol.ocf_multicast import ( OcfResponderPortDiscoveryResult, discover_ocf_responder_ports, @@ -246,3 +250,15 @@ def test_observe_refresh_task_keeps_current_consumer_surface(): ObserveRefreshTask.run_forever, ["self", "stop"], ) + + +def test_ocf_secure_port_discovery_has_a_small_composable_surface(): + _assert_compatible_signature(discover_ocf_secure_ports, ["host"]) + result = OcfSecurePortDiscoveryResult( + ports=(5684,), + attempts=1, + response_received=True, + ) + + assert result.found + assert result.ports == (5684,) From 2aa384fe296c3fc1a1cd1547b834497a8a49683d Mon Sep 17 00:00:00 2001 From: hoon <230467962+atc722@users.noreply.github.com> Date: Fri, 21 Aug 2026 16:37:06 +0900 Subject: [PATCH 3/3] docs(protocol): clarify known public discovery port --- README.md | 13 ++++++++++--- smartthings_local/protocol/ocf_discovery.py | 17 ++++++++++++----- 2 files changed, 22 insertions(+), 8 deletions(-) diff --git a/README.md b/README.md index 59aa195..6527ffb 100644 --- a/README.md +++ b/README.md @@ -349,14 +349,21 @@ candidates = advertisement.ports or fallback_ports probe = probe_dtls_ports(appliance_host, candidates) ``` +`discovery_port` is the target's already-known public CoAP request port. Its +5683 default is only a convenience: this function does not scan or use +multicast to locate a different public port. If the appliance does not listen +on 5683, locate that public port separately and pass it explicitly as +`discovery_port=...`. + `discover_ocf_secure_ports()` first reads the public `/oic/res` directory and uses only `coaps://` endpoints whose literal host matches the correlated response source. If that first lookup yields no correlated response or no usable secure endpoint, the same overall deadline also bounds a filtered `/oic/res?rt=oic.r.doxm` fallback for Samsung's legacy secure-port policy. It -accepts Samsung's dynamic plaintext response source port while still requiring -the resolved target address and CoAP token, and assembles Block2 responses -within fixed time, block-count, and payload limits. +accepts a different dynamic response source port after the request reaches the +known public port, while still requiring the resolved target address and CoAP +token, and assembles Block2 responses within fixed time, block-count, and +payload limits. Directory discovery and the DTLS probe have separate jobs: discovery can learn a device-advertised port outside the caller's fixed fallback set, while diff --git a/smartthings_local/protocol/ocf_discovery.py b/smartthings_local/protocol/ocf_discovery.py index f0c4a8d..9c7a8cc 100644 --- a/smartthings_local/protocol/ocf_discovery.py +++ b/smartthings_local/protocol/ocf_discovery.py @@ -1,10 +1,12 @@ """Bounded discovery of OCF-advertised secure UDP ports. -Samsung appliances normally receive public CoAP discovery on UDP 5683, but -some firmware sends the response from a different source port. This module -therefore uses unconnected UDP sockets, validates the resolved target address -and CoAP token, then pins the first valid response endpoint for the remainder -of each bounded Block2 transfer. +This known-host API requires the target's public CoAP request port to already +be known. UDP 5683 is a convenience default, not a port-discovery mechanism; +callers must locate and explicitly pass a different public port when the +target does not listen there. A response may still originate from another +source port, so this module uses unconnected UDP sockets, validates the +resolved target address and CoAP token, then pins the first valid response +endpoint for the remainder of each bounded Block2 transfer. Discovery first reads the unfiltered ``/oic/res`` directory and accepts only secure ``eps`` entries bound to that response source. If the representation @@ -587,6 +589,11 @@ def discover_ocf_secure_ports( family=socket.AF_UNSPEC): """Discover secure ports advertised by a target's public OCF directory. + ``discovery_port`` is the target's already-known public CoAP request port. + Its 5683 default is only a convenience; this function does not scan or use + multicast to locate a different public port before sending the request. + Callers must locate any such port separately and pass it explicitly. + Name resolution happens synchronously first. ``timeout`` then bounds both explicit directory lookups and every Block2 continuation. An advertisement is only a candidate; callers should prove it with