diff --git a/README.md b/README.md index 59dee29..b155a27 100644 --- a/README.md +++ b/README.md @@ -43,3 +43,24 @@ missing. Keep real values in a `.env` you do not commit. Script output is written to be safe to paste into a public thread: identifiers are reported as a comparison result or a hash prefix rather than printed. Read the output before you post it anyway. + +## Tests for the #115 Python PSK probe + +The tested environment uses Python 3.12.13, pytest 9.1.1, pyOpenSSL 26.4.0, +cryptography 50.0.2, cbor2 6.1.5, and OpenSSL 4.0.3. +These versions describe the tested environment, not minimum requirements. +The OpenSSL fixture uses pyOpenSSL's private CFFI bindings. + +Create a virtual environment and install the tested dependencies: + +```sh +python3.12 -m venv .venv +.venv/bin/python -m pip install pytest==9.1.1 pyOpenSSL==26.4.0 cryptography==50.0.2 cbor2==6.1.5 +cd pr-115-purepy-psk +PYTHONDONTWRITEBYTECODE=1 ../.venv/bin/python -B -m pytest -p no:cacheprovider test_dtls_psk.py test_oven_psk_test.py -q +``` + +The expected result is 26 passing tests. +The tests require no appliance credentials or hardware. +The OpenSSL tests use memory BIOs. +The probe tests use localhost UDP, including replies from another source port. diff --git a/pr-115-purepy-psk/dtls_psk.py b/pr-115-purepy-psk/dtls_psk.py index 22b391e..9a37b01 100644 --- a/pr-115-purepy-psk/dtls_psk.py +++ b/pr-115-purepy-psk/dtls_psk.py @@ -343,10 +343,10 @@ def _send_client_hello(self, cookie: bytes) -> None: + bytes([1, 0]) ) body += struct.pack("!H", len(extensions)) + extensions - # RFC 6347 4.2.1: the initial ClientHello and the HelloVerifyRequest are - # excluded from the transcript; the cookie ClientHello starts it. + # Keep ClientHello when the server skips HelloVerifyRequest. + # The cookie path resets the transcript before its replacement hello. self._start_flight() - self._emit_handshake(_HT_CLIENT_HELLO, bytes(body), transcript=bool(cookie)) + self._emit_handshake(_HT_CLIENT_HELLO, bytes(body), transcript=True) def _send_client_flight(self) -> None: """ClientKeyExchange + ChangeCipherSpec + Finished.""" @@ -469,11 +469,17 @@ def _drain_inbox(self) -> None: def _handle_record(self, record: bytes) -> None: content_type = record[0] - if record[1:3] != _VERSION: - return epoch = int.from_bytes(record[3:5], "big") - seq = int.from_bytes(record[5:11], "big") fragment = record[_RECORD_HEADER:] + if record[1:3] != _VERSION: + # DTLS 1.2 servers may frame HelloVerifyRequest as DTLS 1.0. + # Accept that framing only for the initial plaintext cookie reply. + if not (record[1:3] == b"\xfe\xff" and epoch == 0 + and content_type == _CT_HANDSHAKE + and self._state == "sent_hello" + and fragment[:1] == bytes([_HT_HELLO_VERIFY_REQUEST])): + return + seq = int.from_bytes(record[5:11], "big") expected_epoch = 1 if self._read_active else 0 if epoch != expected_epoch: diff --git a/pr-115-purepy-psk/oven_psk_test.py b/pr-115-purepy-psk/oven_psk_test.py index b2f8073..d294c2c 100755 --- a/pr-115-purepy-psk/oven_psk_test.py +++ b/pr-115-purepy-psk/oven_psk_test.py @@ -14,24 +14,29 @@ 3. a close_notify, then exit It never writes to the appliance, never touches a security resource, and never -retries. Stop the bridge or integration first if one is running, since the +starts another session or retries the GET. DTLS flights can retransmit. +Stop the bridge or integration first if one is running, since the appliance will already have a session with it. Usage: + # Runtime dependencies: cryptography and cbor2. Use a virtual environment. + python3 -m pip install cryptography cbor2 export PSK_HOST=192.0.2.100 export PSK_PORT=49154 export PSK_IDENTITY_HEX=0102... # 32 hex chars, the 16-byte UUID export PSK_KEY_HEX=... # 32 or 64 hex chars - export PSK_EXPECTED_DI=... # optional, the di you expect + export PSK_EXPECTED_DI=... # required, the di you expect + export PSK_LOCAL_PORT=... # optional, stable local UDP port python3 oven_psk_test.py -Output is written to be safe to paste into the thread: the identity and the -device id are reported as a comparison result or a hash prefix, never printed. +Output omits addresses, credentials, device identifiers, and their fingerprints. """ from __future__ import annotations -import hashlib +import io + +import cbor2 import os import socket import struct @@ -65,10 +70,6 @@ def _hexbytes(name: str, lengths: tuple[int, ...]) -> bytes: return value -def _digest(value: bytes) -> str: - return hashlib.sha256(value).hexdigest()[:16] - - def _coap_get(path: tuple[str, ...], token: bytes, message_id: int) -> bytes: """Encode a CON GET. Uri-Path is option 11; deltas are 11 then 0.""" out = bytearray() @@ -89,146 +90,200 @@ def _coap_get(path: tuple[str, ...], token: bytes, message_id: int) -> bytes: return bytes(out) -def _coap_parse(datagram: bytes) -> tuple[str, bytes]: - """Return the response code as 'c.dd' and the payload.""" - if len(datagram) < 4: - return "?", b"" +def _coap_parse(datagram: bytes) -> tuple[int, int, int, bytes, bytes]: + """Validate framing and return type, code, message ID, token, and payload.""" + if len(datagram) < 4 or datagram[0] >> 6 != 1: + raise ValueError("invalid CoAP header") token_length = datagram[0] & 0x0F + if token_length > 8 or len(datagram) < 4 + token_length: + raise ValueError("invalid CoAP token") + kind = (datagram[0] >> 4) & 3 code = datagram[1] + message_id = int.from_bytes(datagram[2:4], "big") + token = datagram[4:4 + token_length] offset = 4 + token_length + payload = b"" while offset < len(datagram): - if datagram[offset] == 0xFF: - offset += 1 - break - delta = datagram[offset] >> 4 - length = datagram[offset] & 0x0F + header = datagram[offset] offset += 1 - for nibble in (delta, length): - if nibble == 13: - offset += 1 - elif nibble == 14: - offset += 2 - if length == 13: - length = datagram[offset - 1] + 13 - elif length == 14: - length = 0 # not expected on these responses - offset += length - return f"{code >> 5}.{code & 0x1F:02d}", datagram[offset:] + if header == 0xFF: + if offset == len(datagram): + raise ValueError("empty CoAP payload") + payload = datagram[offset:] + break + values = [] + for nibble in (header >> 4, header & 15): + if nibble == 15: + raise ValueError("reserved CoAP option nibble") + width = 1 if nibble == 13 else 2 if nibble == 14 else 0 + if offset + width > len(datagram): + raise ValueError("truncated CoAP option") + value = int.from_bytes(datagram[offset:offset + width], "big") + values.append(nibble if not width else value + (13 if width == 1 else 269)) + offset += width + offset += values[1] + if offset > len(datagram): + raise ValueError("truncated CoAP option value") + if code == 0 and (token or len(datagram) != 4): + raise ValueError("invalid empty CoAP message") + return kind, code, message_id, token, payload def _split_records(buf: bytes) -> list[bytes]: """One DTLS record per datagram; the appliance drops packed records.""" out, offset = [], 0 - while offset + 13 <= len(buf): + while offset < len(buf): + if offset + 13 > len(buf): + raise ValueError("truncated DTLS record") length = int.from_bytes(buf[offset + 11:offset + 13], "big") - if offset + 13 + length > len(buf): - break - out.append(buf[offset:offset + 13 + length]) - offset += 13 + length + end = offset + 13 + length + if end > len(buf): + raise ValueError("truncated DTLS record") + out.append(buf[offset:end]) + offset = end return out -def _flush(client, sock) -> None: +def _flush(client, sock, target: tuple[str, int]) -> bool: try: pending = client.bio_read(65535) except WantRead: - return - for record in _split_records(pending): - sock.send(record) + return False + records = _split_records(pending) + for record in records: + sock.sendto(record, target) + return bool(records) -def main() -> int: - host = _env("PSK_HOST") - port = int(_env("PSK_PORT") or 0) - identity = _hexbytes("PSK_IDENTITY_HEX", (16,)) - key = _hexbytes("PSK_KEY_HEX", (16, 32)) - expected_di = _env("PSK_EXPECTED_DI", required=False) - - print(f"host : {host}:{port}") - print(f"identity : 16 bytes, sha256:{_digest(identity)}, " - f"contains a zero byte: {'yes' if 0 in identity else 'no'}") - print(f"key : {len(key)} bytes, sha256:{_digest(key)}") - - client = DtlsPskClient(identity, key, mtu=1400) - sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) - sock.connect((host, port)) - sock.settimeout(0.5) +def _receive(client, sock, target: tuple[str, int]) -> None: + datagram, source = sock.recvfrom(65535) + if source[0] == target[0]: + client.bio_write(datagram) + +class ProbeFailure(Exception): + """A fixed diagnostic that contains no peer data or configuration values.""" + + +HANDSHAKE_TIMEOUT = 20 +REPLY_TIMEOUT = 10 + + +def _probe(client, sock, target: tuple[str, int], expected_di: str) -> None: started = time.monotonic() - deadline = started + 20 - completed = False + deadline = started + HANDSHAKE_TIMEOUT while time.monotonic() < deadline: try: client.do_handshake() - completed = True break except WantRead: pass - except Exception as exc: # noqa: BLE001 - reported, not handled - print(f"handshake : FAILED ({type(exc).__name__}: {exc})") - sock.close() - return 1 - _flush(client, sock) + except Exception as exc: + raise ProbeFailure(f"handshake failed ({type(exc).__name__})") from None + _flush(client, sock, target) try: - client.bio_write(sock.recv(65535)) + _receive(client, sock, target) except socket.timeout: remaining = client.DTLSv1_get_timeout() if remaining is not None and remaining <= 0: client.DTLSv1_handle_timeout() - - if not completed: - print(f"handshake : TIMED OUT after {time.monotonic() - started:.1f}s") - sock.close() - return 1 + else: + raise ProbeFailure("handshake timed out") + _flush(client, sock, target) print(f"handshake : OK in {time.monotonic() - started:.1f}s " - f"(DTLS 1.2, ECDHE-PSK-AES128-CBC-SHA256)") - - client.send(_coap_get(("oic", "d"), b"\x01\x02\x03\x04", 0x1234)) - _flush(client, sock) + "(DTLS 1.2, ECDHE-PSK-AES128-CBC-SHA256)") - payload, code = b"", "?" - reply_deadline = time.monotonic() + 10 - while time.monotonic() < reply_deadline: + token = os.urandom(4) + message_id = int.from_bytes(os.urandom(2), "big") + client.send(_coap_get(("oic", "d"), token, message_id)) + _flush(client, sock, target) + deadline = time.monotonic() + REPLY_TIMEOUT + while time.monotonic() < deadline: try: - code, payload = _coap_parse(client.recv(65535)) - break - except WantRead: - pass + kind, code, reply_id, reply_token, payload = _coap_parse(client.recv(65535)) except ZeroReturn: - print("GET /oic/d : peer closed before replying") - break - try: - client.bio_write(sock.recv(65535)) - except socket.timeout: + raise ProbeFailure("peer closed before GET response") from None + except ValueError: + raise ProbeFailure("GET response had invalid CoAP framing") from None + except WantRead: + try: + _receive(client, sock, target) + except socket.timeout: + pass continue + if kind == 3 and reply_id == message_id: + raise ProbeFailure("GET request was reset") + if code == 0: + continue # An empty ACK is not the response. + if reply_token != token or kind == 3 or (kind == 2 and reply_id != message_id): + continue + if kind == 0: + client.send(struct.pack("!BBH", 0x60, 0, reply_id)) + _flush(client, sock, target) + if code != 69: + raise ProbeFailure("GET did not return 2.05") + stream = io.BytesIO(payload) + try: + document = cbor2.CBORDecoder(stream).decode() + except Exception: + raise ProbeFailure("GET payload was invalid CBOR") from None + if stream.read(1) or not isinstance(document, dict): + raise ProbeFailure("GET payload was not one CBOR map") + if document.get("di") != expected_di: + raise ProbeFailure("GET device ID did not match") + print("GET /oic/d : OK (2.05, CBOR map, expected device ID matched)") + return + raise ProbeFailure("GET timed out") + - print(f"GET /oic/d : {code}, {len(payload)} byte payload") - if payload: +def main() -> int: + client = sock = None + result = 1 + stage = "configuration" + completed = False + try: + host = _env("PSK_HOST") + port = int(_env("PSK_PORT")) + local_port = int(_env("PSK_LOCAL_PORT", required=False) or 0) + if not 1 <= port <= 65535 or not 0 <= local_port <= 65535: + raise ValueError("invalid UDP port") + identity = _hexbytes("PSK_IDENTITY_HEX", (16,)) + key = _hexbytes("PSK_KEY_HEX", (16, 32)) + expected_di = _env("PSK_EXPECTED_DI") + target = (socket.gethostbyname(host), port) + client = DtlsPskClient(identity, key, mtu=1400) + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.bind(("0.0.0.0", local_port)) + sock.settimeout(0.5) + print(f"identity : 16 bytes, contains a zero byte: {'yes' if 0 in identity else 'no'}") + stage = "probe" + _probe(client, sock, target, expected_di) + completed = True + result = 0 + except ProbeFailure as exc: + print(f"{stage:<14}: FAILED ({exc})") + except Exception as exc: + # Exception text can contain credentials, addresses, or appliance data. + print(f"{stage:<14}: FAILED ({type(exc).__name__})") + finally: try: - import cbor2 - - document = cbor2.loads(payload) - except Exception: # noqa: BLE001 - absence of cbor2 is not a failure - document = None - if isinstance(document, dict): - keys = sorted(str(k) for k in document) - print(f" decoded keys: {', '.join(keys)}") - di = document.get("di") - if isinstance(di, str): - if expected_di: - print(" di matches PSK_EXPECTED_DI: " - f"{'YES' if di == expected_di else 'NO'}") - else: - print(f" di sha256 : {_digest(di.encode())}") - else: - print(" payload did not decode as a CBOR map " - "(install cbor2 to decode)") - - client.shutdown() - _flush(client, sock) - sock.close() - print("close_notify : sent") - return 0 + if client is not None: + client.shutdown() + sent = sock is not None and _flush(client, sock, target) + print(f"close_notify : {'sent' if sent else 'not queued'}") + if completed and not sent: + result = 1 + except Exception as exc: + print(f"close_notify : FAILED ({type(exc).__name__})") + result = 1 + finally: + if sock is not None: + try: + sock.close() + except Exception as exc: + print(f"socket cleanup: FAILED ({type(exc).__name__})") + result = 1 + return result def _cap_memory(limit_mb: int = 512) -> None: diff --git a/pr-115-purepy-psk/test_dtls_psk.py b/pr-115-purepy-psk/test_dtls_psk.py new file mode 100644 index 0000000..ca5a0b3 --- /dev/null +++ b/pr-115-purepy-psk/test_dtls_psk.py @@ -0,0 +1,94 @@ +"""Real DTLS interoperability with synthetic credentials and memory BIOs.""" +import struct + +import pytest +from OpenSSL import SSL, _util + +from dtls_psk import DtlsPskClient + + +@pytest.mark.parametrize("use_cookie", [False, True]) +def test_openssl_psk_interoperability(use_cookie): + identity, key = bytes(range(16)), b"k" * 16 + context = SSL.Context(SSL.DTLS_METHOD) + context.set_cipher_list(b"ECDHE-PSK-AES128-CBC-SHA256:@SECLEVEL=0") + + @_util.ffi.callback( + "unsigned int(*)(SSL *, const char *, unsigned char *, unsigned int)" + ) + def psk_callback(ssl, received_identity, output, capacity): + if capacity < len(key): + return 0 + _util.ffi.memmove(output, key, len(key)) + return len(key) + + _util.lib.SSL_CTX_set_psk_server_callback(context._context, psk_callback) + if use_cookie: + context.set_cookie_generate_callback(lambda connection: b"offline-cookie") + context.set_cookie_verify_callback( + lambda connection, cookie: cookie == b"offline-cookie" + ) + context.set_options(SSL.OP_COOKIE_EXCHANGE) + client = DtlsPskClient(identity, key) + server = SSL.Connection(context, None) + server.set_accept_state() + completed = [False, False] + saw_identity = False + first_server = None + for turn in range(80): + for index, sender, receiver in ((0, client, server), (1, server, client)): + try: + sender.do_handshake() + completed[index] = True + except SSL.WantReadError: + pass + try: + data = sender.bio_read(65535) + except SSL.WantReadError: + continue + offset = 0 + while offset + 13 <= len(data): + length = int.from_bytes(data[offset + 11:offset + 13], "big") + record = data[offset:offset + 13 + length] + offset += 13 + length + if index == 1 and first_server is None and record[0] == 22: + first_server = (record[1:3], record[13]) + if (index == 0 and record[0] == 22 + and record[3:5] == b"\0\0" and record[13] == 16): + assert record[25:27] == b"\0\x10" + assert record[27:43] == identity + saw_identity = True + receiver.bio_write(data) + if all(completed): + break + assert all(completed), (use_cookie, completed) + assert saw_identity + assert first_server == ( + (b"\xfe\xff", 3) if use_cookie else (b"\xfe\xfd", 2) + ) + client.send(b"client payload") + server.bio_write(client.bio_read(65535)) + assert server.recv(65535) == b"client payload" + server.send(b"server payload") + client.bio_write(server.bio_read(65535)) + assert client.recv(65535) == b"server payload" + client.shutdown() + server.bio_write(client.bio_read(65535)) + with pytest.raises(SSL.ZeroReturnError): + server.recv(65535) + + +@pytest.mark.parametrize("record_version, accepted", [(b"\xfe\xff", False), (b"\xfe\xfd", True)]) +def test_server_hello_requires_dtls12_record_header(record_version, accepted): + client = DtlsPskClient(bytes(range(16)), b"k" * 16) + with pytest.raises(SSL.WantReadError): + client.do_handshake() + client.bio_read(65535) + body = b"\xfe\xfd" + bytes(range(32)) + b"\0\xc0\x37\0" + length = len(body).to_bytes(3, "big") + handshake = b"\x02" + length + b"\0" * 5 + length + body + record = b"\x16" + record_version + b"\0" * 8 + struct.pack("!H", len(handshake)) + handshake + client.bio_write(record) + with pytest.raises(SSL.WantReadError): + client.do_handshake() + assert client._state == ("got_server_hello" if accepted else "sent_hello") diff --git a/pr-115-purepy-psk/test_oven_psk_test.py b/pr-115-purepy-psk/test_oven_psk_test.py new file mode 100644 index 0000000..859e738 --- /dev/null +++ b/pr-115-purepy-psk/test_oven_psk_test.py @@ -0,0 +1,283 @@ +"""Local transport and acceptance tests; no appliance or credentials required.""" +import errno +import socket +import struct +import threading + +import cbor2 +import pytest + +import oven_psk_test as probe +from dtls_psk import DtlsPskClient + + +def record(payload): + return b"\x17\xfe\xfd" + b"\0" * 8 + len(payload).to_bytes(2, "big") + payload + + +class FixtureClient: + """Plaintext records isolate the probe from the DTLS engine.""" + instances = [] + shutdown_fails = False + + def __init__(self, *args, **kwargs): + self.outbox = b"" + self.inbox = [] + self.state = "start" + self.closed = False + self.instances.append(self) + + def do_handshake(self): + if self.state == "start": + self.outbox += record(b"hello") + self.state = "waiting" + if self.inbox: + assert self.inbox.pop(0) == b"hello" + self.state = "established" + if self.state != "established": + raise probe.WantRead() + + def bio_read(self, capacity): + if not self.outbox: + raise probe.WantRead() + value, self.outbox = self.outbox, b"" + return value + + def bio_write(self, data): + self.inbox.append(data[13:]) + + def send(self, data): + self.outbox += record(data) + + def recv(self, capacity): + if not self.inbox: + raise probe.WantRead() + return self.inbox.pop(0) + + def shutdown(self): + self.closed = True + if self.shutdown_fails: + raise RuntimeError("secret key or host must not reach output") + self.outbox += record(b"close_notify") + + def DTLSv1_get_timeout(self): + return None + + +@pytest.fixture +def setup(monkeypatch): + FixtureClient.instances = [] + monkeypatch.setattr(FixtureClient, "shutdown_fails", False) + monkeypatch.setattr(probe, "DtlsPskClient", FixtureClient) + monkeypatch.setattr(probe, "REPLY_TIMEOUT", 0.02) + monkeypatch.setenv("PSK_HOST", "127.0.0.1") + monkeypatch.setenv("PSK_IDENTITY_HEX", "00" * 16) + monkeypatch.setenv("PSK_KEY_HEX", "11" * 16) + monkeypatch.setenv("PSK_EXPECTED_DI", "fixture-device") + monkeypatch.delenv("PSK_LOCAL_PORT", raising=False) + return monkeypatch + + +def response(kind, code, mid, token, document): + payload = document if isinstance(document, bytes) else cbor2.dumps(document) + return bytes([(1 << 6) | (kind << 4) | len(token), code]) + struct.pack("!H", mid) + token + b"\xff" + payload + + +def run_peer(monkeypatch, mode): + """Reply from another socket, as the appliance changes UDP source port.""" + listener = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + listener.bind(("127.0.0.1", 0)) + sender.bind(("127.0.0.1", 0)) + listener.settimeout(2) + monkeypatch.setenv("PSK_PORT", str(listener.getsockname()[1])) + received, errors = [], [] + + def serve(): + try: + hello, target = listener.recvfrom(65535) + assert hello[13:] == b"hello" + sender.sendto(record(b"hello"), target) + request, source = listener.recvfrom(65535) + assert source == target + kind, code, mid, token, payload = probe._coap_parse(request[13:]) + assert (kind, code, payload) == (0, 1, b"") + received.append(("source", source)) + if mode == "timeout": + pass + elif mode == "separate": + response_mid = (mid + 1) & 0xFFFF + sender.sendto(record(struct.pack("!BBH", 0x60, 0, mid)), target) + sender.sendto(record(response(0, 69, response_mid, token, {"di": "fixture-device"})), target) + ack, _ = listener.recvfrom(65535) + assert ack[13:] == struct.pack("!BBH", 0x60, 0, response_mid) + received.append("ACK") + else: + code = 128 if mode == "error" else 69 + body = {"di": "another-device"} if mode == "mismatch" else {"di": "fixture-device"} + if mode == "missing_di": + body = {} + if mode == "invalid_cbor": + body = b"\xff" + if mode == "unrelated": + sender.sendto(record(response(2, 69, mid, b"bad", body)), target) + sender.sendto(record(response(2, code, mid, token, body)), target) + close, _ = listener.recvfrom(65535) + assert close[13:] == b"close_notify" + received.append("close_notify") + except Exception as exc: + errors.append(exc) + finally: + listener.close() + sender.close() + + thread = threading.Thread(target=serve) + thread.start() + result = probe.main() + thread.join(3) + assert not thread.is_alive() + assert not errors + return result, received + + +@pytest.mark.parametrize("mode", ["success", "separate", "unrelated"]) +def test_udp_reply_from_different_port_and_correlated_response(setup, mode, capsys): + result, received = run_peer(setup, mode) + assert result == 0 + assert received[-1] == "close_notify" + if mode == "separate": + assert "ACK" in received + output = capsys.readouterr().out + assert "127.0.0.1" not in output + assert "fixture-device" not in output + assert "111111" not in output + assert "sha256" not in output + + +@pytest.mark.parametrize("mode", ["timeout", "error", "mismatch", "missing_di", "invalid_cbor"]) +def test_failed_read_returns_nonzero_and_closes(setup, mode): + result, received = run_peer(setup, mode) + assert result == 1 + assert received[-1] == "close_notify" + assert FixtureClient.instances[-1].closed + + +def test_discards_datagrams_from_foreign_host(): + class Socket: + def recvfrom(self, capacity): + return b"foreign", ("127.0.0.2", 4567) + + class Client: + def bio_write(self, data): + pytest.fail("foreign host entered DTLS engine") + + probe._receive(Client(), Socket(), ("127.0.0.1", 1234)) + + +def test_local_port_is_bound(setup): + reservation = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + reservation.bind(("127.0.0.1", 0)) + port = reservation.getsockname()[1] + reservation.close() + setup.setenv("PSK_LOCAL_PORT", str(port)) + result, received = run_peer(setup, "success") + assert result == 0 + assert received[0][1][1] == port + + +def test_local_port_bind_failure_cleans_up_without_sending(setup, capsys): + class Socket: + closed = False + sent = [] + + def bind(self, address): + raise OSError(errno.EADDRINUSE, "private bind address") + + def sendto(self, data, address): + self.sent.append(data) + + def close(self): + self.closed = True + + class Client(DtlsPskClient): + shutdown_attempted = False + + def shutdown(self): + self.shutdown_attempted = True + super().shutdown() + + sock = Socket() + client = Client(bytes(range(16)), b"k" * 16) + setup.setenv("PSK_PORT", "1234") + setup.setenv("PSK_LOCAL_PORT", "4321") + setup.setattr(probe.socket, "socket", lambda *args: sock) + setup.setattr(probe, "DtlsPskClient", lambda *args, **kwargs: client) + setup.setattr(probe, "_probe", lambda *args: pytest.fail("probe ran after bind failure")) + assert probe.main() == 1 + assert sock.closed and client.shutdown_attempted + assert not sock.sent + assert "private bind address" not in capsys.readouterr().out + + +def test_shutdown_failure_still_closes_socket_and_fails(setup, capsys): + class Socket: + closed = False + def bind(self, address): + pass + def settimeout(self, value): + pass + def close(self): + self.closed = True + + sock = Socket() + setup.setenv("PSK_PORT", "1234") + setup.setattr(probe.socket, "socket", lambda *args: sock) + setup.setattr(probe, "_probe", lambda *args: None) + setup.setattr(FixtureClient, "shutdown_fails", True) + assert probe.main() == 1 + assert sock.closed + assert "secret key" not in capsys.readouterr().out + + +def test_handshake_error_still_attempts_shutdown_and_closes(setup, capsys): + class Socket: + closed = False + def bind(self, address): + pass + def settimeout(self, value): + pass + def sendto(self, data, address): + return len(data) + def close(self): + self.closed = True + + sock = Socket() + setup.setenv("PSK_PORT", "1234") + setup.setattr(probe.socket, "socket", lambda *args: sock) + def fail(*args): + raise RuntimeError("private device ID or PSK") + setup.setattr(FixtureClient, "do_handshake", fail) + assert probe.main() == 1 + assert sock.closed and FixtureClient.instances[-1].closed + assert "private device" not in capsys.readouterr().out + + +@pytest.mark.parametrize("datagram", [b"", b"\x40\x45\0\0\xfd", b"\x40\x45\0\0\xff", b"\x49\x45\0\0", b"\x40\x45\0\0\x01"]) +def test_rejects_invalid_coap(datagram): + with pytest.raises(ValueError): + probe._coap_parse(datagram) + + +@pytest.mark.parametrize("datagram", [ + b"\x60\0\x12\x34\xffpayload", + b"\x60\0\x12\x34\0", + b"\x61\0\x12\x34x", +]) +def test_rejects_empty_coap_with_payload_option_or_token(datagram): + with pytest.raises(ValueError, match="invalid empty CoAP message"): + probe._coap_parse(datagram) + + +def test_parses_extended_option_length(): + datagram = b"\x60\x45\x12\x34\x0e\0\0" + b"x" * 269 + b"\xffpayload" + assert probe._coap_parse(datagram) == (2, 69, 0x1234, b"", b"payload")