diff --git a/README.md b/README.md index a033d85..23dabb0 100644 --- a/README.md +++ b/README.md @@ -262,6 +262,34 @@ The resolver retains candidate order and the socket setup tries the next candidate after a family, bind, or connect failure. Resolution and socket setup failures raise the redacted `EndpointError` documented above. +### Dynamic plaintext OCF response ports + +Some OCF devices listen for multicast discovery on UDP 5683 but send their +response from a different port that changes after a power cycle. A caller that +already knows the device's IPv4 address can discover those plaintext response +port candidates on one explicit LAN interface: + +```python +from smartthings_local.protocol.ocf_multicast import ( + discover_ocf_responder_ports, +) + +result = discover_ocf_responder_ports( + "192.0.2.20", + interface_address="192.0.2.10", +) +for discovery_port in result.ports: + pass # use for a bounded, source-bound /oic/res lookup +``` + +The call sends unfiltered current OCF and legacy IoTivity directory requests, +plus a legacy DOXM-filtered fallback, under one deadline. It accepts only +token-correlated replies from the expected host, closes its multicast socket +before returning, and omits addresses and ports from its result representation. +Returned ports are unauthenticated candidates, not DTLS endpoints; directory +parsing, DTLS liveness, and authenticated device identity remain separate +checks. + For a full worked integration, the higher-level `smartthings_local.ocf` layer (`StateCache`, `PollScheduler`, `KeepaliveTask`, `ObserveRefreshTask`) coordinates tiered polling and OBSERVE on top of a session. The MQTT bridge demo below wires all of it together. ### What the demo bridge gives you diff --git a/smartthings_local/protocol/ocf_multicast.py b/smartthings_local/protocol/ocf_multicast.py new file mode 100644 index 0000000..d49a485 --- /dev/null +++ b/smartthings_local/protocol/ocf_multicast.py @@ -0,0 +1,322 @@ +"""Bounded discovery of a known host's plaintext OCF response port. + +Some OCF devices receive multicast discovery on UDP 5683 but reply from an +ephemeral port. This module records only token-correlated response ports from +the caller's expected IPv4 address. The results are candidates: callers still +need directory parsing, a DTLS probe, and authenticated identity validation. +""" + +from __future__ import annotations + +import ipaddress +import math +import secrets +import selectors +import socket +import time +from dataclasses import dataclass + +from ..errors import MalformedMessageError +from .coap import ( + ACCEPT, + CF_CBOR, + METHOD_GET, + TYPE_ACK, + TYPE_CON, + TYPE_NON, + URI_PATH, + URI_QUERY, + build_coap, + parse_coap, +) + +__all__ = [ + "OcfResponderPortDiscoveryResult", + "discover_ocf_responder_ports", +] + +_OCF_MULTICAST_GROUP = socket.inet_ntoa(bytes((224, 0, 1, 187))) +_OCF_DISCOVERY_PORT = 5683 +_OCF_CBOR = (10_000).to_bytes(2, "big") +_OCF_CONTENT_FORMAT_VERSION = 2049 +_OCF_VERSION_1_0 = (2048).to_bytes(2, "big") +_CONTENT = 0x45 +_MAX_DATAGRAM_BYTES = 8192 +_MAX_DATAGRAMS_PER_ROUND = 64 +_MAX_PORTS = 8 + + +@dataclass(frozen=True, slots=True, repr=False) +class OcfResponderPortDiscoveryResult: + """Redacted result of one known-host multicast discovery operation.""" + + ports: tuple[int, ...] + attempts: int + responses: int + error_code: str | None = None + + @property + def found(self) -> bool: + """Return whether at least one response port was discovered.""" + return bool(self.ports) + + def __repr__(self) -> str: + return ( + "OcfResponderPortDiscoveryResult(" + f"found={self.found!r}, port_count={len(self.ports)}, " + f"attempts={self.attempts}, responses={self.responses}, " + f"error_code={self.error_code!r})" + ) + + +def _validate_address(value: object, name: str) -> tuple[str, bytes]: + if not isinstance(value, str): + raise TypeError(f"{name} must be an IPv4 address string") + try: + address = ipaddress.IPv4Address(value) + except ipaddress.AddressValueError as exc: + raise ValueError(f"{name} must be a valid IPv4 address") from exc + if address.is_multicast or address.is_unspecified or address.is_reserved: + raise ValueError(f"{name} must be a unicast IPv4 address") + return str(address), address.packed + + +def _validate_options( + *, + discovery_port: object, + timeout: object, + rounds: object, +) -> tuple[int, float, int]: + 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") + timeout_value = float(timeout) + if not math.isfinite(timeout_value) or not 0 < timeout_value <= 30: + raise ValueError("timeout must be greater than zero and at most 30") + if isinstance(rounds, bool) or not isinstance(rounds, int): + raise TypeError("rounds must be an integer") + if not 1 <= rounds <= 4: + raise ValueError("rounds must be between one and four") + return discovery_port, timeout_value, rounds + + +def _request( + token: bytes, + message_id: int, + *, + versioned: bool, + filtered: bool, +) -> bytes: + options = [ + (URI_PATH, b"oic"), + (URI_PATH, b"res"), + (ACCEPT, _OCF_CBOR if versioned else CF_CBOR), + ] + if filtered: + options.append((URI_QUERY, b"rt=oic.r.doxm")) + if versioned: + options.append((_OCF_CONTENT_FORMAT_VERSION, _OCF_VERSION_1_0)) + return build_coap(TYPE_NON, METHOD_GET, message_id, token, options) + + +def _result( + ports: tuple[int, ...], + attempts: int, + responses: int, + error_code: str | None = None, +) -> OcfResponderPortDiscoveryResult: + return OcfResponderPortDiscoveryResult( + ports=ports, + attempts=attempts, + responses=responses, + error_code=error_code, + ) + + +def discover_ocf_responder_ports( + target_address: str, + *, + interface_address: str, + discovery_port: int = _OCF_DISCOVERY_PORT, + timeout: float = 3.0, + rounds: int = 2, +) -> OcfResponderPortDiscoveryResult: + """Find plaintext OCF response ports for one known IPv4 host. + + Each round sends modern OCF and legacy IoTivity NON requests to the + link-local multicast group. Only a 2.05 response with a request token and + the exact target source address contributes a candidate. One monotonic + deadline bounds all rounds, and every socket is closed before return. + """ + + _target_address, target_key = _validate_address(target_address, "target_address") + interface_address, interface_key = _validate_address( + interface_address, "interface_address" + ) + discovery_port, timeout, rounds = _validate_options( + discovery_port=discovery_port, + timeout=timeout, + rounds=rounds, + ) + + try: + selector = selectors.DefaultSelector() + except (OSError, ValueError): + return _result((), 0, 0, "interface_unavailable") + active = None + try: + active = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP) + active.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, interface_key) + active.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1) + active.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 0) + active.bind((interface_address, 0)) + active.setblocking(False) + selector.register(active, selectors.EVENT_READ) + except (OSError, ValueError): + if active is not None: + try: + active.close() + except OSError: + pass + selector.close() + return _result((), 0, 0, "interface_unavailable") + + started = time.monotonic() + deadline = started + timeout + accepted_tokens: set[bytes] = set() + observations: set[tuple[bytes, int]] = set() + ports: list[int] = [] + seen_ports: set[int] = set() + attempts = 0 + responses = 0 + too_many_ports = False + + try: + for round_number in range(rounds): + if time.monotonic() >= deadline: + break + # Preserve the unfiltered modern and legacy requests used by the + # installed appliance generations. Older media firmware can omit + # usable endpoint policy from its large unfiltered directory but + # answer the smaller legacy DOXM-filtered lookup, so send that as + # a third bounded fallback rather than narrowing every request. + for versioned, filtered in ( + (True, False), + (False, False), + (False, True), + ): + token = secrets.token_bytes(8) + while token in accepted_tokens: + token = secrets.token_bytes(8) + accepted_tokens.add(token) + request = _request( + token, + secrets.randbits(16), + versioned=versioned, + filtered=filtered, + ) + try: + sent = active.sendto( + request, + (_OCF_MULTICAST_GROUP, discovery_port), + ) + except OSError: + continue + attempts += 1 + if sent != len(request): + continue + + round_deadline = started + timeout * (round_number + 1) / rounds + datagrams = 0 + while datagrams < _MAX_DATAGRAMS_PER_ROUND: + remaining = min(deadline, round_deadline) - time.monotonic() + if remaining <= 0: + break + try: + events = selector.select(remaining) + except (OSError, ValueError): + break + if not events: + break + try: + datagram, source = active.recvfrom(_MAX_DATAGRAM_BYTES + 1) + except (BlockingIOError, OSError): + continue + datagrams += 1 + if len(datagram) > _MAX_DATAGRAM_BYTES: + continue + if ( + len(datagram) < 4 + or datagram[0] >> 6 != 1 + or datagram[0] & 0x0F > 8 + or 4 + (datagram[0] & 0x0F) > len(datagram) + ): + continue + if not isinstance(source, tuple) or len(source) != 2: + continue + source_host, source_port = source + if not isinstance(source_host, str): + continue + try: + source_key = socket.inet_pton(socket.AF_INET, source_host) + except OSError: + continue + if source_key != target_key: + continue + if ( + isinstance(source_port, bool) + or not isinstance(source_port, int) + or not 1 <= source_port <= 65535 + ): + continue + try: + message_type, code, mid, token, _options, payload = parse_coap( + datagram + ) + except (IndexError, ValueError, MalformedMessageError): + continue + if ( + token not in accepted_tokens + or code != _CONTENT + or message_type not in (TYPE_NON, TYPE_CON) + or not payload + ): + continue + if message_type == TYPE_CON: + try: + active.sendto(build_coap(TYPE_ACK, 0, mid, b"", []), source) + except OSError: + pass + observation = (token, source_port) + if observation in observations: + continue + observations.add(observation) + responses += 1 + if source_port in seen_ports: + continue + if len(ports) >= _MAX_PORTS: + too_many_ports = True + continue + seen_ports.add(source_port) + ports.append(source_port) + + if too_many_ports: + return _result((), attempts, responses, "ambiguous_response") + if ports: + return _result(tuple(ports), attempts, responses) + if attempts == 0: + return _result((), attempts, responses, "interface_unavailable") + return _result((), attempts, responses, "no_response") + finally: + try: + selector.unregister(active) + except (KeyError, OSError, ValueError): + pass + try: + active.close() + except OSError: + pass + selector.close() diff --git a/tests/test_import_isolation.py b/tests/test_import_isolation.py index 8efcdbb..a6d50a2 100644 --- a/tests/test_import_isolation.py +++ b/tests/test_import_isolation.py @@ -17,6 +17,7 @@ def test_smartthings_local_imports_without_mqtt_demo_present(tmp_path): import_lines = [ "import smartthings_local.protocol.coap", + "import smartthings_local.protocol.ocf_multicast", "import smartthings_local.protocol.dtls_session", "import smartthings_local.ocf.state_cache", "import smartthings_local.ocf.poll_scheduler", diff --git a/tests/test_ocf_multicast.py b/tests/test_ocf_multicast.py new file mode 100644 index 0000000..a3ff544 --- /dev/null +++ b/tests/test_ocf_multicast.py @@ -0,0 +1,392 @@ +"""Known-host OCF multicast responder-port discovery tests.""" + +from __future__ import annotations + +from types import SimpleNamespace + +import pytest + +from smartthings_local.protocol.coap import ( + ACCEPT, + TYPE_CON, + TYPE_NON, + URI_PATH, + URI_QUERY, + build_coap, + parse_coap, +) +from smartthings_local.protocol.ocf_multicast import ( + _OCF_MULTICAST_GROUP, + OcfResponderPortDiscoveryResult, + discover_ocf_responder_ports, +) + +_TARGET = "192.0.2.20" +_OTHER = "192.0.2.21" +_INTERFACE = "192.0.2.10" + + +class _FakeSocket: + def __init__(self, responder=None): + self.responder = responder + self.incoming = [] + self.sent = [] + self.options = [] + self.bound = None + self.blocking = None + self.closed = False + + def setsockopt(self, level, option, value): + self.options.append((level, option, value)) + + def bind(self, address): + self.bound = address + + def setblocking(self, enabled): + self.blocking = enabled + + def sendto(self, datagram, destination): + self.sent.append((datagram, destination)) + if self.responder is not None: + self.incoming.extend(self.responder(datagram, destination)) + return len(datagram) + + def recvfrom(self, _size): + return self.incoming.pop(0) + + def close(self): + self.closed = True + + +class _FakeSelector: + def __init__(self, active): + self.active = active + self.closed = False + + def register(self, *_args): + return None + + def unregister(self, *_args): + return None + + def select(self, _timeout): + if not self.active.incoming: + return [] + return [(SimpleNamespace(fileobj=self.active), selectors_event_read())] + + def close(self): + self.closed = True + + +def selectors_event_read(): + return 1 + + +def _response( + datagram, _destination=None, *, host=_TARGET, port=43123, message_type=TYPE_NON +): + _mtype, _code, mid, token, _options, _payload = parse_coap(datagram) + response_mid = mid if message_type != TYPE_CON else (mid + 1) & 0xFFFF + return [ + ( + build_coap(message_type, 0x45, response_mid, token, [], b"directory"), + (host, port), + ) + ] + + +@pytest.fixture +def patch_socket(monkeypatch): + created = [] + + def install(responder=None): + active = _FakeSocket(responder) + selector = _FakeSelector(active) + created.append((active, selector)) + monkeypatch.setattr( + "smartthings_local.protocol.ocf_multicast.socket.socket", + lambda *_args: active, + ) + monkeypatch.setattr( + "smartthings_local.protocol.ocf_multicast.selectors.DefaultSelector", + lambda: selector, + ) + return active, selector + + return install + + +def test_sends_proven_directory_requests_and_filtered_fallback_on_interface( + patch_socket, +): + active, selector = patch_socket(_response) + + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + + assert result.ports == (43123,) + assert result.attempts == 3 + assert result.responses == 3 + assert active.bound == (_INTERFACE, 0) + assert active.blocking is False + assert active.closed + assert selector.closed + assert all( + destination == (_OCF_MULTICAST_GROUP, 5683) for _, destination in active.sent + ) + + requests = [parse_coap(datagram) for datagram, _ in active.sent] + assert all( + [value for number, value in item[4] if number == URI_PATH] == [b"oic", b"res"] + for item in requests + ) + queries = [ + [value for number, value in item[4] if number == URI_QUERY] for item in requests + ] + assert queries == [[], [], [b"rt=oic.r.doxm"]] + accepts = [ + [value for number, value in item[4] if number == ACCEPT] for item in requests + ] + assert accepts == [[b"\x27\x10"], [b"\x3c"], [b"\x3c"]] + assert [value for number, value in requests[0][4] if number == 2049] == [ + b"\x08\x00" + ] + assert [value for number, value in requests[1][4] if number == 2049] == [] + assert [value for number, value in requests[2][4] if number == 2049] == [] + + +@pytest.mark.parametrize( + ("accepted_accept", "accepted_query"), + ( + (b"\x27\x10", ()), + (b"\x3c", ()), + (b"\x3c", (b"rt=oic.r.doxm",)), + ), +) +def test_each_directory_request_profile_can_find_the_responder( + patch_socket, accepted_accept, accepted_query +): + def responder(datagram, destination): + _mtype, _code, _mid, _token, options, _payload = parse_coap(datagram) + accept = next(value for number, value in options if number == ACCEPT) + query = tuple(value for number, value in options if number == URI_QUERY) + if (accept, query) != (accepted_accept, accepted_query): + return [] + return _response(datagram, destination) + + patch_socket(responder) + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + + assert result.ports == (43123,) + assert result.responses == 1 + + +def test_accepts_only_token_correlated_content_from_the_target(patch_socket): + def responder(datagram, _destination): + responses = _response(datagram) + _mtype, _code, mid, token, _options, _payload = parse_coap(datagram) + responses.extend( + [ + (build_coap(TYPE_NON, 0x45, mid, b"wrong", [], b"x"), (_TARGET, 49999)), + (build_coap(TYPE_NON, 0x44, mid, token, [], b"x"), (_TARGET, 49998)), + (build_coap(TYPE_NON, 0x45, mid, token, [], b"x"), (_OTHER, 49997)), + (build_coap(TYPE_NON, 0x45, mid, token, [], b""), (_TARGET, 49996)), + ] + ) + return responses + + patch_socket(responder) + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + + assert result.ports == (43123,) + assert result.responses == 3 + + +def test_ignores_malformed_and_oversized_datagrams(patch_socket): + def responder(datagram, _destination): + valid = _response(datagram) + invalid_version = bytes([valid[0][0][0] & 0x3F]) + valid[0][0][1:] + invalid_token_length = bytes([0x49]) + valid[0][0][1:] + return [ + (b"\x40", (_TARGET, 49999)), + (invalid_version, (_TARGET, 49998)), + (invalid_token_length, (_TARGET, 49997)), + (b"x" * 8193, (_TARGET, 49996)), + *valid, + ] + + patch_socket(responder) + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + + assert result.ports == (43123,) + assert result.responses == 3 + + +def test_acknowledges_confirmable_responses(patch_socket): + active, _selector = patch_socket( + lambda datagram, _destination: _response(datagram, message_type=TYPE_CON) + ) + + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + + assert result.found + acknowledgements = [ + (parse_coap(datagram), destination) + for datagram, destination in active.sent + if parse_coap(datagram)[0] == 2 + ] + assert len(acknowledgements) == 3 + assert all(item[0][1] == 0 and item[0][3] == b"" for item in acknowledgements) + assert all( + destination == (_TARGET, 43123) for _item, destination in acknowledgements + ) + + +def test_collects_a_small_bounded_candidate_set(patch_socket): + response_number = 0 + + def responder(datagram, _destination): + nonlocal response_number + port = 40000 + response_number % 8 + response_number += 1 + return _response(datagram, port=port) + + patch_socket(responder) + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=4, + ) + + assert result.ports == tuple(range(40000, 40008)) + assert result.error_code is None + + +def test_fails_closed_when_too_many_distinct_ports_answer(patch_socket): + next_port = 40000 + + def responder(datagram, _destination): + nonlocal next_port + responses = [ + *_response(datagram, port=next_port), + *_response(datagram, port=next_port + 1), + ] + next_port += 2 + return responses + + patch_socket(responder) + result = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=4, + ) + + assert result.ports == () + assert result.responses == 24 + assert result.error_code == "ambiguous_response" + + +def test_no_response_and_interface_failures_are_fixed_results( + patch_socket, monkeypatch +): + active, _selector = patch_socket() + no_response = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + rounds=1, + ) + assert no_response == OcfResponderPortDiscoveryResult( + ports=(), + attempts=3, + responses=0, + error_code="no_response", + ) + assert active.closed + + class BrokenSocket: + def setsockopt(self, *_args): + raise OSError("synthetic") + + def close(self): + return None + + monkeypatch.setattr( + "smartthings_local.protocol.ocf_multicast.socket.socket", + lambda *_args: BrokenSocket(), + ) + unavailable = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + ) + assert unavailable.error_code == "interface_unavailable" + assert unavailable.attempts == 0 + + monkeypatch.setattr( + "smartthings_local.protocol.ocf_multicast.selectors.DefaultSelector", + lambda: (_ for _ in ()).throw(OSError("synthetic")), + ) + unavailable = discover_ocf_responder_ports( + _TARGET, + interface_address=_INTERFACE, + ) + assert unavailable.error_code == "interface_unavailable" + assert unavailable.attempts == 0 + + +@pytest.mark.parametrize( + ("kwargs", "exception"), + [ + ({"target_address": 123}, TypeError), + ({"target_address": "not-an-address"}, ValueError), + ({"target_address": _OCF_MULTICAST_GROUP}, ValueError), + ({"interface_address": "0.0.0.0"}, ValueError), + ({"discovery_port": True}, TypeError), + ({"discovery_port": 0}, ValueError), + ({"timeout": float("nan")}, ValueError), + ({"rounds": 0}, ValueError), + ({"rounds": 5}, ValueError), + ], +) +def test_rejects_invalid_options_without_opening_a_socket( + monkeypatch, kwargs, exception +): + values = { + "target_address": _TARGET, + "interface_address": _INTERFACE, + **kwargs, + } + socket_factory = pytest.fail + monkeypatch.setattr( + "smartthings_local.protocol.ocf_multicast.socket.socket", + socket_factory, + ) + with pytest.raises(exception): + discover_ocf_responder_ports(**values) + + +def test_result_repr_omits_ports_and_addresses(): + result = OcfResponderPortDiscoveryResult(ports=(43123,), attempts=2, responses=1) + + rendered = repr(result) + assert "43123" not in rendered + assert _TARGET not in rendered + assert "port_count=1" in rendered diff --git a/tests/test_public_api_contract.py b/tests/test_public_api_contract.py index 712d393..b070bec 100644 --- a/tests/test_public_api_contract.py +++ b/tests/test_public_api_contract.py @@ -18,6 +18,10 @@ ConnectCancellation, DtlsCoapSession, ) +from smartthings_local.protocol.ocf_multicast import ( + OcfResponderPortDiscoveryResult, + discover_ocf_responder_ports, +) from smartthings_local.protocol.owner_psk import derive_mfg_certificate_owner_psk @@ -57,6 +61,32 @@ def test_dtls_session_constructor_keeps_file_memory_and_local_port_inputs(): assert auth_parameter.default is None +def test_known_host_multicast_discovery_has_a_bounded_explicit_interface_api(): + parameters = inspect.signature(discover_ocf_responder_ports).parameters + assert list(parameters) == [ + "target_address", + "interface_address", + "discovery_port", + "timeout", + "rounds", + ] + assert parameters["target_address"].default is inspect.Parameter.empty + for name in ("interface_address", "discovery_port", "timeout", "rounds"): + assert parameters[name].kind is inspect.Parameter.KEYWORD_ONLY + assert parameters["interface_address"].default is inspect.Parameter.empty + assert parameters["discovery_port"].default == 5683 + assert parameters["timeout"].default == 3.0 + assert parameters["rounds"].default == 2 + + result = OcfResponderPortDiscoveryResult( + ports=(43123,), + attempts=2, + responses=1, + ) + assert result.found is True + assert result.ports == (43123,) + + def test_certificate_auth_is_a_public_authentication_provider(): provider = CertificateAuth.from_files("/synthetic/cert.pem", "/synthetic/key") assert isinstance(provider, AuthenticationProvider)