diff --git a/smartthings_local/protocol/coap.py b/smartthings_local/protocol/coap.py index 5a7b7ba..741829f 100644 --- a/smartthings_local/protocol/coap.py +++ b/smartthings_local/protocol/coap.py @@ -13,6 +13,7 @@ URI_PATH = 11 URI_QUERY = 15 OBSERVE = 6 +ETAG = 4 CONTENT_FORMAT = 12 ACCEPT = 17 BLOCK2 = 23 @@ -121,6 +122,14 @@ def block_value(num, more, szx): return struct.pack('>I', v)[1:] +def block_fields(value): + """Decode a CoAP Block-N option value. Inverse of block_value(). + Returns (num, more, szx). An empty value means block 0, no more, + SZX=0 — RFC 7959 §2.2 allows a zero-length option to elide it.""" + v = int.from_bytes(value, 'big') + return v >> 4, (v >> 3) & 1, v & 0x07 + + 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 7165ed0..4b34487 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -12,11 +12,18 @@ or interleaved one-shot / OBSERVE traffic mis-attributes. * Multi-block GET requires the SAME CoAP token across every block of the response ("token-stable Block2"). Fresh-token-per-block - is silently dropped by the server. + is silently dropped by the server, and a transfer that opens at + NUM>0 under a token the server has not seen gets no reply at all. Reader thread owns the UDP socket. Callers issue get()/post() and block on a per-token Event the reader signals. OBSERVE notifications are delivered via the on_notification callback. + +A notification carries only the first block of a large representation +(RFC 7959 §2.6). Because a continuation cannot borrow the observation's +token (§3.4) and this server will not continue a transfer it did not +start, such a notification is withheld and the resource is re-read from +block 0 on a fresh one-shot token by a worker thread. See #39. """ import errno import math @@ -35,11 +42,12 @@ SessionTimeoutError, ) from .coap import ( - URI_PATH, URI_QUERY, OBSERVE, CONTENT_FORMAT, ACCEPT, BLOCK2, SIZE2, + 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, fmt_code, + encode_options, parse_coap, build_coap, block_value, block_fields, + fmt_code, split_dtls as _split_dtls, ) from .auth import ( @@ -72,6 +80,11 @@ _BLOCK_MAX_ATTEMPTS = 3 _BLOCK_ACK_TIMEOUT = 4.0 +# How often a block wait re-checks that the reader is still alive. Short +# enough that a mid-transfer reader death fails fast instead of burning +# the whole per-block timeout, long enough to stay off the CPU. +_BLOCK_LIVENESS_POLL_S = 0.25 + # Inter-request pacing: minimum seconds between CoAP CON sends on one session. # Samsung's RT-OCF stacks drop requests when hit faster than their firmware # ceiling (dryer ~14 req/s, oven ~8 req/s, dishwasher unknown). 5 req/s @@ -79,6 +92,22 @@ # once the ceiling is measured empirically. _DEFAULT_RATE_LIMIT_RPS = 5.0 +# Maximum hrefs held for OBSERVE refetch at once. A notification storm +# on more resources than this is already past what the 5/s ceiling can +# drain, so the excess is dropped rather than queued indefinitely. +_MAX_PENDING_REFETCH = 16 + +# Timeout for one notification refetch. Generous relative to a poll: +# the resource is known large (that is why it blocked) and the worker +# is serialized, so a slow one delays only later refetches. +_REFETCH_TIMEOUT_S = 15.0 + + +class _EtagChanged(Exception): + """Internal: the server's ETag changed partway through a Block2 + transfer, so the blocks in hand are from two different versions.""" + + # ICMP errors a connected UDP socket surfaces on the next recv. On these # appliances they show up while the device is rebooting, while it holds an # orphaned association, or across a router blip, and the next datagram @@ -241,6 +270,11 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, self.endpoint = None self._send_lock = threading.Lock() + # Guards the MID/token counters and _pending. The refetch worker + # makes the session its own second concurrent get() caller, so + # two threads can mint tokens at once; without this they can + # collide and one transfer silently absorbs the other's blocks. + self._state_lock = threading.Lock() # Randomize MID and token counter starting points so reconnects # don't reuse identifiers from previous sessions — Samsung's # RT-OCF appears to remember observer state across DTLS @@ -257,6 +291,14 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, # token (bytes) → href (str) self._observe_tokens = {} + # OBSERVE refetch queue: href → sequence number of the newest + # notification that asked for it. Drained by a worker thread + # because _dispatch_coap cannot block (see _queue_refetch). + self._refetch_cond = threading.Condition() + self._refetch_pending = {} + self._refetch_seq = 0 + self._refetch_thread = None + self._stop = threading.Event() self._reader_thread = None # Set while the reader owns the socket. Cleared when it exits for @@ -407,6 +449,8 @@ def join(self): """Block until the reader thread exits (i.e. socket dies).""" if self._reader_thread is not None: self._reader_thread.join() + if self._refetch_thread is not None: + self._refetch_thread.join() def _send_observe_dereg(self, tok, path_segs): """Send a single OBSERVE deregister GET (Observe option = 1) @@ -438,6 +482,11 @@ def close(self): time.sleep(0.1) self._stop.set() + # Wake the refetch worker so it sees _stop instead of sitting on + # its condition for up to a second after the socket is gone. + with self._refetch_cond: + self._refetch_pending.clear() + self._refetch_cond.notify_all() if self.conn is not None: try: self.conn.shutdown() @@ -448,10 +497,12 @@ def close(self): self.sock.close() except Exception: pass - for tok, (ev, container) in list(self._pending.items()): + with self._state_lock: + pending = list(self._pending.items()) + self._pending.clear() + for tok, (ev, container) in pending: container.setdefault('err', SessionClosedError()) ev.set() - self._pending.clear() self._observe_tokens.clear() self.sock = None self.conn = None @@ -461,27 +512,30 @@ def close(self): # ---- send / receive plumbing ------------------------------------- def _next_mid(self): - self._mid = (self._mid + 1) & 0xFFFF - return self._mid + with self._state_lock: + self._mid = (self._mid + 1) & 0xFFFF + return self._mid def _next_tok(self): - self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF - # 4-byte tokens — fits within tkl=8 cap with headroom and - # avoids collisions across long-running OBSERVE subscriptions. - return self._tok_counter.to_bytes(4, 'big') + with self._state_lock: + self._tok_counter = (self._tok_counter + 1) & 0xFFFFFFFF + # 4-byte tokens — fits within tkl=8 cap with headroom and + # avoids collisions across long-running OBSERVE subscriptions. + return self._tok_counter.to_bytes(4, 'big') def _next_observe_tok(self): - # Single-byte tokens for OBSERVE registrations. Samsung - # RT-OCF accepts these but silently drops TKL=4 OBSERVE - # registrations. Counter is randomly seeded per session so - # reconnects don't collide with stale observer state Samsung - # may still be holding from the previous run. - self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF - # Avoid 0x00 — some CoAP stacks treat an all-zero token as - # equivalent to "no token" / empty (TKL=0). - if self._observe_tok_counter == 0: - self._observe_tok_counter = 1 - return bytes([self._observe_tok_counter]) + with self._state_lock: + # Single-byte tokens for OBSERVE registrations. Samsung + # RT-OCF accepts these but silently drops TKL=4 OBSERVE + # registrations. Counter is randomly seeded per session so + # reconnects don't collide with stale observer state Samsung + # may still be holding from the previous run. + self._observe_tok_counter = (self._observe_tok_counter + 1) & 0xFF + # Avoid 0x00 — some CoAP stacks treat an all-zero token as + # equivalent to "no token" / empty (TKL=0). + if self._observe_tok_counter == 0: + self._observe_tok_counter = 1 + return bytes([self._observe_tok_counter]) def _send_dgram(self, datagram): """Send a CoAP datagram. Holds the send lock for the @@ -583,9 +637,15 @@ def _reader_loop(self): # Reader no longer owns the socket — callers must fail fast. self._reader_running.clear() # Make sure pending waiters don't hang if the reader dies. - for tok, (ev, container) in list(self._pending.items()): + with self._state_lock: + pending = list(self._pending.items()) + for tok, (ev, container) in pending: container.setdefault('err', SessionClosedError()) ev.set() + # Nothing will answer a refetch now either. + with self._refetch_cond: + self._refetch_pending.clear() + self._refetch_cond.notify_all() def _dispatch_coap(self, datagram): try: @@ -615,7 +675,8 @@ def _dispatch_coap(self, datagram): return # Pending one-shot? Resolve and return. - rec = self._pending.get(tok) + with self._state_lock: + rec = self._pending.get(tok) if rec is not None: ev, container = rec container['code'] = code @@ -633,6 +694,16 @@ def _dispatch_coap(self, datagram): logger.warning("observe %s: non-2.05 %s", href, fmt_code(code)) return + # RFC 7959 §2.6: a notification carries only the first block + # of the representation. Handing the callback a partial CBOR + # buffer is what #39 was about, so anything with M=1 (or a + # block past the first) goes to the refetch worker instead. + b2 = [v for n, v in ropts if n == BLOCK2] + if b2: + num, more, _ = block_fields(b2[0]) + if more or num: + self._queue_refetch(href) + return cb = self.on_notification if cb is not None: try: @@ -644,6 +715,110 @@ def _dispatch_coap(self, datagram): # Stale token (post-reconnect or unknown) — drop quietly. + # ---- OBSERVE refetch --------------------------------------------- + + @staticmethod + def _log_refetch(msg, *args): + """Refetch outcomes are debug-level in normal operation, which is + below the bridge's INFO default, so a healthy session stays quiet. + DEBUG_BRIDGE=1 promotes them to INFO for hardware validation: + that shows which token the re-read used and whether it completed, + without also turning on every per-block retransmit line.""" + (logger.info if DEBUG_BRIDGE else logger.debug)(msg, *args) + + def _queue_refetch(self, href): + """Queue a blockwise notification for re-reading. + + Called from the reader thread, so it must not block: _dispatch_coap + runs there and _blockwise_get waits on an Event only that same + thread can set, which would deadlock the session outright. Latest + wins per href — a burst of notifications on one resource collapses + into a single re-read of its final state.""" + with self._refetch_cond: + if (href not in self._refetch_pending + and len(self._refetch_pending) >= _MAX_PENDING_REFETCH): + self._log_refetch( + "refetch %s dropped: queue full (%d pending)", + href, len(self._refetch_pending)) + return + self._refetch_seq += 1 + self._refetch_pending[href] = self._refetch_seq + self._refetch_cond.notify() + self._start_refetch_worker() + + def _start_refetch_worker(self): + """Start the refetch worker on first use. Sessions that never see + a blockwise notification never grow the thread.""" + if self._refetch_thread is not None or self._stop.is_set(): + return + with self._state_lock: + if self._refetch_thread is not None or self._stop.is_set(): + return + self._refetch_thread = threading.Thread( + target=self._refetch_loop, daemon=True, + name=f'stl-refetch-{self.host}') + self._refetch_thread.start() + + def _refetch_loop(self): + """Re-read blockwise-notified resources, one at a time. + + Serialized on purpose. Each re-read is a multi-block transfer and + _blockwise_get paces between blocks, so running one at a time is + what keeps a notification storm under the firmware's request + ceiling.""" + while self._refetch_alive(): + with self._refetch_cond: + while not self._refetch_pending and self._refetch_alive(): + self._refetch_cond.wait(1.0) + if not self._refetch_pending: + return + href, seq = next(iter(self._refetch_pending.items())) + del self._refetch_pending[href] + self._refetch_one(href, seq) + + def _refetch_alive(self): + """False once the session is closing or the reader has died. A + refetch needs the reader to resolve its token, so outliving it + would leave join() waiting on a thread with nothing to do.""" + if self._stop.is_set(): + return False + return self._reader_thread is None or self._reader_running.is_set() + + def _refetch_one(self, href, seq): + """Re-read one href from block 0 and deliver it if it is still + the freshest thing we know about that resource.""" + self.pace() + segs = [s for s in href.split('/') if s] + try: + code, payload, blocks, tok = self._blockwise_get( + segs, (), _REFETCH_TIMEOUT_S) + except Exception as e: + # Device silent, session gone, ETag never settled, block cap + # hit. Whatever the reason, dropping the notification is the + # contract: the poll tiers still carry freshness, and handing + # over the first block is the bug this replaced. + self._log_refetch("refetch %s failed: %s", href, e) + return + if code != 0x45: + self._log_refetch("refetch %s returned %s", href, fmt_code(code)) + return + with self._refetch_cond: + # A newer notification landed while we were reading. That one + # has its own refetch queued, so this result is already stale. + if self._refetch_pending.get(href, 0) > seq: + self._log_refetch( + "refetch %s tok=%s blocks=%d bytes=%d superseded", + href, tok.hex(), blocks, len(payload)) + return + self._log_refetch("refetch %s tok=%s blocks=%d bytes=%d ok", + href, tok.hex(), blocks, len(payload)) + cb = self.on_notification + if cb is not None: + try: + cb(href, payload) + except Exception as e: + logger.warning("notification callback %s: %s", href, e) + # ---- request primitives ------------------------------------------ def get(self, path_segs, query=(), timeout=10.0): @@ -654,76 +829,183 @@ def get(self, path_segs, query=(), timeout=10.0): token, and dropping a fresh token on block 1+ silently drops the request.""" self._check_live() + code, blob, _blocks, _tok = self._blockwise_get( + path_segs, query, timeout) + return code, blob + + def _blockwise_get(self, path_segs, query=(), timeout=10.0): + """Shared token-stable Block2 reassembly (RFC 7959 §2.4). + + Returns (code, payload, block_count, token). The last two are + diagnostics for the refetch log; get() drops them. + + Mints one fresh 4-byte token and holds it across every block of + the transfer. Also the notification-refetch primitive: RFC 7959 + §3.4 forbids continuing a blockwise notification on the + observation's token, and Samsung's RT-OCF drops a transfer that + opens at NUM>0 under a token it has not seen, so a truncated + notification is recovered by re-reading from block 0 through + this same path rather than by a §2.6 continuation. + + Restarts once if the server's ETag changes mid-transfer, then + gives up: RFC 7959 §2.4 requires the client to compare ETags + when the server supplies them. None of the tested appliances + emit option 4, so on those this is inert.""" + try: + return self._blockwise_get_once(path_segs, query, timeout) + except _EtagChanged: + logger.debug("GET %s /%s: ETag changed mid-transfer, restarting", + self.host, '/'.join(path_segs)) + try: + return self._blockwise_get_once(path_segs, query, timeout) + except _EtagChanged: + logger.debug( + "GET %s /%s: representation kept changing mid-transfer", + self.host, '/'.join(path_segs)) + raise BlockwiseError() from None + + def _blockwise_get_once(self, path_segs, query, timeout): + """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 - last_opts = [] + etag = None deadline = time.time() + timeout szx = BLOCK_SZX # server may negotiate down; track per-transfer while True: if num > 0: self.pace() - container = {} - for attempt in range(_BLOCK_MAX_ATTEMPTS): - ev = threading.Event() - container = {} - self._pending[tok] = (ev, container) - try: - mid = self._next_mid() - 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))) - self._send_dgram( - build_coap(TYPE_CON, METHOD_GET, mid, tok, opts)) - per_wait = min(_BLOCK_ACK_TIMEOUT, - max(0.1, deadline - time.time())) - if ev.wait(per_wait): - break # got a response - remaining = deadline - time.time() - if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1: - logger.debug( - "GET %s /%s block %d: timed out after %d attempt(s)", - self.host, '/'.join(path_segs), num, attempt + 1, - ) - raise SessionTimeoutError() - logger.debug( - "GET %s /%s block %d: attempt %d/%d timeout, retrying", - self.host, '/'.join(path_segs), num, - attempt + 1, _BLOCK_MAX_ATTEMPTS, - ) - finally: - self._pending.pop(tok, None) + 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 - last_opts = ropts - blob += payload # 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 + return code, blob, blocks, tok + + # 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() + + blob += payload b2 = [v for n, v in ropts if n == BLOCK2] - more = 0 - if b2: - bv = int.from_bytes(b2[0], 'big') - more = (bv >> 3) & 1 - server_szx = bv & 0x07 - if server_szx != szx: - szx = server_szx + if not b2: + break + _, more, server_szx = block_fields(b2[0]) if not more: break - num += 1 + 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 + return last_code, blob, blocks, tok + + 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. + + A response whose Block2 NUM is not the one we asked for is a + retransmit of an earlier block, not the next one. Concatenating + it would corrupt the buffer, so keep waiting on the same + attempt budget instead.""" + for attempt in range(_BLOCK_MAX_ATTEMPTS): + ev = threading.Event() + container = {} + with self._state_lock: + self._pending[tok] = (ev, container) + try: + mid = self._next_mid() + 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))) + self._send_dgram( + build_coap(TYPE_CON, METHOD_GET, mid, tok, opts)) + while True: + per_wait = min(_BLOCK_ACK_TIMEOUT, + max(0.1, deadline - time.time())) + if not self._wait_for_block(ev, per_wait): + break # attempt timed out + if 'err' in container or self._block_num_matches( + container, num): + return container + logger.debug( + "GET %s /%s block %d: stale block, still waiting", + self.host, '/'.join(path_segs), num) + ev.clear() + container.clear() + if deadline - time.time() <= 0: + break + remaining = deadline - time.time() + if remaining <= 0 or attempt == _BLOCK_MAX_ATTEMPTS - 1: + logger.debug( + "GET %s /%s block %d: timed out after %d attempt(s)", + self.host, '/'.join(path_segs), num, attempt + 1, + ) + raise SessionTimeoutError() + logger.debug( + "GET %s /%s block %d: attempt %d/%d timeout, retrying", + self.host, '/'.join(path_segs), num, + attempt + 1, _BLOCK_MAX_ATTEMPTS, + ) + finally: + with self._state_lock: + self._pending.pop(tok, None) + 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 + while True: + slice_s = min(_BLOCK_LIVENESS_POLL_S, deadline - time.time()) + if slice_s <= 0: + return False + if ev.wait(slice_s): + return True + 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: + return num == 0 + return block_fields(b2[0])[0] == num def post(self, path_segs, body_cbor, timeout=8.0): """Single-frame POST with a CBOR-encoded body. Returns @@ -738,7 +1020,8 @@ def post(self, path_segs, body_cbor, timeout=8.0): body_cbor) ev = threading.Event() container = {} - self._pending[tok] = (ev, container) + with self._state_lock: + self._pending[tok] = (ev, container) try: self._send_dgram(datagram) if not ev.wait(timeout): @@ -747,7 +1030,8 @@ def post(self, path_segs, body_cbor, timeout=8.0): raise container['err'] return container['code'], container['payload'] finally: - self._pending.pop(tok, None) + with self._state_lock: + self._pending.pop(tok, None) def ping(self): """RFC 7252 §4.4 CoAP Ping — empty CON, no token, no payload. diff --git a/tests/test_coap_wire.py b/tests/test_coap_wire.py index b1e4195..baca36f 100644 --- a/tests/test_coap_wire.py +++ b/tests/test_coap_wire.py @@ -2,7 +2,8 @@ from smartthings_local.errors import MalformedMessageError from smartthings_local.protocol.coap import ( - build_coap, parse_coap, encode_options, block_value, fmt_code, + build_coap, parse_coap, encode_options, block_value, block_fields, + fmt_code, TYPE_CON, METHOD_GET, URI_PATH, ACCEPT, CF_CBOR, BLOCK2, ) @@ -43,6 +44,23 @@ def test_block_value_promotes_to_two_bytes_when_num_is_large(): assert len(v) == 2 +@pytest.mark.parametrize('num, more, szx', [ + (0, 0, 0), + (0, 1, 6), + (2, 1, 6), + (1, 0, 4), + (0xFFF, 0, 0), + (0xFFFFF, 1, 7), +]) +def test_block_fields_inverts_block_value(num, more, szx): + assert block_fields(block_value(num, more, szx)) == (num, more, szx) + + +def test_block_fields_treats_empty_value_as_block_zero(): + # RFC 7959 §2.2: a zero-length Block option means num=0, m=0, szx=0. + assert block_fields(b'') == (0, 0, 0) + + def test_fmt_code_formats_class_dot_detail(): assert fmt_code(0x45) == '2.05' assert fmt_code(0x84) == '4.04' diff --git a/tests/test_observe_block2.py b/tests/test_observe_block2.py new file mode 100644 index 0000000..cb8b4db --- /dev/null +++ b/tests/test_observe_block2.py @@ -0,0 +1,508 @@ +"""Blockwise OBSERVE notifications (QuiteYellow/SmartThings-Local#39). + +A notification carries only the first block of the representation +(RFC 7959 §2.6). Before this, _dispatch_coap handed that first block +straight to on_notification and the consumer decoded a truncated CBOR +buffer. These tests pin the replacement: a truncated notification is +withheld, the resource is re-read from block 0 under a fresh one-shot +token, and only the reassembled representation reaches the callback. + +The re-read starts at block 0 rather than continuing at NUM=1 for two +reasons, both recorded on #39: RFC 7959 §3.4 forbids continuing on the +observation's token, and Samsung's RT-OCF drops a transfer that opens +at NUM>0 under a token it has not seen. +""" +import logging +import socket +import threading +import time + +import pytest +from OpenSSL import SSL + +from smartthings_local.errors import BlockwiseError +from smartthings_local.protocol import dtls_session +from smartthings_local.protocol.coap import ( + BLOCK2, ETAG, METHOD_GET, OBSERVE, TYPE_ACK, TYPE_NON, + block_fields, block_value, build_coap, parse_coap, +) +from smartthings_local.protocol.dtls_session import DtlsCoapSession + +SZX = 6 # 1024-byte blocks, the only size these appliances honour +_LOGGER_NAME = "smartthings_local.protocol.dtls_session" + + +class _NullAuth: + """Structural AuthenticationProvider — never configured, we skip connect().""" + + def configure_context(self, _context): + return None + + +class _LoopbackConn: + """SSL.Connection stand-in that answers requests from a script. + + `responder` is called with each parsed request and returns a list of + CoAP datagrams to hand back (possibly empty, to model a silent + device). Responses surface on recv() the way decrypted records do. + """ + + def __init__(self, responder): + self._responder = responder + self._inbox = [] + self._lock = threading.Lock() + self.sent = [] + + # -- client -> device + def send(self, datagram): + self.sent.append(parse_coap(datagram)) + for reply in self._responder(parse_coap(datagram)): + with self._lock: + self._inbox.append(reply) + return len(datagram) + + def bio_read(self, _n): + return b"" + + # -- device -> client + def inject(self, datagram): + """Push a device-initiated frame (an OBSERVE notification).""" + with self._lock: + self._inbox.append(datagram) + + def bio_write(self, _datagram): + return None + + def recv(self, _n): + with self._lock: + if self._inbox: + return self._inbox.pop(0) + raise SSL.WantReadError() + + def shutdown(self): + return None + + def pending(self): + with self._lock: + return len(self._inbox) + + +class _PumpSock: + """UDP socket stand-in. recv() returns a dummy datagram whenever the + connection has something decrypted waiting, so the reader loop keeps + pumping; otherwise it times out like a real socket.""" + + def __init__(self, conn): + self._conn = conn + self.closed = False + + def settimeout(self, _value): + return None + + def recv(self, _n): + for _ in range(20): + if self.closed: + raise OSError("closed") + if self._conn.pending(): + return b"\x00" + time.sleep(0.005) + raise socket.timeout() + + def send(self, data): + return len(data) + + def close(self): + self.closed = True + + +def _make_session(responder, **kwargs): + calls = [] + sess = DtlsCoapSession( + "host", 1234, auth=_NullAuth(), + on_notification=lambda href, payload: calls.append((href, payload)), + **kwargs) + sess.conn = _LoopbackConn(responder) + sess.sock = _PumpSock(sess.conn) + sess.start_reader() + return sess, calls + + +def _notification(tok, payload, *, block2=None, mtype=TYPE_NON, obs=1): + opts = [(OBSERVE, bytes([obs]))] + if block2 is not None: + opts.append((BLOCK2, block2)) + return build_coap(mtype, 0x45, 0x1234, tok, opts, payload) + + +def _content(tok, mid, payload, *, block2=None, etag=None): + opts = [] + if etag is not None: + opts.append((ETAG, etag)) + if block2 is not None: + opts.append((BLOCK2, block2)) + return build_coap(TYPE_ACK, 0x45, mid, tok, opts, payload) + + +def _requested_block(request): + """(num, szx) the request asked for, or (0, None) with no Block2.""" + _, _, _, _, opts, _ = request + b2 = [v for n, v in opts if n == BLOCK2] + if not b2: + return 0, None + num, _, szx = block_fields(b2[0]) + return num, szx + + +def _wait_for(predicate, timeout=3.0): + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.01) + return False + + +def _close(sess): + sess.close() + sess.join() + + +# -------------------------------------------------------------------- +# The no-regression case + + +def test_single_block_notification_is_delivered_inline(): + sess, calls = _make_session(lambda request: []) + try: + tok = sess.subscribe(["oven", "vs", "0"]) + sess.conn.inject(_notification(tok, b"\xa1\x01\x02")) + + assert _wait_for(lambda: calls) + assert calls == [("/oven/vs/0", b"\xa1\x01\x02")] + # The subscribe GET is the only thing we sent — no refetch. + assert len(sess.conn.sent) == 1 + finally: + _close(sess) + + +def test_complete_block_zero_notification_is_delivered_inline(): + """Block2 present but M=0 and NUM=0 means the whole representation + fit in one block. Nothing to fetch back.""" + sess, calls = _make_session(lambda request: []) + try: + tok = sess.subscribe(["oven", "vs", "0"]) + sess.conn.inject( + _notification(tok, b"\xa1\x01\x02", + block2=block_value(0, 0, SZX))) + + assert _wait_for(lambda: calls) + assert calls == [("/oven/vs/0", b"\xa1\x01\x02")] + assert len(sess.conn.sent) == 1 + finally: + _close(sess) + + +# -------------------------------------------------------------------- +# The fix + + +def test_truncated_notification_is_refetched_and_reassembled(): + blocks = [b"A" * 1024, b"B" * 40] + + def responder(request): + _mtype, code, mid, tok, opts, _ = request + if code != METHOD_GET or any(n == OBSERVE for n, _ in opts): + return [] # the subscribe registration itself + num, _ = _requested_block(request) + more = 1 if num + 1 < len(blocks) else 0 + return [_content(tok, mid, blocks[num], + block2=block_value(num, more, SZX))] + + sess, calls = _make_session(responder) + try: + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, blocks[0], + block2=block_value(0, 1, SZX))) + + assert _wait_for(lambda: calls), "callback never fired" + assert calls == [("/mode/vs/0", b"".join(blocks))] + finally: + _close(sess) + + +def test_refetch_uses_a_fresh_one_shot_token_not_the_observe_token(): + """RFC 7959 §3.4: the requests for additional blocks cannot use the + token of the observation relationship.""" + blocks = [b"A" * 1024, b"B" * 40] + + def responder(request): + _mtype, _code, mid, tok, opts, _ = request + if any(n == OBSERVE for n, _ in opts): + return [] + num, _ = _requested_block(request) + more = 1 if num + 1 < len(blocks) else 0 + return [_content(tok, mid, blocks[num], + block2=block_value(num, more, SZX))] + + sess, calls = _make_session(responder) + try: + observe_tok = sess.subscribe(["mode", "vs", "0"]) + assert len(observe_tok) == 1, "OBSERVE registrations use 1-byte tokens" + sess.conn.inject( + _notification(observe_tok, blocks[0], + block2=block_value(0, 1, SZX))) + assert _wait_for(lambda: calls) + + refetch = [r for r in sess.conn.sent + if not any(n == OBSERVE for n, _ in r[4])] + assert refetch, "no refetch request was sent" + tokens = {r[3] for r in refetch} + assert observe_tok not in tokens + assert all(len(t) == 4 for t in tokens), "one-shot tokens are 4-byte" + assert len(tokens) == 1, "the transfer must hold one token throughout" + + # And the transfer restarts at block 0 rather than continuing at 1. + assert _requested_block(refetch[0])[0] == 0 + assert [_requested_block(r)[0] for r in refetch] == [0, 1] + # No Observe option on a continuation request. + assert not any(n == OBSERVE for r in refetch for n, _ in r[4]) + finally: + _close(sess) + + +def test_silent_device_drops_the_notification_without_delivering_a_partial(): + sess, calls = _make_session(lambda request: []) + try: + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, b"A" * 1024, + block2=block_value(0, 1, SZX))) + # Give the worker a chance to try and fail. _BLOCK_ACK_TIMEOUT is + # 4s per attempt, so we only need to see that nothing partial got + # through in the meantime. + assert not _wait_for(lambda: calls, timeout=0.6) + assert sess._reader_thread.is_alive(), "reader must survive" + finally: + _close(sess) + + +def test_non_2xx_refetch_is_dropped_rather_than_delivered(): + def responder(request): + _mtype, _code, mid, tok, opts, _ = request + if any(n == OBSERVE for n, _ in opts): + return [] + return [build_coap(TYPE_ACK, 0x84, mid, tok, [], b"")] + + sess, calls = _make_session(responder) + try: + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, b"A" * 1024, + block2=block_value(0, 1, SZX))) + assert not _wait_for(lambda: calls, timeout=0.6) + finally: + _close(sess) + + +def test_notification_burst_collapses_to_one_refetch_per_resource(): + blocks = [b"A" * 1024, b"B" * 40] + gate = threading.Event() + + def responder(request): + _mtype, _code, mid, tok, opts, _ = request + if any(n == OBSERVE for n, _ in opts): + return [] + gate.wait(2.0) # hold the first transfer open + num, _ = _requested_block(request) + more = 1 if num + 1 < len(blocks) else 0 + return [_content(tok, mid, blocks[num], + block2=block_value(num, more, SZX))] + + sess, calls = _make_session(responder) + try: + tok = sess.subscribe(["mode", "vs", "0"]) + for seq in range(5): + sess.conn.inject( + _notification(tok, blocks[0], obs=seq + 1, + block2=block_value(0, 1, SZX))) + # Five notifications, one queue entry: latest wins per href. + assert _wait_for(lambda: sess._refetch_pending or sess.conn.sent[1:]) + assert len(sess._refetch_pending) <= 1 + gate.set() + + assert _wait_for(lambda: calls) + assert _wait_for( + lambda: not sess._refetch_pending and len(calls) >= 1) + time.sleep(0.2) + # Two transfers at most: the one in flight when the burst landed, + # plus one for the final state. + starts = [r for r in sess.conn.sent + if not any(n == OBSERVE for n, _ in r[4]) + and _requested_block(r)[0] == 0] + assert len(starts) <= 2, f"{len(starts)} refetches for one burst" + assert calls[-1] == ("/mode/vs/0", b"".join(blocks)) + finally: + gate.set() + _close(sess) + + +def test_close_during_a_queued_refetch_stops_the_worker(): + sess, _calls = _make_session(lambda request: []) + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, b"A" * 1024, block2=block_value(0, 1, SZX))) + assert _wait_for(lambda: sess._refetch_thread is not None) + + sess.close() + sess.join() # hangs if the worker outlives the session + assert not sess._refetch_thread.is_alive() + + +def test_refetch_worker_exits_when_the_reader_dies(): + sess, _calls = _make_session(lambda request: []) + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, b"A" * 1024, block2=block_value(0, 1, SZX))) + assert _wait_for(lambda: sess._refetch_thread is not None) + + # Kill the reader the way a socket error does, without close(). + sess.sock.closed = True + assert _wait_for(lambda: not sess._reader_running.is_set(), timeout=5.0) + sess._refetch_thread.join(6.0) + assert not sess._refetch_thread.is_alive() + sess.close() + + +def test_debug_bridge_promotes_the_refetch_outcome_to_info(monkeypatch, caplog): + """The hardware validation for #39 reads this line to confirm which + token the re-read used, so it has to survive the bridge's INFO + default. Without DEBUG_BRIDGE it stays at debug.""" + monkeypatch.setattr(dtls_session, "DEBUG_BRIDGE", True) + blocks = [b"A" * 1024, b"B" * 40] + + def responder(request): + _mtype, _code, mid, tok, opts, _ = request + if any(n == OBSERVE for n, _ in opts): + return [] + num, _ = _requested_block(request) + more = 1 if num + 1 < len(blocks) else 0 + return [_content(tok, mid, blocks[num], + block2=block_value(num, more, SZX))] + + sess, calls = _make_session(responder) + try: + with caplog.at_level(logging.INFO, logger=_LOGGER_NAME): + tok = sess.subscribe(["mode", "vs", "0"]) + sess.conn.inject( + _notification(tok, blocks[0], block2=block_value(0, 1, SZX))) + assert _wait_for(lambda: calls) + + line = next((r.getMessage() for r in caplog.records + if r.getMessage().startswith("refetch /mode/vs/0")), None) + assert line is not None, "no refetch line at INFO" + assert "blocks=2" in line + assert f"bytes={sum(len(b) for b in blocks)}" in line + assert line.endswith("ok") + # The token in the line is the one-shot token, not the observe one. + assert f"tok={tok.hex()} " not in line + finally: + _close(sess) + + +# -------------------------------------------------------------------- +# Shared Block2 loop hardening + + +def test_stale_block_number_is_not_concatenated(): + """A retransmit of block 0 arriving while we wait for block 1 must + not be appended as if it were block 1.""" + served = [] + + def responder(request): + _mtype, _code, mid, tok, _opts, _ = request + num, _ = _requested_block(request) + served.append(num) + if num == 0: + return [_content(tok, mid, b"A" * 1024, + block2=block_value(0, 1, SZX))] + # Answer the block-1 request with a duplicate of block 0 first. + return [ + _content(tok, mid, b"A" * 1024, block2=block_value(0, 1, SZX)), + _content(tok, mid, b"B" * 40, block2=block_value(1, 0, SZX)), + ] + + sess, _calls = _make_session(responder) + try: + code, payload = sess.get(["mode", "vs", "0"], timeout=5.0) + assert code == 0x45 + assert payload == b"A" * 1024 + b"B" * 40 + finally: + _close(sess) + + +def test_etag_change_mid_transfer_restarts_then_fails(): + etags = [b"\x01", b"\x02", b"\x03", b"\x04"] + + def responder(request): + _mtype, _code, mid, tok, _opts, _ = request + num, _ = _requested_block(request) + # A different ETag on every single response: the representation + # never settles, so reassembly can never be consistent. + etag = etags.pop(0) if etags else b"\xff" + payload = b"A" * 1024 if num == 0 else b"B" * 40 + more = 1 if num == 0 else 0 + return [_content(tok, mid, payload, etag=etag, + block2=block_value(num, more, SZX))] + + sess, _calls = _make_session(responder) + try: + with pytest.raises(BlockwiseError): + sess.get(["mode", "vs", "0"], timeout=5.0) + finally: + _close(sess) + + +def test_stable_etag_across_blocks_reassembles(): + def responder(request): + _mtype, _code, mid, tok, _opts, _ = request + num, _ = _requested_block(request) + payload = b"A" * 1024 if num == 0 else b"B" * 40 + more = 1 if num == 0 else 0 + return [_content(tok, mid, payload, etag=b"\x77", + block2=block_value(num, more, SZX))] + + sess, _calls = _make_session(responder) + try: + code, payload = sess.get(["mode", "vs", "0"], timeout=5.0) + assert code == 0x45 + assert payload == b"A" * 1024 + b"B" * 40 + finally: + _close(sess) + + +def test_szx_downshift_asks_for_the_block_after_what_we_have(): + """Server answers block 0 at SZX=6 (1024B) then drops to SZX=4 + (256B). Block numbers index the new size, so the next request is + block 4, not block 1.""" + requested = [] + + def responder(request): + _mtype, _code, mid, tok, _opts, _ = request + num, szx = _requested_block(request) + requested.append((num, szx)) + if num == 0: + return [_content(tok, mid, b"A" * 1024, + block2=block_value(0, 1, 4))] + return [_content(tok, mid, b"B" * 100, + block2=block_value(num, 0, 4))] + + sess, _calls = _make_session(responder) + try: + code, payload = sess.get(["mode", "vs", "0"], timeout=5.0) + assert code == 0x45 + assert payload == b"A" * 1024 + b"B" * 100 + # 1024 bytes in hand at 256B blocks = blocks 0..3 done, ask for 4. + assert requested[1] == (4, 4) + finally: + _close(sess)