diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index aeeb305..3e0034f 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -18,6 +18,7 @@ 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. """ +import errno import os import socket import threading @@ -72,6 +73,21 @@ _BLOCK_ACK_TIMEOUT = 4.0 # once the ceiling is measured empirically. _DEFAULT_RATE_LIMIT_RPS = 5.0 +# 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 +# usually works. UDP delivery was never guaranteed, so treat them as +# advisory and keep reading. Unconnected sockets never see any of this, +# which is why the reader survived them before the connected-socket change +# in d677c72 (v0.1.3). +_ADVISORY_ERRNOS = frozenset( + value for value in ( + getattr(errno, name, None) + for name in ('ECONNREFUSED', 'EHOSTUNREACH', 'ENETUNREACH', + 'EHOSTDOWN', 'ENETDOWN') + ) if value is not None +) + class DtlsCoapSession: """Single sustained DTLS-CoAP session. @@ -168,6 +184,10 @@ class DtlsCoapSession: self._stop = threading.Event() self._reader_thread = None + # Set while the reader owns the socket. Cleared when it exits for + # any reason, so callers fail fast through _check_live() instead of + # waiting out a request timeout against a session nobody is reading. + self._reader_running = threading.Event() self._last_send_ts = 0.0 def pace(self) -> None: @@ -253,11 +273,27 @@ class DtlsCoapSession: """Spawn the reader thread. Must be called after connect().""" if self.sock is None: raise RuntimeError("connect() before start_reader()") + self._reader_running.set() t = threading.Thread(target=self._reader_loop, daemon=True, name='dtls-reader') t.start() self._reader_thread = t + def _check_live(self): + """Raise if the session cannot carry a request. A dead reader is + as fatal as a closed connection: the socket may still accept + sends, but no response will ever be dispatched, so waiting out the + request timeout only delays the inevitable SessionClosedError. + + Callers that never start a reader (config-flow style) keep the old + behaviour — only the conn check applies while _reader_thread is + None.""" + if self.conn is None: + raise SessionClosedError() + if self._reader_thread is not None and \ + not self._reader_running.is_set(): + raise SessionClosedError() + def join(self): """Block until the reader thread exits (i.e. socket dies).""" if self._reader_thread is not None: @@ -374,7 +410,23 @@ class DtlsCoapSession: d = sock.recv(65535) except socket.timeout: continue - except (OSError, ValueError): + except OSError as e: + if self._stop.is_set(): + return # close() got here first + if e.errno in _ADVISORY_ERRNOS: + logger.debug("reader: advisory %s from %s, continuing", + errno.errorcode.get(e.errno, e.errno), + self.host) + continue + logger.warning("reader exiting: socket error %s from %s", + errno.errorcode.get(e.errno, e.errno), + self.host) + return + except ValueError: + # recv on a socket closed underneath the reader. + if not self._stop.is_set(): + logger.warning("reader exiting: socket closed " + "underneath it") return if not d: continue @@ -419,6 +471,8 @@ class DtlsCoapSession: if exit_reader: return finally: + # 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()): container.setdefault('err', SessionClosedError()) @@ -490,8 +544,7 @@ class DtlsCoapSession: response — Samsung's server keys per-transfer state on the token, and dropping a fresh token on block 1+ silently drops the request.""" - if self.conn is None: - raise SessionClosedError() + self._check_live() tok = self._next_tok() blob = b'' num = 0 @@ -566,8 +619,7 @@ class DtlsCoapSession: def post(self, path_segs, body_cbor, timeout=8.0): """Single-frame POST with a CBOR-encoded body. Returns (code, payload_bytes). body_cbor must already be encoded.""" - if self.conn is None: - raise SessionClosedError() + self._check_live() tok = self._next_tok() mid = self._next_mid() opts = [(URI_PATH, s.encode()) for s in path_segs] @@ -600,8 +652,7 @@ class DtlsCoapSession: Real half-open-session detection lives in PollScheduler's `last_success_ts`, surfaced through KeepaliveTask's `liveness_fn`.""" - if self.conn is None: - raise SessionClosedError() + self._check_live() mid = self._next_mid() self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', [])) return mid @@ -618,8 +669,7 @@ class DtlsCoapSession: tokens via subscribe. Brief race window where a notify on the old token gets dropped as 'stale' — acceptable for a 6h-scale safety net.""" - if self.conn is None: - raise SessionClosedError() + self._check_live() for tok, href in list(self._observe_tokens.items()): segs = [s for s in href.split('/') if s] try: @@ -642,8 +692,7 @@ class DtlsCoapSession: Returns the token used (in case the caller wants to deregister later).""" - if self.conn is None: - raise SessionClosedError() + self._check_live() tok = self._next_observe_tok() href = '/' + '/'.join(path_segs) # Register the token BEFORE sending — otherwise the device diff --git a/tests/test_dtls_session_reader_death.py b/tests/test_dtls_session_reader_death.py new file mode 100644 index 0000000..3a5dfc3 --- /dev/null +++ b/tests/test_dtls_session_reader_death.py @@ -0,0 +1,173 @@ +"""Reader-thread death visibility (QuiteYellow/SmartThings-Local#37). + +A connected UDP socket surfaces ICMP errors on recv; before this the +reader exited silently on the first one and every later request waited +out its full timeout against a session nobody was reading. These tests +pin the three behaviours that fixed it: advisory ICMP errnos keep the +reader alive, a real socket error exits with a WARNING and clears +_reader_running, and callers then fail fast with SessionClosedError. +""" +import errno +import logging +import socket +import threading +import time + +import pytest +from OpenSSL import SSL + +from smartthings_local.errors import SessionClosedError +from smartthings_local.protocol.dtls_session import DtlsCoapSession + +_LOGGER_NAME = "smartthings_local.protocol.dtls_session" + + +class _NullAuth: + """Structural AuthenticationProvider — never configured, we skip connect().""" + + def configure_context(self, _context): + return None + + +class _FakeConn: + """Minimal stand-in for SSL.Connection: each datagram written to the + BIO surfaces as one decrypted packet on the next recv(), then + WantReadError like a drained DTLS record buffer.""" + + def __init__(self): + self._decrypted = [] + + def bio_write(self, datagram): + self._decrypted.append(datagram) + + def recv(self, _n): + if self._decrypted: + return self._decrypted.pop(0) + raise SSL.WantReadError() + + def bio_read(self, _n): + return b"" + + def send(self, _datagram): + return None + + def shutdown(self): + return None + + +class _FakeSock: + """Scripted UDP socket. Each step is bytes to return, an exception to + raise, or a callable to run (then a timeout, so the loop re-checks + _stop). An exhausted script blocks like a real recv timeout; once + close()d it raises EBADF the way a closed fd does.""" + + def __init__(self, steps=()): + self._steps = list(steps) + self.closed = False + self.timeout = None + + def settimeout(self, value): + self.timeout = value + + def recv(self, _n): + if self.closed: + raise OSError(errno.EBADF, "bad file descriptor") + if self._steps: + step = self._steps.pop(0) + if callable(step): + step() + raise socket.timeout() + if isinstance(step, BaseException): + raise step + return step + time.sleep(0.01) + raise socket.timeout() + + def send(self, data): + return len(data) + + def close(self): + self.closed = True + + +def _make_session(): + sess = DtlsCoapSession("host", 1234, auth=_NullAuth()) + sess.conn = _FakeConn() + return sess + + +def _run_reader(sess, steps, timeout=2.0): + sess.sock = _FakeSock(steps) + sess.start_reader() + sess._reader_thread.join(timeout) + assert not sess._reader_thread.is_alive(), "reader thread did not exit" + + +def test_advisory_icmp_error_does_not_kill_reader(caplog): + sess = _make_session() + dispatched = [] + sess._dispatch_coap = dispatched.append + + with caplog.at_level(logging.DEBUG, logger=_LOGGER_NAME): + _run_reader(sess, [ + OSError(errno.ECONNREFUSED, "connection refused"), + b"\x60\x00\x00\x00", # survives, gets dispatched + lambda: sess._stop.set(), # end the loop cleanly + ]) + + assert dispatched == [b"\x60\x00\x00\x00"] + assert not sess._reader_running.is_set() + assert any(r.levelno == logging.DEBUG and "advisory" in r.getMessage() + for r in caplog.records) + # An advisory errno is not a real exit — no WARNING. + assert not any(r.levelno >= logging.WARNING for r in caplog.records) + + +def test_fatal_socket_error_exits_with_warning(caplog): + sess = _make_session() + + with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME): + _run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")]) + + assert not sess._reader_running.is_set() + warnings = [r for r in caplog.records if r.levelno == logging.WARNING] + assert len(warnings) == 1 + assert "reader exiting" in warnings[0].getMessage() + + +def test_request_fails_fast_after_reader_death(): + sess = _make_session() + _run_reader(sess, [OSError(errno.EBADF, "bad file descriptor")]) + assert not sess._reader_running.is_set() + + start = time.monotonic() + with pytest.raises(SessionClosedError): + sess.get(["oic", "d"], timeout=10.0) + elapsed = time.monotonic() - start + # The whole point: no waiting out the request timeout. + assert elapsed < 1.0, f"get() waited {elapsed:.2f}s instead of failing fast" + + +def test_close_does_not_log_warning_on_teardown(caplog): + sess = _make_session() + sess.sock = _FakeSock() # empty script: blocks on recv + sess.start_reader() + time.sleep(0.05) # let the reader reach recv + + with caplog.at_level(logging.WARNING, logger=_LOGGER_NAME): + sess.close() + sess._reader_thread.join(2.0) + + assert not sess._reader_thread.is_alive() + assert not sess._reader_running.is_set() + assert not any(r.levelno >= logging.WARNING for r in caplog.records) + + +def test_check_live_without_reader_matches_old_conn_guard(): + sess = _make_session() # conn set, reader never started + assert sess._reader_thread is None + sess._check_live() # must not raise — config-flow behaviour + + sess.conn = None + with pytest.raises(SessionClosedError): + sess._check_live()