From 2f83321d38a92b50760ab120210704b38da9c51a Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 31 Oct 2026 02:29:50 -0400 Subject: [PATCH 1/3] feat(application): implement DTLS 1.3 unified-header records (#2467) - Parse and make the unified-header ciphertext record (RFC 9147 4.1) in DTLS: first octet 001CSLEE, 8- or 16-bit sequence number, optional length. - Dispatch on the first octet's 001 prefix ahead of legacy content types. - A record with a connection ID, or a length past the buffer, degrades to Raw. - Add DTLSCiphertext header schema and data model. - Update the DTLS docs and the constant-home allowlist for the EE bit offset. Tests: 11 new unit tests, all passing; DTLS modules at 100% coverage. --- .../pcapkit/protocols/application/dtls.rst | 35 ++- pcapkit/protocols/application/dtls.py | 189 ++++++++++++++-- pcapkit/protocols/data/application/dtls.py | 30 ++- pcapkit/protocols/schema/application/dtls.py | 51 ++++- .../test_rfc_constants_one_home_1739_unit.py | 3 + .../test_dtls13_unified_header_unit.py | 204 ++++++++++++++++++ 6 files changed, 475 insertions(+), 37 deletions(-) create mode 100644 tests/protocols/application/test_dtls13_unified_header_unit.py diff --git a/docs/source/pcapkit/protocols/application/dtls.rst b/docs/source/pcapkit/protocols/application/dtls.rst index ff5c75334c..4bb57106d4 100644 --- a/docs/source/pcapkit/protocols/application/dtls.rst +++ b/docs/source/pcapkit/protocols/application/dtls.rst @@ -6,8 +6,8 @@ DTLS - Datagram Transport Layer Security :mod:`pcapkit.protocols.application.dtls` contains :class:`~pcapkit.protocols.application.dtls.DTLS` only, 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 records -(:rfc:`9147`) -- at the record layer, plus the epoch/sequence-number fields and +(:rfc:`4347`), DTLS 1.2 (:rfc:`6347`) 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. @@ -34,12 +34,18 @@ every version and :meth:`~pcapkit.protocols.application.dtls.DTLS.read` sets - ``DTLS 1.3`` * - any other ``0xFExx`` - ``DTLS`` + * - unified header (first octet ``001CSLEE``, :rfc:`9147#section-4.1`) + - ``DTLS 1.3`` DTLS 1.3 never puts its own ``0xFEFC`` on the record layer -(:rfc:`9147#section-4`), so it is told apart only by the +(:rfc:`9147#section-4`), so a legacy-format record is told apart only by the ``supported_versions`` peek. A version the table does not name keeps the generic ``DTLS`` alias rather than claiming a specific one. +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. + Known Limitations ----------------- @@ -49,14 +55,14 @@ Known Limitations and its :attr:`~pcapkit.protocols.data.application.dtls.DTLSHandshake.body` kept as raw octets. -**DTLS 1.3's unified record header is not decoded** (:issue:`2467`). It has -no content-type octet (:rfc:`9147#section-4`), so it degrades to -:class:`~pcapkit.protocols.misc.raw.Raw`. DTLS 1.3's cleartext records use -the legacy ``DTLSPlaintext`` header and are decoded. +**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`). -**Connection ID records are not decoded** (``tls12_cid``, :rfc:`9146`): the -connection ID's length is not carried in the record, so they degrade to -:class:`~pcapkit.protocols.misc.raw.Raw`. +**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 +ones on the wire. .. versionadded:: 1.6.0 @@ -76,6 +82,7 @@ connection ID's length is not carried in the record, so they degrade to .. automethod:: read .. automethod:: make + .. automethod:: unpack .. automethod:: __length_hint__ @@ -90,6 +97,10 @@ Header Schema :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.schema.application.dtls.DTLSCiphertext + :members: + :show-inheritance: + Data Models ----------- @@ -99,6 +110,10 @@ Data Models :members: :show-inheritance: +.. autoclass:: pcapkit.protocols.data.application.dtls.DTLSCiphertext + :members: + :show-inheritance: + .. autoclass:: pcapkit.protocols.data.application.dtls.DTLSHandshake :members: :show-inheritance: diff --git a/pcapkit/protocols/application/dtls.py b/pcapkit/protocols/application/dtls.py index 21555b084f..41ecc25e23 100644 --- a/pcapkit/protocols/application/dtls.py +++ b/pcapkit/protocols/application/dtls.py @@ -7,10 +7,11 @@ :mod:`pcapkit.protocols.application.dtls` contains :class:`~pcapkit.protocols.application.dtls.DTLS` only, 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 records (:rfc:`9147`). -: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. +1.2 (:rfc:`6347`), and DTLS 1.3's legacy-format and unified-header records +(:rfc:`9147`). :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. Known Limitations ----------------- @@ -21,14 +22,14 @@ and its :attr:`~pcapkit.protocols.data.application.dtls.DTLSHandshake.body` kept as raw octets. -**DTLS 1.3's unified record header is not decoded** (:issue:`2467`). It has -no content-type octet (:rfc:`9147#section-4`), so it degrades to -:class:`~pcapkit.protocols.misc.raw.Raw`. DTLS 1.3's cleartext records use -the legacy ``DTLSPlaintext`` header and are decoded. +**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`). -**Connection ID records are not decoded** (``tls12_cid``, :rfc:`9146`): the -connection ID's length is not carried in the record, so they degrade to -:class:`~pcapkit.protocols.misc.raw.Raw`. +**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 +ones on the wire. .. versionadded:: 1.6.0 @@ -38,7 +39,10 @@ from pcapkit.const.tls.content_type import ContentType as Enum_ContentType from pcapkit.protocols.application.ssl import 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 +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 @@ -60,6 +64,37 @@ #: Legacy record versions a DTLS 1.3 record may carry: DTLS 1.2's always, and #: DTLS 1.0's on an initial ``ClientHello`` (:rfc:`9147#section-4`). _DTLS13_LEGACY_VERSIONS = frozenset({_DTLSV1_0_VERSION, _DTLSV1_2_VERSION}) +#: Top three bits of a unified-header first octet, ``001`` (:rfc:`9147#section-4.1`). +_UNIFIED_PREFIX = 0b001 + + +def _is_unified_header(octet: int) -> bool: + """Whether ``octet`` opens a unified-header record. + + Legacy records open with a content type of 20 through 27, whose top three + bits are ``000``, so the ``001`` prefix never matches one of them. + + Args: + octet: First octet of the record. + + Returns: + Whether the octet's top three bits are the unified-header prefix. + + """ + return octet >> 5 == _UNIFIED_PREFIX + + +def _ciphertext_header_length(flags: 'dict[str, int]') -> 'int': + """Length of a unified header, from its first octet's decoded flags. + + Args: + flags: Decoded first octet, as the header schema holds it. + + Returns: + First octet, sequence number and, if present, the length field. + + """ + return 1 + (2 if flags['sequence_long'] else 1) + (2 if flags['length_present'] else 0) class DTLS(SSL[Data_DTLS, Schema_DTLS], schema=Schema_DTLS, data=Data_DTLS): @@ -67,7 +102,7 @@ class DTLS(SSL[Data_DTLS, Schema_DTLS], schema=Schema_DTLS, data=Data_DTLS): - Datagram Transport Layer Security 1.0 [:rfc:`4347`] - Datagram Transport Layer Security 1.2 [:rfc:`6347`] - - Datagram Transport Layer Security 1.3, legacy-format records only [:rfc:`9147`] + - Datagram Transport Layer Security 1.3, legacy-format and unified-header records [:rfc:`9147`] Note: The subpackage does not track the dispatch tier: DTLS is dispatched @@ -96,12 +131,15 @@ def alias(self) -> 'Literal["DTLS", "DTLS 1.0", "DTLS 1.2", "DTLS 1.3"]': return self._alias @property - def length(self) -> 'Literal[13]': - """Header length of current protocol -- the fixed record header, - :attr:`~pcapkit.protocols.data.application.dtls.DTLS.length`'s - fragment excluded. + 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 + header's length follows from its first octet. """ + schema = getattr(self, '__header__', None) + if isinstance(schema, Schema_DTLSCiphertext): + return _ciphertext_header_length(schema.flags) return 13 ########################################################################## @@ -148,7 +186,10 @@ def make(self, content: 'Any' = b'', data: 'Optional[bytes]' = None, payload: 'bytes | ProtocolBase | Schema' = b'', - **kwargs: 'Any') -> 'Schema_DTLS': + ciphertext: 'bool' = False, + sequence_long: 'bool' = False, + length_present: 'bool' = True, + **kwargs: 'Any') -> 'Schema_DTLS | Schema_DTLSCiphertext': """Make (construct) packet data, for every tier: ``version`` is written as is. Args: @@ -159,25 +200,50 @@ def make(self, version: Legacy record version (``major << 8 | minor``), e.g. ``0xFEFD`` for DTLS 1.2 and (the legacy value DTLS 1.3 keeps here) DTLS 1.3. - epoch: Epoch. - sequence_number: Sequence number within ``epoch`` (48-bit). + epoch: Epoch. A unified-header record carries only its low two + bits, so a larger value is refused. + sequence_number: Sequence number within ``epoch`` (48-bit); for a + unified-header record, 8 or 16 bits as ``sequence_long`` says. content: Record content: a :obj:`tuple` of :class:`~pcapkit.protocols.data.application.dtls.DTLSHandshake` for ``handshake``, an :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. + or raw :obj:`bytes`, as :meth:`read` would decode it. For a + unified-header 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. + ciphertext: Make a unified-header record (:rfc:`9147#section-4.1`) + instead of a legacy one; ``content_type``, ``version`` and + ``data`` do not apply. + sequence_long: Whether a unified-header record's sequence number is + 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. **kwargs: Arbitrary keyword arguments. Returns: Constructed packet data. """ + if ciphertext: + return Schema_DTLSCiphertext( + flags={ + 'fixed': _UNIFIED_PREFIX, + 'connection_id': 0, + 'sequence_long': int(sequence_long), + 'length_present': int(length_present), + 'epoch': epoch, + }, + sequence_number=sequence_number, + length=len(content) if length_present else None, + fragment=content, + payload=payload, + ) + content_type_ = cast('Enum_ContentType', self._make_index( content_type, content_type_default, namespace=content_type_namespace, reversed=content_type_reversed, pack=False)) @@ -193,6 +259,29 @@ def make(self, payload=payload, ) + def unpack(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS | Data_DTLSCiphertext': + """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. + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + """ + octets = self._data[:1] + if self.__header__ is None and octets and _is_unified_header(octets[0]): + packet = kwargs.get('__packet__', {}) # packet data + self.__header__ = Schema_DTLSCiphertext.unpack(self._file, length, packet) # type: ignore[assignment] + return self.read(length, **kwargs) + return super().unpack(length, **kwargs) + ########################################################################## # Data models. ########################################################################## @@ -206,7 +295,7 @@ def __length_hint__(self) -> 'Literal[13]': ########################################################################## @classmethod - def _make_data(cls, data: 'Data_DTLS') -> 'dict[str, Any]': # type: ignore[override] + def _make_data(cls, data: 'Data_DTLS | Data_DTLSCiphertext') -> 'dict[str, Any]': # type: ignore[override] """Create key-value pairs from ``data`` for protocol construction. Args: @@ -216,6 +305,16 @@ def _make_data(cls, data: 'Data_DTLS') -> 'dict[str, Any]': # type: ignore[over Key-value pairs for protocol construction. """ + if isinstance(data, Data_DTLSCiphertext): + return { + 'ciphertext': True, + 'epoch': data.epoch, + 'sequence_number': data.sequence_number, + 'sequence_long': data.sequence_long, + 'length_present': data.length is not None, + 'content': data.content, + 'payload': cls._make_payload(data), + } return { 'content_type': data.content_type, 'version': data.version, @@ -225,9 +324,11 @@ def _make_data(cls, data: 'Data_DTLS') -> 'dict[str, Any]': # type: ignore[over 'payload': cls._make_payload(data), } - def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS': # pylint: disable=unused-argument + def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_DTLS | Data_DTLSCiphertext': # pylint: disable=unused-argument """Read a DTLS record, identifying its concrete tier as it goes. + A unified-header record is handed to :meth:`_read_dtls_ciphertext`. + Structure of DTLS record layer [:rfc:`9147#section-4`]: .. code-block:: text @@ -262,6 +363,9 @@ def _read_dtls_record(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> malformed for its content type. """ + if isinstance(self.__header__, Schema_DTLSCiphertext): + return self._read_dtls_ciphertext(length, **kwargs) + schema = self.__header__ available = length if length is not None else len(self) self._validate_tls_record(schema, available=available, dtls=True) @@ -290,3 +394,44 @@ 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 _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`). + + Args: + length: Length of packet data. + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If the header carries a connection ID, whose length + is negotiated and not in the record; if the header or its + declared length runs past the buffer; or if there is no + ciphertext after the header. + + """ + self._alias = 'DTLS 1.3' + schema = self.__header__ + flags = schema.flags + available = length if length is not None else len(self) + + if flags['connection_id']: + raise ProtocolError(f'{self.alias}: invalid format: connection ID records are not decoded') + needed = self.length + (schema.length or 0) + if needed > available: + raise ProtocolError(f'{self.alias}: invalid format: record of {needed} octets ' + 'runs past the end of the buffer') + if not schema.fragment: + raise ProtocolError(f'{self.alias}: invalid format: no ciphertext after the record header') + + dtls = Data_DTLSCiphertext( + epoch=flags['epoch'], + sequence_number=schema.sequence_number, + sequence_long=bool(flags['sequence_long']), + length=schema.length, + content=schema.fragment, + ) + remaining = available - (self.length + len(schema.fragment)) + return self._decode_next_layer(dtls, -1, remaining) diff --git a/pcapkit/protocols/data/application/dtls.py b/pcapkit/protocols/data/application/dtls.py index 664fd3b4ee..96e65ee8c0 100644 --- a/pcapkit/protocols/data/application/dtls.py +++ b/pcapkit/protocols/data/application/dtls.py @@ -8,7 +8,7 @@ from pcapkit.protocols.data.protocol import Protocol if TYPE_CHECKING: - from typing import Any, Union + from typing import Any, Optional, Union from pcapkit.const.tls.content_type import ContentType as Enum_ContentType from pcapkit.const.tls.handshake_type import HandshakeType as Enum_HandshakeType @@ -16,6 +16,7 @@ __all__ = [ 'DTLS', + 'DTLSCiphertext', 'DTLSHandshake', ] @@ -89,3 +90,30 @@ class DTLS(Protocol): if TYPE_CHECKING: def __init__(self, content_type: 'Enum_ContentType', version: 'int', epoch: 'int', sequence_number: 'int', length: 'int', content: 'Any') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,line-too-long + + +@info_final +class DTLSCiphertext(Protocol): + """Data model for a DTLS 1.3 ciphertext record (:rfc:`9147#section-4`). + + The unified header carries neither a content type nor a version, and only + the low two bits of the epoch, so those fields are absent here: the + content type is inside :attr:`content`. + + """ + + #: Low two bits of the epoch; the full 16-bit epoch is not on the wire. + epoch: 'int' + #: Sequence number as carried, 8 or 16 bits per :attr:`sequence_long`. It + #: is header-protected, so it is the value on the wire, not the record's. + sequence_number: 'int' + #: Whether :attr:`sequence_number` is 16 bits rather than 8. + sequence_long: 'bool' + #: Length of :attr:`content`, or :data:`None` when the record runs to the + #: end of the datagram. + length: 'Optional[int]' + #: Encrypted record content, undecoded. + content: 'bytes' + + 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 diff --git a/pcapkit/protocols/schema/application/dtls.py b/pcapkit/protocols/schema/application/dtls.py index ae6bfba1c1..c336799159 100644 --- a/pcapkit/protocols/schema/application/dtls.py +++ b/pcapkit/protocols/schema/application/dtls.py @@ -5,12 +5,12 @@ from typing import TYPE_CHECKING from pcapkit.const.tls.content_type import ContentType as Enum_ContentType -from pcapkit.corekit.fields.misc import PayloadField -from pcapkit.corekit.fields.numbers import EnumField, NumberField, UInt16Field -from pcapkit.corekit.fields.strings import BytesField +from pcapkit.corekit.fields.misc import ConditionalField, PayloadField, SwitchField +from pcapkit.corekit.fields.numbers import EnumField, NumberField, UInt8Field, UInt16Field +from pcapkit.corekit.fields.strings import BitField, BytesField from pcapkit.protocols.schema.schema import Schema, schema_final -__all__ = ['DTLS'] +__all__ = ['DTLS', 'DTLSCiphertext'] if TYPE_CHECKING: from pcapkit.protocols.protocol import ProtocolBase @@ -50,3 +50,46 @@ class DTLS(Schema): def __init__(self, content_type: 'Enum_ContentType', version: 'int', epoch: 'int', sequence_number: 'int', length: 'int', fragment: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... + + +@schema_final +class DTLSCiphertext(Schema): + """Header schema for a DTLS 1.3 ciphertext record, which has a unified header. + + The first octet is ``001CSLEE`` (:rfc:`9147#section-4.1`): ``C`` names a + connection ID, ``S`` an 8- or 16-bit sequence number, ``L`` a length field, + and ``EE`` the low two bits of the epoch. The record has no content type or + version octet, since both are inside the ciphertext, so + :class:`DTLS `'s legacy + header cannot describe it. + + """ + + #: Unified-header first octet: fixed prefix, ``C``, ``S``, ``L`` and ``EE``. + flags: 'dict[str, int]' = BitField(length=1, namespace={ + 'fixed': (0, 3), + 'connection_id': (3, 1), + 'sequence_long': (4, 1), + 'length_present': (5, 1), + 'epoch': (6, 2), + }) + #: Sequence number, 8 or 16 bits as ``S`` says (header-protected on the wire). + sequence_number: 'int' = SwitchField( + selector=lambda pkt: UInt16Field() if pkt['flags']['sequence_long'] else UInt8Field(), + ) + #: Length of :attr:`fragment`; absent when ``L`` is clear. + length: 'int | None' = ConditionalField( + UInt16Field(), + condition=lambda pkt: bool(pkt['flags']['length_present']), + ) + #: Ciphertext, undecoded. Without a length it runs to the end of the buffer. + fragment: 'bytes' = SwitchField( + selector=lambda pkt: (BytesField(length=pkt['length']) + if pkt['flags']['length_present'] else PayloadField()), + ) + #: Octets after the record, e.g. a further record coalesced in the same + #: buffer. Empty when :attr:`fragment` runs to the end of the buffer. + payload: 'bytes' = PayloadField() + + 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 diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index dbbf8353c0..a6ee9abaab 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -211,6 +211,9 @@ ('pcapkit/protocols/schema/application/dtls.py', 'NumberField(length=6)'): "DTLS's 48-bit sequence_number (:rfc:`9147#section-4`), not the IPv6 Next " 'Header offset', + ('pcapkit/protocols/schema/application/dtls.py', "{'epoch': (6, 2)}"): + "the EE bits of a DTLS 1.3 unified header, at bit 6 of its first octet " + '(:rfc:`9147#section-4.1`), not the IPv6 Next Header offset', ('pcapkit/protocols/internet/ipv6.py', "schema.hextet['version'] != 6"): 'the IP version number, not the Next Header offset', ('pcapkit/protocols/internet/ipv6.py', 'self._import_next_layer(version=6)'): diff --git a/tests/protocols/application/test_dtls13_unified_header_unit.py b/tests/protocols/application/test_dtls13_unified_header_unit.py new file mode 100644 index 0000000000..c888783253 --- /dev/null +++ b/tests/protocols/application/test_dtls13_unified_header_unit.py @@ -0,0 +1,204 @@ +# -*- coding: utf-8 -*- +"""Unit tests for DTLS 1.3's unified-header ciphertext records (:issue:`2467`). + +RFC 9147 section 4.1 gives an encrypted record a unified header, first octet +``001CSLEE``: ``C`` names a connection ID, ``S`` an 8- or 16-bit sequence +number, ``L`` a length field, and ``EE`` the low two bits of the epoch. + +""" +from __future__ import annotations + +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: First octets for the four flag combinations the tests use, with ``C`` clear +#: unless named: ``001 0 S L EE``. +_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 + +#: A 12-octet stand-in for an AEAD ciphertext. +_FRAGMENT = bytes(range(0xA0, 0xAC)) + + +def _unified(first: int, sequence: bytes, length: 'bytes | None', fragment: bytes) -> bytes: + """A unified-header record: the first octet, the sequence number, the + length if one is carried, then the ciphertext.""" + return bytes([first]) + sequence + (length or b'') + fragment + + +def _udp_to_5684(payload: bytes) -> bytes: + """An 8-octet UDP header to port 5684, where DTLS is bound, plus ``payload``.""" + return ((50000).to_bytes(2, 'big') + (5684).to_bytes(2, 'big') + + (8 + len(payload)).to_bytes(2, 'big') + b'\x00\x00' + payload) + + +class UnifiedHeaderParseTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_length_present_with_an_8_bit_sequence_number(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + wire = _unified(_L_SHORT_SEQ_EPOCH1, b'\x2a', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + + parsed = DTLS(wire, len(wire)) + self.assertEqual(parsed.alias, 'DTLS 1.3') + self.assertEqual(parsed.length, 4) # first octet, sequence number, length + self.assertEqual(parsed.info.epoch, 1) + self.assertEqual(parsed.info.sequence_number, 0x2A) + self.assertFalse(parsed.info.sequence_long) + self.assertEqual(parsed.info.length, len(_FRAGMENT)) + self.assertEqual(parsed.info.content, _FRAGMENT) + self.assertEqual(bytes(DTLS.from_data(parsed.info)), wire) + + def test_length_present_with_a_16_bit_sequence_number(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + wire = _unified(_L_LONG_SEQ_EPOCH2, b'\x12\x34', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + + parsed = DTLS(wire, len(wire)) + self.assertEqual(parsed.alias, 'DTLS 1.3') + self.assertEqual(parsed.length, 5) + self.assertEqual(parsed.info.epoch, 2) + self.assertEqual(parsed.info.sequence_number, 0x1234) + self.assertTrue(parsed.info.sequence_long) + self.assertEqual(parsed.info.content, _FRAGMENT) + self.assertEqual(bytes(DTLS.from_data(parsed.info)), wire) + + def test_without_a_length_the_record_runs_to_the_end_of_the_datagram(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + wire = _unified(_NO_LENGTH_EPOCH3, b'\x07', None, _FRAGMENT) + + parsed = DTLS(wire, len(wire)) + self.assertEqual(parsed.alias, 'DTLS 1.3') + self.assertEqual(parsed.length, 2) # first octet and sequence number only + self.assertEqual(parsed.info.epoch, 3) + self.assertIsNone(parsed.info.length) + self.assertEqual(parsed.info.content, _FRAGMENT) + self.assertEqual(bytes(DTLS.from_data(parsed.info)), wire) + + def test_a_coalesced_record_after_the_length_is_not_swallowed(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + trailing = b'\xde\xad' + wire = _unified(_L_SHORT_SEQ_EPOCH1, b'\x07', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + trailing + + parsed = DTLS(wire, len(wire)) + self.assertEqual(parsed.info.content, _FRAGMENT) + self.assertEqual(bytes(parsed.payload), trailing) + + def test_a_connection_id_is_refused_rather_than_misread(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + # The connection ID's length is negotiated, not carried, so the + # record cannot be cut at the right place. + wire = _unified(_CID_PRESENT, b'\x07', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + + with self.assertRaises(ProtocolError): + DTLS(wire, len(wire)) + + def test_a_length_past_the_end_of_the_datagram_is_refused(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + wire = _unified(_L_SHORT_SEQ_EPOCH1, b'\x07', (0x20).to_bytes(2, 'big'), _FRAGMENT) + + with self.assertRaises(ProtocolError): + DTLS(wire, len(wire)) + + def test_a_header_with_no_ciphertext_is_refused(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + wire = _unified(_NO_LENGTH_EPOCH3, b'\x07', None, b'') + + with self.assertRaises(ProtocolError): + DTLS(wire, len(wire)) + + def test_udp_port_5684_dispatches_the_unified_record(self) -> None: + from pcapkit.protocols.transport.udp import UDP + + wire = _udp_to_5684(_unified(_L_SHORT_SEQ_EPOCH1, b'\x2a', + len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT)) + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = UDP(wire, len(wire)) + self.assertEqual(str(proto.protochain), 'UDP:DTLS 1.3') + + def test_a_refused_connection_id_record_degrades_to_raw_on_dispatch(self) -> None: + from pcapkit.protocols.transport.udp import UDP + + # The genuine record is asserted alongside, so the refusal cannot be + # satisfied by a binding that resolves to ``Raw`` unconditionally. + good = _unified(_L_SHORT_SEQ_EPOCH1, b'\x2a', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + refused = _unified(_CID_PRESENT, b'\x2a', len(_FRAGMENT).to_bytes(2, 'big'), _FRAGMENT) + for name, payload, expected in (('a genuine record', good, 'UDP:DTLS 1.3'), + ('a connection ID', refused, 'UDP:Raw')): + with self.subTest(payload=name): + wire = _udp_to_5684(payload) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = UDP(wire, len(wire)) + self.assertEqual(str(proto.protochain), expected) + + +class UnifiedHeaderMakeTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_make_writes_each_flag_combination(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + cases = ( + ('length, 8-bit sequence', dict(epoch=1, sequence_number=0x2A, sequence_long=False, + length_present=True), _L_SHORT_SEQ_EPOCH1), + ('length, 16-bit sequence', dict(epoch=2, sequence_number=0x1234, sequence_long=True, + length_present=True), _L_LONG_SEQ_EPOCH2), + ('no length', dict(epoch=3, sequence_number=0x07, sequence_long=False, + length_present=False), _NO_LENGTH_EPOCH3), + ) + for name, options, first in cases: + with self.subTest(case=name): + sequence_len = 2 if options['sequence_long'] else 1 + expected = bytes([first]) + options['sequence_number'].to_bytes(sequence_len, 'big') + if options['length_present']: + expected += len(_FRAGMENT).to_bytes(2, 'big') + expected += _FRAGMENT + + made = bytes(DTLS(ciphertext=True, content=_FRAGMENT, **options)) + self.assertEqual(made, expected) + + def test_make_then_parse_round_trips(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + for sequence_long in (False, True): + for length_present in (False, True): + with self.subTest(sequence_long=sequence_long, length_present=length_present): + made = DTLS(ciphertext=True, epoch=2, sequence_number=0x5A, + sequence_long=sequence_long, length_present=length_present, + content=_FRAGMENT) + parsed = DTLS(bytes(made), len(bytes(made))) + self.assertEqual(parsed.alias, 'DTLS 1.3') + self.assertEqual(parsed.info.epoch, 2) + self.assertEqual(parsed.info.sequence_number, 0x5A) + self.assertEqual(parsed.info.sequence_long, sequence_long) + self.assertEqual(parsed.info.content, _FRAGMENT) + self.assertEqual(parsed.info.length, len(_FRAGMENT) if length_present else None) + + def test_an_epoch_wider_than_two_bits_is_refused(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import FieldValueError + + with self.assertRaises(FieldValueError): + DTLS(ciphertext=True, epoch=4, content=_FRAGMENT) + + +if __name__ == '__main__': + unittest.main() From e8011a8b91555b340a7980de443c73e9736a5f21 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 31 Oct 2026 02:57:31 -0400 Subject: [PATCH 2/3] fix(application): refuse DTLS 1.3 unified-header records under the sample length (#2467) - Refuse a unified-header ciphertext shorter than 16 octets, the record-number sample (RFC 9147 4.2.3), so the datagram degrades to Raw on UDP 5684. - Raise the test fixture to 16 octets; the old 12-octet stand-in was not a record the RFC allows. - Allowlist the new constant in the one-home test: it is not a pcap header. Tests: new short-ciphertext cases fail without the check and pass with it. --- pcapkit/protocols/application/dtls.py | 13 ++++-- .../test_rfc_constants_one_home_1739_unit.py | 3 ++ .../test_dtls13_unified_header_unit.py | 42 ++++++++++++++++++- 3 files changed, 52 insertions(+), 6 deletions(-) diff --git a/pcapkit/protocols/application/dtls.py b/pcapkit/protocols/application/dtls.py index 41ecc25e23..984008b7ca 100644 --- a/pcapkit/protocols/application/dtls.py +++ b/pcapkit/protocols/application/dtls.py @@ -66,6 +66,9 @@ _DTLS13_LEGACY_VERSIONS = frozenset({_DTLSV1_0_VERSION, _DTLSV1_2_VERSION}) #: Top three bits of a unified-header first octet, ``001`` (:rfc:`9147#section-4.1`). _UNIFIED_PREFIX = 0b001 +#: 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 def _is_unified_header(octet: int) -> bool: @@ -408,8 +411,8 @@ def _read_dtls_ciphertext(self, length: 'Optional[int]' = None, **kwargs: 'Any') Raises: ProtocolError: If the header carries a connection ID, whose length is negotiated and not in the record; if the header or its - declared length runs past the buffer; or if there is no - ciphertext after the header. + declared length runs past the buffer; or if the ciphertext is + shorter than the record-number sample (:rfc:`9147#section-4.2.3`). """ self._alias = 'DTLS 1.3' @@ -423,8 +426,10 @@ def _read_dtls_ciphertext(self, length: 'Optional[int]' = None, **kwargs: 'Any') if needed > available: raise ProtocolError(f'{self.alias}: invalid format: record of {needed} octets ' 'runs past the end of the buffer') - if not schema.fragment: - raise ProtocolError(f'{self.alias}: invalid format: no ciphertext after the record header') + # Receivers MUST reject shorter records as if they failed deprotection. + if len(schema.fragment or b'') < _CIPHERTEXT_MIN_LENGTH: + raise ProtocolError(f'{self.alias}: invalid format: ciphertext is shorter than the ' + 'record-number sample (RFC 9147 section 4.2.3)') dtls = Data_DTLSCiphertext( epoch=flags['epoch'], diff --git a/tests/project/test_rfc_constants_one_home_1739_unit.py b/tests/project/test_rfc_constants_one_home_1739_unit.py index a6ee9abaab..533488c070 100644 --- a/tests/project/test_rfc_constants_one_home_1739_unit.py +++ b/tests/project/test_rfc_constants_one_home_1739_unit.py @@ -214,6 +214,9 @@ ('pcapkit/protocols/schema/application/dtls.py', "{'epoch': (6, 2)}"): "the EE bits of a DTLS 1.3 unified header, at bit 6 of its first octet " '(:rfc:`9147#section-4.1`), not the IPv6 Next Header offset', + ('pcapkit/protocols/application/dtls.py', '_CIPHERTEXT_MIN_LENGTH = 16'): + "the shortest DTLS 1.3 ciphertext, the record-number sample length " + '(:rfc:`9147#section-4.2.3`), not a pcap record header', ('pcapkit/protocols/internet/ipv6.py', "schema.hextet['version'] != 6"): 'the IP version number, not the Next Header offset', ('pcapkit/protocols/internet/ipv6.py', 'self._import_next_layer(version=6)'): diff --git a/tests/protocols/application/test_dtls13_unified_header_unit.py b/tests/protocols/application/test_dtls13_unified_header_unit.py index c888783253..9d07c1acc9 100644 --- a/tests/protocols/application/test_dtls13_unified_header_unit.py +++ b/tests/protocols/application/test_dtls13_unified_header_unit.py @@ -20,8 +20,9 @@ _NO_LENGTH_EPOCH3 = 0x23 # S=0, L=0, EE=11 _CID_PRESENT = 0x38 # C=1, S=0, L=1, EE=00 -#: A 12-octet stand-in for an AEAD ciphertext. -_FRAGMENT = bytes(range(0xA0, 0xAC)) +#: A 16-octet stand-in for an AEAD ciphertext, the shortest the record-number +#: sample allows (:rfc:`9147#section-4.2.3`). +_FRAGMENT = bytes(range(0xA0, 0xB0)) def _unified(first: int, sequence: bytes, length: 'bytes | None', fragment: bytes) -> bytes: @@ -121,6 +122,30 @@ def test_a_header_with_no_ciphertext_is_refused(self) -> None: with self.assertRaises(ProtocolError): DTLS(wire, len(wire)) + def test_a_ciphertext_shorter_than_the_sample_is_refused(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + from pcapkit.utilities.exceptions import ProtocolError + + # RFC 9147 section 4.2.3: a record under 16 octets of ciphertext MUST + # be rejected as if it had failed deprotection. + short = _FRAGMENT[:15] + for name, length in (('length present', len(short).to_bytes(2, 'big')), ('length absent', None)): + with self.subTest(case=name): + wire = _unified(_L_SHORT_SEQ_EPOCH1 if length else _NO_LENGTH_EPOCH3, b'\x07', length, short) + with self.assertRaises(ProtocolError): + DTLS(wire, len(wire)) + + def test_a_ciphertext_of_the_minimum_length_is_accepted(self) -> None: + from pcapkit.protocols.application.dtls import DTLS + + for name, first, length in (('length present', _L_SHORT_SEQ_EPOCH1, len(_FRAGMENT).to_bytes(2, 'big')), + ('length absent', _NO_LENGTH_EPOCH3, None)): + with self.subTest(case=name): + wire = _unified(first, b'\x07', length, _FRAGMENT) + parsed = DTLS(wire, len(wire)) + self.assertEqual(parsed.alias, 'DTLS 1.3') + self.assertEqual(parsed.info.content, _FRAGMENT) + def test_udp_port_5684_dispatches_the_unified_record(self) -> None: from pcapkit.protocols.transport.udp import UDP @@ -149,6 +174,19 @@ def test_a_refused_connection_id_record_degrades_to_raw_on_dispatch(self) -> Non self.assertEqual(str(proto.protochain), expected) + def test_a_datagram_the_record_format_forbids_degrades_to_raw_on_dispatch(self) -> None: + from pcapkit.protocols.transport.udp import UDP + + # First octet in the 001 range and the header fits, but the ciphertext + # is under the sample length, so it is not a DTLS 1.3 record. + wire = _udp_to_5684(_unified(_L_SHORT_SEQ_EPOCH1, b'\x2a', + len(_FRAGMENT[:15]).to_bytes(2, 'big'), _FRAGMENT[:15])) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = UDP(wire, len(wire)) + self.assertEqual(str(proto.protochain), 'UDP:Raw') + + class UnifiedHeaderMakeTests(unittest.TestCase): def setUp(self) -> None: reimport_once_per_class(self) From 27e5bcfe5b90c0eb41b84dc0656b1ae082ccfabc Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 31 Oct 2026 04:07:22 -0400 Subject: [PATCH 3/3] fix(protocols): re-export DTLSCiphertext from the data and schema packages (#2467) - pcapkit/protocols/data/application/__init__.py and pcapkit/protocols/schema/application/__init__.py do not import DTLSCiphertext, which their dtls __all__ lists. - Re-export it as DTLS_Ciphertext, matching DTLS_Handshake. - Add the alias to the data and schema aggregator __all__ lists. Fixes test_package_reexports_1807_unit (test_every_package_reexports_every_child_export). --- pcapkit/protocols/data/__init__.py | 2 +- pcapkit/protocols/data/application/__init__.py | 3 ++- pcapkit/protocols/schema/__init__.py | 2 +- pcapkit/protocols/schema/application/__init__.py | 3 ++- 4 files changed, 6 insertions(+), 4 deletions(-) diff --git a/pcapkit/protocols/data/__init__.py b/pcapkit/protocols/data/__init__.py index 9c700e139b..344d1f8d0d 100644 --- a/pcapkit/protocols/data/__init__.py +++ b/pcapkit/protocols/data/__init__.py @@ -240,7 +240,7 @@ 'UDP', # Datagram Transport Layer Security - 'DTLS', 'DTLS_Handshake', + 'DTLS', 'DTLS_Ciphertext', 'DTLS_Handshake', # File Transfer Protocol 'FTP', diff --git a/pcapkit/protocols/data/application/__init__.py b/pcapkit/protocols/data/application/__init__.py index 5ce80b3e29..e687bd4e1c 100644 --- a/pcapkit/protocols/data/application/__init__.py +++ b/pcapkit/protocols/data/application/__init__.py @@ -3,6 +3,7 @@ # Datagram Transport Layer Security from pcapkit.protocols.data.application.dtls import DTLS +from pcapkit.protocols.data.application.dtls import DTLSCiphertext as DTLS_Ciphertext from pcapkit.protocols.data.application.dtls import DTLSHandshake as DTLS_Handshake # File Transfer Protocol @@ -64,7 +65,7 @@ __all__ = [ # Datagram Transport Layer Security - 'DTLS', 'DTLS_Handshake', + 'DTLS', 'DTLS_Ciphertext', 'DTLS_Handshake', # File Transfer Protocol 'FTP', diff --git a/pcapkit/protocols/schema/__init__.py b/pcapkit/protocols/schema/__init__.py index 7dffb2cae6..b8c21234ac 100644 --- a/pcapkit/protocols/schema/__init__.py +++ b/pcapkit/protocols/schema/__init__.py @@ -172,7 +172,7 @@ 'SCTP_ProtocolViolationCause', # Application Layer Protocols - 'DTLS', + 'DTLS', 'DTLS_Ciphertext', 'FTP', 'NGAP', 'OSPF', 'OSPF_CryptographicAuthentication', diff --git a/pcapkit/protocols/schema/application/__init__.py b/pcapkit/protocols/schema/application/__init__.py index 1ace1f1a17..8f2780d46f 100644 --- a/pcapkit/protocols/schema/application/__init__.py +++ b/pcapkit/protocols/schema/application/__init__.py @@ -3,6 +3,7 @@ # Datagram Transport Layer Security from pcapkit.protocols.schema.application.dtls import DTLS +from pcapkit.protocols.schema.application.dtls import DTLSCiphertext as DTLS_Ciphertext # File Transfer Protocol from pcapkit.protocols.schema.application.ftp import FTP @@ -40,7 +41,7 @@ __all__ = [ # Datagram Transport Layer Security - 'DTLS', + 'DTLS', 'DTLS_Ciphertext', # File Transfer Protocol 'FTP',