diff --git a/docs/source/contributing/conventions/protocol-family-grouping.rst b/docs/source/contributing/conventions/protocol-family-grouping.rst index 76bdbc98c3..387c9d735c 100644 --- a/docs/source/contributing/conventions/protocol-family-grouping.rst +++ b/docs/source/contributing/conventions/protocol-family-grouping.rst @@ -6,7 +6,7 @@ Protocol Family Grouping A protocol family is a base class named after the family plus its member classes. What each class returns from :meth:`~pcapkit.protocols.protocol.Protocol.id` is what the selectors on :class:`~pcapkit.protocols.protocol.Protocol` match against. The -rulings were set on :pr:`2441`, :pr:`2443`, :pr:`2446` and :issue:`2462`. +rulings were set on :pr:`2441`, :pr:`2443`, :pr:`2446`, :issue:`2462` and :issue:`2517`. The ``id()`` Criterion ~~~~~~~~~~~~~~~~~~~~~~ @@ -101,6 +101,16 @@ A family-inclusive member is selected by any sibling's registry key (a key of and by the key of each family whose ``id()`` lists it: ``AH`` and ``ESP`` by ``IPsec`` and by ``IPv6_Ext``. +Chained Records +~~~~~~~~~~~~~~~ + +**TLS records sharing a TCP segment are chained layers, not a list on one layer** +(:issue:`2517`): each record dissected after the first is the previous record's payload, +so its dynamic alias names its own version -- a ``ServerHello`` then a +``ChangeCipherSpec`` read ``TCP:TLS 1.3:TLS 1.2``. DTLS and SSLv2 do not chain: two DTLS +records in one datagram read ``DTLS 1.2:Raw`` (:issue:`2535`), and an SSLv2 hello +followed by a further record reads ``SSLv2:Raw``. + .. _protocol-family-canonical-names: Canonical Names diff --git a/docs/source/pcapkit/protocols/application/tls.rst b/docs/source/pcapkit/protocols/application/tls.rst index 4162411c7b..d5a22757ec 100644 --- a/docs/source/pcapkit/protocols/application/tls.rst +++ b/docs/source/pcapkit/protocols/application/tls.rst @@ -83,12 +83,31 @@ Decoded Fields * Everything else -- the same types once encrypted, ``application_data`` and every other content type -- as opaque :obj:`bytes`. +Records in One Segment +---------------------- + +A whole record after the first in a TCP segment is dissected as a further +:class:`~pcapkit.protocols.application.tls.TLS` layer, the previous record's +payload (:issue:`2517`): a ``ServerHello`` followed by a ``ChangeCipherSpec`` +reads ``TCP:TLS 1.3:TLS 1.2``. The exceptions: + +* A whole record that fails to decode becomes + :class:`~pcapkit.protocols.misc.raw.Raw` together with everything after it, + with no SSLv2 trial. +* Octets that open no record -- an unsupported content type, a legacy + version outside ``0x03xx``, or a header cut short -- stay raw, with no error. +* A parse told to stop at TLS (``protocol='TLS'`` or ``layer='Application'``) + keeps the rest of the segment raw. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + Known Limitation ---------------- **A record split across two TCP segments is not reassembled**, and decodes as -:class:`~pcapkit.protocols.misc.raw.Raw`. A record may be up to 16 KiB -against a typical 1460-octet MSS, so this is routine. +:class:`~pcapkit.protocols.misc.raw.Raw`, as does the partial record a segment +ends inside. A record may be up to 16 KiB against a typical 1460-octet MSS, so +this is routine. .. versionadded:: 1.6.0 @@ -108,6 +127,7 @@ against a typical 1460-octet MSS, so this is routine. .. automethod:: read .. automethod:: make + .. automethod:: unpack .. automethod:: __length_hint__ diff --git a/pcapkit/protocols/application/application.py b/pcapkit/protocols/application/application.py index 7005c9298c..dd46a07c04 100644 --- a/pcapkit/protocols/application/application.py +++ b/pcapkit/protocols/application/application.py @@ -44,8 +44,9 @@ class Application(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable on a protocol number. What they do permit is the ``-1`` sentinel, which asks for the rest of the packet undissected and resolves to :class:`~pcapkit.protocols.misc.raw.Raw` -- trailing bytes are not a further - protocol layer -- or to :class:`~pcapkit.protocols.misc.null.NoPayload` when - nothing remains. + protocol layer, unless the subclass dissects them, as + :class:`~pcapkit.protocols.application.tls.TLS` does for a further record -- + or to :class:`~pcapkit.protocols.misc.null.NoPayload` when nothing remains. """ diff --git a/pcapkit/protocols/application/tls.py b/pcapkit/protocols/application/tls.py index 5df02dedfd..1aac70756a 100644 --- a/pcapkit/protocols/application/tls.py +++ b/pcapkit/protocols/application/tls.py @@ -33,12 +33,31 @@ * Everything else -- the same types once encrypted, ``application_data`` and every other content type -- as opaque :obj:`bytes`. +Records in One Segment +---------------------- + +A whole record after the first in a TCP segment is dissected as a further +:class:`TLS` layer, the previous record's payload (:issue:`2517`): a +``ServerHello`` followed by a ``ChangeCipherSpec`` reads +``TCP:TLS 1.3:TLS 1.2``. The exceptions: + +* A whole record that fails to decode becomes + :class:`~pcapkit.protocols.misc.raw.Raw` together with everything after it, + with no SSLv2 trial. +* Octets that open no record -- an unsupported content type, a legacy + version outside ``0x03xx``, or a header cut short -- stay raw, with no error. +* A parse told to stop at TLS (``protocol='TLS'`` or ``layer='Application'``) + keeps the rest of the segment raw. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + Known Limitation ---------------- **A record split across two TCP segments is not reassembled**, and decodes as -:class:`~pcapkit.protocols.misc.raw.Raw`. A record may be up to 16 KiB -against a typical 1460-octet MSS, so this is routine. +:class:`~pcapkit.protocols.misc.raw.Raw`, as does the partial record a segment +ends inside. A record may be up to 16 KiB against a typical 1460-octet MSS, so +this is routine. .. versionadded:: 1.6.0 @@ -46,7 +65,9 @@ from typing import TYPE_CHECKING, cast from pcapkit.const.tls.content_type import ContentType as Enum_ContentType -from pcapkit.protocols.application.ssl import SSL, TLS13_VERSION +from pcapkit.corekit.decorators import _beholding, beholder +from pcapkit.protocols.application.ssl import (_TLS_VERSION_MAJOR, _WIRE_CONTENT_TYPES, SSL, + TLS13_VERSION) from pcapkit.protocols.application.sslv2 import SSLv2 from pcapkit.protocols.data.application.sslv2 import ClientHello as SSLv2_ClientHello from pcapkit.protocols.data.application.sslv2 import ServerHello as SSLv2_ServerHello @@ -54,6 +75,7 @@ from pcapkit.protocols.data.application.tls import TLS as Data_TLS from pcapkit.protocols.schema.application.tls import TLS as Schema_TLS from pcapkit.utilities.exceptions import ProtocolError +from pcapkit.utilities.warnings import hold_warnings if TYPE_CHECKING: from enum import IntEnum as StdlibEnum @@ -68,6 +90,10 @@ __all__ = ['TLS', 'SSLv3'] +#: Length of the record header: content type, legacy version and length +#: (:rfc:`9846#section-5.1`). +_RECORD_HEADER_LENGTH = 5 + #: Legacy record version SSLv3 carries, the one tier with its own :attr:`~TLS.name`. _SSLV3_VERSION = 0x0300 @@ -85,6 +111,28 @@ _TLS13_LEGACY_VERSIONS = frozenset({0x0301, 0x0303}) +def _opens_whole_record(data: 'bytes') -> 'bool': + """Whether ``data`` opens with a whole TLS record. + + Only the record header is read, against the same checks as + :meth:`SSL._validate_tls_record + `: a content + type a genuine record carries, a ``0x03xx`` legacy version, and a fragment + that ends inside ``data``. + + Args: + data: Octets after a record. + + Returns: + Whether a further record opens ``data`` and ends inside it. + + """ + if len(data) < _RECORD_HEADER_LENGTH: + return False + return (data[0] in _WIRE_CONTENT_TYPES and data[1] == _TLS_VERSION_MAJOR + and _RECORD_HEADER_LENGTH + int.from_bytes(data[3:5], 'big') <= len(data)) + + class TLS(SSL[Data_TLS, Schema_TLS], schema=Schema_TLS, data=Data_TLS): """This class implements all protocols in the TLS/SSLv3+ record-layer family. @@ -108,7 +156,7 @@ class TLS(SSL[Data_TLS, Schema_TLS], schema=Schema_TLS, data=Data_TLS): #: Tier :meth:`read` identified, e.g. ``'TLS 1.2'``; ``'TLS'`` until then. _alias = 'TLS' # type: Literal["TLS", "SSLv3", "TLS 1.0", "TLS 1.1", "TLS 1.2", "TLS 1.3"] - #: Header length of the SSLv2 record :meth:`read` adopted; ``None`` for TLS's own. + #: Header length of the SSLv2 record :meth:`unpack` adopted; ``None`` for TLS's own. _adopted_length = None # type: Optional[int] ########################################################################## @@ -129,13 +177,13 @@ def alias(self) -> 'Literal["TLS", "SSLv3", "TLS 1.0", "TLS 1.1", "TLS 1.2", "TL def length(self) -> 'int': """Header length of current protocol -- the fixed record header, :attr:`~pcapkit.protocols.data.application.tls.TLS.length`'s fragment - excluded; or, once :meth:`read` has adopted an SSLv2 record, that + excluded; or, once :meth:`unpack` has adopted an SSLv2 record, that record's own 2- or 3-octet header. """ if self._adopted_length is not None: return self._adopted_length - return 5 + return _RECORD_HEADER_LENGTH ########################################################################## # Methods. @@ -153,13 +201,8 @@ def id(cls) -> 'tuple[Literal["TLS"]]': # type: ignore[override] """ return ('TLS',) - def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Any': - """Read Transport Layer Security (TLS), or the SSLv2 record port 443 may carry. - - TLS is tried first. Only when :meth:`_read_tls_record` rejects the - record is the payload offered to - :class:`~pcapkit.protocols.application.sslv2.SSLv2`, whose own header - checks are strict, so a genuine TLS record is never taken as SSLv2. + def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_TLS': + """Read Transport Layer Security (TLS). Args: length: Length of packet data. @@ -169,19 +212,14 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Any': Parsed packet data. Raises: - ProtocolError: If the payload is neither a TLS record (see - :meth:`_read_tls_record`) nor an SSLv2 record. + ProtocolError: If the payload is not a TLS record (see + :meth:`_read_tls_record`), which :meth:`unpack` then offers + to :class:`~pcapkit.protocols.application.sslv2.SSLv2`. """ if length is None: length = len(self) - try: - return self._read_tls_record(length, **kwargs) - except ProtocolError: - parsed = self._read_sslv2_fallback(length, **kwargs) - if parsed is None: - raise - return parsed + return self._read_tls_record(length, **kwargs) def make(self, content_type: 'Enum_ContentType | StdlibEnum | AenumEnum | str | int' = Enum_ContentType.application_data, # pylint: disable=line-too-long @@ -233,11 +271,56 @@ def make(self, payload=payload, ) + def unpack(self, length: 'Optional[int]' = None, *, chained: 'bool' = False, # pylint: disable=arguments-differ + **kwargs: 'Any') -> 'Any': + """Unpack (parse) packet data: TLS, or the SSLv2 record port 443 may carry. + + TLS is tried first: the record header and :meth:`read`. Only when + that trial rejects the first record of a segment is the payload offered + to :class:`~pcapkit.protocols.application.sslv2.SSLv2`, whose own header + checks are strict, so a genuine TLS record is never taken as SSLv2. + + Args: + length: Length of packet data. + chained: Whether the record follows another in its segment; such + a record gets no SSLv2 trial, since SSLv2 only opens a + connection (:issue:`2517`). + **kwargs: Arbitrary keyword arguments. + + Returns: + Parsed packet data. + + Raises: + ProtocolError: If the payload is neither a TLS record (see + :meth:`_read_tls_record`) nor an SSLv2 record. + + Note: + The TLS trial's warnings are dropped when SSLv2 parses the record + (:issue:`2525`), and the SSLv2 trial's when it fails. When the record + is not parsed, the TLS trial's are dropped only under + :func:`~pcapkit.corekit.decorators.beholder` (:issue:`2308`). + + """ + with hold_warnings() as trial: + try: + return super().unpack(length, **kwargs) + except ProtocolError: + parsed = None + if not chained: + trial.discard = True + with hold_warnings() as fallback: + parsed = self._read_sslv2_fallback(len(self) if length is None else length, **kwargs) + fallback.discard = parsed is None + if parsed is None: + trial.discard = _beholding.get() + raise + return parsed + @classmethod def from_data(cls, data: 'Any', **kwargs: 'Any') -> 'Any': """Create protocol instance from data. - An SSLv2 record that :meth:`read` adopted from port 443 is rebuilt as + An SSLv2 record that :meth:`unpack` adopted from port 443 is rebuilt as the SSLv2 record it was parsed from, since TLS's record layout cannot carry its 2- or 3-octet header. @@ -261,7 +344,7 @@ def from_data(cls, data: 'Any', **kwargs: 'Any') -> 'Any': def __length_hint__(self) -> 'Literal[5]': """Return an estimated length for the object.""" - return 5 + return _RECORD_HEADER_LENGTH # type: ignore[return-value] ########################################################################## # Utilities. @@ -390,6 +473,57 @@ def _adopt_dispatched(self, obj: 'Any') -> 'Any': self._adopted_length = obj.length # pylint: disable=attribute-defined-outside-init return super()._adopt_dispatched(obj) + def _import_next_layer(self, proto: 'int', length: 'Optional[int]' = None, *, # type: ignore[override] + packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase': + """Import next layer extractor: a further whole record, or the rest undissected. + + A TCP segment may carry several records (:issue:`2517`). So the ``-1`` + remainder is dissected as this class when :func:`_opens_whole_record` + finds a whole record opening it, and is otherwise kept as + :class:`~pcapkit.protocols.misc.raw.Raw`, as + :meth:`Application._import_next_layer + ` + keeps it. A partial record, which the segment ends inside, is the + reassembly limit, and stays raw. + + Arguments: + proto: next layer protocol index; only the ``-1`` sentinel + (the rest of the segment) is accepted + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack `) + + Returns: + Instance of this class for a further record, or as + :meth:`Application._import_next_layer + ` + returns otherwise. + + """ + if proto == -1 and not self._sigterm and _opens_whole_record(self._get_payload()[:length]): + return self._import_next_record(proto, length, packet=packet) # type: ignore[misc,call-arg] + return super()._import_next_layer(proto, length, packet=packet) + + @beholder # type: ignore[arg-type] + def _import_next_record(self, proto: 'int', length: 'Optional[int]' = None, *, # type: ignore[misc] + packet: 'Optional[dict[str, Any]]' = None) -> 'ProtocolBase': + """Import the further record :meth:`_import_next_layer` found. + + Arguments: + proto: next layer protocol index, the ``-1`` sentinel + length: valid (*non-padding*) length + packet: packet info (passed from :meth:`self.unpack `) + + Returns: + Instance of this class, or of :class:`~pcapkit.protocols.misc.raw.Raw` + when :func:`~pcapkit.corekit.decorators.beholder` replaces a record + that fails to parse, with no SSLv2 trial (see :meth:`unpack`). + + """ + file_ = self._get_payload() + return self._parse_next_layer(type(self), file_, len(file_) if length is None else length, + alias=proto, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx, chained=True) + class SSLv3(TLS, schema=Schema_TLS, data=Data_TLS): diff --git a/tests/protocols/application/test_ssl_tls_dtls_unit.py b/tests/protocols/application/test_ssl_tls_dtls_unit.py index 5038d8feca..0ef883dcae 100644 --- a/tests/protocols/application/test_ssl_tls_dtls_unit.py +++ b/tests/protocols/application/test_ssl_tls_dtls_unit.py @@ -1062,6 +1062,95 @@ def test_sslv2_hellos_on_443_rebuild_byte_exact(self) -> None: self.assertEqual(str(rebuilt.protochain), 'TCP:SSLv2') +class SegmentRecordTests(unittest.TestCase): + """A whole record after the first in a TCP segment is a further TLS layer + (#2517); see ``test_tls_dtls_roundtrip_unit`` for the round trips.""" + + CCS = b'\x14\x03\x03\x00\x01\x01' + APPLICATION_DATA = b'\x17\x03\x03\x00\x05hello' + #: A whole handshake record whose ClientHello is too short for its fixed fields. + MALFORMED_HELLO = b'\x16\x03\x03\x00\x10\x01\x00\x00\x0c\x03\x03' + bytes(10) + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_frame_layer_limit_bounds_the_records(self) -> None: + """Each record is a layer, so the limit stops the walk with one warning.""" + from pcapkit.protocols.transport.tcp import TCP + from pcapkit.utilities import config + from pcapkit.utilities.warnings import ProtocolWarning + + raw = _TCP_TO_443 + self.CCS * (config.FRAME_LAYER_LIMIT + 6) + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + proto = TCP(raw, len(raw)) + self.assertEqual(str(proto.protochain), + ':'.join(['TCP'] + ['TLS 1.2'] * config.FRAME_LAYER_LIMIT + ['Raw'])) + self.assertEqual([item.category for item in caught], [ProtocolWarning]) + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + self.assertEqual(bytes(TCP.from_data(proto.info)), raw) + + def test_a_parse_stopping_at_tls_keeps_the_rest_raw(self) -> None: + from pcapkit.protocols.application.tls import TLS + + wire = self.CCS + self.APPLICATION_DATA + self.assertEqual(str(TLS(wire, len(wire)).protochain), 'TLS 1.2:TLS 1.2') + self.assertEqual(str(TLS(wire, len(wire), protocol='TLS').protochain), 'TLS 1.2:Raw') + + def test_octets_opening_no_record_stay_raw_without_an_error(self) -> None: + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.tcp import TCP + + for name, tail in (('content type 0', b'\x00\x03\x03\x00\x01\x01'), + ('an SSLv2 record version', b'\x14\x00\x02\x00\x01\x01'), + ('a header cut short', self.CCS[:4])): + with self.subTest(tail=name): + raw = _TCP_TO_443 + self.CCS + tail + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + proto = TCP(raw, len(raw)) + self.assertEqual(str(proto.protochain), 'TCP:TLS 1.2:Raw') + self.assertIs(type(proto.payload.payload), Raw) + self.assertIsNone(proto.payload.payload.info.error) + self.assertEqual(caught, []) + + def test_a_whole_record_that_fails_to_decode_is_raw(self) -> None: + """A ClientHello too short for its fixed fields: ``beholder`` keeps it as ``Raw``.""" + from pcapkit.protocols.transport.tcp import TCP + + raw = _TCP_TO_443 + self.CCS + self.MALFORMED_HELLO + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + proto = TCP(raw, len(raw)) + rebuilt = bytes(TCP.from_data(proto.info)) + self.assertEqual(str(proto.protochain), 'TCP:TLS 1.2:Raw') + self.assertRegex(proto.payload.payload.info.error, r'^TLS 1\.2: invalid format: client_hello') + self.assertEqual(caught, []) + self.assertEqual(rebuilt, raw) + + def test_only_the_first_record_gets_the_sslv2_trial(self) -> None: + """A later record that fails to decode is ``Raw`` with no SSLv2 trial; + the first keeps its trial (#2517).""" + from unittest import mock + + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.transport.tcp import TCP + + for name, segment, chain, calls in ( + ('after a record', self.CCS + self.MALFORMED_HELLO, 'TCP:TLS 1.2:Raw', 0), + ('first in the segment', self.MALFORMED_HELLO, 'TCP:Raw', 1)): + with self.subTest(record=name): + raw = _TCP_TO_443 + segment + with mock.patch.object(TLS, '_read_sslv2_fallback', autospec=True, + side_effect=TLS._read_sslv2_fallback) as trial: + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = TCP(raw, len(raw)) + self.assertEqual(str(proto.protochain), chain) + self.assertEqual(trial.call_count, calls) + + class ExportTests(unittest.TestCase): """The family is exported, and registered, the way every sibling application protocol is -- :class:`~pcapkit.protocols.application.ngap.NGAP` diff --git a/tests/protocols/application/test_tls_dtls_roundtrip_unit.py b/tests/protocols/application/test_tls_dtls_roundtrip_unit.py index 1c76d67907..1aea6eb140 100644 --- a/tests/protocols/application/test_tls_dtls_roundtrip_unit.py +++ b/tests/protocols/application/test_tls_dtls_roundtrip_unit.py @@ -16,7 +16,8 @@ :meth:`~pcapkit.protocols.application.tls.TLS.make`/ :meth:`~pcapkit.protocols.application.dtls.DTLS.make` and reading them back (the make-then-parse direction) -- so neither direction is a tautology of -the other. +the other. Segments of several TLS records are checked the same way, each +record a layer of its own (#2517). """ from __future__ import annotations @@ -91,6 +92,23 @@ def _server_hello_body(extensions: 'bytes | None' = b'') -> bytes: 'alert at record version 0x0302': _tls_record(21, b'\x02\x0a', version=b'\x03\x02'), } +#: A ``ServerHello`` selecting TLS 1.3 through ``supported_versions`` (type 43). +_TLS13_SERVER_HELLO = _tls_record(22, _handshake(2, _server_hello_body(extensions=b'\x00\x2b\x00\x02\x03\x04'))) + +#: Segments of several records each, every one dissected as a TLS layer (#2517). +TLS_SEGMENTS = { + 'change_cipher_spec + alert': _tls_record(20, b'\x01') + _tls_record(21, b'\x02\x0a'), + # The shape of frame 2 of Wireshark's tls13-rfc8446.pcap. + 'TLS 1.3 ServerHello + change_cipher_spec + application_data x2': ( + _TLS13_SERVER_HELLO + _tls_record(20, b'\x01') + + _tls_record(23, bytes(range(40))) + _tls_record(23, bytes(range(40, 100)))), + 'handshake coalescing two messages + application_data': ( + _tls_record(22, _handshake(0, b'') + _handshake(14, b'')) + _tls_record(23, b'hello')), +} + +#: A record the segment ends inside: 20 octets of a 55-octet record. +_PARTIAL_RECORD = _tls_record(23, bytes(50))[:20] + def _dtls_client_hello_body(cipher_suites: bytes = b'\x13\x01', extensions: 'bytes | None' = b'', cookie: bytes = b'') -> bytes: @@ -250,6 +268,117 @@ def test_alert_record_with_excess_octets_keeps_its_octets(self) -> None: self.assertEqual(bytes(TLS.from_data(parsed.info)), wire) +def _split_records(wire: bytes) -> 'list[bytes]': + """``wire`` cut at each record's declared length, as the specification frames it.""" + records = [] + while wire: + end = 5 + int.from_bytes(wire[3:5], 'big') + records.append(wire[:end]) + wire = wire[end:] + return records + + +def _layers(proto: 'object') -> 'tuple[list, object]': + """Each TLS layer of ``proto``'s chain, in order, and the layer that ends it.""" + from pcapkit.protocols.application.tls import TLS + + layers = [] + while isinstance(proto, TLS): + layers.append(proto) + proto = proto.payload + return layers, proto + + +class TestTLSSegmentRoundTrip(unittest.TestCase): + """Pin every record of a segment as its own layer, rebuilt both directions (#2517).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_every_record_is_a_tls_layer(self) -> None: + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.misc.null import NoPayload + + for name, wire in TLS_SEGMENTS.items(): + with self.subTest(segment=name): + records = _split_records(wire) + layers, rest = _layers(TLS(wire, len(wire))) + self.assertEqual(len(layers), len(records)) + self.assertIsInstance(rest, NoPayload) + for layer, record in zip(layers, records): + self.assertEqual(int(layer.info.content_type), record[0]) + self.assertEqual(layer.info.version, int.from_bytes(record[1:3], 'big')) + self.assertEqual(layer.info.length, len(record) - 5) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.transport.tcp import TCP + + tcp_to_443 = bytes.fromhex('c35001bb00000001000000015010ffff00000000') + for name, wire in TLS_SEGMENTS.items(): + with self.subTest(segment=name): + parsed = TLS(wire, len(wire)) + rebuilt = TLS.from_data(parsed.info) + self.assertEqual(bytes(rebuilt).hex(), wire.hex()) + self.assertEqual(len(_layers(rebuilt)[0]), len(_split_records(wire))) + + segment = tcp_to_443 + wire + tcp = TCP(segment, len(segment)) + self.assertEqual(str(tcp.protochain), f'TCP:{parsed.protochain}') + self.assertEqual(bytes(TCP.from_data(tcp.info)).hex(), segment.hex()) + + def test_make_then_parse_is_byte_exact(self) -> None: + """Each record made from its parsed content, carrying the next as ``payload``.""" + from pcapkit.protocols.application.tls import TLS + + for name, wire in TLS_SEGMENTS.items(): + with self.subTest(segment=name): + parsed = TLS(wire, len(wire)) + layers, _ = _layers(parsed) + maker = object.__new__(TLS) + schema = b'' # type: object + for layer in reversed(layers): + schema = maker.make(content_type=layer.info.content_type, version=layer.info.version, + content=layer.info.content, payload=schema) + made = bytes(schema) # type: ignore[call-overload] + self.assertEqual(made.hex(), wire.hex()) + self.assertEqual(str(TLS(made, len(made)).protochain), str(parsed.protochain)) + + def test_records_match_the_specification(self) -> None: + """Each record's own tier and content, per its header (RFC 9846 section 5.1).""" + from pcapkit.const.tls.content_type import ContentType + from pcapkit.protocols.application.tls import TLS + + wire = TLS_SEGMENTS['TLS 1.3 ServerHello + change_cipher_spec + application_data x2'] + layers, _ = _layers(TLS(wire, len(wire))) + self.assertEqual([layer.alias for layer in layers], ['TLS 1.3', 'TLS 1.2', 'TLS 1.2', 'TLS 1.2']) + self.assertEqual([layer.info.content_type for layer in layers], + [ContentType.handshake, ContentType.change_cipher_spec, + ContentType.application_data, ContentType.application_data]) + self.assertEqual(layers[1].info.content.msg, 1) + self.assertEqual([layer.info.content for layer in layers[2:]], [bytes(range(40)), bytes(range(40, 100))]) + + # The issue's example: the alert decodes, rather than staying 7 raw octets. + wire = TLS_SEGMENTS['change_cipher_spec + alert'] + layers, _ = _layers(TLS(wire, len(wire))) + self.assertEqual((int(layers[1].info.content.level), int(layers[1].info.content.description)), (2, 10)) + + def test_trailing_partial_record_stays_raw(self) -> None: + """The record a segment ends inside is the reassembly limit, kept as its own octets.""" + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.misc.raw import Raw + + for name, segment in TLS_SEGMENTS.items(): + with self.subTest(segment=name): + wire = segment + _PARTIAL_RECORD + parsed = TLS(wire, len(wire)) + layers, rest = _layers(parsed) + self.assertEqual(len(layers), len(_split_records(segment))) + self.assertIs(type(rest), Raw) + self.assertEqual(bytes(rest), _PARTIAL_RECORD) + self.assertEqual(bytes(TLS.from_data(parsed.info)).hex(), wire.hex()) + + class TestDTLSRoundTrip(unittest.TestCase): """Pin the rebuild of parsed DTLS records, both directions.""" diff --git a/tests/protocols/application/test_tls_sslv2_fallback_warnings_2525_unit.py b/tests/protocols/application/test_tls_sslv2_fallback_warnings_2525_unit.py new file mode 100644 index 0000000000..1159448a63 --- /dev/null +++ b/tests/protocols/application/test_tls_sslv2_fallback_warnings_2525_unit.py @@ -0,0 +1,141 @@ +# -*- coding: utf-8 -*- +"""GitHub issue #2525: an SSLv2 hello on TCP 443 reports nothing of TLS's rejected trial. + +:meth:`TLS.unpack ` tries TLS +first, and offers the record to SSLv2 only once TLS rejects it. An SSLv2 +CLIENT-HELLO advertising ``0x0300``, ``0x0301`` or ``0x0303`` puts that version's +octets where TLS reads its record length, so TLS's header read warns +``packet length < 0`` and TLS rejects the record. Unheld, that warning replays +with the kept SSLv2 layer, and under an ``error`` filter it turns the valid hello +into ``Raw``. A failed SSLv2 trial's warnings are dropped too, and a direct parse +that both trials reject reports TLS's alone, as #2308 rules for a trial outside +``beholder``. + +:mod:`pcapkit` is imported inside each test, after +:func:`~tests._support.reimport_once_per_class`. + +""" + +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class +from tests.utilities._harness import capture + +#: Test tier (#1958): exercises the warning machinery, which free-threaded +#: builds make context-aware. +__tier__ = 'ALL' + +#: A 20-octet TCP header, 50000 to 443, ACK. +TCP_TO_443 = struct.pack('!HHIIBBHHH', 50000, 443, 1, 1, 5 << 4, 0x10, 65535, 0, 0) +#: The versions the issue reports, all of them TLS-era values SSLv2 hellos carry. +VERSIONS = (0x0300, 0x0301, 0x0303) +#: An SSLv2 CLIENT-HELLO cut 4 octets short of its RECORD-LENGTH. +CUT_HELLO = (b'\x80\x1c' + b'\x01\x00\x02\x00\x03\x00\x00\x00\x10' + + b'\x01\x00\x80' + bytes(12)) +#: A TLS record header declaring a fragment far past the octets that follow it. +LONG_RECORD = b'\x17\x03\x03\xff\xffhello' + + +def _client_hello(version: 'int', *, two_byte_header: 'bool') -> 'bytes': + """An SSLv2 CLIENT-HELLO advertising ``version``, with three cipher specs.""" + specs = bytes.fromhex('010080' '000004' '00002f') + challenge = bytes(range(16)) + body = (b'\x01' + struct.pack('!HHHH', version, len(specs), 0, len(challenge)) + + specs + challenge) + if two_byte_header: + return struct.pack('!H', 0x8000 | len(body)) + body + return struct.pack('!HB', len(body), 0) + body + + +class TestSSLv2FallbackWarnings(unittest.TestCase): + """Pin that a trial the result does not keep reports none of its warnings.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _tcp(payload: 'bytes', action: 'str') -> 'tuple[object, list[warnings.WarningMessage]]': + """Parse ``payload`` on TCP 443 under ``action``, with the warnings it reports.""" + from pcapkit.protocols.transport.tcp import TCP + + data = TCP_TO_443 + payload + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter(action) + parsed = TCP(data, len(data)) + return parsed, caught + + def test_tls_header_read_warns_on_each_hello(self) -> None: + """Premise: TLS's record header alone warns on each hello the issue reports.""" + from pcapkit.protocols.schema.application.tls import TLS as Schema_TLS + + for version in VERSIONS: + for two_byte_header in (True, False): + with self.subTest(version=hex(version), two_byte_header=two_byte_header): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + Schema_TLS.unpack(_client_hello(version, two_byte_header=two_byte_header)) + self.assertTrue(any(str(item.message).startswith('packet length < 0') + for item in caught), [str(item.message) for item in caught]) + + def test_hello_is_sslv2_and_warns_nothing(self) -> None: + """Under ``always``, each hello is SSLv2 and reports nothing on either channel.""" + from pcapkit.utilities.logging import logger + + for version in VERSIONS: + for two_byte_header in (True, False): + with self.subTest(version=hex(version), two_byte_header=two_byte_header): + with capture(logger) as recorder: + parsed, caught = self._tcp(_client_hello(version, two_byte_header=two_byte_header), + 'always') + self.assertEqual(str(parsed.protochain), 'TCP:SSLv2') # type: ignore[attr-defined] + self.assertEqual(int(parsed.payload.info.body.client_version), version) # type: ignore[attr-defined] + self.assertEqual([str(item.message) for item in caught], []) + self.assertEqual([record.getMessage() for record in recorder.records + if record.getMessage().startswith('packet length')], []) + + def test_error_filter_keeps_the_hello(self) -> None: + """Under ``error``, TLS's held warning cannot turn the hello into ``Raw``.""" + for version in VERSIONS: + for two_byte_header in (True, False): + with self.subTest(version=hex(version), two_byte_header=two_byte_header): + parsed, caught = self._tcp(_client_hello(version, two_byte_header=two_byte_header), + 'error') + self.assertEqual(str(parsed.protochain), 'TCP:SSLv2') # type: ignore[attr-defined] + self.assertEqual(caught, []) + + def test_rejected_sslv2_trial_warns_nothing(self) -> None: + """A cut hello fails the SSLv2 trial too, which drops its warnings under either filter.""" + for action in ('always', 'error'): + with self.subTest(action=action): + parsed, caught = self._tcp(CUT_HELLO, action) + self.assertEqual(str(parsed.protochain), 'TCP:Raw') # type: ignore[attr-defined] + self.assertEqual(parsed.payload.info.error, # type: ignore[attr-defined] + 'TLS: invalid format: unrecognised content type 128') + self.assertEqual([str(item.message) for item in caught], []) + + def test_both_trials_rejected_under_beholder_warn_nothing(self) -> None: + """On TCP 443 ``beholder`` replaces the layer, so TLS's trial warnings go with it.""" + for action in ('always', 'error'): + with self.subTest(action=action): + parsed, caught = self._tcp(LONG_RECORD, action) + self.assertEqual(str(parsed.protochain), 'TCP:Raw') # type: ignore[attr-defined] + self.assertEqual(parsed.payload.info.error, # type: ignore[attr-defined] + 'TLS: invalid format: record length 65535 runs past the end of the buffer') + self.assertEqual([str(item.message) for item in caught], []) + + def test_both_trials_rejected_outside_beholder_report_tls_alone(self) -> None: + """A direct parse raises TLS's error and reports TLS's trial warning, not SSLv2's.""" + from pcapkit.protocols.application.tls import TLS + from pcapkit.utilities.exceptions import ProtocolError + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + with self.assertRaisesRegex(ProtocolError, 'record length 65535 runs past'): + TLS(LONG_RECORD, len(LONG_RECORD)) + self.assertEqual([str(item.message) for item in caught], ['packet length < 0: -65530']) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/test_construction_keyword_check_unit.py b/tests/protocols/test_construction_keyword_check_unit.py index 2c445738ac..c8d980076e 100644 --- a/tests/protocols/test_construction_keyword_check_unit.py +++ b/tests/protocols/test_construction_keyword_check_unit.py @@ -348,6 +348,27 @@ def keywords(method: 'Any') -> 'set[str]': self.assertNotIn('extension', keywords(HIP.make)) self.assertIn('extension', _declared_keywords(HIP)) + def test_tls_declares_chained_on_unpack_and_not_on_make(self) -> None: + """The same asymmetry on ``unpack``: :meth:`TLS.unpack + ` declares ``chained``, + which a further record in the segment is parsed with (#2517), and + :meth:`TLS.make ` does not. + + """ + from pcapkit.protocols.application.tls import TLS + from pcapkit.protocols.protocol import _declared_keywords + + def keywords(method: 'Any') -> 'set[str]': + return { + name for name, parameter in inspect.signature(method).parameters.items() + if parameter.kind in (parameter.POSITIONAL_OR_KEYWORD, + parameter.KEYWORD_ONLY) + } + + self.assertIn('chained', keywords(TLS.unpack)) + self.assertNotIn('chained', keywords(TLS.make)) + self.assertIn('chained', _declared_keywords(TLS)) + def test_a_keyword_only_read_declares_is_accepted_end_to_end(self) -> None: """The same shape, constructed rather than inspected.