diff --git a/README.md b/README.md index 52c2f41..113faea 100644 --- a/README.md +++ b/README.md @@ -183,6 +183,70 @@ python -m smartthings_local.protocol.dtls_probe "$APPLIANCE_IP" 5684 49153 49154 `live` means a DTLS server answered its first flight; `dead` means silent or not DTLS. Once you have the client cert (Part 2), add the explicit `--diagnostic` flag to run the stateful diagnostic drive, which reports `completed` (cert accepted) or `rejected` with the server's fatal alert. Diagnostic mode can allocate appliance-side DTLS state and is never used by discovery or reconnect. An `unsupported_certificate` / `unknown_ca` alert means the endpoint is reachable but this certificate profile was rejected. It is not a reason to disable verification or keep retrying. The same bounded stateless API gates the bridge's reconnect loop and, when `OCF_PORT` is unset, probes both standard 5684 and ports 49152–49160. +Consumers can discover ports outside that fallback range through the public, +read-only OCF resource directory before probing them: + +```python +from smartthings_local.protocol.dtls_probe import probe_dtls_ports +from smartthings_local.protocol.ocf_discovery import discover_ocf_secure_ports + +fallback_ports = (5684, *range(49152, 49161)) +advertisement = discover_ocf_secure_ports(appliance_host) +candidates = advertisement.ports or fallback_ports +probe = probe_dtls_ports(appliance_host, candidates) +``` + +`discover_ocf_secure_ports()` sends only +`GET /oic/res?rt=oic.r.doxm`. It accepts Samsung's dynamic plaintext response +source port while still requiring the resolved target address and CoAP token, +and assembles Block2 responses within fixed time, block-count, and payload +limits. Its default three-second timeout bounds all socket I/O in total (DNS +resolution is synchronous and outside that budget); retries do not multiply +that deadline. The example probes the fixed fallback range only when discovery +does not return an advertised port, so this remains an explicit consumer +policy rather than an automatic or widened scan. An advertised port remains +only a candidate: require a successful stateless DTLS probe before attempting +authentication. + +Some Samsung hosts expose multiple logical OCF devices from one IPv4 address, +so the root `/oic/sec/doxm` identity is not always the appliance identity. When +the SmartThings OCF `di` UUID is available, use the identity-aware multicast +variant on one explicit LAN interface: + +```python +from smartthings_local.protocol.ocf_discovery import ( + discover_ocf_secure_ports_multicast, +) + +advertisement = discover_ocf_secure_ports_multicast( + "11111111-2222-3333-4444-555555555555", + interface_address="192.0.2.10", +) +``` + +This API performs exactly two IPv4 multicast NON discovery rounds, with a +six-second collection window per round by default (about 12 seconds maximum) +and accepts at most 64 datagrams per round. It succeeds only when the same sole +source advertises the same ports in both rounds, reads links only from the exact +normalized `di` container, binds legacy `p.sec` / `port` values to that +response source, and accepts an `eps` URI only when its host equals the response +source. The result exposes the matched address and ports to the caller but +redacts the address, UUID, and port values from its `repr`. The existing unicast +API remains appropriate when the target host is already identity-safe. + +The unicast and multicast entry points are explicit alternatives. Neither +function invokes the other or starts an automatic fallback, so adding these +APIs does not add discovery time to existing callers. A consumer chooses the +known-host unicast path or the identity-aware multicast path for its own flow, +then applies its separate DTLS and authentication gates. + +On a host that exposes multiple logical OCF devices, the two results may +legitimately contain different ports. Do not merge them automatically: +known-host unicast expresses trust in the caller-supplied host, while +identity-aware multicast selects the one directory container with the requested +`di`. That UUID match selects a candidate endpoint; it does not replace DTLS +liveness and authenticated device-identity checks. + ### Tested combinations | Appliance class | Model family | Confirmed | @@ -509,6 +573,7 @@ smartthings_local/ The installable library — `pip install sm coap.py CoAP wire protocol: message encode/decode, token handling dtls_session.py DTLS session: handshake, client-cert auth (file or in-memory PEM), Block2, liveness dtls_probe.py Stateless DTLS liveness + opt-in stateful diagnostic + ocf_discovery.py Bounded public OCF secure-port discovery ocf_root_ca.pem Samsung OCF root CA, bundled for handshake verification ocf/ OCF resource + state layer (reusable) __init__.py diff --git a/smartthings_local/protocol/coap.py b/smartthings_local/protocol/coap.py index 5a7b7ba..1005a64 100644 --- a/smartthings_local/protocol/coap.py +++ b/smartthings_local/protocol/coap.py @@ -61,9 +61,22 @@ def encode_options(opts): def parse_coap(data): """Decode a CoAP datagram. Returns (mtype, code, mid, token, - options, payload). options is a list of (num, value_bytes).""" + options, payload). options is a list of (num, value_bytes). + + The decoder is intentionally strict because some callers use it on + unauthenticated UDP datagrams. Truncated headers, tokens, extended option + fields, option values, and empty payload markers are classified rather + than leaking ``IndexError`` or being accepted as partial messages. + """ + if not isinstance(data, (bytes, bytearray, memoryview)): + raise MalformedMessageError() + data = bytes(data) + if len(data) < 4 or data[0] >> 6 != 1: + raise MalformedMessageError() mt = (data[0] >> 4) & 0x03 tkl = data[0] & 0x0F + if tkl > 8 or len(data) < 4 + tkl: + raise MalformedMessageError() code = data[1] mid = int.from_bytes(data[2:4], 'big') tok = data[4:4 + tkl] @@ -74,27 +87,39 @@ def parse_coap(data): while i < len(data): b = data[i] if b == 0xFF: + if i + 1 >= len(data): + raise MalformedMessageError() payload = data[i + 1:] break d_nib, l_nib = b >> 4, b & 0x0F i += 1 if d_nib == 13: + if i >= len(data): + raise MalformedMessageError() delta = 13 + data[i]; i += 1 elif d_nib == 14: + if i + 2 > len(data): + raise MalformedMessageError() delta = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2 elif d_nib == 15: raise MalformedMessageError() else: delta = d_nib if l_nib == 13: + if i >= len(data): + raise MalformedMessageError() length = 13 + data[i]; i += 1 elif l_nib == 14: + if i + 2 > len(data): + raise MalformedMessageError() length = 269 + int.from_bytes(data[i:i + 2], 'big'); i += 2 elif l_nib == 15: raise MalformedMessageError() else: length = l_nib num = prev + delta + if i + length > len(data): + raise MalformedMessageError() opts.append((num, data[i:i + length])) i += length prev = num diff --git a/smartthings_local/protocol/ocf_discovery.py b/smartthings_local/protocol/ocf_discovery.py new file mode 100644 index 0000000..b29dc73 --- /dev/null +++ b/smartthings_local/protocol/ocf_discovery.py @@ -0,0 +1,991 @@ +"""Bounded discovery of OCF-advertised secure UDP ports. + +Samsung appliances normally receive public CoAP discovery on UDP 5683, but +some firmware sends the response from a different source port. This module +therefore uses unconnected UDP sockets, validates the resolved target address +and CoAP token, then pins the first valid response endpoint for the remainder +of a bounded Block2 transfer. + +Only NON ``GET /oic/res?rt=oic.r.doxm`` is sent. Retries keep the token and +use a fresh message ID. No ownership, credential, or other OCF security +resource is written. +""" + +from __future__ import annotations + +import io +import math +import secrets +import selectors +import socket +import time +import uuid +from dataclasses import dataclass +from urllib.parse import urlsplit + +import cbor2 + +from ..errors import MalformedMessageError +from .coap import ( + ACCEPT, + BLOCK2, + CF_CBOR, + CONTENT_FORMAT, + METHOD_GET, + SIZE2, + TYPE_ACK, + TYPE_CON, + TYPE_NON, + TYPE_RST, + URI_PATH, + URI_QUERY, + block_value, + build_coap, + parse_coap, +) +from .endpoint import ResolvedUdpEndpoint, resolve_udp_endpoints + +__all__ = [ + 'OcfMulticastSecurePortDiscoveryResult', + 'OcfSecurePortDiscoveryResult', + 'discover_ocf_secure_ports', + 'discover_ocf_secure_ports_multicast', +] + +_DISCOVERY_PORT = 5683 +_IPV4_OCF_MULTICAST_GROUP = socket.inet_ntoa(bytes((224, 0, 1, 187))) +_MULTICAST_ROUNDS = 2 +_MAX_MULTICAST_RESPONSES_PER_ROUND = 64 +_MAX_ENDPOINTS = 8 +_MAX_PORTS = 8 +_MAX_BLOCKS = 32 +_MAX_DATAGRAM_BYTES = 8192 +_MAX_PAYLOAD_BYTES = 65536 +_MAX_CONTAINERS = 64 +_MAX_LINKS = 256 +_MAX_ENDPOINT_URIS_PER_LINK = 32 +_OCF_CBOR_CONTENT_FORMAT = 10000 +_ETAG = 4 +_CONTENT = 0x45 +_UNSET = object() + + +@dataclass(frozen=True, slots=True, repr=False) +class OcfSecurePortDiscoveryResult: + """Redacted outcome of one public OCF resource-directory lookup. + + ``attempts`` counts logical request attempts rather than destination + addresses. The custom representation deliberately omits the discovered + ports, target address, and wire data. + """ + + ports: tuple[int, ...] + attempts: int + response_received: bool + error_code: str | None = None + + @property + def found(self): + """Return whether at least one validated secure port was advertised.""" + return bool(self.ports) + + def __repr__(self): + return ( + 'OcfSecurePortDiscoveryResult(' + f'found={self.found!r}, port_count={len(self.ports)}, ' + f'attempts={self.attempts}, ' + f'response_received={self.response_received!r}, ' + f'error_code={self.error_code!r})' + ) + + +@dataclass(frozen=True, slots=True, repr=False) +class OcfMulticastSecurePortDiscoveryResult: + """Redacted result of identity-aware IPv4 multicast discovery. + + ``address`` and ``ports`` are available for the caller's next bounded + probe, but the custom representation omits both. The target UUID is used + only during discovery and is not retained in the result. + """ + + address: str | None + ports: tuple[int, ...] + rounds: int + responses: int + error_code: str | None = None + + @property + def found(self): + """Return whether one stable target advertisement was found.""" + return self.address is not None and bool(self.ports) + + def __repr__(self): + return ( + 'OcfMulticastSecurePortDiscoveryResult(' + f'found={self.found!r}, port_count={len(self.ports)}, ' + f'rounds={self.rounds}, responses={self.responses}, ' + f'error_code={self.error_code!r})' + ) + + +@dataclass(slots=True, repr=False) +class _Route: + sock: socket.socket + endpoint: ResolvedUdpEndpoint + host_key: tuple[bytes, int] + + +@dataclass(frozen=True, slots=True, repr=False) +class _ResponseBlock: + number: int + more: bool + szx: int | None + payload: bytes + etag: bytes | None + content_format: int | None + size2: int | None + + +def _validate_options(discovery_port, timeout, retries, family): + if isinstance(discovery_port, bool) or not isinstance(discovery_port, int): + raise TypeError('discovery_port must be an integer') + if not 1 <= discovery_port <= 65535: + raise ValueError('discovery_port must be between 1 and 65535') + if isinstance(timeout, bool) or not isinstance(timeout, (int, float)): + raise TypeError('timeout must be a number') + if not math.isfinite(timeout) or not 0 < timeout <= 30: + raise ValueError('timeout must be greater than zero and at most 30') + if isinstance(retries, bool) or not isinstance(retries, int): + raise TypeError('retries must be an integer') + if not 0 <= retries <= 4: + raise ValueError('retries must be between zero and four') + if isinstance(family, bool) or not isinstance(family, int): + raise TypeError('family must be an address-family integer') + if family not in (socket.AF_UNSPEC, socket.AF_INET, socket.AF_INET6): + raise ValueError('family must be AF_UNSPEC, AF_INET, or AF_INET6') + + +def _host_key(family, sockaddr): + """Return canonical address bytes plus an IPv6 scope ID.""" + expected_length = 2 if family == socket.AF_INET else 4 + if not isinstance(sockaddr, tuple) or len(sockaddr) != expected_length: + return None + host = sockaddr[0] + if not isinstance(host, str): + return None + if family == socket.AF_INET6: + host = host.split('%', 1)[0] + try: + packed = socket.inet_pton(family, host) + except OSError: + return None + scope_id = sockaddr[3] if family == socket.AF_INET6 else 0 + if isinstance(scope_id, bool) or not isinstance(scope_id, int): + return None + return packed, scope_id + + +def _peer_key(family, sockaddr): + host_key = _host_key(family, sockaddr) + if host_key is None: + return None + port = sockaddr[1] + if isinstance(port, bool) or not isinstance(port, int): + return None + if not 1 <= port <= 65535: + return None + return family, host_key[0], port, host_key[1] + + +def _open_routes(endpoints, selector): + routes = [] + for endpoint in endpoints[:_MAX_ENDPOINTS]: + key = _host_key(endpoint.family, endpoint.sockaddr) + if key is None: + continue + sock = None + try: + sock = socket.socket( + endpoint.family, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.bind(endpoint.bind_address(0)) + sock.setblocking(False) + route = _Route(sock, endpoint, key) + selector.register(sock, selectors.EVENT_READ, route) + routes.append(route) + except (OSError, ValueError): + if sock is not None: + try: + sock.close() + except OSError: + pass + return routes + + +def _option_values(options, number): + return [value for option_number, value in options + if option_number == number] + + +def _decode_uint_option(values, *, max_length): + if len(values) > 1: + return _UNSET + if not values: + return None + value = values[0] + if len(value) > max_length: + return _UNSET + return int.from_bytes(value, 'big') + + +def _decode_response_block( + datagram, *, token, expected_number, expected_szx): + """Classify one correlated CoAP response without retaining wire data.""" + try: + mtype, code, mid, response_token, options, payload = \ + parse_coap(datagram) + except MalformedMessageError: + return 'malformed', None, None + + if mtype == TYPE_RST: + return 'ignore', None, None + + if mtype == TYPE_ACK: + return 'ignore', None, None + + if response_token != token or code != _CONTENT: + return 'ignore', None, None + if mtype not in (TYPE_CON, TYPE_NON): + return 'ignore', None, None + + ack_mid = mid if mtype == TYPE_CON else None + block_values = _option_values(options, BLOCK2) + if len(block_values) > 1: + return 'malformed', None, ack_mid + + if block_values: + encoded = block_values[0] + if len(encoded) > 3: + return 'malformed', None, ack_mid + value = int.from_bytes(encoded, 'big') + number = value >> 4 + more = bool((value >> 3) & 1) + szx = value & 0x07 + if szx > 6: + return 'malformed', None, ack_mid + else: + number = 0 + more = False + szx = None + + if number < expected_number: + return 'duplicate', None, ack_mid + if number != expected_number: + return 'malformed', None, ack_mid + if expected_number > 0 and szx is None: + return 'malformed', None, ack_mid + if expected_szx is not None and szx != expected_szx: + return 'malformed', None, ack_mid + if szx is not None: + block_size = 1 << (szx + 4) + if len(payload) > block_size or (more and len(payload) != block_size): + return 'malformed', None, ack_mid + + etag_values = _option_values(options, _ETAG) + if len(etag_values) > 1: + return 'malformed', None, ack_mid + etag = etag_values[0] if etag_values else None + if etag is not None and not 1 <= len(etag) <= 8: + return 'malformed', None, ack_mid + + content_format = _decode_uint_option( + _option_values(options, CONTENT_FORMAT), max_length=2) + if content_format is _UNSET or content_format not in ( + None, int.from_bytes(CF_CBOR, 'big'), + _OCF_CBOR_CONTENT_FORMAT): + return 'malformed', None, ack_mid + + size2 = _decode_uint_option( + _option_values(options, SIZE2), max_length=4) + if size2 is _UNSET or ( + size2 is not None and size2 > _MAX_PAYLOAD_BYTES): + return 'malformed', None, ack_mid + + return 'block', _ResponseBlock( + number=number, + more=more, + szx=szx, + payload=payload, + etag=etag, + content_format=content_format, + size2=size2, + ), ack_mid + + +def _decode_cbor(payload): + stream = io.BytesIO(payload) + try: + value = cbor2.CBORDecoder(stream).decode() + except Exception: # noqa: BLE001 - untrusted CBOR must fail closed + return _UNSET + if stream.tell() != len(payload): + return _UNSET + return value + + +def _normalize_uuid(value): + """Return one canonical UUID value without rendering it.""" + try: + if isinstance(value, uuid.UUID): + return value + if isinstance(value, bytes): + if len(value) == 16: + return uuid.UUID(bytes=value) + value = value.decode('ascii') + if not isinstance(value, str): + return None + folded = value.casefold() + for prefix in ('urn:uuid:', 'uuid:'): + if folded.startswith(prefix): + value = value[len(prefix):] + break + return uuid.UUID(value) + except (UnicodeDecodeError, ValueError, AttributeError): + return None + + +def _resource_links(value): + """Return a shallow, bounded OCF link sequence or ``None``.""" + containers = value if isinstance(value, list) else [value] + if not all(isinstance(container, dict) for container in containers): + return None + + links = [] + for container in containers: + if 'links' in container: + nested = container.get('links') + if not isinstance(nested, list): + return None + candidates = nested + elif 'href' in container: + candidates = [container] + else: + candidates = [] + for link in candidates: + if not isinstance(link, dict): + return None + links.append(link) + if len(links) > _MAX_LINKS: + return None + return links + + +def _endpoint_uri_port(value): + if not isinstance(value, str): + return None + try: + parsed = urlsplit(value) + port = parsed.port + except ValueError: + return None + if (parsed.scheme != 'coaps' or not parsed.hostname + or parsed.username is not None or parsed.password is not None + or parsed.path or parsed.query or parsed.fragment): + return None + return 5684 if port is None else port + + +def _endpoint_uri_port_for_ipv4_source(value, source_key): + """Return a secure URI port only when its host is the response source.""" + if not isinstance(value, str): + return None + try: + parsed = urlsplit(value) + port = parsed.port + endpoint_key = socket.inet_pton(socket.AF_INET, parsed.hostname or '') + except (OSError, ValueError): + return None + if (parsed.scheme != 'coaps' + or endpoint_key != source_key + or parsed.username is not None or parsed.password is not None + or parsed.path or parsed.query or parsed.fragment): + return None + return 5684 if port is None else port + + +def _target_secure_ports_from_payload(payload, target_uuid, source_key): + """Classify one bounded directory payload for an exact target UUID. + + The status is ``target`` (with zero or more ports), ``absent``, or + ``malformed``. Only links nested inside the matching top-level container + are considered. Legacy policy ports are implicitly bound to ``source_key``; + modern endpoint URIs must explicitly name that same IPv4 address. + """ + value = _decode_cbor(payload) + if value is _UNSET: + return 'malformed', () + containers = value if isinstance(value, list) else [value] + if (not containers or len(containers) > _MAX_CONTAINERS + or not all(isinstance(container, dict) + for container in containers)): + return 'malformed', () + + matches = [ + container for container in containers + if _normalize_uuid(container.get('di')) == target_uuid + ] + if not matches: + return 'absent', () + if len(matches) != 1: + return 'malformed', () + + links = matches[0].get('links') + if (not isinstance(links, list) or len(links) > _MAX_LINKS + or not all(isinstance(link, dict) for link in links)): + return 'malformed', () + + ports = [] + seen = set() + + def add_port(port): + if (isinstance(port, bool) or not isinstance(port, int) + or not 1 <= port <= 65535 or port in seen): + return + seen.add(port) + ports.append(port) + + for link in links: + if link.get('href') != '/oic/sec/doxm': + continue + resource_types = link.get('rt') + if isinstance(resource_types, str): + resource_types = [resource_types] + if (not isinstance(resource_types, list) + or 'oic.r.doxm' not in resource_types): + continue + + policy = link.get('p') + if isinstance(policy, dict) and policy.get('sec') is True: + # A legacy policy has no host. Returning it with the datagram's + # source address is the binding; it is never associated with a + # different responder or a root-unicast identity. + add_port(policy.get('port')) + + endpoints = link.get('eps') + if isinstance(endpoints, list): + for endpoint in endpoints[:_MAX_ENDPOINT_URIS_PER_LINK]: + if not isinstance(endpoint, dict): + continue + add_port(_endpoint_uri_port_for_ipv4_source( + endpoint.get('ep'), source_key)) + + if len(ports) >= _MAX_PORTS: + break + return 'target', tuple(sorted(ports[:_MAX_PORTS])) + + +def _secure_ports_from_payload(payload): + value = _decode_cbor(payload) + if value is _UNSET: + return None + links = _resource_links(value) + if links is None: + return None + + ports = [] + seen = set() + + def add_port(port): + if (isinstance(port, bool) or not isinstance(port, int) + or not 1 <= port <= 65535 or port in seen): + return + seen.add(port) + ports.append(port) + + for link in links: + if link.get('href') != '/oic/sec/doxm': + continue + resource_types = link.get('rt') + if isinstance(resource_types, str): + resource_types = [resource_types] + if (not isinstance(resource_types, list) + or 'oic.r.doxm' not in resource_types): + continue + + policy = link.get('p') + if isinstance(policy, dict) and policy.get('sec') is True: + add_port(policy.get('port')) + + endpoints = link.get('eps') + if isinstance(endpoints, list): + for endpoint in endpoints[:_MAX_ENDPOINT_URIS_PER_LINK]: + if not isinstance(endpoint, dict): + continue + add_port(_endpoint_uri_port(endpoint.get('ep'))) + + if len(ports) >= _MAX_PORTS: + break + return tuple(ports[:_MAX_PORTS]) + + +def _build_request(token, mid, expected_number, szx): + options = [ + (URI_PATH, b'oic'), + (URI_PATH, b'res'), + (URI_QUERY, b'rt=oic.r.doxm'), + (ACCEPT, CF_CBOR), + ] + if expected_number > 0: + options.append((BLOCK2, block_value(expected_number, 0, szx))) + return build_coap(TYPE_NON, METHOD_GET, mid, token, options) + + +def _validate_multicast_options( + target_uuid, interface_address, discovery_port, round_timeout): + if not isinstance(target_uuid, (str, bytes, uuid.UUID)): + raise TypeError('target_uuid must be a UUID string or bytes value') + normalized_uuid = _normalize_uuid(target_uuid) + if normalized_uuid is None: + raise ValueError('target_uuid must be a valid UUID') + if not isinstance(interface_address, str): + raise TypeError('interface_address must be an IPv4 string') + try: + interface_key = socket.inet_pton(socket.AF_INET, interface_address) + except OSError as exc: + raise ValueError( + 'interface_address must be a valid IPv4 address') from exc + if (interface_key == b'\x00\x00\x00\x00' + or interface_key == b'\xff\xff\xff\xff' + or 224 <= interface_key[0] <= 239): + raise ValueError('interface_address must be a unicast IPv4 address') + _validate_options( + discovery_port, round_timeout, 0, socket.AF_INET) + return normalized_uuid, socket.inet_ntop(socket.AF_INET, interface_key), \ + interface_key + + +def _decode_multicast_response(datagram, token): + """Return one token-correlated, single-block NON response payload.""" + try: + mtype, code, _mid, response_token, _options, _payload = \ + parse_coap(datagram) + except MalformedMessageError: + return 'malformed', None + if response_token != token or code != _CONTENT or mtype != TYPE_NON: + return 'ignore', None + + status, block, _ack_mid = _decode_response_block( + datagram, + token=token, + expected_number=0, + expected_szx=None, + ) + if status != 'block': + return status, None + if (block.more or block.number != 0 + or (block.size2 is not None + and block.size2 != len(block.payload))): + return 'malformed', None + return 'payload', block.payload + + +def _multicast_result( + address, ports, rounds, responses, error_code=None): + return OcfMulticastSecurePortDiscoveryResult( + address=address, + ports=ports, + rounds=rounds, + responses=responses, + error_code=error_code, + ) + + +def discover_ocf_secure_ports_multicast( + target_uuid, *, interface_address, discovery_port=_DISCOVERY_PORT, + round_timeout=6.0): + """Find one exact OCF device identity on one IPv4 LAN interface. + + This is an explicit entry point and never invokes the known-host unicast + discovery function. + + Exactly two NON multicast discovery rounds are sent. A successful result + requires the same sole response source and the same non-empty secure-port + set in both rounds. Only the matching top-level ``di`` container is read; + legacy policy ports are bound to its response source, and ``eps`` hosts + must equal that source. No OCF security resource is read or written. + """ + normalized_uuid, interface_address, interface_key = \ + _validate_multicast_options( + target_uuid, interface_address, discovery_port, round_timeout) + + selector = selectors.DefaultSelector() + sock = None + try: + sock = socket.socket( + socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + sock.setsockopt( + socket.IPPROTO_IP, socket.IP_MULTICAST_IF, interface_key) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0) + sock.bind((interface_address, 0)) + sock.setblocking(False) + selector.register(sock, selectors.EVENT_READ) + except (OSError, ValueError): + if sock is not None: + try: + sock.close() + except OSError: + pass + selector.close() + return _multicast_result( + None, (), 0, 0, 'interface_unavailable') + + used_tokens = set() + used_mids = set() + round_matches = [] + inconsistent_sources = set() + rounds = 0 + responses = 0 + target_seen = False + target_with_ports = False + valid_directory_response = False + saw_malformed = False + + try: + for _round_number in range(_MULTICAST_ROUNDS): + token = secrets.token_bytes(8) + while token in used_tokens: + token = ( + (int.from_bytes(token, 'big') + 1) & ((1 << 64) - 1) + ).to_bytes(8, 'big') + used_tokens.add(token) + mid = secrets.randbits(16) + while mid in used_mids: + mid = (mid + 1) & 0xFFFF + used_mids.add(mid) + request = _build_request(token, mid, 0, None) + try: + sent_length = sock.sendto( + request, + (_IPV4_OCF_MULTICAST_GROUP, discovery_port), + ) + except OSError: + break + if sent_length != len(request): + break + rounds += 1 + + matches = {} + deadline = time.monotonic() + float(round_timeout) + datagrams = 0 + while datagrams < _MAX_MULTICAST_RESPONSES_PER_ROUND: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + try: + events = selector.select(remaining) + except (OSError, ValueError): + events = [] + if not events: + break + try: + datagram, source = sock.recvfrom( + _MAX_DATAGRAM_BYTES + 1) + except (BlockingIOError, OSError): + continue + datagrams += 1 + source_host_key = _host_key(socket.AF_INET, source) + if (_peer_key(socket.AF_INET, source) is None + or source_host_key is None + or source_host_key[0] == b'\x00\x00\x00\x00' + or source_host_key[0] == b'\xff\xff\xff\xff' + or 224 <= source_host_key[0][0] <= 239): + continue + if len(datagram) > _MAX_DATAGRAM_BYTES: + saw_malformed = True + continue + + status, payload = _decode_multicast_response( + datagram, token) + if status == 'ignore': + continue + if status != 'payload': + saw_malformed = True + continue + responses += 1 + target_status, ports = _target_secure_ports_from_payload( + payload, normalized_uuid, source_host_key[0]) + if target_status == 'malformed': + saw_malformed = True + continue + valid_directory_response = True + if target_status == 'absent': + continue + + target_seen = True + target_with_ports = target_with_ports or bool(ports) + source_key = source_host_key[0] + address = socket.inet_ntop(socket.AF_INET, source_key) + previous = matches.get(source_key) + candidate = (address, ports) + if previous is not None and previous != candidate: + inconsistent_sources.add(source_key) + continue + matches[source_key] = candidate + round_matches.append(matches) + + if rounds != _MULTICAST_ROUNDS: + return _multicast_result( + None, (), rounds, responses, 'interface_unavailable') + + sources = set().union(*(set(matches) for matches in round_matches)) + if inconsistent_sources or len(sources) > 1: + return _multicast_result( + None, (), rounds, responses, 'ambiguous_target') + if len(sources) == 1 and all(len(matches) == 1 + for matches in round_matches): + source_key = next(iter(sources)) + first = round_matches[0][source_key] + second = round_matches[1][source_key] + if first == second and first[1]: + return _multicast_result( + first[0], first[1], rounds, responses) + + if not target_seen: + if not responses: + error_code = 'no_ocf_response' + elif valid_directory_response: + error_code = 'target_not_found' + elif saw_malformed: + error_code = 'malformed_ocf_response' + else: + error_code = 'target_not_found' + elif not target_with_ports: + error_code = 'no_secure_ports' + else: + error_code = 'target_not_stable' + return _multicast_result( + None, (), rounds, responses, error_code) + finally: + try: + selector.unregister(sock) + except (KeyError, OSError, ValueError): + pass + try: + sock.close() + except OSError: + pass + selector.close() + + +def _result(ports, attempts, response_received, error_code=None): + return OcfSecurePortDiscoveryResult( + ports=ports, + attempts=attempts, + response_received=response_received, + error_code=error_code, + ) + + +def discover_ocf_secure_ports( + host, *, discovery_port=_DISCOVERY_PORT, timeout=3.0, retries=1, + family=socket.AF_UNSPEC): + """Discover secure ports advertised by a target's public OCF directory. + + This is an explicit known-host entry point and never starts multicast as + an automatic fallback. + + Name resolution happens synchronously first. ``timeout`` then bounds all + socket I/O, including a token-stable Block2 transfer. A port advertisement + is only a candidate; callers should prove it with + :func:`smartthings_local.protocol.dtls_probe.probe_dtls_ports` before a + DTLS handshake. + """ + _validate_options(discovery_port, timeout, retries, family) + try: + endpoints = resolve_udp_endpoints( + host, discovery_port, family=family) + except OSError: + return _result((), 0, False, 'endpoint_unavailable') + + selector = selectors.DefaultSelector() + routes = _open_routes(endpoints, selector) + if not routes: + selector.close() + return _result((), 0, False, 'endpoint_unavailable') + + token = secrets.token_bytes(8) + started = time.monotonic() + deadline = started + float(timeout) + wait_slice = min(1.0, float(timeout) / (retries + 1)) + attempts = 0 + response_received = False + saw_malformed = False + pinned_route = None + pinned_peer = None + pinned_destination = None + payload = bytearray() + expected_number = 0 + expected_szx = None + expected_etag = _UNSET + expected_content_format = _UNSET + expected_size2 = None + used_mids = set() + + try: + while expected_number < _MAX_BLOCKS: + block = None + sent_for_block = False + for block_attempt in range(retries + 1): + if time.monotonic() >= deadline: + break + mid = secrets.randbits(16) + while mid in used_mids: + mid = (mid + 1) & 0xFFFF + used_mids.add(mid) + request = _build_request( + token, mid, expected_number, expected_szx) + send_routes = [pinned_route] if pinned_route else routes + sent = False + for route in send_routes: + try: + destination = ( + pinned_destination + if route is pinned_route and pinned_destination + else route.endpoint.sockaddr + ) + sent_length = route.sock.sendto( + request, destination) + sent = sent or sent_length == len(request) + except OSError: + continue + attempts += 1 + if not sent: + continue + sent_for_block = True + + now = time.monotonic() + attempt_deadline = ( + deadline if block_attempt == retries + else min(deadline, now + wait_slice) + ) + while True: + remaining = attempt_deadline - time.monotonic() + if remaining <= 0: + break + try: + events = selector.select(remaining) + except (OSError, ValueError): + events = [] + if not events: + break + for key, _mask in events: + route = key.data + try: + datagram, source = route.sock.recvfrom( + _MAX_DATAGRAM_BYTES + 1) + except (BlockingIOError, OSError): + continue + source_host_key = _host_key( + route.endpoint.family, source) + peer_key = _peer_key(route.endpoint.family, source) + if (source_host_key != route.host_key + or peer_key is None): + continue + if pinned_peer is not None and ( + route is not pinned_route + or peer_key != pinned_peer): + continue + if len(datagram) > _MAX_DATAGRAM_BYTES: + saw_malformed = True + continue + + status, candidate, ack_mid = _decode_response_block( + datagram, + token=token, + expected_number=expected_number, + expected_szx=expected_szx, + ) + if ack_mid is not None: + ack = build_coap( + TYPE_ACK, 0, ack_mid, b'', []) + try: + route.sock.sendto(ack, source) + except OSError: + pass + if status in ('ignore', 'duplicate'): + continue + if status == 'malformed': + saw_malformed = True + continue + + response_received = True + if pinned_peer is None: + pinned_route = route + pinned_peer = peer_key + pinned_destination = tuple(source) + block = candidate + break + if block is not None: + break + if block is not None: + break + + if block is None: + if not sent_for_block: + error_code = 'endpoint_unavailable' + elif saw_malformed: + error_code = 'malformed_ocf_response' + else: + error_code = 'no_ocf_response' + return _result( + (), attempts, response_received, error_code) + + if expected_number == 0: + expected_szx = block.szx + expected_etag = block.etag + expected_content_format = block.content_format + expected_size2 = block.size2 + else: + if (block.etag != expected_etag + or block.content_format != expected_content_format): + return _result( + (), attempts, True, 'malformed_ocf_response') + if block.size2 is not None: + if (expected_size2 is not None + and block.size2 != expected_size2): + return _result( + (), attempts, True, 'malformed_ocf_response') + expected_size2 = block.size2 + + if len(payload) + len(block.payload) > _MAX_PAYLOAD_BYTES: + return _result( + (), attempts, True, 'malformed_ocf_response') + payload.extend(block.payload) + if not block.more: + if (expected_size2 is not None + and len(payload) != expected_size2): + return _result( + (), attempts, True, 'malformed_ocf_response') + ports = _secure_ports_from_payload(bytes(payload)) + if ports is None: + return _result( + (), attempts, True, 'malformed_ocf_response') + if not ports: + return _result((), attempts, True, 'no_secure_ports') + return _result(ports, attempts, True) + + expected_number += 1 + + return _result((), attempts, response_received, + 'malformed_ocf_response') + finally: + for route in routes: + try: + selector.unregister(route.sock) + except (KeyError, OSError, ValueError): + pass + try: + route.sock.close() + except OSError: + pass + selector.close() diff --git a/tests/test_coap_wire.py b/tests/test_coap_wire.py index b1e4195..1e1452d 100644 --- a/tests/test_coap_wire.py +++ b/tests/test_coap_wire.py @@ -2,8 +2,16 @@ from smartthings_local.errors import MalformedMessageError from smartthings_local.protocol.coap import ( - build_coap, parse_coap, encode_options, block_value, fmt_code, - TYPE_CON, METHOD_GET, URI_PATH, ACCEPT, CF_CBOR, BLOCK2, + ACCEPT, + CF_CBOR, + METHOD_GET, + TYPE_CON, + URI_PATH, + block_value, + build_coap, + encode_options, + fmt_code, + parse_coap, ) @@ -56,3 +64,29 @@ def test_reserved_option_nibbles_raise_classified_value_error(option_header): parse_coap(datagram) assert isinstance(exc.value, ValueError) + + +@pytest.mark.parametrize( + 'datagram', + ( + b'', + b'\x40\x01\x00', + b'\x80\x01\x00\x01', # unsupported CoAP version + b'\x49\x01\x00\x01' + b'x' * 9, # reserved token length + b'\x44\x01\x00\x01abc', # truncated token + b'\x40\x01\x00\x01\xd0', # truncated extended delta + b'\x40\x01\x00\x01\xe0\x00', + b'\x40\x01\x00\x01\x0d', # truncated extended length + b'\x40\x01\x00\x01\x0e\x00', + b'\x40\x01\x00\x01\x03ab', # truncated option value + b'\x40\x01\x00\x01\xff', # empty payload marker + ), +) +def test_truncated_or_structurally_invalid_datagrams_are_classified(datagram): + with pytest.raises(MalformedMessageError): + parse_coap(datagram) + + +def test_non_bytes_coap_input_is_classified(): + with pytest.raises(MalformedMessageError): + parse_coap('not wire bytes') diff --git a/tests/test_import_isolation.py b/tests/test_import_isolation.py index 8efcdbb..2152dc0 100644 --- a/tests/test_import_isolation.py +++ b/tests/test_import_isolation.py @@ -18,6 +18,7 @@ def test_smartthings_local_imports_without_mqtt_demo_present(tmp_path): import_lines = [ "import smartthings_local.protocol.coap", "import smartthings_local.protocol.dtls_session", + "import smartthings_local.protocol.ocf_discovery", "import smartthings_local.ocf.state_cache", "import smartthings_local.ocf.poll_scheduler", "import smartthings_local.ocf.keepalive", diff --git a/tests/test_ocf_discovery.py b/tests/test_ocf_discovery.py new file mode 100644 index 0000000..2ffb5f3 --- /dev/null +++ b/tests/test_ocf_discovery.py @@ -0,0 +1,863 @@ +"""Public OCF secure-port discovery stays bounded and source-correlated.""" + +import socket +import threading +import traceback +import uuid +from dataclasses import FrozenInstanceError +from types import SimpleNamespace + +import cbor2 +import pytest + +from smartthings_local.errors import EndpointError +from smartthings_local.protocol import ocf_discovery as discovery +from smartthings_local.protocol.coap import ( + ACCEPT, + BLOCK2, + CF_CBOR, + CONTENT_FORMAT, + METHOD_GET, + SIZE2, + TYPE_ACK, + TYPE_CON, + TYPE_NON, + TYPE_RST, + URI_PATH, + URI_QUERY, + block_value, + build_coap, + parse_coap, +) + + +def _doxm_link(port=49872): + return { + 'href': '/oic/sec/doxm', + 'rt': ['oic.r.doxm'], + 'p': {'sec': True, 'port': port}, + } + + +def _payload(*links, padding=''): + value = {'links': list(links)} + if padding: + value['padding'] = padding + return cbor2.dumps(value) + + +def _identity_payload(device_id, *links): + return cbor2.dumps({'di': device_id, 'links': list(links)}) + + +class _FakeMulticastSocket: + def __init__(self, response_factory): + self.response_factory = response_factory + self.responses = [] + self.requests = [] + self.socket_options = [] + self.bound = None + self.recv_count = 0 + self.closed = False + + def setsockopt(self, level, option, value): + self.socket_options.append((level, option, value)) + + def bind(self, address): + self.bound = address + + def setblocking(self, _blocking): + pass + + def sendto(self, request, destination): + self.requests.append((request, destination)) + self.responses.extend( + self.response_factory(request, len(self.requests))) + return len(request) + + def recvfrom(self, _size): + if not self.responses: + raise BlockingIOError + self.recv_count += 1 + return self.responses.pop(0) + + def close(self): + self.closed = True + + +class _FakeSelector: + def __init__(self): + self.sock = None + + def register(self, sock, _events): + self.sock = sock + + def unregister(self, sock): + assert sock is self.sock + + def select(self, _timeout): + if self.sock.responses: + return [(SimpleNamespace(data=None), 1)] + return [] + + def close(self): + pass + + +def _option_map(options): + result = {} + for number, value in options: + result.setdefault(number, []).append(value) + return result + + +def _uint_bytes(value): + length = max(1, (value.bit_length() + 7) // 8) + return value.to_bytes(length, 'big') + + +def test_extracts_only_valid_legacy_and_modern_secure_doxm_ports(): + links = [ + _doxm_link(), + { + 'href': '/oic/sec/doxm', + 'rt': 'oic.r.doxm', + 'p': {'sec': True, 'port': 49872}, + 'eps': [ + {'ep': 'coaps://192.0.2.20:49873'}, + {'ep': 'coaps://[2001:db8::20]'}, + {'ep': 'coap://192.0.2.20:49874'}, + {'ep': 'coaps+tcp://192.0.2.20:49875'}, + {'ep': 'coaps://user:secret@192.0.2.20:49876'}, + {'ep': 'coaps://192.0.2.20:49877/path'}, + ], + }, + { + 'href': '/oic/sec/doxm', + 'rt': ['oic.r.doxm'], + 'p': {'sec': False, 'port': 49901}, + }, + { + 'href': '/oic/sec/pstat', + 'rt': ['oic.r.pstat'], + 'p': {'sec': True, 'port': 49902}, + }, + _doxm_link(True), + _doxm_link('49878'), + _doxm_link(0), + _doxm_link(65536), + ] + + assert discovery._secure_ports_from_payload(_payload(*links)) == ( + 49872, 49873, 5684) + + +def test_direct_link_array_is_supported_and_port_count_is_bounded(): + links = [_doxm_link(port) for port in range(49870, 49880)] + + ports = discovery._secure_ports_from_payload(cbor2.dumps(links)) + + assert ports == tuple(range(49870, 49878)) + + +@pytest.mark.parametrize( + 'payload', + ( + b'not-cbor', + cbor2.dumps({'links': 'not-a-list'}), + cbor2.dumps([{'links': [None]}]), + cbor2.dumps({'links': []}) + cbor2.dumps(1), + ), +) +def test_malformed_or_trailing_cbor_is_rejected(payload): + assert discovery._secure_ports_from_payload(payload) is None + + +def test_valid_directory_without_secure_doxm_has_no_ports(): + value = {'links': [{'href': '/oic/d', 'rt': ['oic.wk.d']}]} + + assert discovery._secure_ports_from_payload(cbor2.dumps(value)) == () + + +def test_multicast_extraction_is_identity_and_response_source_scoped(): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + other = uuid.UUID('00000000-0000-0000-0000-000000000000') + source = socket.inet_pton(socket.AF_INET, '192.0.2.20') + payload = cbor2.dumps([ + { + 'di': str(other), + 'links': [_doxm_link(49901)], + }, + { + 'di': target.bytes, + 'links': [ + _doxm_link(49872), + { + 'href': '/oic/sec/doxm', + 'rt': 'oic.r.doxm', + 'eps': [ + {'ep': 'coaps://192.0.2.20:49873'}, + {'ep': 'coaps://192.0.2.21:49874'}, + {'ep': 'coap://192.0.2.20:49875'}, + ], + }, + ], + }, + ]) + + status, ports = discovery._target_secure_ports_from_payload( + payload, target, source) + + assert status == 'target' + assert ports == (49872, 49873) + + +def test_multicast_target_uuid_is_normalized_but_never_fuzzy_matched(): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + source = socket.inet_pton(socket.AF_INET, '192.0.2.20') + + assert discovery._normalize_uuid( + f'URN:UUID:{str(target).upper()}') == target + assert discovery._normalize_uuid(target.bytes) == target + assert discovery._target_secure_ports_from_payload( + _identity_payload(str(target) + '0', _doxm_link()), + target, + source, + ) == ('absent', ()) + + +def test_multicast_duplicate_target_containers_fail_closed(): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + source = socket.inet_pton(socket.AF_INET, '192.0.2.20') + payload = cbor2.dumps([ + {'di': str(target), 'links': [_doxm_link(49872)]}, + {'di': str(target), 'links': [_doxm_link(49873)]}, + ]) + + assert discovery._target_secure_ports_from_payload( + payload, target, source) == ('malformed', ()) + + +def test_multicast_never_treats_the_identity_container_as_a_link(): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + source = socket.inet_pton(socket.AF_INET, '192.0.2.20') + payload = cbor2.dumps({ + 'di': str(target), + **_doxm_link(49872), + }) + + assert discovery._target_secure_ports_from_payload( + payload, target, source) == ('malformed', ()) + + +def test_response_type_and_token_matrix(): + token = b'12345678' + request_mid = 0x1234 + payload = _payload(_doxm_link()) + + non_response = build_coap( + TYPE_NON, 0x45, 0x7000, token, [], payload) + status, block, ack_mid = discovery._decode_response_block( + non_response, + token=token, + expected_number=0, + expected_szx=None, + ) + assert status == 'block' + assert block.payload == payload + assert ack_mid is None + + con_response = build_coap( + TYPE_CON, 0x45, 0x7001, token, [], payload) + status, _block, ack_mid = discovery._decode_response_block( + con_response, + token=token, + expected_number=0, + expected_szx=None, + ) + assert status == 'block' + assert ack_mid == 0x7001 + + ack_response = build_coap( + TYPE_ACK, 0x45, request_mid, token, [], payload) + assert discovery._decode_response_block( + ack_response, + token=token, + expected_number=0, + expected_szx=None, + )[0] == 'ignore' + + wrong_token = build_coap( + TYPE_NON, 0x45, 0x7002, b'87654321', [], payload) + assert discovery._decode_response_block( + wrong_token, + token=token, + expected_number=0, + expected_szx=None, + )[0] == 'ignore' + + empty_ack = build_coap(TYPE_ACK, 0, request_mid, b'', []) + assert discovery._decode_response_block( + empty_ack, + token=token, + expected_number=0, + expected_szx=None, + )[0] == 'ignore' + + reset = build_coap(TYPE_RST, 0, request_mid, b'', []) + assert discovery._decode_response_block( + reset, + token=token, + expected_number=0, + expected_szx=None, + )[0] == 'ignore' + + +@pytest.mark.parametrize( + 'options,payload,expected_number,expected_szx', + ( + ([(BLOCK2, b'\x00'), (BLOCK2, b'\x00')], b'x', 0, None), + ([(BLOCK2, b'\x07')], b'x', 0, None), + ([(BLOCK2, block_value(1, 0, 0))], b'x', 0, None), + ([(BLOCK2, block_value(0, 1, 0))], b'x' * 15, 0, None), + ([(BLOCK2, block_value(1, 0, 1))], b'x', 1, 0), + ([(CONTENT_FORMAT, b'\x00')], b'x', 0, None), + ([(SIZE2, _uint_bytes(65537))], b'x', 0, None), + ), +) +def test_malformed_blockwise_metadata_is_rejected( + options, payload, expected_number, expected_szx): + token = b'12345678' + response = build_coap( + TYPE_NON, 0x45, 0x7000, token, options, payload) + + status, _block, _ack_mid = discovery._decode_response_block( + response, + token=token, + expected_number=expected_number, + expected_szx=expected_szx, + ) + + assert status == 'malformed' + + +def test_dynamic_source_port_and_two_block_response_are_supported(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + responder.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + responder.settimeout(2.0) + assert listener.getsockname()[1] != responder.getsockname()[1] + + body = _payload(_doxm_link(), padding='x' * 300) + block_size = 256 + assert block_size < len(body) <= block_size * 2 + etag = b'test' + errors = [] + + def respond(): + try: + first_request, client = listener.recvfrom(8192) + mtype, code, first_mid, token, options, request_payload = \ + parse_coap(first_request) + option_map = _option_map(options) + assert mtype == TYPE_NON + assert code == METHOD_GET + assert len(token) == 8 + assert request_payload == b'' + assert option_map[URI_PATH] == [b'oic', b'res'] + assert option_map[URI_QUERY] == [b'rt=oic.r.doxm'] + assert option_map[ACCEPT] == [CF_CBOR] + assert BLOCK2 not in option_map + + common = [ + (CONTENT_FORMAT, CF_CBOR), + (discovery._ETAG, etag), + (SIZE2, _uint_bytes(len(body))), + ] + first_response = build_coap( + TYPE_CON, + 0x45, + 0x7001, + token, + [*common, (BLOCK2, block_value(0, 1, 4))], + body[:block_size], + ) + responder.sendto(first_response, client) + first_ack, ack_peer = responder.recvfrom(8192) + ack_type, ack_code, ack_mid, ack_token, ack_options, ack_body = \ + parse_coap(first_ack) + assert ack_peer == client + assert (ack_type, ack_code, ack_mid) == (TYPE_ACK, 0, 0x7001) + assert ack_token == b'' and ack_options == [] and ack_body == b'' + + second_request, second_client = responder.recvfrom(8192) + mtype, code, second_mid, second_token, options, request_payload = \ + parse_coap(second_request) + option_map = _option_map(options) + assert second_client == client + assert (mtype, code) == (TYPE_NON, METHOD_GET) + assert second_token == token + assert second_mid != first_mid + assert request_payload == b'' + assert option_map[BLOCK2] == [block_value(1, 0, 4)] + + second_response = build_coap( + TYPE_CON, + 0x45, + 0x7002, + token, + [*common, (BLOCK2, block_value(1, 0, 4))], + body[block_size:], + ) + responder.sendto(second_response, client) + second_ack, _ack_peer = responder.recvfrom(8192) + assert parse_coap(second_ack)[:4] == ( + TYPE_ACK, 0, 0x7002, b'') + except Exception as exc: # pragma: no cover - asserted below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=1.5, + retries=1, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=3.0) + listener.close() + responder.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (49872,) + assert result.response_received + assert result.error_code is None + assert result.attempts == 2 + + +def test_block_transfer_pins_first_valid_response_source_port(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + first_responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + other_responder = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + first_responder.bind(('127.0.0.1', 0)) + other_responder.bind(('127.0.0.1', 0)) + listener.settimeout(1.0) + first_responder.settimeout(1.0) + body = _payload(_doxm_link(), padding='x' * 10) + assert 64 < len(body) <= 128 + errors = [] + + def respond(): + try: + request, client = listener.recvfrom(8192) + _mtype, _code, _mid, token, _options, _body = \ + parse_coap(request) + first_responder.sendto( + build_coap( + TYPE_NON, 0x45, 0x7101, token, + [(BLOCK2, block_value(0, 1, 2))], body[:64]), + client, + ) + request, second_client = first_responder.recvfrom(8192) + assert second_client == client + other_responder.sendto( + build_coap( + TYPE_NON, 0x45, 0x7102, token, + [(BLOCK2, block_value(1, 0, 2))], body[64:]), + client, + ) + except Exception as exc: # pragma: no cover - asserted below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.4, + retries=0, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + first_responder.close() + other_responder.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == () + assert result.response_received + assert result.error_code == 'no_ocf_response' + + +def test_retry_keeps_token_and_changes_message_id(): + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(('127.0.0.1', 0)) + listener.settimeout(2.0) + errors = [] + + def respond(): + try: + first, client = listener.recvfrom(8192) + second, second_client = listener.recvfrom(8192) + first_parsed = parse_coap(first) + second_parsed = parse_coap(second) + assert second_client == client + assert first_parsed[0] == second_parsed[0] == TYPE_NON + assert first_parsed[3] == second_parsed[3] + assert first_parsed[2] != second_parsed[2] + listener.sendto( + build_coap( + TYPE_NON, + 0x45, + 0x7201, + second_parsed[3], + [], + _payload(_doxm_link()), + ), + client, + ) + except Exception as exc: # pragma: no cover - asserted below + errors.append(exc) + + thread = threading.Thread(target=respond) + thread.start() + try: + result = discovery.discover_ocf_secure_ports( + '127.0.0.1', + discovery_port=listener.getsockname()[1], + timeout=0.5, + retries=1, + family=socket.AF_INET, + ) + finally: + thread.join(timeout=2.0) + listener.close() + + assert not thread.is_alive() + assert errors == [] + assert result.ports == (49872,) + assert result.attempts == 2 + + +def test_ipv4_multicast_discovery_requires_two_stable_identity_rounds( + monkeypatch): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + other = uuid.UUID('00000000-0000-0000-0000-000000000000') + target_source = ('192.0.2.20', 41000) + other_source = ('192.0.2.21', 42000) + + def response_factory(request, round_number): + mtype, code, mid, token, options, payload = parse_coap(request) + option_map = _option_map(options) + assert (mtype, code) == (TYPE_NON, METHOD_GET) + assert payload == b'' + assert option_map[URI_PATH] == [b'oic', b'res'] + assert option_map[URI_QUERY] == [b'rt=oic.r.doxm'] + assert option_map[ACCEPT] == [CF_CBOR] + target_payload = _identity_payload( + str(target), + _doxm_link(49872), + { + 'href': '/oic/sec/doxm', + 'rt': ['oic.r.doxm'], + 'eps': [ + {'ep': 'coaps://192.0.2.20:49873'}, + {'ep': 'coaps://192.0.2.21:49874'}, + ], + }, + ) + unrelated_payload = _identity_payload( + str(other), _doxm_link(49901)) + return [ + ( + build_coap( + TYPE_NON, 0x45, mid + 1, b'badtoken', [], + target_payload), + target_source, + ), + ( + build_coap( + TYPE_CON, 0x45, mid + 2, token, [], target_payload), + target_source, + ), + ( + build_coap( + TYPE_NON, 0x45, mid + 3, token, [], + unrelated_payload), + other_source, + ), + ( + build_coap( + TYPE_NON, 0x45, mid + 4, token, [], target_payload), + target_source, + ), + ] + + fake_socket = _FakeMulticastSocket(response_factory) + socket_calls = [] + + def open_socket(*args): + socket_calls.append(args) + return fake_socket + + def unexpected_unicast(*_args, **_kwargs): + pytest.fail('explicit multicast discovery invoked unicast fallback') + + monkeypatch.setattr(discovery.socket, 'socket', open_socket) + monkeypatch.setattr( + discovery, 'discover_ocf_secure_ports', unexpected_unicast) + monkeypatch.setattr( + discovery.selectors, 'DefaultSelector', _FakeSelector) + + result = discovery.discover_ocf_secure_ports_multicast( + f'urn:uuid:{target}', + interface_address='192.0.2.10', + round_timeout=0.1, + ) + + assert result.address == target_source[0] + assert result.ports == (49872, 49873) + assert result.rounds == 2 + assert result.responses == 4 + assert result.error_code is None + assert result.found + assert len(socket_calls) == 1 + assert len(fake_socket.requests) == 2 + assert fake_socket.requests[0][1] == ( + discovery._IPV4_OCF_MULTICAST_GROUP, 5683) + first = parse_coap(fake_socket.requests[0][0]) + second = parse_coap(fake_socket.requests[1][0]) + assert first[2] != second[2] + assert first[3] != second[3] + assert fake_socket.bound == ('192.0.2.10', 0) + assert ( + socket.IPPROTO_IP, + socket.IP_MULTICAST_IF, + socket.inet_pton(socket.AF_INET, '192.0.2.10'), + ) in fake_socket.socket_options + assert ( + socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1, + ) in fake_socket.socket_options + assert ( + socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0, + ) in fake_socket.socket_options + assert fake_socket.closed + rendered = repr(result) + assert target_source[0] not in rendered + assert str(target) not in rendered + assert '49872' not in rendered + + +def test_ipv4_multicast_discovery_rejects_changing_target_source( + monkeypatch): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + sources = [('192.0.2.20', 41000), ('192.0.2.21', 42000)] + + def response_factory(request, round_number): + _mtype, _code, mid, token, _options, _payload = parse_coap(request) + payload = _identity_payload(str(target), _doxm_link(49872)) + return [( + build_coap(TYPE_NON, 0x45, mid + 1, token, [], payload), + sources[round_number - 1], + )] + + fake_socket = _FakeMulticastSocket(response_factory) + monkeypatch.setattr( + discovery.socket, 'socket', lambda *_args: fake_socket) + monkeypatch.setattr( + discovery.selectors, 'DefaultSelector', _FakeSelector) + + result = discovery.discover_ocf_secure_ports_multicast( + target, + interface_address='192.0.2.10', + round_timeout=0.1, + ) + + assert not result.found + assert result.address is None + assert result.ports == () + assert result.error_code == 'ambiguous_target' + + +def test_ipv4_multicast_no_response_stops_without_unicast_fallback( + monkeypatch): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + fake_socket = _FakeMulticastSocket( + lambda _request, _round_number: []) + + def unexpected_unicast(*_args, **_kwargs): + pytest.fail('explicit multicast discovery invoked unicast fallback') + + monkeypatch.setattr( + discovery.socket, 'socket', lambda *_args: fake_socket) + monkeypatch.setattr( + discovery.selectors, 'DefaultSelector', _FakeSelector) + monkeypatch.setattr( + discovery, 'discover_ocf_secure_ports', unexpected_unicast) + + result = discovery.discover_ocf_secure_ports_multicast( + target, + interface_address='192.0.2.10', + round_timeout=0.1, + ) + + assert not result.found + assert result.rounds == 2 + assert result.responses == 0 + assert result.error_code == 'no_ocf_response' + assert len(fake_socket.requests) == 2 + + +def test_ipv4_multicast_rounds_are_time_and_datagram_bounded(monkeypatch): + target = uuid.UUID('11111111-2222-3333-4444-555555555555') + source = ('192.0.2.20', 41000) + + def response_factory(request, _round_number): + _mtype, _code, mid, _token, _options, _payload = parse_coap(request) + return [ + ( + build_coap( + TYPE_NON, + 0x45, + (mid + offset + 1) & 0xFFFF, + b'wrong-token', + [], + _identity_payload(str(target), _doxm_link()), + ), + source, + ) + for offset in range( + discovery._MAX_MULTICAST_RESPONSES_PER_ROUND + 1) + ] + + class RecordingSelector(_FakeSelector): + def __init__(self): + super().__init__() + self.timeouts = [] + + def select(self, timeout): + self.timeouts.append(timeout) + return super().select(timeout) + + fake_socket = _FakeMulticastSocket(response_factory) + fake_selector = RecordingSelector() + monkeypatch.setattr( + discovery.socket, 'socket', lambda *_args: fake_socket) + monkeypatch.setattr( + discovery.selectors, 'DefaultSelector', lambda: fake_selector) + + round_timeout = 0.1 + result = discovery.discover_ocf_secure_ports_multicast( + target, + interface_address='192.0.2.10', + round_timeout=round_timeout, + ) + + assert not result.found + assert result.rounds == 2 + assert len(fake_socket.requests) == 2 + assert fake_socket.recv_count == ( + discovery._MAX_MULTICAST_RESPONSES_PER_ROUND * 2) + assert len(fake_selector.timeouts) == fake_socket.recv_count + assert all( + 0 < timeout <= round_timeout for timeout in fake_selector.timeouts) + + +@pytest.mark.parametrize( + ('target_uuid', 'interface_address', 'round_timeout', 'error_type'), + ( + ('not-a-uuid', '192.0.2.10', 1.0, ValueError), + (object(), '192.0.2.10', 1.0, TypeError), + ('11111111-2222-3333-4444-555555555555', 'not-an-ip', 1.0, + ValueError), + ('11111111-2222-3333-4444-555555555555', 1, 1.0, TypeError), + ('11111111-2222-3333-4444-555555555555', + discovery._IPV4_OCF_MULTICAST_GROUP, 1.0, + ValueError), + ('11111111-2222-3333-4444-555555555555', '192.0.2.10', 30.1, + ValueError), + ), +) +def test_invalid_multicast_options_fail_before_network( + target_uuid, interface_address, round_timeout, error_type): + with pytest.raises(error_type): + discovery.discover_ocf_secure_ports_multicast( + target_uuid, + interface_address=interface_address, + round_timeout=round_timeout, + ) + + +def test_resolution_failure_and_result_repr_are_redacted(monkeypatch): + remote_host = 'private-appliance.invalid' + + def fail(host, port, *, family): + assert host == remote_host + assert port == 5683 + assert family == socket.AF_INET6 + raise EndpointError() + + def unexpected_multicast(*_args, **_kwargs): + pytest.fail('known-host discovery invoked multicast fallback') + + monkeypatch.setattr(discovery, 'resolve_udp_endpoints', fail) + monkeypatch.setattr( + discovery, + 'discover_ocf_secure_ports_multicast', + unexpected_multicast, + ) + + result = discovery.discover_ocf_secure_ports( + remote_host, family=socket.AF_INET6) + rendered = repr(result) + ''.join( + traceback.format_exception(EndpointError())) + + assert result.error_code == 'endpoint_unavailable' + assert result.attempts == 0 + assert remote_host not in rendered + assert '49872' not in repr( + discovery.OcfSecurePortDiscoveryResult((49872,), 1, True)) + + +def test_result_is_immutable_and_ipv6_scope_is_part_of_source_identity(): + result = discovery.OcfSecurePortDiscoveryResult((49872,), 1, True) + + with pytest.raises(FrozenInstanceError): + result.attempts = 2 + assert discovery._host_key( + socket.AF_INET, ('192.0.2.20', 5683)) != discovery._host_key( + socket.AF_INET, ('192.0.2.21', 5683)) + assert discovery._host_key( + socket.AF_INET6, ('2001:db8::20', 5683, 0, 7)) != \ + discovery._host_key( + socket.AF_INET6, ('2001:db8::20', 5683, 0, 8)) + + +@pytest.mark.parametrize( + ('keyword', 'value', 'error_type'), + ( + ('discovery_port', 0, ValueError), + ('discovery_port', True, TypeError), + ('timeout', 0, ValueError), + ('timeout', float('nan'), ValueError), + ('timeout', True, TypeError), + ('retries', 5, ValueError), + ('retries', True, TypeError), + ('family', 9999, ValueError), + ('family', 'AF_INET', TypeError), + ), +) +def test_invalid_options_fail_before_network(keyword, value, error_type): + with pytest.raises(error_type): + discovery.discover_ocf_secure_ports( + '192.0.2.20', **{keyword: value}) diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index f40e5b4..9867ec2 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -12,6 +12,12 @@ PskAuth, ) from smartthings_local.protocol.dtls_session import DtlsCoapSession +from smartthings_local.protocol.ocf_discovery import ( + OcfMulticastSecurePortDiscoveryResult, + OcfSecurePortDiscoveryResult, + discover_ocf_secure_ports, + discover_ocf_secure_ports_multicast, +) def _assert_compatible_signature(callable_object, expected: list[str]) -> None: @@ -134,3 +140,39 @@ def test_observe_refresh_task_keeps_current_consumer_surface(): ObserveRefreshTask.run_forever, ["self", "stop"], ) + + +def test_ocf_secure_port_discovery_has_a_small_composable_surface(): + _assert_compatible_signature(discover_ocf_secure_ports, ["host"]) + result = OcfSecurePortDiscoveryResult( + ports=(5684,), + attempts=1, + response_received=True, + ) + + assert result.found + assert result.ports == (5684,) + + +def test_identity_aware_multicast_discovery_is_explicitly_ipv4_scoped(): + _assert_compatible_signature( + discover_ocf_secure_ports_multicast, + ['target_uuid', 'interface_address'], + ) + interface_parameter = inspect.signature( + discover_ocf_secure_ports_multicast).parameters['interface_address'] + assert interface_parameter.kind is inspect.Parameter.KEYWORD_ONLY + assert interface_parameter.default is inspect.Parameter.empty + assert inspect.signature( + discover_ocf_secure_ports_multicast + ).parameters['round_timeout'].default == 6.0 + + result = OcfMulticastSecurePortDiscoveryResult( + address='192.0.2.20', + ports=(5684,), + rounds=2, + responses=2, + ) + assert result.found + assert result.address == '192.0.2.20' + assert result.ports == (5684,)