From 746ddfc59e43b783782b982bbbf13cf269a5b19f Mon Sep 17 00:00:00 2001 From: Jason Morcos Date: Sat, 29 Aug 2026 13:19:30 -0700 Subject: [PATCH] Add BLE OCF framing codec --- README.md | 55 +++ smartthings_local/protocol/ble_ocf.py | 492 +++++++++++++++++++++ tests/test_ble_ocf.py | 595 ++++++++++++++++++++++++++ tests/test_import_isolation.py | 1 + 4 files changed, 1143 insertions(+) create mode 100644 smartthings_local/protocol/ble_ocf.py create mode 100644 tests/test_ble_ocf.py diff --git a/README.md b/README.md index 8c8b6e5..1e7fdff 100644 --- a/README.md +++ b/README.md @@ -192,6 +192,61 @@ messages = decoder.feed(received_chunk) The module deliberately does not open a TCP/TLS/Bluetooth connection, choose a carrier, or perform setup and ownership operations. Those remain caller policy. +## BLE OCF framing + +`smartthings_local.protocol.ble_ocf` provides the pure fragmentation layer +used by IoTivity's GATT transport. The two-byte header carries a start flag, +source and destination virtual ports, and a secure flag. A start frame also +carries the PDU's four-byte big-endian total length. The format and full-frame +fragmentation behavior are documented in Samsung's public IoTivity sources: +[`cafragmentation.h`](https://github.com/Samsung/TizenRT/blob/0df9b54dfd35d9aaba2c16eb2ef9f4b4b6a5f545/external/iotivity/iotivity_1.2-rel/resource/csdk/connectivity/inc/cafragmentation.h), +[`cafragmentation.c`](https://github.com/Samsung/TizenRT/blob/0df9b54dfd35d9aaba2c16eb2ef9f4b4b6a5f545/external/iotivity/iotivity_1.2-rel/resource/csdk/connectivity/src/adapter_util/cafragmentation.c), +and the adapter's +[`caleadapter.c`](https://github.com/Samsung/TizenRT/blob/0df9b54dfd35d9aaba2c16eb2ef9f4b4b6a5f545/external/iotivity/iotivity_1.2-rel/resource/csdk/connectivity/src/bt_le_adapter/caleadapter.c). + +The BLE payload is the same reliable-transport CoAP message described above. +For example, a caller that already owns GATT connection policy can wrap a +plaintext discovery request and strictly reassemble response characteristic +values: + +```python +from smartthings_local.protocol.ble_ocf import ( + AdaptiveBleOcfReassembler, + fragment_pdu, +) +from smartthings_local.protocol.coap_tcp import build_coap_tcp_get + +request_pdu = build_coap_tcp_get("/oic/res", token=b"\x01") +request_frames = fragment_pdu( + request_pdu, + mtu=20, + source_port=1, + destination_port=0, + secure=False, +) + +decoder = AdaptiveBleOcfReassembler(max_pdu_size=64 * 1024) +response = decoder.feed(received_characteristic_value) +if response is not None: + response_pdu = response.pdu +``` + +Here `mtu` means IoTivity's maximum complete characteristic-value frame size, +not the raw ATT MTU. The default ATT MTU of 23 normally leaves 20 bytes for a +characteristic value. The adaptive reassembler infers a peer's usable frame +size from each first frame, rejects inconsistent continuation metadata and +lengths, and discards partial state after an error. + +The two-byte IoTivity header has no fragment sequence number. A duplicated or +reordered full-size continuation with otherwise identical metadata is +therefore indistinguishable at this layer; IoTivity relies on GATT's ordered +delivery. The codec does reject duplicate starts, orphan continuations, +detectable missing or shortened fragments, and changed port or secure flags. + +The secure bit is transport metadata; this codec does not encrypt or +authenticate the PDU. It also does not connect to Bluetooth, select GATT +characteristics, discover credentials, or perform setup or ownership work. + Reads retransmit each Block2 request; writes send once. Where a lost write has been shown to be the cause rather than a device that is simply refusing load, `write_max_attempts` lets `post()` retransmit inside the caller's own diff --git a/smartthings_local/protocol/ble_ocf.py b/smartthings_local/protocol/ble_ocf.py new file mode 100644 index 0000000..5161226 --- /dev/null +++ b/smartthings_local/protocol/ble_ocf.py @@ -0,0 +1,492 @@ +"""Pure IoTivity BLE transport framing for OCF PDUs. + +This module implements only the wire codec and reassembly state machine. It +does not open a Bluetooth connection or read or write a GATT characteristic. + +The format follows IoTivity's ``cafragmentation.h`` and +``cafragmentation.c``: + +* byte 0: start flag in bit 7, source port in bits 0 through 6; +* byte 1: secure flag in bit 7, destination port in bits 0 through 6; +* the first frame then carries the total PDU length as a four-byte, + big-endian unsigned integer; +* continuation frames carry only the two-byte header before their payload. + +IoTivity fills every non-final frame to the negotiated transport frame size. +The reassembler enforces that invariant, metadata consistency, and a +conservative PDU-size bound. The header has no sequence number, so no framing +codec can distinguish reordered or duplicated full-size continuation frames; +IoTivity relies on GATT's ordered transport for that property. +""" + +from __future__ import annotations + +import struct +from dataclasses import dataclass +from typing import Final + +__all__ = [ + "BLE_FIRST_FRAME_OVERHEAD", + "BLE_HEADER_SIZE", + "BLE_LENGTH_HEADER_SIZE", + "BLE_MAX_MTU", + "BLE_MAX_PORT", + "BLE_MIN_SOURCE_PORT", + "BLE_MULTICAST_PORT", + "BLE_WIRE_MAX_PDU_SIZE", + "DEFAULT_MAX_PDU_SIZE", + "AdaptiveBleOcfReassembler", + "BleOcfCodecError", + "BleOcfHeader", + "BleOcfInterleavedFrameError", + "BleOcfReassembler", + "ReassembledBleOcfPdu", + "decode_header", + "encode_header", + "fragment_pdu", +] + +BLE_HEADER_SIZE: Final = 2 +BLE_LENGTH_HEADER_SIZE: Final = 4 +BLE_FIRST_FRAME_OVERHEAD: Final = BLE_HEADER_SIZE + BLE_LENGTH_HEADER_SIZE +BLE_MIN_SOURCE_PORT: Final = 1 +BLE_MAX_PORT: Final = 127 +BLE_MULTICAST_PORT: Final = 0 +BLE_MAX_MTU: Final = 0xFFFF +BLE_WIRE_MAX_PDU_SIZE: Final = 0xFFFFFFFF + +# A four-byte length field can describe almost 4 GiB, but accepting such a +# declaration by default would make a local peer an easy memory-exhaustion +# vector. OCF control PDUs are far smaller. Callers may select another bound, +# up to the on-wire uint32 maximum, when constructing the codec. +DEFAULT_MAX_PDU_SIZE: Final = 1024 * 1024 + + +class BleOcfCodecError(ValueError): + """An invalid or unsupported BLE OCF frame.""" + + +class BleOcfInterleavedFrameError(BleOcfCodecError): + """A frame belongs to a different PDU than the active reassembly.""" + + +@dataclass(frozen=True, slots=True) +class BleOcfHeader: + """Semantic representation of the two-byte IoTivity BLE header.""" + + start: bool + source_port: int + secure: bool + destination_port: int + + +@dataclass(frozen=True, slots=True, repr=False) +class ReassembledBleOcfPdu: + """A complete OCF PDU and the transport metadata that carried it.""" + + pdu: bytes + source_port: int + destination_port: int + secure: bool + + def __repr__(self) -> str: + """Return transport metadata without exposing the PDU bytes.""" + return ( + "ReassembledBleOcfPdu(" + f"pdu_length={len(self.pdu)}, source_port={self.source_port}, " + f"destination_port={self.destination_port}, secure={self.secure})" + ) + + +def _byte_length(value: object, *, name: str) -> int: + if not isinstance(value, (bytes, bytearray, memoryview)): + raise TypeError(f"{name} must be bytes-like") + try: + return value.nbytes if isinstance(value, memoryview) else len(value) + except ValueError: + raise TypeError(f"{name} must be an active bytes-like value") from None + + +def _coerce_bytes( + value: object, + *, + name: str, + max_length: int | None = None, + limit_name: str = "maximum", +) -> bytes: + length = _byte_length(value, name=name) + if max_length is not None and length > max_length: + raise BleOcfCodecError( + f"{name} length {length} exceeds {limit_name} {max_length}" + ) + try: + return bytes(value) + except ValueError: + raise TypeError(f"{name} must be an active bytes-like value") from None + + +def _validate_port(port: object, *, source: bool) -> int: + label = "source_port" if source else "destination_port" + minimum = BLE_MIN_SOURCE_PORT if source else BLE_MULTICAST_PORT + if isinstance(port, bool) or not isinstance(port, int): + raise TypeError(f"{label} must be an integer") + if not minimum <= port <= BLE_MAX_PORT: + raise BleOcfCodecError( + f"{label} must be in the range {minimum}..{BLE_MAX_PORT}" + ) + return port + + +def _validate_flag(value: object, *, name: str) -> bool: + if not isinstance(value, bool): + raise TypeError(f"{name} must be a bool") + return value + + +def _validate_mtu(mtu: object) -> int: + if isinstance(mtu, bool) or not isinstance(mtu, int): + raise TypeError("mtu must be an integer") + # A non-empty PDU needs the two-byte header, four-byte length, and at + # least one payload byte in its first frame. + if not BLE_FIRST_FRAME_OVERHEAD < mtu <= BLE_MAX_MTU: + raise BleOcfCodecError( + f"mtu must be in the range {BLE_FIRST_FRAME_OVERHEAD + 1}..{BLE_MAX_MTU}" + ) + return mtu + + +def _validate_max_pdu_size(max_pdu_size: object) -> int: + if isinstance(max_pdu_size, bool) or not isinstance(max_pdu_size, int): + raise TypeError("max_pdu_size must be an integer") + if not 1 <= max_pdu_size <= BLE_WIRE_MAX_PDU_SIZE: + raise BleOcfCodecError( + f"max_pdu_size must be in the range 1..{BLE_WIRE_MAX_PDU_SIZE}" + ) + return max_pdu_size + + +def encode_header( + *, + start: bool, + source_port: int, + secure: bool, + destination_port: int, +) -> bytes: + """Encode the two-byte IoTivity BLE transport header.""" + + start = _validate_flag(start, name="start") + secure = _validate_flag(secure, name="secure") + source_port = _validate_port(source_port, source=True) + destination_port = _validate_port(destination_port, source=False) + return bytes( + ( + (0x80 if start else 0) | source_port, + (0x80 if secure else 0) | destination_port, + ) + ) + + +def decode_header(frame: bytes | bytearray | memoryview) -> BleOcfHeader: + """Decode and validate the header at the beginning of ``frame``.""" + + data = _coerce_bytes(frame, name="frame") + if len(data) < BLE_HEADER_SIZE: + raise BleOcfCodecError("BLE OCF frame is shorter than its two-byte header") + + first, second = data[:BLE_HEADER_SIZE] + source_port = first & 0x7F + if source_port < BLE_MIN_SOURCE_PORT: + raise BleOcfCodecError("BLE OCF source port 0 is invalid") + return BleOcfHeader( + start=bool(first & 0x80), + source_port=source_port, + secure=bool(second & 0x80), + destination_port=second & 0x7F, + ) + + +def fragment_pdu( + pdu: bytes | bytearray | memoryview, + *, + mtu: int, + source_port: int, + destination_port: int, + secure: bool, + max_pdu_size: int = DEFAULT_MAX_PDU_SIZE, +) -> tuple[bytes, ...]: + """Fragment one non-empty OCF PDU into IoTivity BLE frames. + + ``mtu`` is the maximum complete characteristic-value frame used by the + IoTivity adapter. It is not the raw ATT MTU; a default ATT MTU of 23, for + example, normally permits a 20-byte characteristic value. + """ + + mtu = _validate_mtu(mtu) + max_pdu_size = _validate_max_pdu_size(max_pdu_size) + source_port = _validate_port(source_port, source=True) + destination_port = _validate_port(destination_port, source=False) + secure = _validate_flag(secure, name="secure") + data = _coerce_bytes( + pdu, + name="OCF PDU", + max_length=max_pdu_size, + ) + + if not data: + raise BleOcfCodecError("OCF PDU must not be empty") + + first_capacity = mtu - BLE_FIRST_FRAME_OVERHEAD + continuation_capacity = mtu - BLE_HEADER_SIZE + first_length = min(len(data), first_capacity) + frames = [ + encode_header( + start=True, + source_port=source_port, + secure=secure, + destination_port=destination_port, + ) + + struct.pack(">I", len(data)) + + data[:first_length] + ] + + continuation_header = encode_header( + start=False, + source_port=source_port, + secure=secure, + destination_port=destination_port, + ) + offset = first_length + while offset < len(data): + end = min(offset + continuation_capacity, len(data)) + frames.append(continuation_header + data[offset:end]) + offset = end + return tuple(frames) + + +class BleOcfReassembler: + """Strict, single-PDU IoTivity BLE reassembly state machine. + + Any malformed or interleaved frame aborts the active PDU before the error + is raised. This fail-closed behavior prevents a later valid-looking tail + from completing data that was already shown to be inconsistent. + """ + + def __init__( + self, + *, + mtu: int, + max_pdu_size: int = DEFAULT_MAX_PDU_SIZE, + ) -> None: + self._mtu = _validate_mtu(mtu) + self._max_pdu_size = _validate_max_pdu_size(max_pdu_size) + self.reset() + + @property + def in_progress(self) -> bool: + """Return whether a partial PDU is currently buffered.""" + + return self._header is not None + + @property + def buffered_bytes(self) -> int: + """Return the number of PDU bytes retained for reassembly.""" + + return len(self._buffer) + + def reset(self) -> None: + """Discard any partial PDU.""" + + self._header: BleOcfHeader | None = None + self._expected_length = 0 + self._buffer = bytearray() + + def feed( + self, + frame: bytes | bytearray | memoryview, + ) -> ReassembledBleOcfPdu | None: + """Consume one complete GATT value and return a PDU when complete.""" + + try: + return self._feed(frame) + except (BleOcfCodecError, TypeError): + self.reset() + raise + + def _feed( + self, + frame: bytes | bytearray | memoryview, + ) -> ReassembledBleOcfPdu | None: + data = _coerce_bytes( + frame, + name="BLE OCF frame", + max_length=self._mtu, + limit_name="MTU", + ) + + header = decode_header(data) + if header.start: + if self.in_progress: + raise BleOcfInterleavedFrameError( + "received a new start frame while another PDU is incomplete" + ) + return self._start(data, header) + return self._continue(data, header) + + def _start( + self, + data: bytes, + header: BleOcfHeader, + ) -> ReassembledBleOcfPdu | None: + if len(data) < BLE_FIRST_FRAME_OVERHEAD: + raise BleOcfCodecError( + "BLE OCF start frame is missing its four-byte length header" + ) + + expected_length = struct.unpack( + ">I", + data[BLE_HEADER_SIZE:BLE_FIRST_FRAME_OVERHEAD], + )[0] + if expected_length == 0: + raise BleOcfCodecError("BLE OCF PDU length must not be zero") + if expected_length > self._max_pdu_size: + raise BleOcfCodecError( + f"declared OCF PDU length {expected_length} exceeds maximum " + f"{self._max_pdu_size}" + ) + + payload = data[BLE_FIRST_FRAME_OVERHEAD:] + expected_payload_length = min( + expected_length, + self._mtu - BLE_FIRST_FRAME_OVERHEAD, + ) + if len(payload) != expected_payload_length: + raise BleOcfCodecError( + "BLE OCF start frame payload length does not match " + "IoTivity fragmentation" + ) + + if len(payload) == expected_length: + return ReassembledBleOcfPdu( + pdu=payload, + source_port=header.source_port, + destination_port=header.destination_port, + secure=header.secure, + ) + + self._header = header + self._expected_length = expected_length + self._buffer.extend(payload) + return None + + def _continue( + self, + data: bytes, + header: BleOcfHeader, + ) -> ReassembledBleOcfPdu | None: + active_header = self._header + if active_header is None: + raise BleOcfCodecError( + "received a continuation frame without a start frame" + ) + if ( + header.start + or header.source_port != active_header.source_port + or header.secure != active_header.secure + or header.destination_port != active_header.destination_port + ): + raise BleOcfInterleavedFrameError( + "continuation frame metadata does not match the active PDU" + ) + + payload = data[BLE_HEADER_SIZE:] + remaining = self._expected_length - len(self._buffer) + expected_payload_length = min( + remaining, + self._mtu - BLE_HEADER_SIZE, + ) + if len(payload) != expected_payload_length: + raise BleOcfCodecError( + "BLE OCF continuation payload length does not match " + "IoTivity fragmentation" + ) + + self._buffer.extend(payload) + if len(self._buffer) < self._expected_length: + return None + + message = ReassembledBleOcfPdu( + pdu=bytes(self._buffer), + source_port=active_header.source_port, + destination_port=active_header.destination_port, + secure=active_header.secure, + ) + self.reset() + return message + + +class AdaptiveBleOcfReassembler: + """Infer IoTivity's transport frame size from each first frame. + + A GATT client receives complete characteristic values, not necessarily the + ATT MTU negotiated by the remote IoTivity stack. IoTivity fills an + incomplete first frame to its usable frame size, so the observed length is + the value needed by :class:`BleOcfReassembler`. A complete single-frame + PDU is valid with that same observed length. This wrapper preserves strict + fragmentation checks without trusting a platform-specific MTU property. + """ + + def __init__( + self, + *, + max_pdu_size: int = DEFAULT_MAX_PDU_SIZE, + ) -> None: + self._max_pdu_size = _validate_max_pdu_size(max_pdu_size) + self._reassembler: BleOcfReassembler | None = None + + @property + def in_progress(self) -> bool: + """Return whether a partial PDU is currently buffered.""" + + return bool(self._reassembler and self._reassembler.in_progress) + + @property + def buffered_bytes(self) -> int: + """Return the number of PDU bytes retained for reassembly.""" + + return self._reassembler.buffered_bytes if self._reassembler else 0 + + def reset(self) -> None: + """Discard any partial PDU and its inferred frame size.""" + + self._reassembler = None + + def feed( + self, + frame: bytes | bytearray | memoryview, + ) -> ReassembledBleOcfPdu | None: + """Consume one frame, inferring a new frame size at PDU boundaries.""" + + try: + data = _coerce_bytes( + frame, + name="BLE OCF frame", + max_length=BLE_MAX_MTU, + limit_name="wire maximum", + ) + if self._reassembler is None: + header = decode_header(data) + if not header.start: + raise BleOcfCodecError( + "received a continuation frame without a start frame" + ) + self._reassembler = BleOcfReassembler( + mtu=len(data), + max_pdu_size=self._max_pdu_size, + ) + completed = self._reassembler.feed(data) + except (BleOcfCodecError, TypeError): + self.reset() + raise + if completed is not None: + self.reset() + return completed diff --git a/tests/test_ble_ocf.py b/tests/test_ble_ocf.py new file mode 100644 index 0000000..4cb8811 --- /dev/null +++ b/tests/test_ble_ocf.py @@ -0,0 +1,595 @@ +"""IoTivity BLE OCF fragmentation and reassembly contracts.""" + +from __future__ import annotations + +from array import array + +import pytest + +from smartthings_local.protocol.ble_ocf import ( + BLE_MAX_MTU, + BLE_WIRE_MAX_PDU_SIZE, + AdaptiveBleOcfReassembler, + BleOcfCodecError, + BleOcfHeader, + BleOcfInterleavedFrameError, + BleOcfReassembler, + ReassembledBleOcfPdu, + decode_header, + encode_header, + fragment_pdu, +) +from smartthings_local.protocol.coap_tcp import ( + CoapTcpMessage, + build_coap_tcp_get, + parse_coap_tcp_message, +) + + +@pytest.mark.parametrize( + ("fields", "encoded"), + ( + ( + { + "start": True, + "source_port": 1, + "secure": False, + "destination_port": 0, + }, + b"\x81\x00", + ), + ( + { + "start": False, + "source_port": 127, + "secure": True, + "destination_port": 42, + }, + b"\x7f\xaa", + ), + ( + { + "start": True, + "source_port": 37, + "secure": True, + "destination_port": 127, + }, + b"\xa5\xff", + ), + ), +) +def test_header_known_vectors_preserve_flags_and_ports(fields, encoded): + assert encode_header(**fields) == encoded + assert decode_header(encoded + b"ignored payload") == BleOcfHeader(**fields) + + +def test_fragmentation_source_vector_uses_big_endian_length_and_mtu(): + assert fragment_pdu( + b"0123456789ABC", + mtu=10, + source_port=5, + destination_port=9, + secure=True, + ) == ( + b"\x85\x89\x00\x00\x00\x0d0123", + b"\x05\x89456789AB", + b"\x05\x89C", + ) + + +@pytest.mark.parametrize("mtu", (7, 8, 10, 20, 23, 64, 512)) +@pytest.mark.parametrize("secure", (False, True)) +def test_round_trip_at_every_fragment_boundary(mtu, secure): + first_capacity = mtu - 6 + continuation_capacity = mtu - 2 + sizes = { + 1, + first_capacity, + first_capacity + 1, + first_capacity + continuation_capacity, + first_capacity + continuation_capacity + 1, + first_capacity + 3 * continuation_capacity, + first_capacity + 3 * continuation_capacity + 1, + } + + for size in sorted(sizes): + pdu = bytes(index % 251 for index in range(size)) + frames = fragment_pdu( + pdu, + mtu=mtu, + source_port=17, + destination_port=0, + secure=secure, + ) + reassembler = BleOcfReassembler(mtu=mtu) + completed = None + + for frame in frames: + assert len(frame) <= mtu + completed = reassembler.feed(frame) + + assert completed == ReassembledBleOcfPdu( + pdu=pdu, + source_port=17, + destination_port=0, + secure=secure, + ) + assert not reassembler.in_progress + assert reassembler.buffered_bytes == 0 + + +def test_non_final_frames_fill_mtu_including_exact_continuation_multiple(): + mtu = 20 + first_capacity = mtu - 6 + continuation_capacity = mtu - 2 + pdu = b"x" * (first_capacity + 3 * continuation_capacity) + + frames = fragment_pdu( + pdu, + mtu=mtu, + source_port=1, + destination_port=1, + secure=False, + ) + + assert len(frames) == 4 + assert {len(frame) for frame in frames} == {mtu} + + +def test_tcp_get_round_trips_through_ble_transport_frames(): + pdu = build_coap_tcp_get( + "/oic/res", + token=b"\x12\x34", + query=("if=oic.if.baseline",), + ) + reassembler = BleOcfReassembler(mtu=20) + + for frame in fragment_pdu( + pdu, + mtu=20, + source_port=1, + destination_port=0, + secure=False, + ): + completed = reassembler.feed(frame) + + assert completed is not None + assert completed.destination_port == 0 + assert parse_coap_tcp_message(completed.pdu) == CoapTcpMessage( + code=0x01, + token=b"\x12\x34", + options=( + (11, b"oic"), + (11, b"res"), + (15, b"if=oic.if.baseline"), + ), + payload=b"", + ) + + +def test_bytes_like_inputs_are_snapshotted(): + mutable_pdu = bytearray(b"payload that fragments") + frames = fragment_pdu( + memoryview(mutable_pdu), + mtu=10, + source_port=3, + destination_port=7, + secure=False, + ) + mutable_pdu[:] = b"z" * len(mutable_pdu) + + reassembler = BleOcfReassembler(mtu=10) + completed = None + for frame in (bytearray(value) for value in frames): + completed = reassembler.feed(frame) + + assert completed is not None + assert completed.pdu == b"payload that fragments" + + +def test_adaptive_reassembler_infers_each_first_frame_size(): + adaptive = AdaptiveBleOcfReassembler(max_pdu_size=1024) + expected = ( + (b"first", 20), + (b"second payload that spans frames", 11), + (b"third", 23), + ) + completed = [] + + for pdu, mtu in expected: + for frame in fragment_pdu( + pdu, + mtu=mtu, + source_port=1, + destination_port=1, + secure=False, + max_pdu_size=1024, + ): + result = adaptive.feed(frame) + if result is not None: + completed.append(result) + + assert completed == [ + ReassembledBleOcfPdu( + pdu=pdu, + source_port=1, + destination_port=1, + secure=False, + ) + for pdu, _mtu in expected + ] + assert not adaptive.in_progress + assert adaptive.buffered_bytes == 0 + + +def test_adaptive_reassembler_resets_after_malformed_stream(): + adaptive = AdaptiveBleOcfReassembler(max_pdu_size=64) + frames = fragment_pdu( + b"long enough to fragment", + mtu=10, + source_port=1, + destination_port=1, + secure=False, + ) + assert adaptive.feed(frames[0]) is None + assert adaptive.buffered_bytes == 4 + + with pytest.raises(BleOcfInterleavedFrameError): + adaptive.feed(b"\x02\x01bad-tail") + + assert not adaptive.in_progress + recovered = adaptive.feed( + fragment_pdu( + b"ok", + mtu=20, + source_port=1, + destination_port=1, + secure=False, + )[0] + ) + assert recovered is not None + assert recovered.pdu == b"ok" + + +def test_adaptive_reassembler_resets_after_oversized_frame(): + adaptive = AdaptiveBleOcfReassembler(max_pdu_size=64) + first = fragment_pdu( + b"long enough to fragment", + mtu=10, + source_port=1, + destination_port=1, + secure=False, + )[0] + assert adaptive.feed(first) is None + + with pytest.raises(BleOcfCodecError, match="wire maximum"): + adaptive.feed(b"x" * (BLE_MAX_MTU + 1)) + + assert not adaptive.in_progress + assert adaptive.buffered_bytes == 0 + + +def test_released_memoryview_resets_partial_reassembly(): + reassembler = BleOcfReassembler(mtu=10) + first = fragment_pdu( + b"long enough to fragment", + mtu=10, + source_port=1, + destination_port=1, + secure=False, + )[0] + assert reassembler.feed(first) is None + released = memoryview(b"unused") + released.release() + + with pytest.raises(TypeError, match="active bytes-like"): + reassembler.feed(released) + + assert not reassembler.in_progress + assert reassembler.buffered_bytes == 0 + + +@pytest.mark.parametrize("source_port", (0, 128, -1)) +def test_header_rejects_invalid_source_port(source_port): + with pytest.raises(BleOcfCodecError, match="source_port"): + encode_header( + start=True, + source_port=source_port, + secure=False, + destination_port=0, + ) + + +@pytest.mark.parametrize("destination_port", (-1, 128)) +def test_header_rejects_invalid_destination_port(destination_port): + with pytest.raises(BleOcfCodecError, match="destination_port"): + encode_header( + start=True, + source_port=1, + secure=False, + destination_port=destination_port, + ) + + +@pytest.mark.parametrize(("name", "value"), (("start", 1), ("secure", 0))) +def test_header_rejects_non_boolean_flags(name, value): + fields = { + "start": True, + "source_port": 1, + "secure": False, + "destination_port": 0, + } + fields[name] = value + with pytest.raises(TypeError, match="bool"): + encode_header(**fields) + + +@pytest.mark.parametrize("mtu", (0, 6, BLE_MAX_MTU + 1)) +def test_fragmenter_rejects_invalid_mtu(mtu): + with pytest.raises(BleOcfCodecError, match="mtu"): + fragment_pdu( + b"x", + mtu=mtu, + source_port=1, + destination_port=0, + secure=False, + ) + + +@pytest.mark.parametrize("mtu", (True, 20.0, "20")) +def test_fragmenter_rejects_non_integer_mtu(mtu): + with pytest.raises(TypeError, match="mtu"): + fragment_pdu( + b"x", + mtu=mtu, + source_port=1, + destination_port=0, + secure=False, + ) + + +def test_pdu_and_maximum_validation_is_bounded(): + with pytest.raises(BleOcfCodecError, match="must not be empty"): + fragment_pdu( + b"", + mtu=20, + source_port=1, + destination_port=0, + secure=False, + ) + with pytest.raises(BleOcfCodecError, match="length 5 exceeds maximum 4"): + fragment_pdu( + b"12345", + mtu=20, + source_port=1, + destination_port=0, + secure=False, + max_pdu_size=4, + ) + with pytest.raises(TypeError, match="bytes-like"): + fragment_pdu( + "not bytes", + mtu=20, + source_port=1, + destination_port=0, + secure=False, + ) + for maximum in (0, BLE_WIRE_MAX_PDU_SIZE + 1): + with pytest.raises(BleOcfCodecError, match="max_pdu_size"): + BleOcfReassembler(mtu=20, max_pdu_size=maximum) + with pytest.raises(TypeError, match="max_pdu_size"): + AdaptiveBleOcfReassembler(max_pdu_size=True) + + +def test_memoryview_bounds_use_bytes_not_element_count(): + wide_view = memoryview(array("I", (1, 2))) + assert len(wide_view) == 2 + assert wide_view.nbytes > 2 + maximum = wide_view.nbytes - 1 + + with pytest.raises( + BleOcfCodecError, + match=rf"length {wide_view.nbytes} exceeds maximum {maximum}", + ): + fragment_pdu( + wide_view, + mtu=20, + source_port=1, + destination_port=0, + secure=False, + max_pdu_size=maximum, + ) + + +@pytest.mark.parametrize( + "frame", + ( + b"", + b"\x81", + b"\x80\x00", # source port zero + b"\x81\x00", # missing length field + b"\x81\x00\x00\x00\x00\x00", # zero-length PDU + b"\x81\x00\x00\x00\x00\x02x", # short completed start + b"\x81\x00\x00\x00\x00\x01xy", # payload exceeds declaration + b"\x01\x00x", # continuation without a start + ), +) +def test_reassembler_rejects_malformed_frames_and_resets(frame): + reassembler = BleOcfReassembler(mtu=10) + + with pytest.raises(BleOcfCodecError): + reassembler.feed(frame) + + assert not reassembler.in_progress + assert reassembler.buffered_bytes == 0 + + +def test_reassembler_rejects_oversized_frame_and_declaration(): + with pytest.raises(BleOcfCodecError, match="exceeds MTU"): + BleOcfReassembler(mtu=10).feed(b"\x81\x00\x00\x00\x00\x05abcde") + with pytest.raises(BleOcfCodecError, match="declared.*exceeds maximum"): + BleOcfReassembler(mtu=10, max_pdu_size=12).feed(b"\x81\x00\x00\x00\x00\x0d0123") + with pytest.raises(BleOcfCodecError, match="wire maximum"): + AdaptiveBleOcfReassembler().feed(b"x" * (BLE_MAX_MTU + 1)) + + +def test_reassembler_rejects_short_middle_and_empty_continuations(): + frames = fragment_pdu( + b"a" * 30, + mtu=10, + source_port=5, + destination_port=9, + secure=True, + ) + + for bad_continuation in (frames[1][:-1], frames[1][:2]): + reassembler = BleOcfReassembler(mtu=10) + assert reassembler.feed(frames[0]) is None + with pytest.raises(BleOcfCodecError, match="continuation payload"): + reassembler.feed(bad_continuation) + assert not reassembler.in_progress + + +def test_reassembler_rejects_interleaved_start_and_resets(): + first = fragment_pdu( + b"first-pdu", + mtu=10, + source_port=5, + destination_port=9, + secure=True, + ) + second = fragment_pdu( + b"second-pdu", + mtu=10, + source_port=6, + destination_port=9, + secure=True, + ) + reassembler = BleOcfReassembler(mtu=10) + + assert reassembler.feed(first[0]) is None + with pytest.raises(BleOcfInterleavedFrameError, match="new start"): + reassembler.feed(second[0]) + + assert not reassembler.in_progress + + +@pytest.mark.parametrize( + "changed_header", + ( + encode_header( + start=False, + source_port=6, + destination_port=9, + secure=True, + ), + encode_header( + start=False, + source_port=5, + destination_port=8, + secure=True, + ), + encode_header( + start=False, + source_port=5, + destination_port=9, + secure=False, + ), + ), +) +def test_reassembler_rejects_changed_continuation_metadata(changed_header): + frames = fragment_pdu( + b"first-pdu", + mtu=10, + source_port=5, + destination_port=9, + secure=True, + ) + reassembler = BleOcfReassembler(mtu=10) + + assert reassembler.feed(frames[0]) is None + with pytest.raises(BleOcfInterleavedFrameError, match="metadata"): + reassembler.feed(changed_header + frames[1][2:]) + + assert not reassembler.in_progress + + +def test_duplicate_start_and_completed_tail_fail_closed(): + frames = fragment_pdu( + b"a payload long enough for several frames", + mtu=10, + source_port=5, + destination_port=9, + secure=False, + ) + reassembler = BleOcfReassembler(mtu=10) + + assert reassembler.feed(frames[0]) is None + with pytest.raises(BleOcfInterleavedFrameError): + reassembler.feed(frames[0]) + + completed = None + for frame in frames: + completed = reassembler.feed(frame) + assert completed is not None + with pytest.raises(BleOcfCodecError, match="without a start"): + reassembler.feed(frames[-1]) + + +def test_missing_fragment_is_incomplete_or_rejected_then_can_be_reset(): + frames = fragment_pdu( + b"a" * 31, + mtu=10, + source_port=5, + destination_port=9, + secure=False, + ) + reassembler = BleOcfReassembler(mtu=10) + + for frame in frames[:-1]: + assert reassembler.feed(frame) is None + assert reassembler.in_progress + assert reassembler.buffered_bytes < 31 + reassembler.reset() + assert not reassembler.in_progress + + assert reassembler.feed(frames[0]) is None + with pytest.raises(BleOcfCodecError, match="continuation payload"): + reassembler.feed(frames[-1]) + assert not reassembler.in_progress + + +def test_failed_stream_accepts_a_fresh_pdu(): + frames = fragment_pdu( + b"recovered", + mtu=10, + source_port=3, + destination_port=7, + secure=False, + ) + reassembler = BleOcfReassembler(mtu=10) + with pytest.raises(BleOcfCodecError): + reassembler.feed(b"\x03\x07bad") + + completed = None + for frame in frames: + completed = reassembler.feed(frame) + assert completed is not None + assert completed.pdu == b"recovered" + + +def test_reassembled_pdu_repr_omits_wire_bytes(): + result = ReassembledBleOcfPdu( + pdu=b"token-and-payload-not-for-repr", + source_port=1, + destination_port=0, + secure=False, + ) + + rendered = repr(result) + + assert rendered == ( + "ReassembledBleOcfPdu(pdu_length=30, source_port=1, " + "destination_port=0, secure=False)" + ) + assert "token-and-payload" not in rendered diff --git a/tests/test_import_isolation.py b/tests/test_import_isolation.py index 4e3d90e..4d599ed 100644 --- a/tests/test_import_isolation.py +++ b/tests/test_import_isolation.py @@ -16,6 +16,7 @@ def test_smartthings_local_imports_without_mqtt_demo_present(tmp_path): shutil.copytree(REPO_ROOT / "smartthings_local", tmp_path / "smartthings_local") import_lines = [ + "import smartthings_local.protocol.ble_ocf", "import smartthings_local.protocol.coap", "import smartthings_local.protocol.coap_tcp", "import smartthings_local.protocol.ocf_multicast",