From a020c98b6e040b59ebfa1c88f53170bb3d05b27c Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 23:03:38 -0500 Subject: [PATCH] feat(application): dissect RFC 9146 connection-ID DTLS records (#2515) - tls12_cid records fell back to Raw: their CID sits before the length and its length is negotiated. DTLS parses them with a DTLSCIDCiphertext schema and data model, keeping the content encrypted. - DTLSContext resolves the length from CIDs the hellos negotiated, matched by value, then from a caller's cid_length; unknown stays Raw, quietly. Learned CIDs live on the instance, and each extraction gets its own. - SSL decodes the connection_id extension; a TLS hello carrying it, DTLS-only per RFC 9146 section 10.2, decodes too. make() takes cid. - Chained records, rebuilds and makes carry the CIDs they need. Tests: an Mbed TLS 4.2.0 capture cross-checked with tshark 4.6.9, plus RFC 9146 Figure 3 vectors; both round trips byte-exact. --- .../pcapkit/protocols/application/dtls.rst | 86 ++- .../pcapkit/protocols/application/ssl.rst | 6 + pcapkit/protocols/application/dtls.py | 424 +++++++++++- pcapkit/protocols/application/ssl.py | 52 ++ pcapkit/protocols/data/application/dtls.py | 29 + pcapkit/protocols/data/application/tls.py | 17 + pcapkit/protocols/schema/application/dtls.py | 46 +- .../test_dtls_connection_id_2515_unit.py | 654 ++++++++++++++++++ 8 files changed, 1277 insertions(+), 37 deletions(-) create mode 100644 tests/protocols/application/test_dtls_connection_id_2515_unit.py diff --git a/docs/source/pcapkit/protocols/application/dtls.rst b/docs/source/pcapkit/protocols/application/dtls.rst index f27a6b8249..8f718586c7 100644 --- a/docs/source/pcapkit/protocols/application/dtls.rst +++ b/docs/source/pcapkit/protocols/application/dtls.rst @@ -4,13 +4,16 @@ DTLS - Datagram Transport Layer Security .. module:: pcapkit.protocols.application.dtls :mod:`pcapkit.protocols.application.dtls` contains -:class:`~pcapkit.protocols.application.dtls.DTLS` only, which implements an +:class:`~pcapkit.protocols.application.dtls.DTLS`, which implements an extractor for Datagram Transport Layer Security (DTLS) [*]_ -- DTLS 1.0 -(:rfc:`4347`), DTLS 1.2 (:rfc:`6347`) and DTLS 1.3's legacy-format and +(:rfc:`4347`), DTLS 1.2 (:rfc:`6347`) and its connection-ID records +(:rfc:`9146`), and DTLS 1.3's legacy-format and unified-header records (:rfc:`9147`) -- at the record layer, plus the epoch/sequence-number fields and the handshake-message fragment header that set DTLS apart from :class:`~pcapkit.protocols.application.tls.TLS`, whose reliable, ordered -transport needs neither. +transport needs neither. It also contains +:class:`~pcapkit.protocols.application.dtls.DTLSContext`, which holds the +connection IDs those records carry. Version Identification ---------------------- @@ -46,6 +49,53 @@ 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. +A ``tls12_cid`` record is ``DTLS 1.2``: :rfc:`9146#section-10.3` limits the +content type to that version, so one with another legacy version is refused. + +Connection IDs +-------------- + +A ``tls12_cid`` record carries a connection ID between its sequence number +and its length (:rfc:`9146#section-4`), and the ID's length is negotiated +rather than carried. +:class:`~pcapkit.protocols.application.dtls.DTLSContext` resolves it, in this +order: + +* a connection ID the record begins with that a ``ClientHello`` or + ``ServerHello`` ``connection_id`` extension negotiated earlier in the + capture, each peer naming the one it receives (:rfc:`9146#section-3`); +* the ``cid_length`` the caller gives + :class:`~pcapkit.protocols.application.dtls.DTLSContext`. + +Without either the record stays :class:`~pcapkit.protocols.misc.raw.Raw`. +A dissected one reports its epoch, sequence number, connection ID and length, +and keeps its encrypted content as octets. The context reaches DTLS through +:mod:`pcapkit.corekit.context`: + +.. code-block:: python + + import pcapkit + from pcapkit.protocols.application.dtls import DTLSContext + + extraction = pcapkit.extract('dtls.pcap', context=DTLSContext(cid_length=8)) + +The connection IDs a +:class:`~pcapkit.protocols.application.dtls.DTLSContext` learns belong to that +instance. A new instance starts with none, and one the caller reuses keeps +what it has learned. An extraction given none gets its own in its own +registry, so it carries what its handshakes negotiate across its capture and +no further. A registry passed as ``__context__`` to several parses carries +its instance across them. + +Matching by value relies on IDs being self-delineating +(:rfc:`9146#section-4`): with ``c1c2c3c4`` registered, a record whose ID is +``c1c2c3c400010203`` reads as carrying ``c1c2c3c4``. + +A rebuild parses each layer's octets again, so ``from_data`` takes the same +``__context__``. Without it the octets still rebuild exactly, and +``DTLS.from_data`` keeps its chain, but a ``tls12_cid`` record under a +rebuilt UDP layer reads as :class:`~pcapkit.protocols.misc.raw.Raw`. + Records in One Datagram ----------------------- @@ -61,8 +111,9 @@ exceptions: :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. + length past the end of the datagram, a unified header with a connection + ID, or a ``tls12_cid`` record whose connection ID is not known, either of + 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` @@ -81,8 +132,10 @@ with :meth:`~pcapkit.protocols.application.dtls.DTLS.read_handshake`. **A unified-header record with a connection ID is not decoded** (:rfc:`9147#section-4.1`, :issue:`2467`). Its length is negotiated, not carried -in the record, so it degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. The -same holds for ``tls12_cid`` records (:rfc:`9146`). +in the record, so it degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. + +**A connection ID negotiated in a fragmented hello is not learned**: the +message is not decoded here, so its ``connection_id`` extension is not read. **A unified-header record carries only the low two bits of the epoch**, and its sequence number is header-protected, so the values it reports are the @@ -113,6 +166,17 @@ ones on the wire. .. automethod:: _make_data +.. autoclass:: pcapkit.protocols.application.dtls.DTLSContext + :no-members: + :show-inheritance: + + .. autoproperty:: connection_ids + .. autoproperty:: cid_length + + .. automethod:: protocol + .. automethod:: register + .. automethod:: match + Header Schema ------------- @@ -126,6 +190,10 @@ Header Schema :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.schema.application.dtls.DTLSCIDCiphertext + :members: + :show-inheritance: + Data Models ----------- @@ -139,6 +207,10 @@ Data Models :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.data.application.dtls.DTLSCIDCiphertext + :members: + :show-inheritance: + .. autoclass:: pcapkit.protocols.data.application.dtls.DTLSHandshake :members: :show-inheritance: diff --git a/docs/source/pcapkit/protocols/application/ssl.rst b/docs/source/pcapkit/protocols/application/ssl.rst index de40f16338..bd765115db 100644 --- a/docs/source/pcapkit/protocols/application/ssl.rst +++ b/docs/source/pcapkit/protocols/application/ssl.rst @@ -50,6 +50,7 @@ accepting bytes that cannot be a genuine record. .. automethod:: _read_ext_alpn .. automethod:: _read_ext_supported_versions .. automethod:: _read_ext_key_share + .. automethod:: _read_ext_connection_id .. automethod:: _make_ext_unknown .. automethod:: _make_ext_server_name @@ -58,6 +59,7 @@ accepting bytes that cannot be a genuine record. .. automethod:: _make_ext_alpn .. automethod:: _make_ext_supported_versions .. automethod:: _make_ext_key_share + .. automethod:: _make_ext_connection_id .. automethod:: _validate_tls_record @@ -157,6 +159,10 @@ Data Models :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.data.application.tls.ConnectionId + :members: + :show-inheritance: + .. autoclass:: pcapkit.protocols.data.application.tls.Alert :members: :show-inheritance: diff --git a/pcapkit/protocols/application/dtls.py b/pcapkit/protocols/application/dtls.py index eeff80680c..36ecade375 100644 --- a/pcapkit/protocols/application/dtls.py +++ b/pcapkit/protocols/application/dtls.py @@ -5,14 +5,53 @@ .. module:: pcapkit.protocols.application.dtls :mod:`pcapkit.protocols.application.dtls` contains -:class:`~pcapkit.protocols.application.dtls.DTLS` only, which implements an +:class:`~pcapkit.protocols.application.dtls.DTLS`, which implements an extractor for the DTLS record-layer family -- DTLS 1.0 (:rfc:`4347`), DTLS -1.2 (:rfc:`6347`), and DTLS 1.3's legacy-format and unified-header records -(:rfc:`9147`). :attr:`UDP.__proto__ ` +1.2 (:rfc:`6347`) and its connection-ID records (:rfc:`9146`), and DTLS 1.3's +legacy-format and unified-header records (:rfc:`9147`) -- and +:class:`~pcapkit.protocols.application.dtls.DTLSContext`, which holds the +connection IDs those records carry. +:attr:`UDP.__proto__ ` binds it at port 5684. As with :class:`~pcapkit.protocols.application.tls.TLS`, :meth:`~DTLS.read` sets :attr:`~DTLS.alias` from the record's legacy version, or to ``DTLS 1.3`` for a unified-header record. +Connection IDs +-------------- + +A ``tls12_cid`` record carries a connection ID between its sequence number +and its length (:rfc:`9146#section-4`), and the ID's length is negotiated +rather than carried. :class:`DTLSContext` resolves it, in this order: + +* a connection ID the record begins with that a ``ClientHello`` or + ``ServerHello`` ``connection_id`` extension negotiated earlier in the + capture, each peer naming the one it receives (:rfc:`9146#section-3`); +* the ``cid_length`` the caller gives :class:`DTLSContext`. + +Without either the record stays :class:`~pcapkit.protocols.misc.raw.Raw`. +A dissected one reports its epoch, sequence number, connection ID and length, +and keeps its encrypted content as octets. The context reaches DTLS through +:mod:`pcapkit.corekit.context`: + +.. code-block:: python + + import pcapkit + from pcapkit.protocols.application.dtls import DTLSContext + + extraction = pcapkit.extract('dtls.pcap', context=DTLSContext(cid_length=8)) + +The connection IDs a :class:`DTLSContext` learns belong to that instance. A +new instance starts with none, and one the caller reuses keeps what it has +learned. An extraction given none gets its own in its own registry, so it +carries what its handshakes negotiate across its capture and no further. +A registry passed as ``__context__`` to several parses carries its instance +across them. + +A rebuild parses each layer's octets again, so ``from_data`` takes the same +``__context__``. Without it the octets still rebuild exactly, and +``DTLS.from_data`` keeps its chain, but a ``tls12_cid`` record under a +rebuilt UDP layer reads as :class:`~pcapkit.protocols.misc.raw.Raw`. + Records in One Datagram ----------------------- @@ -27,8 +66,9 @@ :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. + length past the end of the datagram, a unified header with a connection + ID, or a ``tls12_cid`` record whose connection ID is not known, either of + 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` @@ -49,8 +89,10 @@ **A unified-header record with a connection ID is not decoded** (:rfc:`9147#section-4.1`). Its length is negotiated, not carried in the -record, so it degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. The same -holds for ``tls12_cid`` records (:rfc:`9146`). +record, so it degrades to :class:`~pcapkit.protocols.misc.raw.Raw`. + +**A connection ID negotiated in a fragmented hello is not learned**: the +message is not decoded here, so its ``connection_id`` extension is not read. **A unified-header record carries only the low two bits of the epoch**, and its sequence number is header-protected, so the values it reports are the @@ -62,18 +104,23 @@ from typing import TYPE_CHECKING, cast from pcapkit.const.tls.content_type import ContentType as Enum_ContentType +from pcapkit.const.tls.extension_type import ExtensionType as Enum_ExtensionType +from pcapkit.corekit.context import ProtocolContext 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 DTLSCIDCiphertext as Data_DTLSCIDCiphertext from pcapkit.protocols.data.application.dtls import DTLSCiphertext as Data_DTLSCiphertext +from pcapkit.protocols.data.application.tls import ConnectionId as Data_ConnectionId from pcapkit.protocols.schema.application.dtls import DTLS as Schema_DTLS +from pcapkit.protocols.schema.application.dtls import DTLSCIDCiphertext as Schema_DTLSCIDCiphertext from pcapkit.protocols.schema.application.dtls import DTLSCiphertext as Schema_DTLSCiphertext from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: from enum import IntEnum as StdlibEnum - from typing import Any, Optional, Type + from typing import Any, Iterable, Optional, Type from aenum import IntEnum as AenumEnum from typing_extensions import Literal @@ -82,7 +129,7 @@ from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.schema.schema import Schema -__all__ = ['DTLS'] +__all__ = ['DTLS', 'DTLSContext'] #: Length of a legacy record header: content type, legacy version, epoch, #: sequence number and length (:rfc:`9147#section-4`). @@ -101,6 +148,27 @@ #: Shortest ciphertext a unified-header record may carry: the record-number #: sample is taken from its first 16 octets (:rfc:`9147#section-4.2.3`). _CIPHERTEXT_MIN_LENGTH = 16 +#: Offset of a ``tls12_cid`` record's connection ID, after its content type, +#: legacy version, epoch and sequence number (:rfc:`9146#section-4`). +_CID_OFFSET = 11 +#: Longest connection ID: its length is one octet (:rfc:`9146#section-3`). +_CID_MAX_LENGTH = 255 + + +def _is_cid(value: 'Any') -> 'bool': + """Whether ``value`` can be a record's connection ID. + + A ``tls12_cid`` record's is one or more octets (:rfc:`9146#section-4`), + and at most :data:`_CID_MAX_LENGTH`. + + Args: + value: Candidate connection ID. + + Returns: + Whether ``value`` is :obj:`bytes` of an allowed length. + + """ + return isinstance(value, bytes) and 1 <= len(value) <= _CID_MAX_LENGTH def _is_unified_header(octet: int) -> bool: @@ -132,20 +200,23 @@ 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': +def _opens_whole_record(data: 'bytes', context: 'Optional[DTLSContext]') -> '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`). + that ends inside ``data``. A ``tls12_cid`` header needs DTLS 1.2's version + and a connection ID ``context`` knows the length of. 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. + context: Connection IDs known to the parse, if any. Returns: Whether a further record opens ``data`` and ends inside it. @@ -162,19 +233,132 @@ def _opens_whole_record(data: 'bytes') -> 'bool': return False if not flags['length_present']: return True + elif data[0] == Enum_ContentType.tls12_cid: + cid_length = None if context is None else context.match(data[_CID_OFFSET:]) + if cid_length is None or int.from_bytes(data[1:3], 'big') != _DTLSV1_2_VERSION: + return False + header = _LEGACY_HEADER_LENGTH + cid_length + if len(data) < header: + return False 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. + # Every header ends with its 16-bit length. return header + int.from_bytes(data[header - 2:header], 'big') <= len(data) +class DTLSContext(ProtocolContext): + """Connection IDs for the :class:`DTLS` records that carry one (:rfc:`9146`). + + A ``tls12_cid`` record does not carry its connection ID's length, so + :meth:`match` finds it: from the connection IDs registered here, then from + :attr:`cid_length`. :class:`DTLS` registers those a ``ClientHello`` or + ``ServerHello`` negotiates, and those of the records a constructed one + carries as its payload. + + The IDs belong to the instance: a new one starts with none, one reused + across parses or extractions keeps what it has learned, and nothing is + kept anywhere else. An extraction given no instance gets its own, in its + own registry, the first time a hello negotiates an ID. + + Matching by value relies on IDs being self-delineating + (:rfc:`9146#section-4`): with ``c1c2c3c4`` registered, a record whose ID + is ``c1c2c3c400010203`` reads as carrying ``c1c2c3c4``. + + Args: + *connection_ids: Connection IDs known out of band. + cid_length: Connection ID length of a record that no registered one opens. + + Raises: + ProtocolError: If a connection ID is not 1 to 255 octets, or + ``cid_length`` is not an :class:`int` from 1 to 255. + + """ + + ########################################################################## + # Properties. + ########################################################################## + + @property + def connection_ids(self) -> 'tuple[bytes, ...]': + """Registered connection IDs, in the order registered.""" + return tuple(self.__connection_ids__) + + @property + def cid_length(self) -> 'Optional[int]': + """Connection ID length of a record that no registered one opens.""" + return self.__cid_length__ + + ########################################################################## + # Methods. + ########################################################################## + + @classmethod + def protocol(cls) -> 'tuple[Literal["DTLS"]]': + """Index ID of the protocol this context applies to.""" + return ('DTLS',) + + def register(self, cid: 'bytes') -> 'None': + """Add a connection ID; one already registered is left as it is. + + Args: + cid: Connection ID to add. + + Raises: + ProtocolError: If ``cid`` is not 1 to 255 octets. + + """ + if not _is_cid(cid): + raise ProtocolError(f'DTLS: invalid connection ID: {cid!r}') + if cid not in self.__connection_ids__: + self.__connection_ids__.append(cid) + + def match(self, octets: 'bytes') -> 'Optional[int]': + """Find the length of the connection ID that ``octets`` opens with. + + Each registered connection ID that ``octets`` begins with is a + candidate, longest first, and :attr:`cid_length` comes last. The first + whose record -- a 16-bit length after the ID, then that many octets -- + ends inside ``octets`` is the answer. Failing that it is the first + candidate, so that reading the record reports what runs past the end. + + Args: + octets: A ``tls12_cid`` record from its connection ID on. + + Returns: + The connection ID's length, or :data:`None` with no candidate. + + """ + candidates = sorted({len(cid) for cid in self.__connection_ids__ if octets.startswith(cid)}, + reverse=True) + if self.__cid_length__ is not None: + candidates.append(self.__cid_length__) + for length in candidates: + end = length + 2 + if end + int.from_bytes(octets[length:end], 'big') <= len(octets): + return length + return candidates[0] if candidates else None + + ########################################################################## + # Data models. + ########################################################################## + + def __init__(self, *connection_ids: 'bytes', cid_length: 'Optional[int]' = None) -> 'None': + if cid_length is not None and (isinstance(cid_length, bool) or not isinstance(cid_length, int) + or not 1 <= cid_length <= _CID_MAX_LENGTH): + raise ProtocolError(f'DTLS: invalid connection ID length: {cid_length!r}') + self.__cid_length__ = cid_length # type: Optional[int] + self.__connection_ids__ = [] # type: list[bytes] + for cid in connection_ids: + self.register(cid) + + class DTLS(SSL[Data_DTLS, Schema_DTLS], schema=Schema_DTLS, data=Data_DTLS): """This class implements all protocols in the DTLS record-layer family. - Datagram Transport Layer Security 1.0 [:rfc:`4347`] - - Datagram Transport Layer Security 1.2 [:rfc:`6347`] + - Datagram Transport Layer Security 1.2 [:rfc:`6347`], and its connection-ID records [:rfc:`9146`] - Datagram Transport Layer Security 1.3, legacy-format and unified-header records [:rfc:`9147`] Note: @@ -206,13 +390,16 @@ def alias(self) -> 'Literal["DTLS", "DTLS 1.0", "DTLS 1.2", "DTLS 1.3"]': @property def length(self) -> 'int': """Header length of current protocol -- the record header, the - fragment excluded. Fixed at 13 octets for a legacy record; a unified + fragment excluded. Fixed at 13 octets for a legacy record, and the + connection ID's length more for a ``tls12_cid`` one; a unified header's length follows from its first octet. """ schema = getattr(self, '__header__', None) if isinstance(schema, Schema_DTLSCiphertext): return _ciphertext_header_length(schema.flags) + if isinstance(schema, Schema_DTLSCIDCiphertext): + return _LEGACY_HEADER_LENGTH + len(schema.cid) return _LEGACY_HEADER_LENGTH ########################################################################## @@ -262,7 +449,8 @@ def make(self, ciphertext: 'bool' = False, sequence_long: 'bool' = False, length_present: 'bool' = True, - **kwargs: 'Any') -> 'Schema_DTLS | Schema_DTLSCiphertext': + cid: 'Optional[bytes]' = None, + **kwargs: 'Any') -> 'Schema_DTLS | Schema_DTLSCiphertext | Schema_DTLSCIDCiphertext': """Make (construct) packet data, for every tier: ``version`` is written as is. Args: @@ -283,12 +471,16 @@ def make(self, :class:`~pcapkit.protocols.data.application.tls.Alert`, a :class:`~pcapkit.protocols.data.application.tls.ChangeCipherSpec`, or raw :obj:`bytes`, as :meth:`read` would decode it. For a - unified-header record, the ciphertext as :obj:`bytes`. + unified-header or ``tls12_cid`` record, the ciphertext as + :obj:`bytes`. data: Pre-encoded record fragment. When given, it is used verbatim in place of ``content`` and ``length`` is computed from it. payload: Octets after the record, e.g. a further record - coalesced in the same buffer. + coalesced in the same buffer. The connection IDs of the + ``tls12_cid`` records a :class:`DTLS` payload carries are + registered, so that its octets parse again + (:class:`DTLSContext`). ciphertext: Make a unified-header record (:rfc:`9147#section-4.1`) instead of a legacy one; ``content_type``, ``version`` and ``data`` do not apply. @@ -296,13 +488,22 @@ def make(self, 16 bits rather than 8. length_present: Whether a unified-header record carries its length; if not, it runs to the end of the datagram. + cid: Connection ID of a ``tls12_cid`` record (:rfc:`9146#section-4`), + 1 to 255 octets; no other record takes one. **kwargs: Arbitrary keyword arguments. Returns: Constructed packet data. + Raises: + ProtocolError: If ``cid`` is missing or invalid for a ``tls12_cid`` + record, or given for any other. + """ + self._learn_payload_connection_ids(payload) if ciphertext: + if cid is not None: + raise ProtocolError(f'{self.alias}: invalid connection ID: only a tls12_cid record carries one') return Schema_DTLSCiphertext( flags={ 'fixed': _UNIFIED_PREFIX, @@ -322,6 +523,22 @@ def make(self, reversed=content_type_reversed, pack=False)) fragment = data if data is not None else self._make_tls_content(content_type_, content, dtls=True) + if int(content_type_) == Enum_ContentType.tls12_cid: + if cid is None or not _is_cid(cid): + raise ProtocolError(f'{self.alias}: invalid connection ID: {cid!r} ' + '(a tls12_cid record carries 1 to 255 octets)') + return Schema_DTLSCIDCiphertext( + content_type=content_type_, + version=version, + epoch=epoch, + sequence_number=sequence_number, + cid=cid, + length=len(fragment), + fragment=fragment, + payload=payload, + ) + if cid is not None: + raise ProtocolError(f'{self.alias}: invalid connection ID: only a tls12_cid record carries one') return Schema_DTLS( content_type=content_type_, version=version, @@ -336,9 +553,11 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS """Unpack (parse) packet data, with the header schema its first octet selects. A unified-header record (:rfc:`9147#section-4.1`) is parsed by - :class:`~pcapkit.protocols.schema.application.dtls.DTLSCiphertext` - rather than the legacy header; every other record uses the inherited - behaviour. + :class:`~pcapkit.protocols.schema.application.dtls.DTLSCiphertext`, + and a ``tls12_cid`` one (:rfc:`9146#section-4`) by + :class:`~pcapkit.protocols.schema.application.dtls.DTLSCIDCiphertext` + with the connection ID length :class:`DTLSContext` finds, rather than + the legacy header; every other record uses the inherited behaviour. Args: length: Length of packet data. @@ -347,11 +566,27 @@ def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS Returns: Parsed packet data. + Raises: + ProtocolError: If a ``tls12_cid`` record's connection ID length is + not known. + """ octets = self._data[:1] - if self.__header__ is None and octets and _is_unified_header(octets[0]): + if cast('Optional[Schema_DTLS]', self.__header__) is not None or not octets: + return super().unpack(length, **kwargs) + if _is_unified_header(octets[0]): packet = kwargs.get('__packet__', {}) # packet data - self.__header__ = Schema_DTLSCiphertext.unpack(self._file, length, packet) # type: ignore[assignment] + self.__header__ = Schema_DTLSCiphertext.unpack(self._file, length, packet) # type: ignore[call-arg,misc] + return self.read(length, **kwargs) + if octets[0] == Enum_ContentType.tls12_cid: + context = self._get_context(DTLSContext) + data = self._data if length is None else self._data[:length] + cid_length = None if context is None else context.match(data[_CID_OFFSET:]) + if cid_length is None: + raise ProtocolError(f'{self.alias}: invalid format: no connection ID length is known ' + 'for a tls12_cid record (RFC 9146 section 4)') + packet = {**kwargs.get('__packet__', {}), '__cid_length__': cid_length} + self.__header__ = Schema_DTLSCIDCiphertext.unpack(self._file, length, packet) # type: ignore[call-arg,misc] return self.read(length, **kwargs) return super().unpack(length, **kwargs) @@ -395,7 +630,7 @@ def __length_hint__(self) -> 'Literal[13]': ########################################################################## @classmethod - def _make_data(cls, data: 'Data_DTLS | Data_DTLSCiphertext') -> 'dict[str, Any]': # type: ignore[override] + def _make_data(cls, data: 'Data_DTLS | Data_DTLSCiphertext | Data_DTLSCIDCiphertext') -> 'dict[str, Any]': # type: ignore[override] # pylint: disable=line-too-long """Create key-value pairs from ``data`` for protocol construction. Args: @@ -405,6 +640,16 @@ def _make_data(cls, data: 'Data_DTLS | Data_DTLSCiphertext') -> 'dict[str, Any]' Key-value pairs for protocol construction. """ + if isinstance(data, Data_DTLSCIDCiphertext): + return { + 'content_type': data.content_type, + 'version': data.version, + 'epoch': data.epoch, + 'sequence_number': data.sequence_number, + 'cid': data.cid, + 'content': data.content, + 'payload': cls._make_payload(data), + } if isinstance(data, Data_DTLSCiphertext): return { 'ciphertext': True, @@ -424,10 +669,13 @@ def _make_data(cls, data: 'Data_DTLS | Data_DTLSCiphertext') -> 'dict[str, Any]' 'payload': cls._make_payload(data), } - def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS | Data_DTLSCiphertext': # pylint: disable=unused-argument + def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS | Data_DTLSCiphertext | Data_DTLSCIDCiphertext': # pylint: disable=unused-argument,line-too-long """Read a DTLS record, identifying its concrete tier as it goes. - A unified-header record is handed to :meth:`_read_dtls_ciphertext`. + A unified-header record is handed to :meth:`_read_dtls_ciphertext`, + and a ``tls12_cid`` one to :meth:`_read_dtls_cid_ciphertext`. A + handshake record registers the connection IDs its hellos negotiate + before the next record is read (:meth:`_learn_connection_ids`). Structure of DTLS record layer [:rfc:`9147#section-4`]: @@ -465,6 +713,8 @@ def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> """ if isinstance(self.__header__, Schema_DTLSCiphertext): return self._read_dtls_ciphertext(length, **kwargs) + if isinstance(self.__header__, Schema_DTLSCIDCiphertext): + return self._read_dtls_cid_ciphertext(length, **kwargs) schema = self.__header__ available = length if length is not None else len(self) @@ -483,6 +733,7 @@ def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> content = self._read_tls_content(schema.content_type, schema.fragment, dtls=True, epoch=schema.epoch) + self._learn_connection_ids(content) dtls = Data_DTLS( content_type=schema.content_type, version=schema.version, @@ -495,6 +746,24 @@ def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> remaining = available - (self.length + schema.length) return self._decode_next_layer(dtls, -1, remaining) + def _learn_payload_connection_ids(self, payload: 'Any') -> 'None': + """Register the connection IDs of the ``tls12_cid`` records in ``payload``. + + A constructed record parses its payload's octets again, which needs + the connection IDs that the payload's records carry. + + Args: + payload: Payload given to :meth:`make`. + + """ + cids = [] # type: list[bytes] + layer = payload + while isinstance(layer, DTLS): + if isinstance(layer.info, Data_DTLSCIDCiphertext): + cids.append(layer.info.cid) + layer = layer.payload + self._register_connection_ids(cids) + def _read_dtls_ciphertext(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLSCiphertext': # pylint: disable=unused-argument """Read a unified-header ciphertext record (:rfc:`9147#section-4.1`). @@ -538,6 +807,102 @@ 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 _read_dtls_cid_ciphertext(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLSCIDCiphertext': # pylint: disable=unused-argument,line-too-long + """Read a DTLS 1.2 record with a connection ID. + + Structure of a ``tls12_cid`` record [:rfc:`9146#section-4`]: + + .. code-block:: text + + struct { + ContentType outer_type = tls12_cid; + ProtocolVersion version; + uint16 epoch; + uint48 sequence_number; + opaque cid[cid_length]; + uint16 length; + opaque enc_content[DTLSCiphertext.length]; + } DTLSCiphertext; + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If the legacy version is not DTLS 1.2's, the only + one ``tls12_cid`` applies to (:rfc:`9146#section-10.3`); if the + connection ID is empty; or if the record runs past the end of + the buffer. + + """ + schema = cast('Schema_DTLSCIDCiphertext', self.__header__) + available = length if length is not None else len(self) + + if schema.version != _DTLSV1_2_VERSION: + raise ProtocolError(f'{self.alias}: invalid format: legacy record version {schema.version:#06x} ' + 'is not DTLS 1.2, which tls12_cid needs') + self._alias = 'DTLS 1.2' + if not _is_cid(schema.cid): + raise ProtocolError(f'{self.alias}: invalid format: a tls12_cid record carries no connection ID') + if self.length + schema.length > available: + raise ProtocolError(f'{self.alias}: invalid format: record length {schema.length} runs past ' + 'the end of the buffer') + + dtls = Data_DTLSCIDCiphertext( + content_type=schema.content_type, + version=schema.version, + epoch=schema.epoch, + sequence_number=schema.sequence_number, + cid=schema.cid, + length=schema.length, + content=schema.fragment, + ) + remaining = available - (self.length + schema.length) + # NOTE: the class is generic in the legacy record's data model, which this one is not. + return cast('Data_DTLSCIDCiphertext', self._decode_next_layer(cast('Data_DTLS', dtls), -1, remaining)) + + def _learn_connection_ids(self, content: 'Any') -> 'None': + """Register the connection IDs the hellos in ``content`` negotiate. + + Each peer's ``connection_id`` extension names the connection ID it + receives, and an empty one asks for none (:rfc:`9146#section-3`). + + Args: + content: Decoded record content. + + """ + if not isinstance(content, tuple): + return + self._register_connection_ids( + extension.data.cid + for message in content + for extension in getattr(getattr(message, 'body', None), 'extensions', None) or () + if extension.type == Enum_ExtensionType.connection_id and isinstance(extension.data, Data_ConnectionId) + ) + + def _register_connection_ids(self, cids: 'Iterable[bytes]') -> 'None': + """Register ``cids`` with the parse's :class:`DTLSContext`, adding one if it has none. + + Args: + cids: Connection IDs; any a ``tls12_cid`` record cannot carry are skipped. + + """ + registry = self._exctx + valid = [cid for cid in cids if _is_cid(cid)] + if not valid or registry is None: + return + context = registry.match(self.id(), DTLSContext) + if context is None: + if any(name in registry for name in self.id()): + return # a context of another type holds the name + context = DTLSContext() + registry.register(context) + for cid in valid: + context.register(cid) + 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. @@ -563,7 +928,8 @@ def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # returns otherwise. """ - if proto == -1 and not self._sigterm and _opens_whole_record(self._get_payload()[:length]): + if (proto == -1 and not self._sigterm + and _opens_whole_record(self._get_payload()[:length], self._get_context(DTLSContext))): return self._import_next_record(proto, length, packet=packet) # type: ignore[misc,call-arg] return super()._import_next_layer(proto, length, packet=packet) diff --git a/pcapkit/protocols/application/ssl.py b/pcapkit/protocols/application/ssl.py index 2464abc857..06eb989a64 100644 --- a/pcapkit/protocols/application/ssl.py +++ b/pcapkit/protocols/application/ssl.py @@ -42,6 +42,7 @@ from pcapkit.protocols.data.application.tls import Alert as Data_Alert from pcapkit.protocols.data.application.tls import ChangeCipherSpec as Data_ChangeCipherSpec from pcapkit.protocols.data.application.tls import ClientHello as Data_ClientHello +from pcapkit.protocols.data.application.tls import ConnectionId as Data_ConnectionId from pcapkit.protocols.data.application.tls import EncryptedExtensions as Data_EncryptedExtensions from pcapkit.protocols.data.application.tls import Extension as Data_Extension from pcapkit.protocols.data.application.tls import Handshake as Data_Handshake @@ -282,6 +283,9 @@ class SSL(Application[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract * - :attr:`~pcapkit.const.tls.extension_type.ExtensionType.key_share` - :meth:`~pcapkit.protocols.application.ssl.SSL._read_ext_key_share` - :meth:`~pcapkit.protocols.application.ssl.SSL._make_ext_key_share` + * - :attr:`~pcapkit.const.tls.extension_type.ExtensionType.connection_id` + - :meth:`~pcapkit.protocols.application.ssl.SSL._read_ext_connection_id` + - :meth:`~pcapkit.protocols.application.ssl.SSL._make_ext_connection_id` """ @@ -304,6 +308,7 @@ class SSL(Application[_PT, _ST], Generic[_PT, _ST]): # pylint: disable=abstract Enum_ExtensionType.application_layer_protocol_negotiation: 'alpn', # [RFC 7301] Enum_ExtensionType.supported_versions: 'supported_versions', # [RFC 9846] Enum_ExtensionType.key_share: 'key_share', # [RFC 9846] + Enum_ExtensionType.connection_id: 'connection_id', # [RFC 9146] }, ) # type: DefaultDict[Enum_ExtensionType | int, str | tuple[ExtensionParser, ExtensionConstructor]] @@ -615,6 +620,35 @@ def _read_ext_key_share(self, ext_type: 'Enum_ExtensionType', data: 'bytes', *, return Data_KeyShareServerHello(server_share=entry) return data + def _read_ext_connection_id(self, ext_type: 'Enum_ExtensionType', data: 'bytes', *, # pylint: disable=unused-argument + msg_type: 'Enum_HandshakeType') -> 'Data_ConnectionId': # pylint: disable=unused-argument + """Read a ``connection_id`` extension. + + Structure of ``ConnectionId`` [:rfc:`9146#section-3`]: + + .. code-block:: text + + struct { + opaque cid<0..2^8-1>; + } ConnectionId; + + The extension is DTLS-only (:rfc:`9146#section-10.2`). A TLS hello + carrying it does not conform, and it decodes all the same. + + Args: + ext_type: Extension type. + data: ``extension_data`` octets. + msg_type: Handshake message carrying the extension. + + Returns: + The decoded connection ID. + + Raises: + ProtocolError: If ``data`` is malformed. + + """ + return Data_ConnectionId(cid=self._read_whole_vector(data, size=1, what='connection_id.cid')) + def _make_ext_unknown(self, ext_type: 'Enum_ExtensionType', data: 'Optional[Data_ExtensionData]' = None, *, # pylint: disable=unused-argument msg_type: 'Enum_HandshakeType', **kwargs: 'Any') -> 'bytes': # pylint: disable=unused-argument """Make an extension with no constructor registered. @@ -803,6 +837,24 @@ def _make_ext_key_share(self, ext_type: 'Enum_ExtensionType', # pylint: disable return _make_key_share_entry(self, data.server_share) raise ProtocolError(f'{self.alias}: [Extension {ext_type}] no key_share structure for {msg_type!r}') + def _make_ext_connection_id(self, ext_type: 'Enum_ExtensionType', data: 'Optional[Data_ConnectionId]' = None, *, # pylint: disable=unused-argument + msg_type: 'Enum_HandshakeType', # pylint: disable=unused-argument + cid: 'bytes' = b'', **kwargs: 'Any') -> 'bytes': + """Make a ``connection_id`` extension. + + Args: + ext_type: Extension type. + data: Decoded ``extension_data``. + msg_type: Handshake message carrying the extension. + cid: Connection ID, if ``data`` is not given. + **kwargs: Arbitrary keyword arguments. + + Returns: + Constructed ``extension_data``. + + """ + return self._make_vector(data.cid if data is not None else cid, size=1) + def _adopt_dispatched(self, obj: 'Any') -> 'Any': """Adopt the parse of a sibling tier this dispatcher delegated to. diff --git a/pcapkit/protocols/data/application/dtls.py b/pcapkit/protocols/data/application/dtls.py index 3b04bce4aa..270e83edfb 100644 --- a/pcapkit/protocols/data/application/dtls.py +++ b/pcapkit/protocols/data/application/dtls.py @@ -17,6 +17,7 @@ __all__ = [ 'DTLS', 'DTLSCiphertext', + 'DTLSCIDCiphertext', 'DTLSHandshake', ] @@ -118,3 +119,31 @@ class DTLSCiphertext(Protocol): if TYPE_CHECKING: def __init__(self, epoch: 'int', sequence_number: 'int', sequence_long: 'bool', length: 'Optional[int]', content: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long + + +@info_final +class DTLSCIDCiphertext(Protocol): + """Data model for a DTLS 1.2 record with a connection ID (:rfc:`9146#section-4`). + + The record is always encrypted, so its real content type is inside + :attr:`content` and :attr:`content_type` is ``tls12_cid``. + + """ + + #: Outer content type, always ``tls12_cid``. + content_type: 'Enum_ContentType' + #: Legacy record version (``major << 8 | minor``), DTLS 1.2's. + version: 'int' + #: Epoch -- incremented on every change of cipher state. + epoch: 'int' + #: Sequence number within :attr:`epoch` (48-bit). + sequence_number: 'int' + #: Connection ID, as long as the handshake or the caller says. + cid: 'bytes' + #: Length of :attr:`content`. + length: 'int' + #: Encrypted record content, undecoded. + content: 'bytes' + + if TYPE_CHECKING: + def __init__(self, content_type: 'Enum_ContentType', version: 'int', epoch: 'int', sequence_number: 'int', cid: 'bytes', length: 'int', content: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long diff --git a/pcapkit/protocols/data/application/tls.py b/pcapkit/protocols/data/application/tls.py index 3f0b9c6007..d840773f8d 100644 --- a/pcapkit/protocols/data/application/tls.py +++ b/pcapkit/protocols/data/application/tls.py @@ -23,6 +23,7 @@ 'ServerName', 'ServerNameList', 'NamedGroupList', 'SignatureSchemeList', 'ProtocolNameList', 'SupportedVersionsClientHello', 'SupportedVersionsServerHello', 'KeyShareEntry', 'KeyShareClientHello', 'KeyShareHelloRetryRequest', 'KeyShareServerHello', + 'ConnectionId', 'Alert', 'ChangeCipherSpec', ] @@ -192,6 +193,22 @@ class KeyShareServerHello(ExtensionData): def __init__(self, server_share: 'KeyShareEntry') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements +@info_final +class ConnectionId(ExtensionData): + """Data model for the ``connection_id`` extension (:rfc:`9146#section-3`). + + The extension is DTLS-only (:rfc:`9146#section-10.2`); a TLS hello that + carries it, which does not conform, decodes to this model all the same. + + """ + + #: The connection ID its sender wants to receive; empty when it wants none. + cid: 'bytes' + + if TYPE_CHECKING: + def __init__(self, cid: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements + + @info_final class ClientHello(Data): """Data model for a TLS/DTLS ``ClientHello`` handshake message body.""" diff --git a/pcapkit/protocols/schema/application/dtls.py b/pcapkit/protocols/schema/application/dtls.py index c336799159..901310b4c1 100644 --- a/pcapkit/protocols/schema/application/dtls.py +++ b/pcapkit/protocols/schema/application/dtls.py @@ -10,9 +10,11 @@ from pcapkit.corekit.fields.strings import BitField, BytesField from pcapkit.protocols.schema.schema import Schema, schema_final -__all__ = ['DTLS', 'DTLSCiphertext'] +__all__ = ['DTLS', 'DTLSCiphertext', 'DTLSCIDCiphertext'] if TYPE_CHECKING: + from typing import Any + from pcapkit.protocols.protocol import ProtocolBase @@ -93,3 +95,45 @@ class DTLSCiphertext(Schema): if TYPE_CHECKING: def __init__(self, flags: 'dict[str, int]', sequence_number: 'int', length: 'int | None', fragment: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long + + +@schema_final +class DTLSCIDCiphertext(Schema): + """Header schema for a DTLS 1.2 record with a connection ID (:rfc:`9146#section-4`). + + :class:`DTLS `'s legacy + header with ``cid`` between the sequence number and the length. The record + does not carry the CID's length, so the parser gives it as the + ``__cid_length__`` key of the packet context; packing takes it from ``cid``. + + """ + + #: Outer content type, always ``tls12_cid``. + content_type: 'Enum_ContentType' = EnumField(length=1, namespace=Enum_ContentType) + #: Legacy record version (``major << 8 | minor``). + version: 'int' = UInt16Field() + #: Epoch -- incremented on every change of cipher state. + epoch: 'int' = UInt16Field() + #: Sequence number within :attr:`epoch` (48-bit). + sequence_number: 'int' = NumberField(length=6, signed=False) + #: Connection ID. + cid: 'bytes' = BytesField(length=lambda pkt: pkt['__cid_length__']) + #: Length of :attr:`fragment`. + length: 'int' = UInt16Field() + #: Encrypted record content, undecoded. + fragment: 'bytes' = BytesField(length=lambda pkt: pkt['length']) + #: Octets after the record, e.g. a further record coalesced in the same + #: buffer. + payload: 'bytes' = PayloadField() + + def pre_pack(self, packet: 'dict[str, Any]') -> 'None': + """Give packing the length of :attr:`cid`, which the record does not carry. + + Args: + packet: packet data + + """ + packet['__cid_length__'] = len(packet['cid']) + + if TYPE_CHECKING: + def __init__(self, content_type: 'Enum_ContentType', version: 'int', epoch: 'int', sequence_number: 'int', cid: 'bytes', length: 'int', fragment: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long diff --git a/tests/protocols/application/test_dtls_connection_id_2515_unit.py b/tests/protocols/application/test_dtls_connection_id_2515_unit.py new file mode 100644 index 0000000000..df64817f93 --- /dev/null +++ b/tests/protocols/application/test_dtls_connection_id_2515_unit.py @@ -0,0 +1,654 @@ +# -*- coding: utf-8 -*- +"""DTLS 1.2 records with a connection ID (RFC 9146, #2515). + +A ``tls12_cid`` record carries its connection ID between the sequence number +and the length (RFC 9146 section 4, Figure 3). The ID's length is negotiated in +the ``connection_id`` extension (section 3), not carried, so ``DTLSContext`` +resolves it: from the IDs the capture's hellos negotiate, then from a +caller-supplied ``cid_length``. + +:data:`MBEDTLS` is a real exchange between Mbed TLS 4.2.0's ``ssl_client2`` and +``ssl_server2`` (``dtls=1 cid=1``, PSK, ``TLS-PSK-WITH-AES-128-CCM-8``), recorded +through a loopback UDP relay: the client asks for ``c1c2c3c4`` and the server +for ``a1a2a3a4a5a6a7a8``. tshark 4.6.9 (``-d udp.port==5684,dtls``) reads every +record as :data:`TSHARK` lists it. The other vectors follow Figure 3. + +""" +from __future__ import annotations + +import os +import struct +import sys +import tempfile +import threading +import unittest +import warnings +from typing import TYPE_CHECKING + +from tests._support import close_extractor, reimport_once_per_class +from tests.utilities._harness import capture + +if TYPE_CHECKING: + from typing import Any + +#: Test tier (#1958): checks that the exception hooks are left alone. +__tier__ = 'ALL' + +#: Connection IDs the Mbed TLS client and server ask to receive. +CLIENT_CID = bytes.fromhex('c1c2c3c4') +SERVER_CID = bytes.fromhex('a1a2a3a4a5a6a7a8') + +#: The exchange, one ``(from_client, payload)`` per UDP datagram, in order. +MBEDTLS = tuple((direction == 'C', bytes.fromhex(payload)) for direction, payload in ( + ('C', '16fefd00000000000000000073010000670000000000000067fefd6ae803a4e31740e255176c17d4d7c19d92b9baf6c6' + '94d969206ae50ed241a71b00000004c0a800ff010000390000000e000c0000096c6f63616c686f7374000d000e000c06' + '03060105030501040304010036000504c1c2c3c4001600000017000000230000'), + ('S', '16fefd0000000000000000002f030000230000000000000023fefd206ae803a41747fbaa56813d1553efde9296659fc8' + 'a203e9cf16838e743a00bd79'), + ('C', '16fefd00000000000000010093010000870001000000000087fefd6ae803a4e31740e255176c17d4d7c19d92b9baf6c6' + '94d969206ae50ed241a71b00206ae803a41747fbaa56813d1553efde9296659fc8a203e9cf16838e743a00bd790004c0' + 'a800ff010000390000000e000c0000096c6f63616c686f7374000d000e000c06030601050305010403040100360005' + '04c1c2c3c4001600000017000000230000'), + ('S', '16fefd0000000000000001004e020000420001000000000042fefd6ae803a4066dfc8934f6d38d96f4ce92497b503f64' + 'e42e463c1b9bcc92c4d82200c0a800001aff010001000036000908a1a2a3a4a5a6a7a8001700000023000016fefd0000' + '000000000002000c0e0000000002000000000000'), + ('C', '16fefd0000000000000002001d100000110002000000000011000f436c69656e745f6964656e7469747914fefd000000' + '000000000300010119fefd0001000000000000a1a2a3a4a5a6a7a8003000010000000000001b8aee327bde4168cb0404' + '9d9cddb4af84dcaf56dd038079d4bffb32112f3cddfa545d09ce119162'), + ('S', '16fefd000000000000000300a70400009b000300000000009b000151800095262e883ce0e2ad87ccfb4e54c5026e7500' + '73dddb0c04a230c4e072ac2de31f060db34df670e2b3b8f36b3ea79f9630b202b84a5a916bd373d47332a76d67800833' + '226c1ce974d46458c998c697b851698196ff635d111abd3dd0e71e90c210262d36a45e3d3420942d37713ab31b5ad339' + 'd77a74ae2e1b03c860937fdca5185a0146287ca3b647c39cd267f72b871bdb05ab26948b14fefd00000000000000040001' + '0119fefd0001000000000000c1c2c3c400300001000000000000cabe27c79b3a78d4bdbd0624dc9d12f28e73530701fc' + 'f6b4e854c070744bc84d7e42334c0285b52b'), + ('C', '19fefd0001000000000001a1a2a3a4a5a6a7a8005000010000000000015c14c021e3f2c5f71518f19696e72672aaea43' + '729c1919edb9fd31adade0639b1b45b8ff56ab931d218e191e2e62c51ceca82492002c8b6aa535c7f352b671ceca4ae3' + 'b824f355bc'), + ('S', '19fefd0001000000000001c1c2c3c400a00001000000000001997b0b4964e8217400cd02db2f5d3cbc8d6836729fe801' + '221c047fdb3e81d44c00e116cfa4b9cc242d822a3a85a4379f8ddf6111126b2a1fc22925196eeee7d0f31e3d8bbb6de5' + 'bd99d05098e2f12d70b7b60e3426c6a9b87d8bd384e7180981b3e4616c5e74a08add944e3b9f4b3a4cc235338e5048d5' + '37f9561a282f2abdcd7d33301d82f86e4b47672338f66c5ff4e90c12c9119085a1'), + ('S', '19fefd0001000000000002c1c2c3c4002000010000000000021869d0aea6345fc7eb664b55505f8c5485d78f2af0b08d66'), + ('C', '19fefd0001000000000002a1a2a3a4a5a6a7a8002000010000000000023ca0a04f6ecfb4dfe3b8d3ac7cfbc052fad42c1a' + '2132ec46'), +)) + +#: Per datagram: its protochain, and ``(cid, epoch, sequence number, length)`` +#: of each ``tls12_cid`` record in it, as tshark reports them. +TSHARK = ( + ('UDP:DTLS 1.2', ()), + ('UDP:DTLS 1.2', ()), + ('UDP:DTLS 1.2', ()), + ('UDP:DTLS 1.2:DTLS 1.2', ()), + ('UDP:DTLS 1.2:DTLS 1.2:DTLS 1.2', ((SERVER_CID, 1, 0, 48),)), + ('UDP:DTLS 1.2:DTLS 1.2:DTLS 1.2', ((CLIENT_CID, 1, 0, 48),)), + ('UDP:DTLS 1.2', ((SERVER_CID, 1, 1, 80),)), + ('UDP:DTLS 1.2', ((CLIENT_CID, 1, 1, 160),)), + ('UDP:DTLS 1.2', ((CLIENT_CID, 1, 2, 32),)), + ('UDP:DTLS 1.2', ((SERVER_CID, 1, 2, 32),)), +) + +#: 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 _cid_record(cid: bytes, fragment: bytes, *, version: bytes = b'\xfe\xfd', + epoch: int = 1, sequence_number: int = 1) -> bytes: + """A ``tls12_cid`` record, as RFC 9146 section 4's Figure 3 lays it out.""" + return (b'\x19' + version + epoch.to_bytes(2, 'big') + sequence_number.to_bytes(6, 'big') + + cid + len(fragment).to_bytes(2, 'big') + fragment) + + +def _udp(payload: bytes, from_client: bool = True) -> bytes: + """An 8-octet UDP header between port 50000 and port 5684, plus ``payload``.""" + ports = (50000, 5684) if from_client else (5684, 50000) + return struct.pack('!HHHH', *ports, 8 + len(payload), 0) + payload + + +def _dtls_layers(proto: 'Any') -> 'tuple[list, Any]': + """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 + + +def _cid_layers(proto: 'Any') -> 'tuple': + """``(cid, epoch, sequence number, length)`` of each ``tls12_cid`` layer under ``proto``.""" + from pcapkit.protocols.data.application.dtls import DTLSCIDCiphertext + + return tuple((layer.info.cid, layer.info.epoch, layer.info.sequence_number, layer.info.length) + for layer in _dtls_layers(proto)[0] if isinstance(layer.info, DTLSCIDCiphertext)) + + +def _ethernet_pcap(datagrams: 'tuple[tuple[bool, bytes], ...]') -> str: + """Write ``datagrams`` as an Ethernet/IPv4/UDP PCAP file in a temporary directory.""" + path = os.path.join(tempfile.mkdtemp(prefix='pcapkit-dtls-cid-'), 'dtls-cid.pcap') + client, server = bytes([192, 0, 2, 1]), bytes([192, 0, 2, 2]) + with open(path, 'wb') as file: + file.write(struct.pack('> 16) + header = header[:10] + struct.pack('!H', ~total & 0xFFFF) + header[12:] + frame = bytes(6) + bytes(6) + b'\x08\x00' + header + udp + file.write(struct.pack(' None: + reimport_once_per_class(self) + + def test_each_hello_decodes_the_id_its_sender_receives(self) -> None: + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.data.application.tls import ConnectionId + + for name, (_, wire), cid in (('ClientHello', MBEDTLS[0], CLIENT_CID), + ('ServerHello', MBEDTLS[3], SERVER_CID)): + with self.subTest(hello=name): + parsed = DTLS(wire, len(wire)) + extensions = parsed.info.content[0].body.extensions + found = [ext.data for ext in extensions if ext.type == ExtensionType.connection_id] + self.assertEqual(found, [ConnectionId(cid=cid)]) + self.assertEqual(bytes(DTLS.from_data(parsed.info)).hex(), wire.hex()) + + def test_a_malformed_extension_stays_raw_and_teaches_nothing(self) -> None: + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + + # The ``cid`` vector says 5 octets, but 4 follow it. + bad = MBEDTLS[0][1].replace(bytes.fromhex('0036000504c1c2c3c4'), bytes.fromhex('0036000505c1c2c3c4')) + parsed = DTLS(bad, len(bad)) + ext, = [ext for ext in parsed.info.content[0].body.extensions if ext.type == ExtensionType.connection_id] + self.assertEqual(ext.data, bytes.fromhex('05c1c2c3c4')) + self.assertIsNone(parsed.context.match(['DTLS'], DTLSContext)) + self.assertEqual(bytes(DTLS.from_data(parsed.info)).hex(), bad.hex()) + + def test_the_extension_is_made_from_a_cid(self) -> None: + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.const.tls.handshake_type import HandshakeType + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.data.application.tls import ConnectionId, Extension + + maker = object.__new__(DTLS) + for name, extension in (('from kwargs', (ExtensionType.connection_id, {'cid': b'\x01\x02'})), + ('from data', Extension(type=ExtensionType.connection_id, + data=ConnectionId(cid=b'\x01\x02')))): + with self.subTest(case=name): + made = maker._make_tls_extensions([extension], msg_type=HandshakeType.client_hello) # pylint: disable=protected-access + self.assertEqual(made.hex(), '00360003020102') + + def test_a_tls_hello_carrying_the_extension_decodes_it_too(self) -> None: + """RFC 9146 section 10.2 makes it DTLS-only; TLS shares the decoder regardless.""" + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.data.application.tls import ConnectionId + + extensions = bytes.fromhex('0036000302abcd') + body = (b'\x03\x03' + bytes(32) + b'\x00' + b'\x00\x02\x13\x01' + b'\x01\x00' + + len(extensions).to_bytes(2, 'big') + extensions) + fragment = b'\x01' + len(body).to_bytes(3, 'big') + body + wire = b'\x16\x03\x03' + len(fragment).to_bytes(2, 'big') + fragment + parsed = TLS(wire, len(wire)) + ext, = parsed.info.content[0].body.extensions + self.assertEqual((ext.type, ext.data), (ExtensionType.connection_id, ConnectionId(cid=b'\xab\xcd'))) + self.assertEqual(bytes(TLS.from_data(parsed.info)).hex(), wire.hex()) + + def test_handing_back_the_shipped_entry_is_a_silent_no_op(self) -> None: + from pcapkit.const.tls.extension_type import ExtensionType + from pcapkit.foundation.registry.protocols import register_tls_extension + from pcapkit.protocols.application.ssl import SSL + + shipped = dict(SSL.__extension__) + self.assertEqual(shipped[ExtensionType.connection_id], 'connection_id') + with warnings.catch_warnings(): + warnings.simplefilter('error') + register_tls_extension(ExtensionType.connection_id, 'connection_id') + self.assertEqual(dict(SSL.__extension__), shipped) + + +class CaptureTests(unittest.TestCase): + """The Mbed TLS exchange reads as tshark reads it.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_the_handshake_teaches_each_direction_its_id(self) -> None: + from pcapkit.corekit.context import ContextRegistry + from pcapkit.protocols.application.dtls import DTLSContext + from pcapkit.protocols.transport.udp import UDP + + registry = ContextRegistry() + for index, ((from_client, payload), (chain, records)) in enumerate(zip(MBEDTLS, TSHARK), 1): + with self.subTest(datagram=index): + datagram = _udp(payload, from_client) + proto = UDP(datagram, len(datagram), __context__=registry) + self.assertEqual(str(proto.protochain), chain) + self.assertEqual(_cid_layers(proto.payload), records) + context = registry.match(['DTLS'], DTLSContext) + self.assertEqual(context.connection_ids, (CLIENT_CID, SERVER_CID)) # type: ignore[union-attr] + + def test_an_extraction_carries_the_ids_across_its_frames(self) -> None: + import pcapkit + + path = _ethernet_pcap(MBEDTLS) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extraction = pcapkit.extract(fin=path, nofile=True) + try: + self.assertEqual([str(frame.protochain) for frame in extraction.frame], + [f'Ethernet:IPv4:{chain}' for chain, _ in TSHARK]) + self.assertEqual([_cid_layers(frame['UDP'].payload) for frame in extraction.frame], + [records for _, records in TSHARK]) + finally: + close_extractor(extraction) + + def test_a_caller_supplied_length_reads_records_without_the_handshake(self) -> None: + from pcapkit.protocols.application.dtls import DTLSContext + from pcapkit.protocols.transport.udp import UDP + + for index in range(6, 10): + (from_client, payload), (chain, records) = MBEDTLS[index], TSHARK[index] + with self.subTest(datagram=index + 1): + context = DTLSContext(cid_length=len(records[0][0])) + datagram = _udp(payload, from_client) + proto = UDP(datagram, len(datagram), __context__=context) + self.assertEqual(str(proto.protochain), chain) + self.assertEqual(_cid_layers(proto.payload), records) + + +class InstanceMemoryTests(unittest.TestCase): + """Learned connection IDs belong to one ``DTLSContext`` instance, and nowhere else. + + Each test fails if a class-level, module-level or otherwise shared store + carries IDs from one instance or extraction to another. + + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _extract(self, datagrams: 'tuple', **kwargs: 'Any') -> 'Any': + import pcapkit + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extraction = pcapkit.extract(fin=_ethernet_pcap(datagrams), nofile=True, **kwargs) + self.addCleanup(close_extractor, extraction) + return extraction + + def test_two_fresh_instances_share_nothing(self) -> None: + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + from pcapkit.utilities.exceptions import ProtocolError + + first = DTLSContext() + first.register(SERVER_CID) + wire = MBEDTLS[3][1] # a ServerHello, which teaches the instance its parse holds + DTLS(wire, len(wire), __context__=DTLSContext()) + DTLS(wire, len(wire)) + + second = DTLSContext() + record = MBEDTLS[6][1] + self.assertEqual(first.connection_ids, (SERVER_CID,)) + self.assertEqual(second.connection_ids, ()) + self.assertIsNone(second.match(record[11:])) + for name, kwargs in (('a fresh instance', {'__context__': DTLSContext()}), ('no context', {})): + with self.subTest(parse=name), \ + self.assertRaisesRegex(ProtocolError, 'no connection ID length is known'): + DTLS(record, len(record), **kwargs) + + #: The Mbed TLS ClientHello, asking for ``d1d2d3d4`` instead, so that an + #: extraction of it holds an instance of its own. + OTHER_HELLO = (True, MBEDTLS[0][1].replace(CLIENT_CID, bytes.fromhex('d1d2d3d4'))) + + def test_consecutive_extractions_without_a_context_share_nothing(self) -> None: + """A CID record whose hellos were in the previous extraction stays raw.""" + first = self._extract(MBEDTLS) + self.assertEqual(str(first.frame[6].protochain), 'Ethernet:IPv4:UDP:DTLS 1.2') + for name, datagrams in (('no hellos', (MBEDTLS[6],)), + ('a hello asking for another ID', (self.OTHER_HELLO, MBEDTLS[6]))): + with self.subTest(second=name): + second = self._extract(datagrams) + self.assertEqual(str(second.frame[-1].protochain), 'Ethernet:IPv4:UDP:Raw') + + def test_concurrent_extractions_without_a_context_share_nothing(self) -> None: + """The CID-record extraction runs in its own thread, after the other has learned.""" + learned = threading.Event() + chains = {} # type: dict[str, str] + + def run(name: str, datagrams: 'tuple', wait: bool) -> None: + if wait: + learned.wait(60) + chains[name] = str(self._extract(datagrams).frame[-1].protochain) + learned.set() + + threads = [threading.Thread(target=run, args=('handshake', MBEDTLS[:7], False)), + threading.Thread(target=run, args=('record', (self.OTHER_HELLO, MBEDTLS[6]), True))] + for thread in reversed(threads): + thread.start() + for thread in threads: + thread.join(120) + self.assertEqual(chains, {'handshake': 'Ethernet:IPv4:UDP:DTLS 1.2', + 'record': 'Ethernet:IPv4:UDP:Raw'}) + + def test_a_reused_caller_instance_keeps_its_ids(self) -> None: + from pcapkit.protocols.application.dtls import DTLSContext + + context = DTLSContext() + self._extract(MBEDTLS, context=context) + self.assertEqual(context.connection_ids, (CLIENT_CID, SERVER_CID)) + + again = self._extract((MBEDTLS[6],), context=context) + self.assertEqual(str(again.frame[0].protochain), 'Ethernet:IPv4:UDP:DTLS 1.2') + self.assertEqual(_cid_layers(again.frame[0]['UDP'].payload), TSHARK[6][1]) + + for name, datagrams in (('no hellos', (MBEDTLS[6],)), + ('a hello asking for another ID', (self.OTHER_HELLO, MBEDTLS[6]))): + with self.subTest(fresh=name): + fresh = self._extract(datagrams, context=DTLSContext()) + self.assertEqual(str(fresh.frame[-1].protochain), 'Ethernet:IPv4:UDP:Raw') + + def test_a_constructed_payload_registers_on_the_instance_in_use(self) -> None: + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + from pcapkit.utilities.exceptions import ProtocolError + + def build(**kwargs: 'Any') -> 'Any': + inner = DTLS(content_type=ContentType.tls12_cid, epoch=1, sequence_number=0, cid=SERVER_CID, + content=bytes(range(32))) + return DTLS(content_type=ContentType.change_cipher_spec, epoch=0, sequence_number=3, + content=b'\x01', payload=inner, **kwargs) + + context = DTLSContext() + self.assertEqual(str(build(__context__=context).protochain), 'DTLS 1.2:DTLS 1.2') + self.assertEqual(context.connection_ids, (SERVER_CID,)) + + own = build() + self.assertEqual(own.context.match(['DTLS'], DTLSContext).connection_ids, (SERVER_CID,)) + self.assertEqual(DTLSContext().connection_ids, ()) + record = _cid_record(SERVER_CID, bytes(range(32)), sequence_number=0) + with self.assertRaisesRegex(ProtocolError, 'no connection ID length is known'): + DTLS(record, len(record)) + + +class QuietRefusalTests(unittest.TestCase): + """A record whose connection ID is unknown is ``Raw``, quietly, on dispatch.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + saved_sys, saved_threading = sys.excepthook, threading.excepthook + + def _restore() -> None: + sys.excepthook = saved_sys + threading.excepthook = saved_threading + + self.addCleanup(_restore) + + def _sys_sentinel(*args: 'Any') -> None: + saved_sys(*args) + + def _threading_sentinel(args: 'Any') -> None: + saved_threading(args) + + self.sys_sentinel, self.threading_sentinel = _sys_sentinel, _threading_sentinel + sys.excepthook, threading.excepthook = _sys_sentinel, _threading_sentinel + + def test_dispatch_keeps_the_record_raw_and_logs_nothing(self) -> None: + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.udp import UDP + from pcapkit.utilities import exceptions + from pcapkit.utilities.logging import logger + + for index in range(6, 10): + from_client, payload = MBEDTLS[index] + with self.subTest(datagram=index + 1): + datagram = _udp(payload, from_client) + with capture(logger) as package, capture(exceptions.logger) as module: + proto = UDP(datagram, len(datagram)) + self.assertEqual(str(proto.protochain), 'UDP:Raw') + self.assertIs(type(proto.payload), Raw) + self.assertRegex(str(proto.payload.info.error), + r'^DTLS: invalid format: no connection ID length is known') + self.assertEqual(package.messages + module.messages, []) + self.assertIs(sys.excepthook, self.sys_sentinel) + self.assertIs(threading.excepthook, self.threading_sentinel) + self.assertEqual(bytes(UDP.from_data(proto.info)).hex(), datagram.hex()) + + def test_a_direct_parse_raises(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + wire = MBEDTLS[6][1] + with self.assertRaisesRegex(ProtocolError, 'no connection ID length is known'): + DTLS(wire, len(wire)) + + +class ResolutionTests(unittest.TestCase): + """How ``DTLSContext`` finds a record's connection ID length.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_a_learned_id_comes_before_cid_length(self) -> None: + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + + wire = MBEDTLS[6][1] + parsed = DTLS(wire, len(wire), __context__=DTLSContext(SERVER_CID, cid_length=4)) + self.assertEqual(parsed.info.cid, SERVER_CID) + self.assertEqual(parsed.info.length, 80) + + def test_the_longest_id_whose_record_fits(self) -> None: + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + + fragment = b'\xff\xff' + bytes(14) + for name, cid, ids in ( + ('both ids open it, the longer fits', b'\xab\xcd\xef\x01', (b'\xab', b'\xab\xcd\xef\x01')), + ('the longer opens it but does not fit', b'\xab', (b'\xab', b'\xab\x00\x10')), + ): + with self.subTest(case=name): + wire = _cid_record(cid, fragment) + parsed = DTLS(wire, len(wire), __context__=DTLSContext(*ids)) + self.assertEqual((parsed.info.cid, parsed.info.content), (cid, fragment)) + self.assertEqual(parsed.length, 13 + len(cid)) + + def test_a_record_no_candidate_fits_reports_its_overrun(self) -> None: + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + from pcapkit.utilities.exceptions import ProtocolError + + wire = _cid_record(SERVER_CID, bytes(16))[:-1] + self.assertEqual(DTLSContext(SERVER_CID).match(wire[11:]), len(SERVER_CID)) + with warnings.catch_warnings(), \ + self.assertRaisesRegex(ProtocolError, r'^DTLS 1\.2: invalid format: record length 16 runs past'): + warnings.simplefilter('ignore') # the schema's short read, as for a legacy record + DTLS(wire, len(wire), __context__=DTLSContext(SERVER_CID)) + + def test_only_dtls_1_2_carries_tls12_cid(self) -> None: + """RFC 9146 section 10.3 applies the content type to DTLS 1.2 alone.""" + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + from pcapkit.utilities.exceptions import ProtocolError + + wire = _cid_record(SERVER_CID, bytes(16), version=b'\xfe\xff') + context = DTLSContext(SERVER_CID) + with self.assertRaisesRegex(ProtocolError, r'^DTLS: invalid format: legacy record version 0xfeff'): + DTLS(wire, len(wire), __context__=context) + parsed = DTLS(CCS + wire, len(CCS + wire), __context__=context) + self.assertEqual(str(parsed.protochain), 'DTLS 1.2:Raw') + self.assertIsNone(parsed.payload.info.error) + + def test_an_empty_id_is_not_learned(self) -> None: + """A zero-length ``connection_id`` asks for no ID (RFC 9146 section 3).""" + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + + # An empty ``connection_id``, then an unknown empty extension (0xFFCE), + # in the 9 octets the vector's own extension takes. + wire = MBEDTLS[0][1].replace(bytes.fromhex('0036000504c1c2c3c4'), bytes.fromhex('0036000100ffce0000')) + parsed = DTLS(wire, len(wire)) + self.assertIsNone(parsed.context.match(['DTLS'], DTLSContext)) + self.assertEqual(bytes(DTLS.from_data(parsed.info)).hex(), wire.hex()) + + def test_a_context_of_another_type_is_left_alone(self) -> None: + from pcapkit.corekit.context import ContextRegistry, ProtocolContext + from pcapkit.protocols.application.dtls import DTLS + + class Other(ProtocolContext): + """A context some other code keeps under ``DTLS``.""" + + @classmethod + def protocol(cls) -> 'tuple[str, ...]': + return ('DTLS',) + + registry = ContextRegistry(Other()) + wire = MBEDTLS[3][1] + DTLS(wire, len(wire), __context__=registry) + self.assertIsInstance(registry['DTLS'], Other) + self.assertEqual(len(registry), 1) + + def test_the_context_validates_what_it_is_given(self) -> None: + from pcapkit.protocols.application.dtls import DTLSContext + from pcapkit.utilities.exceptions import ProtocolError + + for name, args, kwargs in (('an empty id', (b'',), {}), ('a 256-octet id', (bytes(256),), {}), + ('a str id', ('ab',), {}), ('cid_length 0', (), {'cid_length': 0}), + ('cid_length 256', (), {'cid_length': 256}), + ('cid_length True', (), {'cid_length': True}), + ('cid_length a str', (), {'cid_length': '8'})): + with self.subTest(case=name), self.assertRaises(ProtocolError): + DTLSContext(*args, **kwargs) # type: ignore[arg-type] + context = DTLSContext(b'\x01', bytes(255), b'\x01', cid_length=255) + self.assertEqual(context.connection_ids, (b'\x01', bytes(255))) + self.assertEqual(context.cid_length, 255) + self.assertEqual(context.protocol(), ('DTLS',)) + self.assertIsNone(DTLSContext().match(bytes(20))) + + +class ChainTests(unittest.TestCase): + """A ``tls12_cid`` record chains with the records sharing its datagram (#2535).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_records_either_side_of_one_are_further_layers(self) -> None: + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + + record = _cid_record(CLIENT_CID, bytes(range(24))) + for name, wire, chain in (('after a legacy record', CCS + record, 'DTLS 1.2:DTLS 1.2'), + ('before a legacy record', record + CCS, 'DTLS 1.2:DTLS 1.2'), + ('two of them', record + record, 'DTLS 1.2:DTLS 1.2')): + with self.subTest(case=name): + parsed = DTLS(wire, len(wire), __context__=DTLSContext(CLIENT_CID)) + self.assertEqual(str(parsed.protochain), chain) + layers, _ = _dtls_layers(parsed) + self.assertEqual(sum(1 for layer in layers if layer.info.content_type == 25), + wire.count(b'\x19\xfe\xfd')) + self.assertEqual(bytes(DTLS.from_data(parsed.info)).hex(), wire.hex()) + + def test_an_unknown_id_after_a_record_stays_raw_without_an_error(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.misc.raw import Raw + + record = _cid_record(CLIENT_CID, bytes(range(24))) + parsed = DTLS(CCS + record, len(CCS + record)) + self.assertEqual(str(parsed.protochain), 'DTLS 1.2:Raw') + self.assertIs(type(parsed.payload), Raw) + self.assertEqual(bytes(parsed.payload), record) + self.assertIsNone(parsed.payload.info.error) + + +class RoundTripTests(unittest.TestCase): + """Byte-exact both ways: parse then rebuild, and make then parse.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + from pcapkit.corekit.context import ContextRegistry + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.protocols.transport.udp import UDP + + registry = ContextRegistry() + for index, (from_client, payload) in enumerate(MBEDTLS, 1): + with self.subTest(datagram=index): + datagram = _udp(payload, from_client) + udp = UDP(datagram, len(datagram), __context__=registry) + rebuilt = DTLS.from_data(udp.payload.info) + self.assertEqual(bytes(rebuilt).hex(), payload.hex()) + self.assertEqual(str(rebuilt.protochain), str(udp.payload.protochain)) + + again = UDP.from_data(udp.info, __context__=registry) + self.assertEqual(bytes(again).hex(), datagram.hex()) + self.assertEqual(str(again.protochain), str(udp.protochain)) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(bytes(UDP.from_data(udp.info)).hex(), datagram.hex()) + + def test_make_then_parse_is_byte_exact(self) -> None: + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.dtls import DTLS, DTLSContext + from pcapkit.protocols.data.application.dtls import DTLSCIDCiphertext + + for index in (4, 5, 6, 7): + wire = MBEDTLS[index][1] + with self.subTest(datagram=index + 1): + parsed = DTLS(wire, len(wire), __context__=DTLSContext(CLIENT_CID, SERVER_CID)) + maker = object.__new__(DTLS) + schema = b'' # type: object + for layer in reversed(_dtls_layers(parsed)[0]): + info = layer.info + extra = {'cid': info.cid} if isinstance(info, DTLSCIDCiphertext) 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, **extra) + made = bytes(schema) # type: ignore[call-overload] + self.assertEqual(made.hex(), wire.hex()) + + again = DTLS(made, len(made), __context__=DTLSContext(CLIENT_CID, SERVER_CID)) + self.assertEqual(str(again.protochain), str(parsed.protochain)) + self.assertEqual(_cid_layers(again), _cid_layers(parsed)) + + made = DTLS(content_type=ContentType.tls12_cid, epoch=3, sequence_number=7, cid=b'\x0a\x0b', + content=bytes(range(20))) + self.assertEqual(bytes(made).hex(), _cid_record(b'\x0a\x0b', bytes(range(20)), epoch=3, + sequence_number=7).hex()) + self.assertEqual((made.alias, made.length, made.info.cid, made.info.length), ('DTLS 1.2', 15, b'\x0a\x0b', 20)) + + def test_a_record_made_with_a_payload_parses_it_again(self) -> None: + """The constructed payload's connection IDs are registered for its parse.""" + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.dtls import DTLS + + inner = DTLS(content_type=ContentType.tls12_cid, epoch=1, sequence_number=0, cid=SERVER_CID, + content=bytes(range(32))) + outer = DTLS(content_type=ContentType.change_cipher_spec, epoch=0, sequence_number=3, + content=b'\x01', payload=inner) + self.assertEqual(str(outer.protochain), 'DTLS 1.2:DTLS 1.2') + self.assertEqual(_cid_layers(outer), ((SERVER_CID, 1, 0, 32),)) + + def test_make_refuses_a_misplaced_or_invalid_id(self) -> None: + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + maker = object.__new__(DTLS) + for name, kwargs in (('none for tls12_cid', {'content_type': ContentType.tls12_cid}), + ('an empty one', {'content_type': ContentType.tls12_cid, 'cid': b''}), + ('a 256-octet one', {'content_type': ContentType.tls12_cid, 'cid': bytes(256)}), + ('one on application_data', {'content_type': ContentType.application_data, + 'cid': b'\x01'}), + ('one on a unified header', {'ciphertext': True, 'content': bytes(16), + 'cid': b'\x01'})): + with self.subTest(case=name), self.assertRaisesRegex(ProtocolError, 'invalid connection ID'): + maker.make(**kwargs) + + +if __name__ == '__main__': + unittest.main()