diff --git a/docs/source/contributing/conventions/protocol-family-grouping.rst b/docs/source/contributing/conventions/protocol-family-grouping.rst index 6341b6a216..765db71b83 100644 --- a/docs/source/contributing/conventions/protocol-family-grouping.rst +++ b/docs/source/contributing/conventions/protocol-family-grouping.rst @@ -230,12 +230,13 @@ are all ``False``; only ``frame.protochain`` matches them. Chained Records ~~~~~~~~~~~~~~~ -**TLS and SSLv2 records sharing a TCP segment are chained layers, not a list on one -layer** (:issue:`2517`, :issue:`2567`): each record dissected after the first is the -previous record's payload, so a TLS record's dynamic alias names its own version -- a -``ServerHello`` then a ``ChangeCipherSpec`` read ``TCP:TLS 1.3:TLS 1.2`` -- and an SSLv2 -``CLIENT-HELLO`` then a ``CLIENT-MASTER-KEY`` read ``TCP:SSLv2:SSLv2``. DTLS does not -chain: two DTLS records in one datagram read ``DTLS 1.2:Raw`` (:issue:`2535`). +**TLS and SSLv2 records sharing a TCP segment, and DTLS records sharing a UDP datagram, +are chained layers, not a list on one layer** (:issue:`2517`, :issue:`2567`, +:issue:`2535`): each record dissected after the first is the previous record's payload, +so a TLS or DTLS record's dynamic alias names its own version -- a ``ServerHello`` then a +``ChangeCipherSpec`` read ``TCP:TLS 1.3:TLS 1.2``, and a DTLS 1.2 record then a DTLS 1.3 +unified-header record read ``UDP:DTLS 1.2:DTLS 1.3`` -- and an SSLv2 ``CLIENT-HELLO`` +then a ``CLIENT-MASTER-KEY`` read ``TCP:SSLv2:SSLv2``. **HTTP/2 frames chain the same way** (:issue:`2536`): a ``SETTINGS`` frame then a ``WINDOW_UPDATE`` read ``TCP:HTTP/2:HTTP/2``. A later whole frame that fails to decode diff --git a/docs/source/pcapkit/protocols/application/dtls.rst b/docs/source/pcapkit/protocols/application/dtls.rst index 8aa3c61c86..ba9f1567cb 100644 --- a/docs/source/pcapkit/protocols/application/dtls.rst +++ b/docs/source/pcapkit/protocols/application/dtls.rst @@ -46,6 +46,28 @@ A unified-header record has no legacy version to read. Its first octet's top three bits are ``001``, which no legacy content type (20 through 27) sets, so the two formats are told apart by that octet alone. +Records in One Datagram +----------------------- + +A whole record after the first in a UDP datagram is dissected as a further +:class:`~pcapkit.protocols.application.dtls.DTLS` layer, the previous record's +payload (:rfc:`6347#section-4.1.1`, :rfc:`9147#section-4.3`, :issue:`2535`): a +``ChangeCipherSpec`` followed by an ``Alert`` reads ``UDP:DTLS 1.2:DTLS 1.2``. +A unified-header record is delimited by its length; one without a length runs +to the end of the datagram, so it is the last (:rfc:`9147#section-4`). The +exceptions: + +* A whole record that fails to decode becomes + :class:`~pcapkit.protocols.misc.raw.Raw` together with everything after it. +* Octets that open no whole record stay raw, with no error: an unsupported + content type, a legacy version outside ``0xFExx``, a header cut short, a + length past the end of the datagram, or a unified header with a connection + ID, which leaves the record's end unknown. +* A parse told to stop at DTLS (``protocol='DTLS'`` or ``layer='Application'``) + keeps the rest of the datagram raw. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + Known Limitations ----------------- diff --git a/pcapkit/protocols/application/application.py b/pcapkit/protocols/application/application.py index 35c844507f..d606f1e024 100644 --- a/pcapkit/protocols/application/application.py +++ b/pcapkit/protocols/application/application.py @@ -45,8 +45,9 @@ class Application(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable for the rest of the packet undissected and resolves to :class:`~pcapkit.protocols.misc.raw.Raw` -- trailing bytes are not a further protocol layer, unless the subclass dissects them, as - :class:`~pcapkit.protocols.application.tls.TLS` and - :class:`~pcapkit.protocols.application.sslv2.SSLv2` do for a further record and + :class:`~pcapkit.protocols.application.tls.TLS`, + :class:`~pcapkit.protocols.application.sslv2.SSLv2` and + :class:`~pcapkit.protocols.application.dtls.DTLS` do for a further record and :class:`~pcapkit.protocols.application.httpv2.HTTPv2` for a further frame -- or to :class:`~pcapkit.protocols.misc.null.NoPayload` when nothing remains. diff --git a/pcapkit/protocols/application/dtls.py b/pcapkit/protocols/application/dtls.py index 2c6018b3ac..eeff80680c 100644 --- a/pcapkit/protocols/application/dtls.py +++ b/pcapkit/protocols/application/dtls.py @@ -13,6 +13,27 @@ :meth:`~DTLS.read` sets :attr:`~DTLS.alias` from the record's legacy version, or to ``DTLS 1.3`` for a unified-header record. +Records in One Datagram +----------------------- + +A whole record after the first in a UDP datagram is dissected as a further +:class:`DTLS` layer, the previous record's payload (:rfc:`6347#section-4.1.1`, +:rfc:`9147#section-4.3`, :issue:`2535`): a ``ChangeCipherSpec`` followed by an +``Alert`` reads ``UDP:DTLS 1.2:DTLS 1.2``. A unified-header record is delimited +by its length; one without a length runs to the end of the datagram, so it is +the last (:rfc:`9147#section-4`). The exceptions: + +* A whole record that fails to decode becomes + :class:`~pcapkit.protocols.misc.raw.Raw` together with everything after it. +* Octets that open no whole record stay raw, with no error: an unsupported + content type, a legacy version outside ``0xFExx``, a header cut short, a + length past the end of the datagram, or a unified header with a connection + ID, which leaves the record's end unknown. +* A parse told to stop at DTLS (``protocol='DTLS'`` or ``layer='Application'``) + keeps the rest of the datagram raw. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + Known Limitations ----------------- @@ -41,7 +62,9 @@ from typing import TYPE_CHECKING, cast from pcapkit.const.tls.content_type import ContentType as Enum_ContentType -from pcapkit.protocols.application.ssl import DTLS13_VERSION, SSL +from pcapkit.corekit.decorators import beholder +from pcapkit.protocols.application.ssl import (_DTLS_VERSION_MAJOR, _WIRE_CONTENT_TYPES, + DTLS13_VERSION, SSL) from pcapkit.protocols.data.application.dtls import DTLS as Data_DTLS from pcapkit.protocols.data.application.dtls import DTLSCiphertext as Data_DTLSCiphertext from pcapkit.protocols.schema.application.dtls import DTLS as Schema_DTLS @@ -61,6 +84,10 @@ __all__ = ['DTLS'] +#: Length of a legacy record header: content type, legacy version, epoch, +#: sequence number and length (:rfc:`9147#section-4`). +_LEGACY_HEADER_LENGTH = 13 + #: Legacy record version DTLS 1.0 carries (:rfc:`4347#section-4.1`). _DTLSV1_0_VERSION = 0xFEFF #: Legacy record version DTLS 1.2 carries, and the value DTLS 1.3 freezes its @@ -105,6 +132,44 @@ def _ciphertext_header_length(flags: 'dict[str, int]') -> 'int': return 1 + (2 if flags['sequence_long'] else 1) + (2 if flags['length_present'] else 0) +def _opens_whole_record(data: 'bytes') -> 'bool': + """Whether ``data`` opens with a whole DTLS record. + + Only the record header is read. A legacy header gets the checks of + :meth:`SSL._validate_tls_record + `: a content + type a genuine record carries, a ``0xFExx`` legacy version, and a fragment + that ends inside ``data``. A unified header must carry no connection ID, + whose negotiated length leaves the record's end unknown, and a fragment + that ends inside ``data``; without a length it runs to the end of ``data``, + as the last record of a datagram does (:rfc:`9147#section-4`). + + Args: + data: Octets after a record. + + Returns: + Whether a further record opens ``data`` and ends inside it. + + """ + if not data: + return False + if _is_unified_header(data[0]): + # NOTE: the ``C``, ``S`` and ``L`` bits of ``001CSLEE``, as the header schema reads them. + flags = {'connection_id': data[0] >> 4 & 1, 'sequence_long': data[0] >> 3 & 1, + 'length_present': data[0] >> 2 & 1} + header = _ciphertext_header_length(flags) + if flags['connection_id'] or len(data) < header: + return False + if not flags['length_present']: + return True + else: + header = _LEGACY_HEADER_LENGTH + if len(data) < header or data[0] not in _WIRE_CONTENT_TYPES or data[1] != _DTLS_VERSION_MAJOR: + return False + # Either header ends with its 16-bit length. + return header + int.from_bytes(data[header - 2:header], 'big') <= len(data) + + class DTLS(SSL[Data_DTLS, Schema_DTLS], schema=Schema_DTLS, data=Data_DTLS): """This class implements all protocols in the DTLS record-layer family. @@ -148,7 +213,7 @@ def length(self) -> 'int': schema = getattr(self, '__header__', None) if isinstance(schema, Schema_DTLSCiphertext): return _ciphertext_header_length(schema.flags) - return 13 + return _LEGACY_HEADER_LENGTH ########################################################################## # Methods. @@ -323,7 +388,7 @@ def read_handshake(cls, message: 'bytes') -> 'Data_DTLSHandshake': def __length_hint__(self) -> 'Literal[13]': """Return an estimated length for the object.""" - return 13 + return _LEGACY_HEADER_LENGTH # type: ignore[return-value] ########################################################################## # Utilities. @@ -472,3 +537,53 @@ def _read_dtls_ciphertext(self, length: 'Optional[int]' = None, **kwargs: 'Any') ) remaining = available - (self.length + len(schema.fragment)) return self._decode_next_layer(dtls, -1, remaining) + + def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # type: ignore[override] + packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase': + """Import next layer extractor: a further whole record, or the rest undissected. + + A UDP datagram may carry several records (:rfc:`9147#section-4.3`, + :issue:`2535`). So the ``-1`` remainder is dissected as this class when + :func:`_opens_whole_record` finds a whole record opening it, and is + otherwise kept as :class:`~pcapkit.protocols.misc.raw.Raw`, as + :meth:`Application._import_next_layer + ` + keeps it. + + Arguments: + proto: next layer protocol index; only the ``-1`` sentinel + (the rest of the datagram) is accepted + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack `) + + Returns: + Instance of this class for a further record, or as + :meth:`Application._import_next_layer + ` + returns otherwise. + + """ + if proto == -1 and not self._sigterm and _opens_whole_record(self._get_payload()[:length]): + return self._import_next_record(proto, length, packet=packet) # type: ignore[misc,call-arg] + return super()._import_next_layer(proto, length, packet=packet) + + @beholder # type: ignore[arg-type] + def _import_next_record(self, proto: 'int', length: 'Optional[int]' = None, *, # type: ignore[misc] + packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase': + """Import the further record :meth:`_import_next_layer` found. + + Arguments: + proto: next layer protocol index, the ``-1`` sentinel + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack `) + + Returns: + Instance of this class, or of :class:`~pcapkit.protocols.misc.raw.Raw` + when :func:`~pcapkit.corekit.decorators.beholder` replaces a record + that fails to parse. + + """ + file_ = self._get_payload() + return self._parse_next_layer(type(self), file_, len(file_) if length is None else length, + alias=proto, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx) diff --git a/tests/protocols/application/test_dtls13_unified_header_unit.py b/tests/protocols/application/test_dtls13_unified_header_unit.py index 9d07c1acc9..9f0282fc19 100644 --- a/tests/protocols/application/test_dtls13_unified_header_unit.py +++ b/tests/protocols/application/test_dtls13_unified_header_unit.py @@ -18,7 +18,7 @@ _L_SHORT_SEQ_EPOCH1 = 0x25 # S=0, L=1, EE=01 _L_LONG_SEQ_EPOCH2 = 0x2E # S=1, L=1, EE=10 _NO_LENGTH_EPOCH3 = 0x23 # S=0, L=0, EE=11 -_CID_PRESENT = 0x38 # C=1, S=0, L=1, EE=00 +_CID_PRESENT = 0x34 # C=1, S=0, L=1, EE=00 #: A 16-octet stand-in for an AEAD ciphertext, the shortest the record-number #: sample allows (:rfc:`9147#section-4.2.3`). @@ -187,6 +187,58 @@ def test_a_datagram_the_record_format_forbids_degrades_to_raw_on_dispatch(self) self.assertEqual(str(proto.protochain), 'UDP:Raw') +class UnifiedHeaderChainTests(unittest.TestCase): + """How a unified-header record chains with the records sharing its datagram + (RFC 9147 section 4.3, #2535).""" + + #: A legacy-format ``change_cipher_spec`` record, DTLS 1.2. + CCS = b'\x14\xfe\xfd\x00\x00' + (1).to_bytes(6, 'big') + b'\x00\x01\x01' + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_a_record_after_one_with_a_length_is_a_further_layer(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + first = _unified(_L_SHORT_SEQ_EPOCH1, b'\x07', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + second = _unified(_L_LONG_SEQ_EPOCH2, b'\x12\x34', (20).to_bytes(2, 'big'), bytes(20)) + for name, wire, chain in (('a unified header', first + second, 'DTLS 1.3:DTLS 1.3'), + ('a legacy header', first + self.CCS, 'DTLS 1.3:DTLS 1.2'), + ('after a legacy header', self.CCS + first, 'DTLS 1.2:DTLS 1.3')): + with self.subTest(next=name): + parsed = DTLS(wire, len(wire)) + self.assertEqual(str(parsed.protochain), chain) + self.assertEqual(bytes(DTLS.from_data(parsed.info)), wire) + parsed = DTLS(first + second, len(first + second)).payload + self.assertEqual((parsed.info.epoch, parsed.info.sequence_number, parsed.info.length), (2, 0x1234, 20)) + + def test_a_record_without_a_length_is_the_last(self) -> None: + """It runs to the end of the datagram, so octets that would open a record are its ciphertext.""" + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.misc.null import NoPayload + + last = _unified(_NO_LENGTH_EPOCH3, b'\x07', None, _FRAGMENT + self.CCS) + wire = self.CCS + last + parsed = DTLS(wire, len(wire)) + self.assertEqual(str(parsed.protochain), 'DTLS 1.2:DTLS 1.3') + self.assertEqual(parsed.payload.info.content, _FRAGMENT + self.CCS) + self.assertIsInstance(parsed.payload.payload, NoPayload) + self.assertEqual(bytes(DTLS.from_data(parsed.info)), wire) + + def test_a_connection_id_after_a_record_stays_raw_without_an_error(self) -> None: + """Its negotiated length leaves the record's end unknown, so no whole record opens there.""" + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.misc.raw import Raw + + refused = _unified(_CID_PRESENT, b'\x07', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + wire = self.CCS * 2 + refused + parsed = DTLS(wire, len(wire)) + self.assertEqual(str(parsed.protochain), 'DTLS 1.2:DTLS 1.2:Raw') + self.assertIs(type(parsed.payload.payload), Raw) + self.assertEqual(bytes(parsed.payload.payload), refused) + self.assertIsNone(parsed.payload.payload.info.error) + + class UnifiedHeaderMakeTests(unittest.TestCase): def setUp(self) -> None: reimport_once_per_class(self) diff --git a/tests/protocols/application/test_ssl_tls_dtls_unit.py b/tests/protocols/application/test_ssl_tls_dtls_unit.py index e7e3144e92..192cc14ec5 100644 --- a/tests/protocols/application/test_ssl_tls_dtls_unit.py +++ b/tests/protocols/application/test_ssl_tls_dtls_unit.py @@ -1151,6 +1151,98 @@ def test_only_the_first_record_gets_the_sslv2_trial(self) -> None: self.assertEqual(trial.call_count, calls) +class DatagramRecordTests(unittest.TestCase): + """A whole record after the first in a UDP datagram is a further DTLS layer + (#2535); see ``test_tls_dtls_roundtrip_unit`` for the round trips.""" + + CCS = b'\x14\xfe\xfd\x00\x00' + (1).to_bytes(6, 'big') + b'\x00\x01\x01' + ALERT = b'\x15\xfe\xfd\x00\x00' + (2).to_bytes(6, 'big') + b'\x00\x02\x02\x0a' + #: A whole handshake record whose unfragmented ClientHello is too short for its fixed fields. + MALFORMED_HELLO = (b'\x16\xfe\xfd\x00\x00' + (3).to_bytes(6, 'big') + b'\x00\x16' + + b'\x01\x00\x00\x0a' + b'\x00\x00' + b'\x00\x00\x00' + b'\x00\x00\x0a' + bytes(10)) + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _udp(self, payload: bytes) -> 'tuple[object, list[warnings.WarningMessage]]': + """Parse ``payload`` on UDP 5684 under ``always``, with the warnings it reports.""" + from pcapkit.protocols.transport.udp import UDP + + raw = _udp_to_5684(payload) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + proto = UDP(raw, len(raw)) + return proto, caught + + def test_frame_layer_limit_bounds_the_records(self) -> None: + """Each record is a layer, so the limit stops the walk with one warning.""" + from pcapkit.protocols.transport.udp import UDP + from pcapkit.utilities import config + from pcapkit.utilities.warnings import ProtocolWarning + + payload = self.CCS * (config.FRAME_LAYER_LIMIT + 6) + proto, caught = self._udp(payload) + self.assertEqual(str(proto.protochain), # type: ignore[attr-defined] + ':'.join(['UDP'] + ['DTLS 1.2'] * config.FRAME_LAYER_LIMIT + ['Raw'])) + self.assertEqual([item.category for item in caught], [ProtocolWarning]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(bytes(UDP.from_data(proto.info)), _udp_to_5684(payload)) # type: ignore[attr-defined] + + def test_a_parse_stopping_at_dtls_keeps_the_rest_raw(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.transport.udp import UDP + + wire = self.CCS + self.ALERT + raw = _udp_to_5684(wire) + self.assertEqual(str(DTLS(wire, len(wire)).protochain), 'DTLS 1.2:DTLS 1.2') + for limit in ({'protocol': 'DTLS'}, {'layer': 'Application'}): + with self.subTest(**limit): + self.assertEqual(str(DTLS(wire, len(wire), **limit).protochain), 'DTLS 1.2:Raw') + proto = UDP(raw, len(raw), **limit) + self.assertEqual(str(proto.protochain), 'UDP:DTLS 1.2:Raw') + self.assertEqual(bytes(proto.payload.payload), self.ALERT) + + def test_octets_opening_no_whole_record_stay_raw_without_an_error(self) -> None: + from pcapkit.protocols.misc.raw import Raw + + for name, tail in (('content type 0', b'\x00' + self.CCS[1:]), + ('tls12_cid, whose connection ID has no fixed length', b'\x19' + self.CCS[1:]), + ('a TLS record version', self.CCS[:1] + b'\x03\x03' + self.CCS[3:]), + ('a header cut short', self.CCS[:12]), + ('a length past the datagram', self.CCS[:11] + b'\x00\x05\x01'), + ('a unified header with a connection ID', b'\x34\x07\x00\x10' + bytes(16)), + ('a unified header cut short', b'\x2c\x00\x01')): + with self.subTest(tail=name): + proto, caught = self._udp(self.CCS + self.ALERT + tail) + self.assertEqual(str(proto.protochain), 'UDP:DTLS 1.2:DTLS 1.2:Raw') # type: ignore[attr-defined] + rest = proto.payload.payload.payload # type: ignore[attr-defined] + self.assertIs(type(rest), Raw) + self.assertEqual(bytes(rest), tail) + self.assertIsNone(rest.info.error) + self.assertEqual(caught, []) + + def test_a_later_whole_record_that_fails_to_decode_is_raw_with_the_rest(self) -> None: + """``beholder`` keeps the record, and everything after it, as ``Raw`` (ruling A on #2517).""" + from pcapkit.protocols.transport.udp import UDP + + too_short = b'\x25\x07\x00\x0f' + bytes(15) # a ciphertext under the record-number sample + for name, bad, error in ( + ('a ClientHello short of its fields', self.MALFORMED_HELLO, + r'^DTLS 1\.2: invalid format: client_hello\.random runs past'), + ('a unified-header record short of the sample', too_short, + r'^DTLS 1\.3: invalid format: ciphertext is shorter than the record-number sample')): + with self.subTest(record=name): + payload = self.CCS + self.ALERT + bad + self.CCS + proto, caught = self._udp(payload) + self.assertEqual(str(proto.protochain), 'UDP:DTLS 1.2:DTLS 1.2:Raw') # type: ignore[attr-defined] + rest = proto.payload.payload.payload # type: ignore[attr-defined] + self.assertEqual(bytes(rest), bad + self.CCS) + self.assertRegex(rest.info.error, error) + self.assertEqual(caught, []) + self.assertEqual(bytes(UDP.from_data(proto.info)), _udp_to_5684(payload)) # type: ignore[attr-defined] + + class ExportTests(unittest.TestCase): """The family is exported, and registered, the way every sibling application protocol is -- :class:`~pcapkit.protocols.application.ngap.NGAP` diff --git a/tests/protocols/application/test_tls_dtls_roundtrip_unit.py b/tests/protocols/application/test_tls_dtls_roundtrip_unit.py index 1aea6eb140..7f6c876051 100644 --- a/tests/protocols/application/test_tls_dtls_roundtrip_unit.py +++ b/tests/protocols/application/test_tls_dtls_roundtrip_unit.py @@ -16,8 +16,9 @@ :meth:`~pcapkit.protocols.application.tls.TLS.make`/ :meth:`~pcapkit.protocols.application.dtls.DTLS.make` and reading them back (the make-then-parse direction) -- so neither direction is a tautology of -the other. Segments of several TLS records are checked the same way, each -record a layer of its own (#2517). +the other. Segments of several TLS records, and datagrams of several DTLS +records, are checked the same way, each record a layer of its own (#2517, +#2535). """ from __future__ import annotations @@ -50,8 +51,7 @@ def _client_hello_body(cipher_suites: bytes = b'\x13\x01', extensions: 'bytes | + tail) -def _server_hello_body(extensions: 'bytes | None' = b'') -> bytes: - version = b'\x03\x03' +def _server_hello_body(extensions: 'bytes | None' = b'', version: bytes = b'\x03\x03') -> bytes: random = bytes(range(32, 64)) session_id = b'' cipher_suite = b'\x13\x01' @@ -180,6 +180,40 @@ def _dtls_handshake(msg_type: int, body: bytes, *, message_seq: int, fragment_of } +def _unified_record(fragment: bytes, *, seq: int, length_present: bool = True) -> bytes: + """A unified-header record at epoch 2, the handshake traffic keys' (RFC 9147 + section 6.1), with an 8-bit sequence number: first octet ``001 0 0 L 10`` + (RFC 9147 section 4).""" + if length_present: + return bytes([0x26, seq]) + len(fragment).to_bytes(2, 'big') + fragment + return bytes([0x22, seq]) + fragment + + +#: A DTLS ``ServerHello`` selecting DTLS 1.3 through ``supported_versions`` (type 43), +#: at the ``legacy_version`` RFC 9147 section 5.4 gives it. +_DTLS13_SH_BODY = _server_hello_body(extensions=b'\x00\x2b\x00\x02\xfe\xfc', version=b'\xfe\xfd') +_DTLS13_SERVER_HELLO = _dtls_record(22, _dtls_handshake( + 2, _DTLS13_SH_BODY, message_seq=0, fragment_offset=0, fragment_length=len(_DTLS13_SH_BODY)), seq=0) + +#: Datagrams of several records each, every one dissected as a DTLS layer (#2535). +DTLS_DATAGRAMS = { + 'change_cipher_spec + alert': _dtls_record(20, b'\x01', seq=1) + _dtls_record(21, b'\x02\x0a', seq=2), + # A DTLS 1.2 client's second flight: ClientKeyExchange, then the epoch-1 Finished. + 'ClientKeyExchange + change_cipher_spec + encrypted Finished': ( + _dtls_record(22, _dtls_handshake(16, bytes(range(33)), message_seq=2, fragment_offset=0, + fragment_length=33), seq=2) + + _dtls_record(20, b'\x01', seq=3) + + _dtls_record(22, (b'\x5c\x91\x0e\xd7' * 12)[:48], epoch=1, seq=0)), + # A DTLS 1.3 server flight: the last record may omit its length (RFC 9147 section 4). + 'DTLS 1.3 ServerHello + unified-header records, the last without a length': ( + _DTLS13_SERVER_HELLO + _unified_record(bytes(range(32)), seq=0) + + _unified_record(bytes(range(32, 80)), seq=1, length_present=False)), +} + +#: A record the datagram ends inside: 20 octets of a 63-octet record. +_PARTIAL_DTLS_RECORD = _dtls_record(23, bytes(50))[:20] + + class TestTLSRoundTrip(unittest.TestCase): """Pin the rebuild of parsed TLS records, both directions.""" @@ -457,5 +491,149 @@ def test_unfragmented_client_hello_with_an_empty_cookie(self) -> None: self.assertEqual(message.body.cookie, b'') +def _split_dtls_records(wire: bytes) -> 'list[bytes]': + """``wire`` cut at each record's end, as RFC 9147 section 4 frames it: a legacy + header's length, a unified header's length, or the end of the datagram.""" + records = [] + while wire: + if wire[0] >> 5 == 0b001: + header = 1 + (2 if wire[0] & 0x08 else 1) + (2 if wire[0] & 0x04 else 0) + end = header + int.from_bytes(wire[header - 2:header], 'big') if wire[0] & 0x04 else len(wire) + else: + end = 13 + int.from_bytes(wire[11:13], 'big') + records.append(wire[:end]) + wire = wire[end:] + return records + + +def _dtls_layers(proto: 'object') -> 'tuple[list, object]': + """Each DTLS layer of ``proto``'s chain, in order, and the layer that ends it.""" + from pcapkit.protocols.application.dtls import DTLS + + layers = [] + while isinstance(proto, DTLS): + layers.append(proto) + proto = proto.payload + return layers, proto + + +class TestDTLSDatagramRoundTrip(unittest.TestCase): + """Pin every record of a datagram as its own layer, rebuilt both directions (#2535).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_every_record_is_a_dtls_layer(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.data.application.dtls import DTLSCiphertext + from pcapkit.protocols.misc.null import NoPayload + + for name, wire in DTLS_DATAGRAMS.items(): + with self.subTest(datagram=name): + records = _split_dtls_records(wire) + layers, rest = _dtls_layers(DTLS(wire, len(wire))) + self.assertEqual(len(layers), len(records)) + self.assertIsInstance(rest, NoPayload) + for layer, record in zip(layers, records): + if isinstance(layer.info, DTLSCiphertext): + self.assertEqual(layer.info.epoch, record[0] & 0x03) + self.assertEqual(layer.info.content, record[layer.length:]) + continue + self.assertEqual(int(layer.info.content_type), record[0]) + self.assertEqual(layer.info.version, int.from_bytes(record[1:3], 'big')) + self.assertEqual(layer.info.epoch, int.from_bytes(record[3:5], 'big')) + self.assertEqual(layer.info.sequence_number, int.from_bytes(record[5:11], 'big')) + self.assertEqual(layer.info.length, len(record) - 13) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.transport.udp import UDP + + for name, wire in DTLS_DATAGRAMS.items(): + with self.subTest(datagram=name): + parsed = DTLS(wire, len(wire)) + rebuilt = DTLS.from_data(parsed.info) + self.assertEqual(bytes(rebuilt).hex(), wire.hex()) + self.assertEqual(len(_dtls_layers(rebuilt)[0]), len(_split_dtls_records(wire))) + + datagram = (50000).to_bytes(2, 'big') + (5684).to_bytes(2, 'big') \ + + (8 + len(wire)).to_bytes(2, 'big') + b'\x00\x00' + wire + udp = UDP(datagram, len(datagram)) + self.assertEqual(str(udp.protochain), f'UDP:{parsed.protochain}') + self.assertEqual(bytes(UDP.from_data(udp.info)).hex(), datagram.hex()) + + def test_make_then_parse_is_byte_exact(self) -> None: + """Each record made from its parsed fields, carrying the next as ``payload``.""" + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.data.application.dtls import DTLSCiphertext + + for name, wire in DTLS_DATAGRAMS.items(): + with self.subTest(datagram=name): + parsed = DTLS(wire, len(wire)) + layers, _ = _dtls_layers(parsed) + maker = object.__new__(DTLS) + schema = b'' # type: object + for layer in reversed(layers): + info = layer.info + if isinstance(info, DTLSCiphertext): + schema = maker.make(ciphertext=True, epoch=info.epoch, + sequence_number=info.sequence_number, + sequence_long=info.sequence_long, + length_present=info.length is not None, + content=info.content, payload=schema) + else: + schema = maker.make(content_type=info.content_type, version=info.version, + epoch=info.epoch, sequence_number=info.sequence_number, + content=info.content, payload=schema) + made = bytes(schema) # type: ignore[call-overload] + self.assertEqual(made.hex(), wire.hex()) + self.assertEqual(str(DTLS(made, len(made)).protochain), str(parsed.protochain)) + + def test_records_match_the_specification(self) -> None: + """Each record's own tier and fields, per its header (RFC 9147 section 4).""" + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.dtls import DTLS + + wire = DTLS_DATAGRAMS['DTLS 1.3 ServerHello + unified-header records, the last without a length'] + layers, _ = _dtls_layers(DTLS(wire, len(wire))) + self.assertEqual([layer.alias for layer in layers], ['DTLS 1.3'] * 3) + self.assertEqual(layers[0].info.content_type, ContentType.handshake) + self.assertEqual(int(layers[0].info.content[0].msg_type), 2) + self.assertEqual([(layer.info.epoch, layer.info.sequence_number) for layer in layers[1:]], + [(2, 0), (2, 1)]) + self.assertEqual([layer.info.length for layer in layers[1:]], [32, None]) + self.assertEqual([layer.info.content for layer in layers[1:]], [bytes(range(32)), bytes(range(32, 80))]) + + wire = DTLS_DATAGRAMS['ClientKeyExchange + change_cipher_spec + encrypted Finished'] + layers, _ = _dtls_layers(DTLS(wire, len(wire))) + self.assertEqual([layer.alias for layer in layers], ['DTLS 1.2'] * 3) + self.assertEqual([(layer.info.epoch, layer.info.sequence_number) for layer in layers], + [(0, 2), (0, 3), (1, 0)]) + self.assertEqual(int(layers[0].info.content[0].msg_type), 16) + self.assertEqual(layers[1].info.content.msg, 1) + self.assertEqual(layers[2].info.content, (b'\x5c\x91\x0e\xd7' * 12)[:48]) + + # The alert decodes, rather than staying 15 raw octets. + wire = DTLS_DATAGRAMS['change_cipher_spec + alert'] + layers, _ = _dtls_layers(DTLS(wire, len(wire))) + self.assertEqual((int(layers[1].info.content.level), int(layers[1].info.content.description)), (2, 10)) + + def test_trailing_partial_record_stays_raw(self) -> None: + """A record the datagram ends inside is kept as its own octets, with no error.""" + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.misc.raw import Raw + + for name in ('change_cipher_spec + alert', 'ClientKeyExchange + change_cipher_spec + encrypted Finished'): + with self.subTest(datagram=name): + wire = DTLS_DATAGRAMS[name] + _PARTIAL_DTLS_RECORD + parsed = DTLS(wire, len(wire)) + layers, rest = _dtls_layers(parsed) + self.assertEqual(len(layers), len(_split_dtls_records(DTLS_DATAGRAMS[name]))) + self.assertIs(type(rest), Raw) + self.assertEqual(bytes(rest), _PARTIAL_DTLS_RECORD) + self.assertIsNone(rest.info.error) # type: ignore[attr-defined] + self.assertEqual(bytes(DTLS.from_data(parsed.info)).hex(), wire.hex()) + + if __name__ == '__main__': unittest.main()