diff --git a/README.md b/README.md index 52c2f41..735733c 100644 --- a/README.md +++ b/README.md @@ -48,6 +48,31 @@ sess.subscribe(["operational", "state", "vs", "0"], # OBSERVE sess.close() ``` +`connect()` uses a 12-second monotonic DTLS handshake deadline by default. A +caller that needs a shorter bounded attempt can pass a positive finite value +without changing later reader timeouts: + +```python +sess.connect(timeout=4.0) +``` + +Connection attempts can also observe a `threading.Event` without changing the +normal two-second DTLS retry cadence: + +```python +import threading + +cancel_connect = threading.Event() +# A lifecycle or shutdown thread may call cancel_connect.set(). +sess.connect(timeout=8.0, cancel=cancel_connect) +``` + +Setting the event stops that connection attempt and closes its temporary UDP +socket. `quiesce_for_close()` stops new session work, interrupts an in-progress +connection, and wakes requests waiting for responses while retaining an +established socket for `close()`. `abort()` additionally closes established +local I/O immediately. Interrupted operations raise `SessionClosedError`. + If the cert/key are minted at runtime and never written to disk (e.g. inside an HA config flow), create the provider from memory instead: @@ -56,6 +81,47 @@ auth = CertificateAuth.from_memory(cert_pem, key_pem) sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth) ``` +Some newer OCF-PKI home appliances require an exact Samsung DTLS offer and +present a hardware certificate whose subject contains the appliance's OCF +UUID. When the caller already has an authorized client certificate and a +previously verified appliance UUID, opt in to both requirements explicitly: + +```python +from smartthings_local.protocol.auth import ( + CertificateAuth, + SamsungServerProfile, +) + +server_profile = SamsungServerProfile.bound_device( + expected_device_uuid, + additional_ca_pem=additional_samsung_ca_pem, +) +auth = CertificateAuth.from_memory( + cert_pem, + key_pem, + server_profile=server_profile, +) +sess = DtlsCoapSession("192.0.2.100", 49154, auth=auth) +``` + +The profile limits the ClientHello to P-256, +`ECDHE-ECDSA-AES128-GCM-SHA256`, and the observed SHA-256/SHA-1 RSA/ECDSA +signature set, disables session tickets, preserves certificate-chain +verification, and requires the exact subject role +`C=KR, O=Samsung Electronics, OU=OCF HA Device` with a common name ending in +the expected UUID. `additional_ca_pem` is optional and accepts only a bounded +PEM CA-certificate chain; it is applied only to this profiled context. Without +a profile, `CertificateAuth` retains its existing verification behavior. + +This API deliberately does not discover, mint, authorize, provision, rotate, +or persist credentials, and it performs no ownership transfer or OCF security +resource writes. In particular, an unowned vendor-OTM appliance such as the +one discussed in [issue #20](https://github.com/QuiteYellow/SmartThings-Local/issues/20) +does not become connectable merely by selecting this profile. The already-owned +new-PKI case in [issue #16](https://github.com/QuiteYellow/SmartThings-Local/issues/16) +still requires an authorized client identity before this connection profile +can be used. + For compatibility, the existing `cert_path` / `key_path` and `cert_pem` / `key_pem` session arguments remain supported without a deprecation warning. They are routed through `CertificateAuth` internally. Do not combine `auth` diff --git a/smartthings_local/protocol/auth.py b/smartthings_local/protocol/auth.py index 0e99712..e0fab08 100644 --- a/smartthings_local/protocol/auth.py +++ b/smartthings_local/protocol/auth.py @@ -3,15 +3,28 @@ from __future__ import annotations import re +import warnings from os import PathLike from pathlib import Path from typing import Protocol, runtime_checkable +from uuid import UUID +from cryptography.x509.oid import ExtensionOID, NameOID from OpenSSL import SSL, _util, crypto _OCF_ROOT_CA = str(Path(__file__).with_name("ocf_root_ca.pem")) _DTLS_CIPHERS = b"ECDHE-ECDSA-AES128-GCM-SHA256:@SECLEVEL=0" _DTLS_PSK_CIPHERS = b"ECDHE-PSK-AES128-CBC-SHA256:@SECLEVEL=0" +_SAMSUNG_SERVER_CURVES = b"prime256v1" +_SAMSUNG_SERVER_SIGNATURE_ALGORITHMS = ( + b"RSA+SHA256:ECDSA+SHA256:RSA+SHA1:ECDSA+SHA1" +) +_SAMSUNG_SERVER_CN_RE = re.compile( + r"\AOCF Device: [^()\r\n]{1,96} " + r"\((?P[0-9a-f]{8}-[0-9a-f]{4}-" + r"[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12})\)\Z", + re.IGNORECASE, +) _PSK_CLIENT_CALLBACK_CDEF = ( "unsigned int (*)(SSL *, char *, char *, unsigned int, " "unsigned char *, unsigned int)" @@ -53,6 +66,224 @@ def configure_context(self, context: SSL.Context) -> None: """Configure a context while this provider remains session-owned.""" +class SamsungServerProfile: + """Opt-in Samsung home-appliance server verification profile.""" + + __slots__ = ("_additional_ca_certificates", "_expected_device_identity") + + def __init__( + self, + *, + expected_device_identity: UUID | str, + additional_ca_pem: str | None = None, + ) -> None: + if type(expected_device_identity) is UUID: + parsed_identity = expected_device_identity + elif type(expected_device_identity) is str: + try: + parsed_identity = UUID(expected_device_identity) + except ValueError: + raise ValueError( + "expected_device_identity must be a canonical non-zero UUID" + ) from None + if expected_device_identity != str(parsed_identity): + raise ValueError( + "expected_device_identity must be a canonical non-zero UUID" + ) + else: + raise TypeError("expected_device_identity must be a UUID or string") + if parsed_identity.int == 0: + raise ValueError( + "expected_device_identity must be a canonical non-zero UUID" + ) + + certificates: tuple[bytes, ...] = () + if additional_ca_pem is not None: + if type(additional_ca_pem) is not str: + raise TypeError("additional_ca_pem must be a string") + try: + raw_ca_pem = additional_ca_pem.encode("ascii") + except UnicodeEncodeError: + raise ValueError( + "additional_ca_pem must contain ASCII PEM certificates" + ) from None + parsed_certificates = tuple(_PEM_CERT_RE.findall(raw_ca_pem)) + if ( + not 1 <= len(parsed_certificates) <= 4 + or len(raw_ca_pem) > 32 * 1024 + or _PEM_CERT_RE.sub(b"", raw_ca_pem).strip() + ): + raise ValueError( + "additional_ca_pem must contain one to four PEM certificates" + ) + try: + loaded_certificates = [ + crypto.load_certificate(crypto.FILETYPE_PEM, certificate) + for certificate in parsed_certificates + ] + basic_constraints = [ + [ + extension + for extension in certificate.to_cryptography().extensions + if extension.oid == ExtensionOID.BASIC_CONSTRAINTS + ] + for certificate in loaded_certificates + ] + except (crypto.Error, ValueError): + raise ValueError( + "additional_ca_pem contains an invalid certificate" + ) from None + if any( + len(constraints) != 1 or not constraints[0].value.ca + for constraints in basic_constraints + ): + raise ValueError( + "additional_ca_pem must contain only CA certificates" + ) + fingerprints = { + crypto.dump_certificate(crypto.FILETYPE_ASN1, certificate) + for certificate in loaded_certificates + } + if len(fingerprints) != len(loaded_certificates): + raise ValueError( + "additional_ca_pem must not contain duplicate certificates" + ) + certificates = parsed_certificates + + object.__setattr__(self, "_expected_device_identity", parsed_identity) + object.__setattr__(self, "_additional_ca_certificates", certificates) + + def __setattr__(self, _name: str, _value: object) -> None: + raise AttributeError("SamsungServerProfile is immutable") + + def __delattr__(self, _name: str) -> None: + raise AttributeError("SamsungServerProfile is immutable") + + @classmethod + def bound_device( + cls, + expected_device_identity: UUID | str, + *, + additional_ca_pem: str | None = None, + ) -> SamsungServerProfile: + """Bind a verified Samsung hardware certificate to one OCF UUID.""" + return cls( + expected_device_identity=expected_device_identity, + additional_ca_pem=additional_ca_pem, + ) + + def __repr__(self) -> str: + """Return a representation without device or trust-chain details.""" + return "SamsungServerProfile()" + + def _configure_context(self, context: SSL.Context) -> None: + curve_setter = getattr(_util.lib, "SSL_CTX_set1_curves_list", None) + if curve_setter is not None: + if curve_setter(context._context, _SAMSUNG_SERVER_CURVES) != 1: + raise RuntimeError( + "OpenSSL rejected the Samsung server certificate profile" + ) + else: + # pyOpenSSL 23.1 does not expose SSL_CTX_set1_curves_list. Its + # public set_tmp_ecdh fallback produces the same single P-256 + # supported-groups ClientHello extension; wire-level tests protect + # that compatibility path. Newer pyOpenSSL uses the exact setter. + try: + with warnings.catch_warnings(): + warnings.simplefilter("ignore", DeprecationWarning) + curve = crypto.get_elliptic_curve("prime256v1") + context.set_tmp_ecdh(curve) + except (AttributeError, TypeError, ValueError, SSL.Error): + raise RuntimeError( + "OpenSSL rejected the Samsung server certificate profile" + ) from None + + signature_setter = getattr( + _util.lib, + "SSL_CTX_set1_sigalgs_list", + None, + ) + if ( + signature_setter is None + or signature_setter( + context._context, + _SAMSUNG_SERVER_SIGNATURE_ALGORITHMS, + ) + != 1 + ): + raise RuntimeError( + "OpenSSL rejected the Samsung server certificate profile" + ) + context.set_options(SSL.OP_NO_TICKET) + + if self._additional_ca_certificates: + store = context.get_cert_store() + try: + for certificate in self._additional_ca_certificates: + store.add_cert( + crypto.load_certificate( + crypto.FILETYPE_PEM, + certificate, + ) + ) + except crypto.Error: + raise RuntimeError( + "OpenSSL rejected the Samsung server trust profile" + ) from None + + def _verify_peer( + self, + _connection, + certificate, + _error, + depth, + ok, + ) -> bool: + if not ok or certificate is None or depth < 0: + return False + if depth > 0: + return True + try: + components = [ + (attribute.oid, attribute.value) + for relative_name in certificate.to_cryptography().subject.rdns + for attribute in relative_name + ] + common_names = [ + value for name, value in components if name == NameOID.COMMON_NAME + ] + organizational_units = [ + value + for name, value in components + if name == NameOID.ORGANIZATIONAL_UNIT_NAME + ] + organizations = [ + value + for name, value in components + if name == NameOID.ORGANIZATION_NAME + ] + countries = [ + value for name, value in components if name == NameOID.COUNTRY_NAME + ] + if ( + len(common_names) != 1 + or organizational_units != ["OCF HA Device"] + or organizations != ["Samsung Electronics"] + or countries != ["KR"] + ): + return False + match = _SAMSUNG_SERVER_CN_RE.fullmatch(common_names[0]) + return ( + match is not None + and UUID(match.group("device_identity")) + == self._expected_device_identity + ) + except Exception: # noqa: BLE001 - never cross the OpenSSL callback + # Verification callbacks cannot propagate Python exceptions into + # OpenSSL. Any malformed or unexpected subject fails closed. + return False + + class CertificateAuth: """Certificate authentication loaded from files or in-memory PEM data. @@ -65,6 +296,7 @@ class CertificateAuth: "_certificate_pem", "_private_key_path", "_private_key_pem", + "_server_profile", ) def __init__( @@ -74,6 +306,7 @@ def __init__( private_key_path: str | PathLike[str] | None = None, certificate_pem: str | None = None, private_key_pem: str | None = None, + server_profile: SamsungServerProfile | None = None, ) -> None: file_supplied = ( certificate_path is not None or private_key_path is not None @@ -101,6 +334,11 @@ def __init__( "must pass either certificate_path/private_key_path or " "certificate_pem/private_key_pem" ) + if ( + server_profile is not None + and type(server_profile) is not SamsungServerProfile + ): + raise TypeError("server_profile must be a SamsungServerProfile") object.__setattr__( self, "_certificate_path", @@ -113,6 +351,7 @@ def __init__( ) object.__setattr__(self, "_certificate_pem", certificate_pem) object.__setattr__(self, "_private_key_pem", private_key_pem) + object.__setattr__(self, "_server_profile", server_profile) def __setattr__(self, _name: str, _value: object) -> None: raise AttributeError("CertificateAuth is immutable") @@ -125,11 +364,14 @@ def from_files( cls, certificate_path: str | PathLike[str], private_key_path: str | PathLike[str], + *, + server_profile: SamsungServerProfile | None = None, ) -> CertificateAuth: """Create a provider backed by certificate-chain and key files.""" return cls( certificate_path=certificate_path, private_key_path=private_key_path, + server_profile=server_profile, ) @classmethod @@ -137,11 +379,14 @@ def from_memory( cls, certificate_pem: str, private_key_pem: str, + *, + server_profile: SamsungServerProfile | None = None, ) -> CertificateAuth: """Create a provider backed by an in-memory PEM chain and key.""" return cls( certificate_pem=certificate_pem, private_key_pem=private_key_pem, + server_profile=server_profile, ) def __repr__(self) -> str: @@ -151,7 +396,14 @@ def __repr__(self) -> str: def configure_context(self, context: SSL.Context) -> None: """Apply the existing certificate authentication profile to a context.""" context.load_verify_locations(_OCF_ROOT_CA) - context.set_verify(SSL.VERIFY_PEER, _verify_peer) + if self._server_profile is None: + context.set_verify(SSL.VERIFY_PEER, _verify_peer) + else: + self._server_profile._configure_context(context) + context.set_verify( + SSL.VERIFY_PEER, + self._server_profile._verify_peer, + ) # @SECLEVEL=0 permits SHA-1 in Samsung's server cert chain (AC14K_M # intermediate is SHA-1 signed). This is the only channel that reaches # the OpenSSL instance cryptography bundles; ctypes and cffi bindings @@ -243,4 +495,9 @@ def configure_context(self, context: SSL.Context) -> None: setter(context._context, self._callback) # noqa: SLF001 -__all__ = ["AuthenticationProvider", "CertificateAuth", "PskAuth"] +__all__ = [ + "AuthenticationProvider", + "CertificateAuth", + "PskAuth", + "SamsungServerProfile", +] diff --git a/smartthings_local/protocol/dtls_session.py b/smartthings_local/protocol/dtls_session.py index aeeb305..e0f6c3a 100644 --- a/smartthings_local/protocol/dtls_session.py +++ b/smartthings_local/protocol/dtls_session.py @@ -18,6 +18,7 @@ on a per-token Event the reader signals. OBSERVE notifications are delivered via the on_notification callback. """ +import math import os import socket import threading @@ -72,6 +73,29 @@ # once the ceiling is measured empirically. _DEFAULT_RATE_LIMIT_RPS = 5.0 + +def _validate_handshake_timeout(timeout, default): + """Return one finite, positive DTLS handshake timeout.""" + value = default if timeout is None else timeout + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise TypeError('timeout must be a number or None') + try: + value = float(value) + except OverflowError: + raise ValueError( + 'timeout must be a positive finite number or None') from None + if not math.isfinite(value) or value <= 0: + raise ValueError('timeout must be a positive finite number or None') + return value + + +def _validate_cancel_event(cancel): + """Require an explicit thread-safe cancellation primitive.""" + if cancel is not None and not isinstance(cancel, threading.Event): + raise TypeError('cancel must be a threading.Event or None') + return cancel + + class DtlsCoapSession: """Single sustained DTLS-CoAP session. @@ -90,6 +114,8 @@ class DtlsCoapSession: """ HANDSHAKE_TIMEOUT_S = 12.0 + HANDSHAKE_SOCKET_TIMEOUT_S = 2.0 + HANDSHAKE_CANCEL_POLL_S = 0.1 READER_RECV_TIMEOUT_S = 1.0 # short so stop_event propagates quickly MAX_BLOCKS = 32 # safety bound for Block2 fetches @@ -145,11 +171,14 @@ def __init__(self, host, port, cert_path=None, key_path=None, *, self.family = family self.sock = None + self._connecting_sock = None + self._socket_state_lock = threading.Lock() self.conn = None self.dest = None self.endpoint = None self._send_lock = threading.Lock() + self._pending_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 @@ -179,35 +208,121 @@ def pace(self) -> None: # ---- lifecycle --------------------------------------------------- - def connect(self): - """DTLS handshake. Blocks up to HANDSHAKE_TIMEOUT_S. Raises - ConnectionError / TimeoutError on failure.""" + def _connect_interrupted(self, cancel): + return self._stop.is_set() or ( + cancel is not None and cancel.is_set() + ) + + def _raise_if_connect_interrupted(self, cancel): + if self._connect_interrupted(cancel): + raise SessionClosedError() + + def _retire_connecting_socket(self, sock): + should_close = False + with self._socket_state_lock: + if self._connecting_sock is sock: + self._connecting_sock = None + should_close = True + if not should_close: + return + try: + sock.close() + except OSError: + pass + + def _recv_handshake_datagram(self, sock, deadline, cancel): + """Wait one normal retry interval while polling cancellation.""" + poll_deadline = min( + deadline, + time.monotonic() + self.HANDSHAKE_SOCKET_TIMEOUT_S, + ) + while True: + self._raise_if_connect_interrupted(cancel) + remaining = poll_deadline - time.monotonic() + if remaining <= 0: + return None + poll_timeout = remaining + if cancel is not None: + poll_timeout = min( + self.HANDSHAKE_CANCEL_POLL_S, + remaining, + ) + sock.settimeout(poll_timeout) + try: + return sock.recv(65535) + except socket.timeout: + if cancel is None: + return None + + def connect( + self, + *, + timeout: float | None = None, + cancel: threading.Event | None = None, + ): + """Perform a cancellable DTLS handshake within a monotonic deadline. + + ``timeout`` overrides ``HANDSHAKE_TIMEOUT_S`` for this call. The + receive poll and retry sleep are both capped by the remaining budget, + so a missing peer cannot add an extra socket timeout after the + deadline. When supplied, ``cancel`` interrupts the attempt without + changing the two-second handshake retry cadence. + """ + handshake_timeout = _validate_handshake_timeout( + timeout, self.HANDSHAKE_TIMEOUT_S) + cancel = _validate_cancel_event(cancel) + self._raise_if_connect_interrupted(cancel) + deadline = time.monotonic() + handshake_timeout ctx = SSL.Context(SSL.DTLS_METHOD) self.auth.configure_context(ctx) + self._raise_if_connect_interrupted(cancel) conn = SSL.Connection(ctx, None) conn.set_connect_state() conn.set_ciphertext_mtu(self.mtu) + self._raise_if_connect_interrupted(cancel) + remaining = deadline - time.monotonic() + if remaining <= 0: + raise SessionTimeoutError() sock, endpoint = open_connected_udp_socket( self.host, self.port, family=self.family, local_port=self.local_port, - timeout=2.0, + timeout=min(self.HANDSHAKE_SOCKET_TIMEOUT_S, remaining), ) dest = endpoint.sockaddr + with self._socket_state_lock: + interrupted = self._connect_interrupted(cancel) + if not interrupted: + self._connecting_sock = sock + if interrupted: + try: + sock.close() + except OSError: + pass + raise SessionClosedError() - t0 = time.time() backend_failed = False - while time.time() - t0 < self.HANDSHAKE_TIMEOUT_S: + while time.monotonic() < deadline: + if self._connect_interrupted(cancel): + self._retire_connecting_socket(sock) + raise SessionClosedError() try: conn.do_handshake() + if self._connect_interrupted(cancel): + self._retire_connecting_socket(sock) + raise SessionClosedError() + if time.monotonic() >= deadline: + continue break except SSL.WantReadError: pass except SSL.Error: - sock.close() + self._retire_connecting_socket(sock) + if self._connect_interrupted(cancel): + raise SessionClosedError() from None backend_failed = True break send_failed = False @@ -215,42 +330,71 @@ def connect(self): o = conn.bio_read(65535) if o: for r in _split_dtls(o): + self._raise_if_connect_interrupted(cancel) if sock.send(r) != len(r): raise OSError('incomplete UDP send') except SSL.WantReadError: pass except OSError: - sock.close() + self._retire_connecting_socket(sock) + if self._connect_interrupted(cancel): + raise SessionClosedError() from None send_failed = True if send_failed: raise EndpointError() from OSError('UDP send failed') + remaining = deadline - time.monotonic() + if remaining <= 0: + continue receive_failed = False try: - d = sock.recv(65535) + d = self._recv_handshake_datagram( + sock, + deadline, + cancel, + ) + self._raise_if_connect_interrupted(cancel) if d: conn.bio_write(d) except socket.timeout: pass except OSError: - sock.close() + self._retire_connecting_socket(sock) + if self._connect_interrupted(cancel): + raise SessionClosedError() from None receive_failed = True if receive_failed: raise EndpointError() from OSError('UDP receive failed') - time.sleep(0.05) + remaining = deadline - time.monotonic() + if remaining > 0: + time.sleep(min(0.05, remaining)) else: - sock.close() + self._retire_connecting_socket(sock) + if self._connect_interrupted(cancel): + raise SessionClosedError() raise SessionTimeoutError() if backend_failed: raise SessionError() from ConnectionError('DTLS backend failed') - self.sock = sock - self.conn = conn - self.dest = dest - self.endpoint = endpoint - self._stop.clear() + with self._socket_state_lock: + interrupted = self._connect_interrupted(cancel) + if self._connecting_sock is sock: + self._connecting_sock = None + if not interrupted: + self.sock = sock + self.conn = conn + self.dest = dest + self.endpoint = endpoint + if interrupted: + try: + sock.close() + except OSError: + pass + raise SessionClosedError() def start_reader(self): """Spawn the reader thread. Must be called after connect().""" + if self._stop.is_set(): + raise SessionClosedError() if self.sock is None: raise RuntimeError("connect() before start_reader()") t = threading.Thread(target=self._reader_loop, @@ -275,6 +419,59 @@ def _send_observe_dereg(self, tok, path_segs): self._send_dgram( build_coap(TYPE_CON, METHOD_GET, mid, tok, opts)) + def _require_active(self): + if self._stop.is_set() or self.conn is None: + raise SessionClosedError() + + def _register_pending(self, tok, ev, container): + with self._pending_lock: + self._require_active() + self._pending[tok] = (ev, container) + + def _pop_pending(self, tok): + with self._pending_lock: + self._pending.pop(tok, None) + + def _wake_pending(self): + with self._pending_lock: + for ev, container in self._pending.values(): + container.setdefault('err', SessionClosedError()) + ev.set() + + def quiesce_for_close(self): + """Stop new work and wake response waiters, preserving an active session.""" + self._stop.set() + with self._socket_state_lock: + connecting_sock = self._connecting_sock + self._connecting_sock = None + if connecting_sock is not None: + try: + connecting_sock.close() + except OSError: + pass + self._wake_pending() + + def abort(self): + """Immediately interrupt local session I/O without peer interaction.""" + self.quiesce_for_close() + with self._socket_state_lock: + sock = self.sock + if sock is not None: + try: + sock.close() + except OSError: + pass + with self._send_lock: + with self._socket_state_lock: + if self.sock is sock: + self.sock = None + self.conn = None + self.dest = None + self.endpoint = None + with self._pending_lock: + self._pending.clear() + self._observe_tokens.clear() + def close(self): """Tear down session. Sends best-effort OBSERVE deregisters first so Samsung's RT-OCF cleans up its observer table — @@ -283,7 +480,8 @@ def close(self): # Send dereg for every active observation while the conn is # still healthy. Tiny sleep lets the records reach the wire # before we shut DTLS down. - if self.conn is not None and self._observe_tokens: + if (not self._stop.is_set() and self.conn is not None + and self._observe_tokens): for tok, href in list(self._observe_tokens.items()): segs = [s for s in href.split('/') if s] try: @@ -292,26 +490,28 @@ def close(self): logger.warning("dereg %s: %s", href, e) time.sleep(0.1) - self._stop.set() - if self.conn is not None: - try: - self.conn.shutdown() - except Exception: - pass - if self.sock is not None: - try: - self.sock.close() - except Exception: - pass - for tok, (ev, container) in list(self._pending.items()): - container.setdefault('err', SessionClosedError()) - ev.set() - self._pending.clear() + self.quiesce_for_close() + with self._send_lock: + with self._socket_state_lock: + conn = self.conn + sock = self.sock + self.sock = None + self.conn = None + self.dest = None + self.endpoint = None + if conn is not None: + try: + conn.shutdown() + except Exception: + pass + if sock is not None: + try: + sock.close() + except Exception: + pass + with self._pending_lock: + self._pending.clear() self._observe_tokens.clear() - self.sock = None - self.conn = None - self.dest = None - self.endpoint = None # ---- send / receive plumbing ------------------------------------- @@ -342,8 +542,7 @@ def _send_dgram(self, datagram): """Send a CoAP datagram. Holds the send lock for the BIO-drain so two writers can't interleave records.""" with self._send_lock: - if self.conn is None: - raise SessionClosedError() + self._require_active() send_failed = False try: self.conn.send(datagram) @@ -358,6 +557,8 @@ def _send_dgram(self, datagram): except SSL.WantReadError: pass except OSError: + if self._stop.is_set(): + raise SessionClosedError() from None send_failed = True if send_failed: raise EndpointError() from OSError('UDP send failed') @@ -420,9 +621,7 @@ def _reader_loop(self): return finally: # Make sure pending waiters don't hang if the reader dies. - for tok, (ev, container) in list(self._pending.items()): - container.setdefault('err', SessionClosedError()) - ev.set() + self._wake_pending() def _dispatch_coap(self, datagram): try: @@ -452,15 +651,17 @@ def _dispatch_coap(self, datagram): return # Pending one-shot? Resolve and return. - rec = self._pending.get(tok) + with self._pending_lock: + rec = self._pending.get(tok) + if rec is not None: + ev, container = rec + container['code'] = code + container['mtype'] = mt + container['mid'] = mid + container['options'] = ropts + container['payload'] = payload + ev.set() if rec is not None: - ev, container = rec - container['code'] = code - container['mtype'] = mt - container['mid'] = mid - container['options'] = ropts - container['payload'] = payload - ev.set() return # OBSERVE notification? @@ -490,8 +691,7 @@ def get(self, path_segs, query=(), timeout=10.0): 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._require_active() tok = self._next_tok() blob = b'' num = 0 @@ -506,7 +706,7 @@ def get(self, path_segs, query=(), timeout=10.0): for attempt in range(_BLOCK_MAX_ATTEMPTS): ev = threading.Event() container = {} - self._pending[tok] = (ev, container) + self._register_pending(tok, ev, container) try: mid = self._next_mid() opts = [(URI_PATH, s.encode()) for s in path_segs] @@ -534,7 +734,7 @@ def get(self, path_segs, query=(), timeout=10.0): attempt + 1, _BLOCK_MAX_ATTEMPTS, ) finally: - self._pending.pop(tok, None) + self._pop_pending(tok) if 'err' in container: raise container['err'] @@ -566,8 +766,7 @@ def get(self, path_segs, query=(), timeout=10.0): 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._require_active() tok = self._next_tok() mid = self._next_mid() opts = [(URI_PATH, s.encode()) for s in path_segs] @@ -577,7 +776,7 @@ def post(self, path_segs, body_cbor, timeout=8.0): body_cbor) ev = threading.Event() container = {} - self._pending[tok] = (ev, container) + self._register_pending(tok, ev, container) try: self._send_dgram(datagram) if not ev.wait(timeout): @@ -586,7 +785,7 @@ def post(self, path_segs, body_cbor, timeout=8.0): raise container['err'] return container['code'], container['payload'] finally: - self._pending.pop(tok, None) + self._pop_pending(tok) def ping(self): """RFC 7252 §4.4 CoAP Ping — empty CON, no token, no payload. @@ -600,8 +799,7 @@ def ping(self): 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._require_active() mid = self._next_mid() self._send_dgram(build_coap(TYPE_CON, 0, mid, b'', [])) return mid @@ -618,20 +816,25 @@ def refresh_observes(self, paths): 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._require_active() for tok, href in list(self._observe_tokens.items()): segs = [s for s in href.split('/') if s] try: self._send_observe_dereg(tok, segs) + except SessionClosedError: + raise except Exception as e: logger.warning("refresh dereg %s: %s", href, e) self._observe_tokens.clear() - time.sleep(0.1) + if self._stop.wait(0.1): + raise SessionClosedError() for path in paths: try: self.subscribe(list(path)) - time.sleep(0.05) + if self._stop.wait(0.05): + raise SessionClosedError() + except SessionClosedError: + raise except Exception as e: logger.warning("refresh subscribe %s: %s", path, e) @@ -642,8 +845,7 @@ def subscribe(self, path_segs): Returns the token used (in case the caller wants to deregister later).""" - if self.conn is None: - raise SessionClosedError() + self._require_active() tok = self._next_observe_tok() href = '/' + '/'.join(path_segs) # Register the token BEFORE sending — otherwise the device @@ -654,6 +856,10 @@ def subscribe(self, path_segs): opts = [(URI_PATH, s.encode()) for s in path_segs] opts.append((OBSERVE, OBSERVE_REGISTER)) opts.append((ACCEPT, CF_CBOR)) - self._send_dgram( - build_coap(TYPE_CON, METHOD_GET, mid, tok, opts)) + try: + self._send_dgram( + build_coap(TYPE_CON, METHOD_GET, mid, tok, opts)) + except Exception: + self._observe_tokens.pop(tok, None) + raise return tok diff --git a/tests/test_certificate_profiles.py b/tests/test_certificate_profiles.py new file mode 100644 index 0000000..5114f2b --- /dev/null +++ b/tests/test_certificate_profiles.py @@ -0,0 +1,588 @@ +"""Synthetic security and wire-contract tests for certificate profiles.""" + +from __future__ import annotations + +from dataclasses import asdict +from datetime import UTC, datetime, timedelta +from types import SimpleNamespace +from uuid import UUID + +import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID +from OpenSSL import SSL, crypto + +import smartthings_local.protocol.auth as auth_module +from smartthings_local.protocol.auth import ( + CertificateAuth, + SamsungServerProfile, +) + +_IDENTITY = UUID(bytes=b"\xab" * 16) +_OTHER_IDENTITY = UUID(bytes=b"\xcd" * 16) + + +def _build_certificate( + *, + subject: x509.Name, + issuer: x509.Name, + public_key, + issuer_key, + serial_number: int, + is_ca: bool | None, +) -> x509.Certificate: + now = datetime.now(UTC) + builder = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(issuer) + .public_key(public_key) + .serial_number(serial_number) + .not_valid_before(now - timedelta(minutes=1)) + .not_valid_after(now + timedelta(hours=1)) + ) + if is_ca is not None: + builder = builder.add_extension( + x509.BasicConstraints(ca=is_ca, path_length=None), + critical=True, + ).add_extension( + x509.KeyUsage( + digital_signature=True, + content_commitment=False, + key_encipherment=False, + data_encipherment=False, + key_agreement=False, + key_cert_sign=is_ca, + crl_sign=is_ca, + encipher_only=None, + decipher_only=None, + ), + critical=True, + ) + if is_ca is False: + builder = builder.add_extension( + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.SERVER_AUTH]), + critical=False, + ) + return builder.sign(issuer_key, hashes.SHA256()) + + +def _make_generated_chain( + identity: UUID, + *, + organizational_unit: str = "OCF HA Device", + intermediate_has_constraints: bool = True, +): + """Create a throwaway three-level chain unrelated to real devices.""" + root_key = ec.generate_private_key(ec.SECP256R1()) + root_name = x509.Name( + [x509.NameAttribute(NameOID.COMMON_NAME, "Synthetic profile root")] + ) + root = _build_certificate( + subject=root_name, + issuer=root_name, + public_key=root_key.public_key(), + issuer_key=root_key, + serial_number=101, + is_ca=True, + ) + + intermediate_key = ec.generate_private_key(ec.SECP256R1()) + intermediate_name = x509.Name( + [ + x509.NameAttribute( + NameOID.COMMON_NAME, + "Synthetic profile intermediate", + ) + ] + ) + intermediate = _build_certificate( + subject=intermediate_name, + issuer=root_name, + public_key=intermediate_key.public_key(), + issuer_key=root_key, + serial_number=102, + is_ca=True if intermediate_has_constraints else None, + ) + + leaf_key = ec.generate_private_key(ec.SECP256R1()) + leaf_name = x509.Name( + [ + x509.NameAttribute(NameOID.COUNTRY_NAME, "KR"), + x509.NameAttribute( + NameOID.ORGANIZATION_NAME, + "Samsung Electronics", + ), + x509.NameAttribute( + NameOID.ORGANIZATIONAL_UNIT_NAME, + organizational_unit, + ), + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: Test ({identity})", + ), + ] + ) + leaf = _build_certificate( + subject=leaf_name, + issuer=intermediate_name, + public_key=leaf_key.public_key(), + issuer_key=intermediate_key, + serial_number=103, + is_ca=False, + ) + + root_pem = root.public_bytes(serialization.Encoding.PEM).decode() + intermediate_pem = intermediate.public_bytes(serialization.Encoding.PEM).decode() + leaf_pem = leaf.public_bytes(serialization.Encoding.PEM).decode() + key_pem = leaf_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ).decode() + return SimpleNamespace( + root_pem=root_pem, + certificate_pem=leaf_pem + intermediate_pem, + private_key_pem=key_pem, + leaf=crypto.load_certificate(crypto.FILETYPE_PEM, leaf_pem), + intermediate=crypto.load_certificate( + crypto.FILETYPE_PEM, + intermediate_pem, + ), + ) + + +@pytest.fixture(scope="module") +def generated_chain(): + return _make_generated_chain(_IDENTITY) + + +def _configured_context(chain, profile=None) -> SSL.Context: + context = SSL.Context(SSL.DTLS_METHOD) + CertificateAuth.from_memory( + chain.certificate_pem, + chain.private_key_pem, + server_profile=profile, + ).configure_context(context) + return context + + +def _first_client_hello(context: SSL.Context) -> bytes: + connection = SSL.Connection(context, None) + connection.set_connect_state() + with pytest.raises(SSL.WantReadError): + connection.do_handshake() + chunks = [] + while True: + try: + chunks.append(connection.bio_read(65535)) + except SSL.WantReadError: + break + assert len(chunks) == 1 + return chunks[0] + + +def _parse_client_hello(datagram: bytes): + """Return cipher suites and extensions from one DTLS ClientHello.""" + assert datagram[0] == 22 + record_length = int.from_bytes(datagram[11:13], "big") + handshake = datagram[13 : 13 + record_length] + assert handshake[0] == 1 + body = handshake[12:] + + offset = 2 + 32 + session_id_length = body[offset] + offset += 1 + session_id_length + cookie_length = body[offset] + offset += 1 + cookie_length + + cipher_length = int.from_bytes(body[offset : offset + 2], "big") + offset += 2 + ciphers = [ + int.from_bytes(body[index : index + 2], "big") + for index in range(offset, offset + cipher_length, 2) + ] + offset += cipher_length + + compression_length = body[offset] + offset += 1 + compression_length + extensions_length = int.from_bytes(body[offset : offset + 2], "big") + offset += 2 + extensions_end = offset + extensions_length + extensions = {} + while offset < extensions_end: + extension_type = int.from_bytes(body[offset : offset + 2], "big") + extension_length = int.from_bytes( + body[offset + 2 : offset + 4], + "big", + ) + offset += 4 + assert extension_type not in extensions + extensions[extension_type] = body[offset : offset + extension_length] + offset += extension_length + assert offset == extensions_end == len(body) + return ciphers, extensions + + +def _vector_values(extension: bytes) -> list[int]: + vector_length = int.from_bytes(extension[:2], "big") + assert vector_length == len(extension) - 2 + return [ + int.from_bytes(extension[index : index + 2], "big") + for index in range(2, len(extension), 2) + ] + + +def _verify_chain(context: SSL.Context, chain) -> None: + crypto.X509StoreContext( + context.get_cert_store(), + chain.leaf, + [chain.intermediate], + ).verify_certificate() + + +def test_profile_accepts_only_canonical_nonzero_identity(): + assert repr(SamsungServerProfile.bound_device(_IDENTITY)) == ( + "SamsungServerProfile()" + ) + assert repr(SamsungServerProfile.bound_device(str(_IDENTITY))) == ( + "SamsungServerProfile()" + ) + + for invalid in ( + UUID(int=0), + str(UUID(int=0)), + str(_IDENTITY).upper(), + "{" + str(_IDENTITY) + "}", + "not-an-identity", + ): + with pytest.raises(ValueError, match="canonical non-zero UUID"): + SamsungServerProfile.bound_device(invalid) + for invalid in (True, 1, b"not-an-identity"): + with pytest.raises(TypeError, match="UUID or string"): + SamsungServerProfile.bound_device(invalid) + + +def test_profile_additional_ca_input_is_bounded_and_parsed(generated_chain): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + assert repr(profile) == "SamsungServerProfile()" + + with pytest.raises(TypeError, match="must be a string"): + SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem.encode(), + ) + for invalid in ( + "", + generated_chain.root_pem + "unexpected trailing material", + generated_chain.root_pem * 5, + generated_chain.root_pem * 2, + generated_chain.root_pem + (" " * (32 * 1024)), + generated_chain.certificate_pem, + "-----BEGIN CERTIFICATE-----\ninvalid\n-----END CERTIFICATE-----", + "non-ascii-\N{SNOWMAN}", + ): + with pytest.raises(ValueError): + SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=invalid, + ) + + +def test_profile_is_immutable_and_has_no_public_identity_or_ca_surface( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + rendered = repr(profile) + assert str(_IDENTITY) not in rendered + assert "BEGIN CERTIFICATE" not in rendered + assert not hasattr(profile, "expected_device_identity") + assert not hasattr(profile, "additional_ca_pem") + with pytest.raises(TypeError): + vars(profile) + with pytest.raises(TypeError): + asdict(profile) + with pytest.raises(AttributeError, match="immutable"): + profile.expected_device_identity = _OTHER_IDENTITY + with pytest.raises(AttributeError, match="immutable"): + del profile._expected_device_identity + + +def test_certificate_auth_requires_the_exact_profile_type(generated_chain): + with pytest.raises(TypeError, match="SamsungServerProfile"): + CertificateAuth.from_memory( + generated_chain.certificate_pem, + generated_chain.private_key_pem, + server_profile=object(), + ) + + +def test_profile_emits_the_exact_client_hello_and_reuses_cold( + generated_chain, +): + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + + first = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + second = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + assert first == second + + ciphers, extensions = first + # Older OpenSSL appends the non-negotiable renegotiation SCSV (0x00ff). + # No other negotiable cipher may enter the profile. + assert ciphers[0] == 0xC02B + assert set(ciphers) <= {0xC02B, 0x00FF} + assert _vector_values(extensions[10]) == [23] + assert _vector_values(extensions[13]) == [ + 0x0401, + 0x0403, + 0x0201, + 0x0203, + ] + assert 35 not in extensions + + +def test_python_floor_curve_fallback_has_the_same_wire_contract( + monkeypatch, + generated_chain, +): + signature_setter = auth_module._util.lib.SSL_CTX_set1_sigalgs_list + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace( + lib=SimpleNamespace( + SSL_CTX_set1_sigalgs_list=signature_setter, + ) + ), + ) + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + + ciphers, extensions = _parse_client_hello( + _first_client_hello(_configured_context(generated_chain, profile)) + ) + + assert ciphers[0] == 0xC02B + assert set(ciphers) <= {0xC02B, 0x00FF} + assert _vector_values(extensions[10]) == [23] + assert _vector_values(extensions[13]) == [ + 0x0401, + 0x0403, + 0x0201, + 0x0203, + ] + assert 35 not in extensions + + +def test_profile_fails_closed_when_exact_openssl_support_is_unavailable( + monkeypatch, +): + class FallbackContext: + _context = object() + + def set_tmp_ecdh(self, _curve): + return None + + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace(lib=SimpleNamespace()), + ) + profile = SamsungServerProfile.bound_device(_IDENTITY) + + with pytest.raises(RuntimeError, match="rejected"): + profile._configure_context(FallbackContext()) + + +def test_default_and_profile_verification_are_selected_per_provider( + monkeypatch, +): + valid_chain = _make_generated_chain(_IDENTITY) + mismatch_chain = _make_generated_chain(_OTHER_IDENTITY) + + class RecordingContext: + def __init__(self): + self._context = object() + self.calls = [] + self.verify_callback = None + + def load_verify_locations(self, path): + self.calls.append(("load_verify_locations", path)) + + def set_verify(self, mode, callback): + self.calls.append(("set_verify", mode)) + self.verify_callback = callback + + def set_cipher_list(self, ciphers): + self.calls.append(("set_cipher_list", ciphers)) + + def use_certificate_chain_file(self, path): + self.calls.append(("use_certificate_chain_file", path)) + + def use_privatekey_file(self, path): + self.calls.append(("use_privatekey_file", path)) + + def check_privatekey(self): + self.calls.append(("check_privatekey",)) + + def set_options(self, options): + self.calls.append(("set_options", options)) + + profile_calls = [] + monkeypatch.setattr( + auth_module, + "_util", + SimpleNamespace( + lib=SimpleNamespace( + SSL_CTX_set1_curves_list=( + lambda handle, value: ( + profile_calls.append(("curves", handle, value)) or 1 + ) + ), + SSL_CTX_set1_sigalgs_list=( + lambda handle, value: ( + profile_calls.append(("signature_algorithms", handle, value)) + or 1 + ) + ), + ) + ), + ) + + default_context = RecordingContext() + CertificateAuth.from_files("/synthetic/cert", "/synthetic/key").configure_context( + default_context + ) + assert default_context.verify_callback(None, None, 0, 0, True) is True + assert default_context.verify_callback(None, None, 0, 0, False) is False + assert not profile_calls + assert all(call[0] != "set_options" for call in default_context.calls) + + profiled_context = RecordingContext() + profile = SamsungServerProfile.bound_device(_IDENTITY) + CertificateAuth.from_files( + "/synthetic/cert", + "/synthetic/key", + server_profile=profile, + ).configure_context(profiled_context) + + assert [call[0] for call in profile_calls] == [ + "curves", + "signature_algorithms", + ] + assert ("set_options", SSL.OP_NO_TICKET) in profiled_context.calls + callback = profiled_context.verify_callback + assert callback(None, valid_chain.leaf, 0, 0, True) is True + assert callback(None, mismatch_chain.leaf, 0, 0, True) is False + assert callback(None, valid_chain.leaf, 0, 0, False) is False + + +def test_profile_identity_verification_rejects_malformed_subjects(): + profile = SamsungServerProfile.bound_device(_IDENTITY) + wrong_role = _make_generated_chain( + _IDENTITY, + organizational_unit="Unexpected Device", + ) + + assert profile._verify_peer(None, wrong_role.leaf, 0, 0, True) is False + assert profile._verify_peer(None, None, 0, 0, True) is False + assert profile._verify_peer(None, wrong_role.leaf, 0, -1, True) is False + assert profile._verify_peer(None, object(), 0, 0, True) is False + assert profile._verify_peer(None, object(), 0, 1, True) is True + + duplicate_subject = x509.Name( + [ + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: First ({_IDENTITY})", + ), + x509.NameAttribute( + NameOID.COMMON_NAME, + f"OCF Device: Other ({_IDENTITY})", + ), + x509.NameAttribute(NameOID.ORGANIZATIONAL_UNIT_NAME, "OCF HA Device"), + x509.NameAttribute(NameOID.ORGANIZATION_NAME, "Samsung Electronics"), + x509.NameAttribute(NameOID.COUNTRY_NAME, "KR"), + ] + ) + duplicate_common_name = SimpleNamespace( + to_cryptography=lambda: SimpleNamespace(subject=duplicate_subject) + ) + assert ( + profile._verify_peer( + None, + duplicate_common_name, + 0, + 0, + True, + ) + is False + ) + + +def test_additional_ca_is_scoped_and_invalid_intermediate_is_rejected( + generated_chain, +): + default_context = _configured_context(generated_chain) + with pytest.raises(crypto.X509StoreContextError): + _verify_chain(default_context, generated_chain) + + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + profiled_context = _configured_context(generated_chain, profile) + _verify_chain(profiled_context, generated_chain) + + missing_constraints = _make_generated_chain( + _IDENTITY, + intermediate_has_constraints=False, + ) + missing_constraints_profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=missing_constraints.root_pem, + ) + missing_constraints_context = _configured_context( + missing_constraints, + missing_constraints_profile, + ) + with pytest.raises(crypto.X509StoreContextError): + _verify_chain(missing_constraints_context, missing_constraints) + + +def test_profile_errors_and_provider_repr_do_not_echo_inputs(generated_chain): + marker = "private" + "-profile-marker" + with pytest.raises(ValueError) as captured: + SamsungServerProfile.bound_device(marker) + assert marker not in str(captured.value) + + profile = SamsungServerProfile.bound_device( + _IDENTITY, + additional_ca_pem=generated_chain.root_pem, + ) + provider = CertificateAuth.from_memory( + generated_chain.certificate_pem, + generated_chain.private_key_pem, + server_profile=profile, + ) + rendered = repr(provider) + repr(profile) + assert str(_IDENTITY) not in rendered + assert generated_chain.root_pem not in rendered + assert generated_chain.private_key_pem not in rendered diff --git a/tests/test_endpoint.py b/tests/test_endpoint.py index 6d6ddf3..4848c49 100644 --- a/tests/test_endpoint.py +++ b/tests/test_endpoint.py @@ -376,7 +376,7 @@ def open_socket(*args, **kwargs): assert open_calls == [(('device.example', 5684), { 'family': socket.AF_INET6, 'local_port': None, - 'timeout': 2.0, + 'timeout': session.HANDSHAKE_SOCKET_TIMEOUT_S, })] session.close() diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index f40e5b4..1a1a40d 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -10,6 +10,7 @@ AuthenticationProvider, CertificateAuth, PskAuth, + SamsungServerProfile, ) from smartthings_local.protocol.dtls_session import DtlsCoapSession @@ -54,6 +55,19 @@ def test_certificate_auth_is_a_public_authentication_provider(): provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key") assert isinstance(provider, AuthenticationProvider) + for factory in (CertificateAuth.from_files, CertificateAuth.from_memory): + profile_parameter = inspect.signature(factory).parameters["server_profile"] + assert profile_parameter.kind is inspect.Parameter.KEYWORD_ONLY + assert profile_parameter.default is None + + +def test_samsung_server_profile_is_public_and_explicitly_bound(): + parameters = inspect.signature(SamsungServerProfile.bound_device).parameters + assert list(parameters) == ["expected_device_identity", "additional_ca_pem"] + assert parameters["expected_device_identity"].default is inspect.Parameter.empty + assert parameters["additional_ca_pem"].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["additional_ca_pem"].default is None + def test_psk_auth_is_a_public_authentication_provider(): provider = PskAuth(identity=b"i" * 16, key=b"k" * 16) @@ -69,6 +83,7 @@ def test_psk_auth_is_a_public_authentication_provider(): def test_dtls_session_keeps_current_consumer_methods(): expected = { + "abort", "close", "connect", "get", @@ -76,11 +91,23 @@ def test_dtls_session_keeps_current_consumer_methods(): "pace", "ping", "post", + "quiesce_for_close", "refresh_observes", "start_reader", "subscribe", } assert expected <= set(dir(DtlsCoapSession)) + _assert_compatible_signature(DtlsCoapSession.connect, ["self"]) + connect_timeout = inspect.signature(DtlsCoapSession.connect).parameters[ + "timeout" + ] + assert connect_timeout.kind is inspect.Parameter.KEYWORD_ONLY + assert connect_timeout.default is None + connect_cancel = inspect.signature(DtlsCoapSession.connect).parameters[ + "cancel" + ] + assert connect_cancel.kind is inspect.Parameter.KEYWORD_ONLY + assert connect_cancel.default is None _assert_compatible_signature( DtlsCoapSession.get, [ diff --git a/tests/test_session_connect_deadline.py b/tests/test_session_connect_deadline.py new file mode 100644 index 0000000..428ffa0 --- /dev/null +++ b/tests/test_session_connect_deadline.py @@ -0,0 +1,299 @@ +"""Deterministic tests for bounded DTLS handshake timing.""" + +from __future__ import annotations + +import socket + +import pytest +from OpenSSL import SSL + +from smartthings_local.errors import SessionTimeoutError +from smartthings_local.protocol import dtls_session +from smartthings_local.protocol.dtls_session import DtlsCoapSession +from smartthings_local.protocol.endpoint import ResolvedUdpEndpoint + + +class _Clock: + def __init__(self): + self.now = 100.0 + + def monotonic(self): + return self.now + + def advance(self, seconds): + self.now += seconds + + +class _Auth: + def __init__(self, clock=None, configure_delay=0.0): + self.clock = clock + self.configure_delay = configure_delay + + def configure_context(self, _context): + if self.clock is not None: + self.clock.advance(self.configure_delay) + + +class _Connection: + def __init__(self, outcomes=None): + self.outcomes = list(outcomes or ()) + self.bio_writes = [] + + def set_connect_state(self): + return None + + def set_ciphertext_mtu(self, _mtu): + return None + + def do_handshake(self): + outcome = self.outcomes.pop(0) if self.outcomes else "want-read" + if outcome == "want-read": + raise SSL.WantReadError() + if isinstance(outcome, Exception): + raise outcome + + def bio_read(self, _size): + raise SSL.WantReadError() + + def bio_write(self, data): + self.bio_writes.append(data) + + +class _Socket: + def __init__(self, clock, inbound=()): + self.clock = clock + self.inbound = list(inbound) + self.timeouts = [] + self.closed = False + + def settimeout(self, timeout): + self.timeouts.append(timeout) + + def send(self, data): + return len(data) + + def recv(self, _size): + if self.inbound: + return self.inbound.pop(0) + self.clock.advance(self.timeouts[-1]) + raise TimeoutError() + + def close(self): + self.closed = True + + +def _session(auth=None): + return DtlsCoapSession( + "device.example", + 5684, + auth=auth or _Auth(), + ) + + +def _install_handshake(monkeypatch, clock, *, outcomes=(), inbound=()): + connection = _Connection(outcomes) + sock = _Socket(clock, inbound) + endpoint = ResolvedUdpEndpoint( + socket.AF_INET, + ("192.0.2.10", 5684), + ) + open_calls = [] + + def open_socket(*args, **kwargs): + open_calls.append((args, kwargs)) + sock.settimeout(kwargs["timeout"]) + return sock, endpoint + + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session.SSL, + "Connection", + lambda *_args: connection, + ) + monkeypatch.setattr( + dtls_session, + "open_connected_udp_socket", + open_socket, + ) + monkeypatch.setattr(dtls_session.time, "monotonic", clock.monotonic) + monkeypatch.setattr(dtls_session.time, "sleep", clock.advance) + monkeypatch.setattr( + dtls_session.time, + "time", + lambda: pytest.fail("wall clock must not control handshake deadlines"), + ) + return connection, sock, endpoint, open_calls + + +@pytest.mark.parametrize("timeout", (True, "1", object())) +def test_connect_timeout_type_is_explicit(timeout): + with pytest.raises(TypeError, match="number or None"): + _session().connect(timeout=timeout) + + +@pytest.mark.parametrize( + "timeout", + ( + 0, + -1, + float("nan"), + float("inf"), + float("-inf"), + 10**1000, + ), +) +def test_connect_timeout_must_be_positive_and_finite(timeout): + with pytest.raises(ValueError, match="positive finite"): + _session().connect(timeout=timeout) + + +def test_connect_timeout_caps_every_blocking_poll(monkeypatch): + clock = _Clock() + _connection, sock, _endpoint, open_calls = _install_handshake( + monkeypatch, + clock, + ) + session = _session() + + with pytest.raises(SessionTimeoutError): + session.connect(timeout=4.75) + + assert clock.now == pytest.approx(104.75) + assert sock.closed + assert open_calls == [ + ( + ("device.example", 5684), + { + "family": socket.AF_UNSPEC, + "local_port": None, + "timeout": session.HANDSHAKE_SOCKET_TIMEOUT_S, + }, + ) + ] + assert sock.timeouts == pytest.approx([2.0, 2.0, 2.0, 0.65]) + + +def test_short_timeout_is_not_rounded_up_to_poll_interval(monkeypatch): + clock = _Clock() + _connection, sock, _endpoint, open_calls = _install_handshake( + monkeypatch, + clock, + ) + + with pytest.raises(SessionTimeoutError): + _session().connect(timeout=0.125) + + assert clock.now == pytest.approx(100.125) + assert open_calls[0][1]["timeout"] == pytest.approx(0.125) + assert sock.timeouts == pytest.approx([0.125, 0.125]) + + +def test_default_timeout_uses_session_constant(monkeypatch): + clock = _Clock() + _connection, sock, _endpoint, _open_calls = _install_handshake( + monkeypatch, + clock, + ) + session = _session() + session.HANDSHAKE_TIMEOUT_S = 0.2 + + with pytest.raises(SessionTimeoutError): + session.connect() + + assert clock.now == pytest.approx(100.2) + assert sock.closed + + +def test_context_setup_consumes_the_same_deadline(monkeypatch): + clock = _Clock() + socket_opened = False + + def open_socket(*_args, **_kwargs): + nonlocal socket_opened + socket_opened = True + raise AssertionError("expired setup must not open a socket") + + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr(dtls_session.SSL, "Connection", lambda *_args: _Connection()) + monkeypatch.setattr(dtls_session, "open_connected_udp_socket", open_socket) + monkeypatch.setattr(dtls_session.time, "monotonic", clock.monotonic) + session = _session(_Auth(clock, configure_delay=0.2)) + + with pytest.raises(SessionTimeoutError): + session.connect(timeout=0.1) + + assert not socket_opened + + +def test_socket_setup_consumes_the_same_deadline(monkeypatch): + clock = _Clock() + connection = _Connection() + connection.do_handshake = lambda: pytest.fail( + "expired socket setup must not start a handshake" + ) + sock = _Socket(clock) + endpoint = ResolvedUdpEndpoint( + socket.AF_INET, + ("192.0.2.10", 5684), + ) + + def open_socket(*_args, **kwargs): + sock.settimeout(kwargs["timeout"]) + clock.advance(0.2) + return sock, endpoint + + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session.SSL, + "Connection", + lambda *_args: connection, + ) + monkeypatch.setattr(dtls_session, "open_connected_udp_socket", open_socket) + monkeypatch.setattr(dtls_session.time, "monotonic", clock.monotonic) + + with pytest.raises(SessionTimeoutError): + _session().connect(timeout=0.1) + + assert sock.closed + + +def test_successful_handshake_preserves_connected_session_state(monkeypatch): + clock = _Clock() + connection, sock, endpoint, _open_calls = _install_handshake( + monkeypatch, + clock, + outcomes=("want-read", "success"), + inbound=(b"synthetic server flight",), + ) + session = _session() + + session.connect(timeout=1.0) + + assert connection.bio_writes == [b"synthetic server flight"] + assert session.conn is connection + assert session.sock is sock + assert session.endpoint is endpoint + assert session.dest == endpoint.sockaddr + assert not sock.closed + + +@pytest.mark.parametrize("success_delay", (0.1, 0.2)) +def test_handshake_success_at_or_after_deadline_is_rejected( + monkeypatch, + success_delay, +): + clock = _Clock() + connection, sock, _endpoint, _open_calls = _install_handshake( + monkeypatch, + clock, + ) + + def late_success(): + clock.advance(success_delay) + + connection.do_handshake = late_success + + with pytest.raises(SessionTimeoutError): + _session().connect(timeout=0.1) + + assert sock.closed diff --git a/tests/test_session_interruption.py b/tests/test_session_interruption.py new file mode 100644 index 0000000..ea1c853 --- /dev/null +++ b/tests/test_session_interruption.py @@ -0,0 +1,565 @@ +"""Deterministic tests for connection cancellation and forced shutdown.""" + +from __future__ import annotations + +import socket +import threading + +import pytest +from OpenSSL import SSL + +from smartthings_local.errors import SessionClosedError +from smartthings_local.protocol import dtls_session +from smartthings_local.protocol.dtls_session import DtlsCoapSession +from smartthings_local.protocol.endpoint import ResolvedUdpEndpoint + + +class _Auth: + def __init__(self, on_configure=None): + self.on_configure = on_configure + + def configure_context(self, _context): + if self.on_configure is not None: + self.on_configure() + + +class _Connection: + def __init__(self, handshake="want-read", outbound=()): + self.handshake = handshake + self.outbound = list(outbound) + self.handshake_calls = 0 + self.bio_writes = [] + + def set_connect_state(self): + return None + + def set_ciphertext_mtu(self, _mtu): + return None + + def do_handshake(self): + self.handshake_calls += 1 + if callable(self.handshake): + self.handshake() + return + if self.handshake == "want-read": + raise SSL.WantReadError() + + def bio_read(self, _size): + if self.outbound: + return self.outbound.pop(0) + raise SSL.WantReadError() + + def bio_write(self, data): + self.bio_writes.append(data) + + +class _Socket: + def __init__(self): + self.timeouts = [] + self.sent = [] + self.close_calls = 0 + + def settimeout(self, timeout): + self.timeouts.append(timeout) + + def send(self, data): + self.sent.append(data) + return len(data) + + def recv(self, _size): + raise TimeoutError() + + def close(self): + self.close_calls += 1 + + +class _BlockingSocket(_Socket): + def __init__(self): + super().__init__() + self.recv_started = threading.Event() + self.closed = threading.Event() + + def recv(self, _size): + self.recv_started.set() + if self.closed.wait(self.timeouts[-1]): + raise OSError("synthetic closed socket") + raise TimeoutError() + + def close(self): + super().close() + self.closed.set() + + +class _BlockingSendSocket(_Socket): + def __init__(self): + super().__init__() + self.send_started = threading.Event() + self.closed = threading.Event() + + def send(self, _data): + self.send_started.set() + assert self.closed.wait(1.0) + raise OSError("synthetic closed socket") + + def close(self): + super().close() + self.closed.set() + + +def _session(auth=None): + return DtlsCoapSession( + "device.example", + 5684, + auth=auth or _Auth(), + ) + + +def _install_connect(monkeypatch, connection, sock, *, on_open=None): + endpoint = ResolvedUdpEndpoint( + socket.AF_INET, + ("192.0.2.10", 5684), + ) + open_calls = [] + + def open_socket(*args, **kwargs): + open_calls.append((args, kwargs)) + sock.settimeout(kwargs["timeout"]) + if on_open is not None: + on_open() + return sock, endpoint + + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session.SSL, + "Connection", + lambda *_args: connection, + ) + monkeypatch.setattr( + dtls_session, + "open_connected_udp_socket", + open_socket, + ) + return endpoint, open_calls + + +def _connect_in_thread(session, **kwargs): + started = threading.Event() + outcome = [] + + def connect(): + started.set() + try: + session.connect(**kwargs) + except BaseException as error: # noqa: BLE001 + outcome.append(error) + + worker = threading.Thread(target=connect, name="test-connect") + worker.start() + assert started.wait(1.0) + return worker, outcome + + +@pytest.mark.parametrize("cancel", (True, object(), "event")) +def test_connect_cancel_type_is_explicit(cancel): + with pytest.raises(TypeError, match="threading.Event or None"): + _session().connect(cancel=cancel) + + +def test_pre_cancelled_connect_stops_before_context_setup(monkeypatch): + cancel = threading.Event() + cancel.set() + monkeypatch.setattr( + dtls_session.SSL, + "Context", + lambda *_args: pytest.fail("cancelled connect configured TLS"), + ) + + with pytest.raises(SessionClosedError): + _session().connect(cancel=cancel) + + +def test_cancel_during_context_setup_stops_before_socket_setup(monkeypatch): + cancel = threading.Event() + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session, + "open_connected_udp_socket", + lambda *_args, **_kwargs: pytest.fail("cancelled connect opened a socket"), + ) + + with pytest.raises(SessionClosedError): + _session(_Auth(cancel.set)).connect(cancel=cancel) + + +def test_cancel_during_connection_setup_stops_before_socket_setup(monkeypatch): + cancel = threading.Event() + connection = _Connection(handshake="success") + connection.set_ciphertext_mtu = lambda _mtu: cancel.set() + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session.SSL, + "Connection", + lambda *_args: connection, + ) + monkeypatch.setattr( + dtls_session, + "open_connected_udp_socket", + lambda *_args, **_kwargs: pytest.fail("cancelled connect opened a socket"), + ) + + with pytest.raises(SessionClosedError): + _session().connect(cancel=cancel) + + +def test_cancel_during_socket_setup_closes_without_handshake(monkeypatch): + cancel = threading.Event() + connection = _Connection(handshake="success") + sock = _Socket() + _install_connect( + monkeypatch, + connection, + sock, + on_open=cancel.set, + ) + + with pytest.raises(SessionClosedError): + _session().connect(cancel=cancel) + + assert sock.close_calls == 1 + assert connection.handshake_calls == 0 + + +def test_cancel_during_receive_does_not_accelerate_handshake_retry(monkeypatch): + cancel = threading.Event() + connection = _Connection() + sock = _BlockingSocket() + _install_connect(monkeypatch, connection, sock) + session = _session() + worker, outcome = _connect_in_thread( + session, + timeout=5.0, + cancel=cancel, + ) + assert sock.recv_started.wait(1.0) + + cancel.set() + worker.join(1.0) + + assert not worker.is_alive() + assert len(outcome) == 1 + assert isinstance(outcome[0], SessionClosedError) + assert connection.handshake_calls == 1 + assert sock.timeouts[0] == session.HANDSHAKE_SOCKET_TIMEOUT_S + assert all( + timeout <= session.HANDSHAKE_CANCEL_POLL_S + for timeout in sock.timeouts[1:] + ) + assert sock.close_calls == 1 + assert session._connecting_sock is None + assert session.sock is None + + +def test_cancel_reported_during_handshake_success_is_rejected(monkeypatch): + cancel = threading.Event() + connection = _Connection(handshake=cancel.set) + sock = _Socket() + _install_connect(monkeypatch, connection, sock) + + with pytest.raises(SessionClosedError): + _session().connect(cancel=cancel) + + assert connection.handshake_calls == 1 + assert sock.close_calls == 1 + + +def test_cancel_during_handshake_send_closes_before_receive(monkeypatch): + cancel = threading.Event() + connection = _Connection(outbound=(b"synthetic record",)) + sock = _Socket() + + def send(data): + sock.sent.append(data) + cancel.set() + return len(data) + + sock.send = send + _install_connect(monkeypatch, connection, sock) + monkeypatch.setattr( + dtls_session, + "_split_dtls", + lambda data: (data,), + ) + + with pytest.raises(SessionClosedError): + _session().connect(cancel=cancel) + + assert sock.sent == [b"synthetic record"] + assert sock.close_calls == 1 + + +def test_per_call_cancellation_does_not_close_session_for_a_retry(monkeypatch): + cancel = threading.Event() + first_connection = _Connection(handshake="success") + second_connection = _Connection(handshake="success") + connections = iter((first_connection, second_connection)) + first_socket = _Socket() + second_socket = _Socket() + sockets = iter((first_socket, second_socket)) + endpoint = ResolvedUdpEndpoint( + socket.AF_INET, + ("192.0.2.10", 5684), + ) + + monkeypatch.setattr(dtls_session.SSL, "Context", lambda *_args: object()) + monkeypatch.setattr( + dtls_session.SSL, + "Connection", + lambda *_args: next(connections), + ) + + def open_socket(*_args, **kwargs): + sock = next(sockets) + sock.settimeout(kwargs["timeout"]) + if sock is first_socket: + cancel.set() + return sock, endpoint + + monkeypatch.setattr( + dtls_session, + "open_connected_udp_socket", + open_socket, + ) + session = _session() + + with pytest.raises(SessionClosedError): + session.connect(cancel=cancel) + cancel.clear() + session.connect(cancel=cancel) + + assert not session._stop.is_set() + assert session.conn is second_connection + assert session.sock is second_socket + assert first_socket.close_calls == 1 + assert second_socket.close_calls == 0 + + +def test_quiesce_interrupts_an_inflight_handshake(monkeypatch): + connection = _Connection() + sock = _BlockingSocket() + _install_connect(monkeypatch, connection, sock) + session = _session() + worker, outcome = _connect_in_thread(session, timeout=5.0) + assert sock.recv_started.wait(1.0) + + session.quiesce_for_close() + worker.join(1.0) + + assert not worker.is_alive() + assert len(outcome) == 1 + assert isinstance(outcome[0], SessionClosedError) + assert session._stop.is_set() + assert session._connecting_sock is None + assert sock.close_calls == 1 + + +@pytest.mark.parametrize( + "operation", + ( + lambda session: session.get(["resource"], timeout=30.0), + lambda session: session.post(["resource"], b"payload", timeout=30.0), + ), +) +def test_quiesce_wakes_blocked_requests_without_closing_transport( + monkeypatch, + operation, +): + session = _session() + connection = object() + sock = _Socket() + session.conn = connection + session.sock = sock + sent = threading.Event() + monkeypatch.setattr(session, "_send_dgram", lambda _data: sent.set()) + outcome = [] + + def request(): + try: + operation(session) + except BaseException as error: # noqa: BLE001 + outcome.append(error) + + worker = threading.Thread(target=request, name="test-request") + worker.start() + assert sent.wait(1.0) + + session.quiesce_for_close() + worker.join(1.0) + + assert not worker.is_alive() + assert len(outcome) == 1 + assert isinstance(outcome[0], SessionClosedError) + assert session.conn is connection + assert session.sock is sock + assert sock.close_calls == 0 + + +def test_quiesce_rejects_new_session_work(): + session = _session() + session.conn = object() + session.sock = _Socket() + + session.quiesce_for_close() + + with pytest.raises(SessionClosedError): + session.ping() + with pytest.raises(SessionClosedError): + session.start_reader() + + +def test_abort_closes_established_transport_once_and_blocks_new_work(): + session = _session() + sock = _Socket() + session.conn = object() + session.sock = sock + session.dest = ("192.0.2.10", 5684) + session.endpoint = ResolvedUdpEndpoint( + socket.AF_INET, + session.dest, + ) + + session.abort() + session.abort() + + assert sock.close_calls == 1 + assert session.conn is None + assert session.sock is None + assert session.dest is None + assert session.endpoint is None + with pytest.raises(SessionClosedError): + session.ping() + + +def test_abort_wakes_a_blocked_request(monkeypatch): + session = _session() + sock = _Socket() + session.conn = object() + session.sock = sock + sent = threading.Event() + monkeypatch.setattr(session, "_send_dgram", lambda _data: sent.set()) + outcome = [] + + def request(): + try: + session.post(["resource"], b"payload", timeout=30.0) + except BaseException as error: # noqa: BLE001 + outcome.append(error) + + worker = threading.Thread(target=request, name="test-abort-request") + worker.start() + assert sent.wait(1.0) + + session.abort() + worker.join(1.0) + + assert not worker.is_alive() + assert len(outcome) == 1 + assert isinstance(outcome[0], SessionClosedError) + assert sock.close_calls == 1 + + +def test_abort_interrupts_a_request_blocked_while_sending(monkeypatch): + session = _session() + connection = _Connection() + connection.send = lambda _data: None + connection.outbound = [b"synthetic record"] + sock = _BlockingSendSocket() + session.conn = connection + session.sock = sock + monkeypatch.setattr( + dtls_session, + "_split_dtls", + lambda data: (data,), + ) + outcome = [] + + def request(): + try: + session.post(["resource"], b"payload", timeout=30.0) + except BaseException as error: # noqa: BLE001 + outcome.append(error) + + worker = threading.Thread(target=request, name="test-abort-send") + worker.start() + assert sock.send_started.wait(1.0) + + session.abort() + worker.join(1.0) + + assert not worker.is_alive() + assert len(outcome) == 1 + assert isinstance(outcome[0], SessionClosedError) + assert sock.close_calls == 1 + + +def test_abort_interrupts_rate_limit_pacing(): + session = _session() + session._min_req_interval = 60.0 + session._last_send_ts = dtls_session.time.monotonic() + started = threading.Event() + finished = threading.Event() + + def pace(): + started.set() + session.pace() + finished.set() + + worker = threading.Thread(target=pace, name="test-pace") + worker.start() + assert started.wait(1.0) + assert not finished.wait(0.05) + + session.abort() + worker.join(1.0) + + assert not worker.is_alive() + assert finished.is_set() + + +def test_abort_stops_a_reader_blocked_on_the_socket(): + session = _session() + sock = _BlockingSocket() + session.conn = object() + session.sock = sock + + session.start_reader() + assert sock.recv_started.wait(1.0) + + session.abort() + session.join() + + assert session._reader_thread is not None + assert not session._reader_thread.is_alive() + assert sock.close_calls == 1 + + +def test_close_after_quiesce_finalizes_preserved_transport(): + session = _session() + connection = _Connection(handshake="success") + connection.shutdown = lambda: None + sock = _Socket() + session.conn = connection + session.sock = sock + session.dest = ("192.0.2.10", 5684) + session.endpoint = ResolvedUdpEndpoint(socket.AF_INET, session.dest) + + session.quiesce_for_close() + + assert session.conn is connection + assert session.sock is sock + assert sock.close_calls == 0 + + session.close() + + assert sock.close_calls == 1 + assert session.conn is None + assert session.sock is None