From 30cc511af1ac347a7d88b3da639ebd30702d4e12 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 11:22:47 -0500 Subject: [PATCH 1/2] feat(application): dissect every whole SSLv2 record in a TCP segment (#2567) - SSLv2._import_next_layer dissects a whole record opening the rest of the segment as a further SSLv2 layer, the previous record's payload, as TLS does; a header cut short or a record the segment ends inside stays Raw. - A later record that fails to decode (RECORD-LENGTH 0) becomes Raw with everything after it, under beholder, carrying read()'s error. - protocol=, layer= and FRAME_LAYER_LIMIT stop the chain; TLS's port-443 SSLv2 fallback chains the same way. - protocol-family-grouping.rst, "Chained Records": SSLv2 chains; DTLS still does not (#2535). sslv2.rst and the module docstring gain a "Records in One Segment" section. - Tests: two and three records, a malformed later record, octets opening no record, the stops, and both round-trip directions. --- .../conventions/protocol-family-grouping.rst | 12 +- .../pcapkit/protocols/application/sslv2.rst | 20 ++ pcapkit/protocols/application/application.py | 3 +- pcapkit/protocols/application/sslv2.py | 95 ++++++++ .../protocols/application/test_sslv2_unit.py | 210 ++++++++++++++++++ 5 files changed, 333 insertions(+), 7 deletions(-) diff --git a/docs/source/contributing/conventions/protocol-family-grouping.rst b/docs/source/contributing/conventions/protocol-family-grouping.rst index 4a53d65aee..b7b91c78cf 100644 --- a/docs/source/contributing/conventions/protocol-family-grouping.rst +++ b/docs/source/contributing/conventions/protocol-family-grouping.rst @@ -104,12 +104,12 @@ 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``. +**TLS and SSLv2 records sharing a TCP segment are chained layers, not a list on one +layer** (:issue:`2517`, :issue:`2567`): each record dissected after the first is the +previous record's payload, so a TLS record's dynamic alias names its own version -- a +``ServerHello`` then a ``ChangeCipherSpec`` read ``TCP:TLS 1.3:TLS 1.2`` -- and an SSLv2 +``CLIENT-HELLO`` then a ``CLIENT-MASTER-KEY`` read ``TCP:SSLv2:SSLv2``. DTLS does not +chain: two DTLS records in one datagram read ``DTLS 1.2:Raw`` (:issue:`2535`). .. _protocol-family-canonical-names: diff --git a/docs/source/pcapkit/protocols/application/sslv2.rst b/docs/source/pcapkit/protocols/application/sslv2.rst index 161d7040a4..8b265d2427 100644 --- a/docs/source/pcapkit/protocols/application/sslv2.rst +++ b/docs/source/pcapkit/protocols/application/sslv2.rst @@ -42,6 +42,26 @@ handshake message bodies; every other body whose declared sub-field lengths do not fit the record (e.g. ciphertext), is kept as raw octets, header and message type decoded. +Records in One Segment +---------------------- + +A whole record after the first in a TCP segment is dissected as a further +:class:`~pcapkit.protocols.application.sslv2.SSLv2` layer, the previous record's +payload, whatever its message type (:issue:`2567`): a ``CLIENT-HELLO`` followed by +a ``CLIENT-MASTER-KEY`` reads ``SSLv2:SSLv2``. The exceptions: + +* A whole record that fails to decode -- a ``RECORD-LENGTH`` of ``0``, which + leaves no ``MSG-TYPE`` -- becomes :class:`~pcapkit.protocols.misc.raw.Raw` + together with everything after it. +* Octets that open no record -- a header cut short, or a record the segment + ends inside -- stay raw, with no error. +* A parse told to stop at SSLv2 (``protocol='SSLv2'`` or ``layer='Application'``) + keeps the rest of the segment raw, except behind the SSLv2 fallback of + :class:`~pcapkit.protocols.application.tls.TLS` on port 443, which is not + handed the stop. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + .. versionadded:: 1.6.0 .. raw:: html diff --git a/pcapkit/protocols/application/application.py b/pcapkit/protocols/application/application.py index dd46a07c04..0534cd9b92 100644 --- a/pcapkit/protocols/application/application.py +++ b/pcapkit/protocols/application/application.py @@ -45,7 +45,8 @@ class Application(ProtocolBase[_PT, _ST], Generic[_PT, _ST]): # pylint: disable for the rest of the packet undissected and resolves to :class:`~pcapkit.protocols.misc.raw.Raw` -- trailing bytes are not a further protocol layer, unless the subclass dissects them, as - :class:`~pcapkit.protocols.application.tls.TLS` does for a further record -- + :class:`~pcapkit.protocols.application.tls.TLS` and + :class:`~pcapkit.protocols.application.sslv2.SSLv2` do for a further record -- or to :class:`~pcapkit.protocols.misc.null.NoPayload` when nothing remains. """ diff --git a/pcapkit/protocols/application/sslv2.py b/pcapkit/protocols/application/sslv2.py index 6898535216..4e52ce6ac9 100644 --- a/pcapkit/protocols/application/sslv2.py +++ b/pcapkit/protocols/application/sslv2.py @@ -37,6 +37,26 @@ .. [*] https://datatracker.ietf.org/doc/html/draft-hickman-netscape-ssl-00 +Records in One Segment +---------------------- + +A whole record after the first in a TCP segment is dissected as a further +:class:`SSLv2` layer, the previous record's payload, whatever its message type +(:issue:`2567`): a ``CLIENT-HELLO`` followed by a ``CLIENT-MASTER-KEY`` reads +``SSLv2:SSLv2``. The exceptions: + +* A whole record that fails to decode -- a ``RECORD-LENGTH`` of ``0``, which + leaves no ``MSG-TYPE`` -- becomes :class:`~pcapkit.protocols.misc.raw.Raw` + together with everything after it. +* Octets that open no record -- a header cut short, or a record the segment + ends inside -- stay raw, with no error. +* A parse told to stop at SSLv2 (``protocol='SSLv2'`` or ``layer='Application'``) + keeps the rest of the segment raw, except behind the SSLv2 fallback of + :class:`~pcapkit.protocols.application.tls.TLS` on port 443, which is not + handed the stop. +* Each record is a layer, so :data:`~pcapkit.utilities.config.FRAME_LAYER_LIMIT` + bounds how many are dissected. + .. versionadded:: 1.6.0 """ @@ -44,6 +64,7 @@ from pcapkit.const.sslv2.handshake_type import HandshakeType as Enum_HandshakeType from pcapkit.const.sslv2.version import Version as Enum_Version +from pcapkit.corekit.decorators import beholder from pcapkit.protocols.application.ssl import SSL from pcapkit.protocols.data.application.sslv2 import ClientHello as Data_ClientHello from pcapkit.protocols.data.application.sslv2 import ServerHello as Data_ServerHello @@ -53,6 +74,7 @@ from pcapkit.protocols.schema.application.sslv2 import SSLv2 as Schema_SSLv2 from pcapkit.protocols.schema.application.sslv2 import \ UnknownHandshakeBody as Schema_UnknownHandshakeBody +from pcapkit.protocols.schema.application.sslv2 import sslv2_record_length from pcapkit.utilities.exceptions import ProtocolError if TYPE_CHECKING: @@ -68,6 +90,28 @@ __all__ = ['SSLv2'] +def _opens_whole_record(data: 'bytes') -> 'bool': + """Whether ``data`` opens with a whole SSLv2 record. + + Only the record header is read: it must be whole, and the record its + ``RECORD-LENGTH`` declares must end inside ``data``. A ``RECORD-LENGTH`` of + ``0`` passes, so that :meth:`SSLv2.read` refuses the record rather than it + staying raw unreported. + + Args: + data: Octets after a record. + + Returns: + Whether a further record opens ``data`` and ends inside it. + + """ + if len(data) < 2: + return False + header_len = 2 if data[0] & 0x80 else 3 + record_length = sslv2_record_length({'indicator': data[0], 'length_low': data[1]}) + return header_len + record_length <= len(data) + + class SSLv2(SSL[Data_SSLv2, Schema_SSLv2], schema=Schema_SSLv2, data=Data_SSLv2): """This class implements Secure Sockets Layer version 2. @@ -475,3 +519,54 @@ def _make_sslv2_server_hello(self, cipher_specs=cipher_specs, connection_id=connection_id, ) + + 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:`2567`). 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. + + """ + file_ = self._get_payload() + return self._parse_next_layer(type(self), file_, len(file_) if length is None else length, + alias=proto, packet=packet, layer=self._exlayer, + protocol=self._exproto, __context__=self._exctx) diff --git a/tests/protocols/application/test_sslv2_unit.py b/tests/protocols/application/test_sslv2_unit.py index 3780d62f35..0a375dd2d9 100644 --- a/tests/protocols/application/test_sslv2_unit.py +++ b/tests/protocols/application/test_sslv2_unit.py @@ -12,6 +12,8 @@ Round trips are checked as ``bytes(SSLv2.from_data(parsed.info)) == raw``, which routes through ``_make_data`` and ``make``; ``parsed.data`` is the constructor's input echoed back, so comparing it to ``raw`` checks nothing. +Segments of several records are checked both ways too, each record a layer of +its own (#2567). """ from __future__ import annotations @@ -32,6 +34,55 @@ RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) +#: A 20-octet TCP header, 50000 to 443, ACK. +TCP_TO_443 = bytes.fromhex('c35001bb00000001000000015010ffff00000000') + +#: CLIENT-HELLO, 2-octet header, RECORD-LENGTH 28: CLIENT-VERSION 0x0002, one +#: cipher spec, no session id, a 16-octet challenge. +CLIENT_HELLO = bytes.fromhex('801c' '01' '0002' '0003' '0000' '0010' '010080') + bytes(16) +#: CLIENT-MASTER-KEY, 2-octet header, RECORD-LENGTH 14: CIPHER-KIND, then +#: CLEAR-KEY-LENGTH 0, ENCRYPTED-KEY-LENGTH 4, KEY-ARG-LENGTH 0 and the key. +CLIENT_MASTER_KEY = bytes.fromhex('800e' '02' '010080' '0000' '0004' '0000' 'deadbeef') +#: SERVER-HELLO, 2-octet header, RECORD-LENGTH 34 (as in ``test_ssl_tls_dtls_unit``). +SERVER_HELLO = bytes.fromhex('80220400010002000400030010aaaaaaaa010080') + b'\xbb' * 16 +#: SERVER-VERIFY, 3-octet header (no escape), RECORD-LENGTH 17, PADDING 7. +SERVER_VERIFY = bytes.fromhex('001107' '05') + bytes(range(16)) +#: SERVER-FINISHED, 2-octet header, RECORD-LENGTH 17. +SERVER_FINISHED = bytes.fromhex('8011' '06') + bytes(range(16, 32)) +#: A 2-octet header declaring RECORD-LENGTH 0, which leaves no MSG-TYPE. +ZERO_LENGTH = b'\x80\x00' + +#: Segments of several records each, every one dissected as an SSLv2 layer (#2567). +SEGMENTS = { + 'CLIENT-HELLO + CLIENT-MASTER-KEY': CLIENT_HELLO + CLIENT_MASTER_KEY, + 'SERVER-HELLO + SERVER-VERIFY (3-octet header) + SERVER-FINISHED': ( + SERVER_HELLO + SERVER_VERIFY + SERVER_FINISHED), +} + + +def _split_records(wire: bytes) -> 'list[tuple[int, bytes]]': + """``wire`` cut at each record's RECORD-LENGTH, with its header length, per the draft's §4.1.""" + records = [] + while wire: + if wire[0] & 0x80: + header_len, record_length = 2, ((wire[0] & 0x7F) << 8) | wire[1] + else: + header_len, record_length = 3, ((wire[0] & 0x3F) << 8) | wire[1] + records.append((header_len, wire[:header_len + record_length])) + wire = wire[header_len + record_length:] + return records + + +def _layers(proto: 'Any') -> 'tuple[list[Any], Any]': + """Each SSLv2 layer of ``proto``'s chain, in order, and the layer that ends it.""" + from pcapkit.protocols.application.sslv2 import SSLv2 + + layers = [] + while isinstance(proto, SSLv2): + layers.append(proto) + proto = proto.payload + return layers, proto + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class TestSSLv2RoundTrip(unittest.TestCase): @@ -682,5 +733,164 @@ def test_make_itself_rejects_out_of_range_session_id_hit(self) -> None: with self.assertRaisesRegex(ProtocolError, r'SSLv2: SESSION-ID-HIT \(256\) out of range'): proto.make(msg_type=HandshakeType.SERVER_HELLO, body={'session_id_hit': 256}) + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestSSLv2SegmentRecords(unittest.TestCase): + """A whole record after the first in a TCP segment is a further SSLv2 layer (#2567).""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_every_record_is_an_sslv2_layer(self) -> None: + """Each record's header form and RECORD-LENGTH, per the draft's §4.1.""" + from pcapkit.protocols.application.sslv2 import SSLv2 + from pcapkit.protocols.misc.null import NoPayload + + for name, wire in SEGMENTS.items(): + with self.subTest(segment=name): + records = _split_records(wire) + parsed = SSLv2(wire, len(wire)) + layers, rest = _layers(parsed) + self.assertEqual(str(parsed.protochain), ':'.join(['SSLv2'] * len(records))) + self.assertIsInstance(rest, NoPayload) + for layer, (header_len, record) in zip(layers, records): + self.assertEqual(layer.length, header_len) + self.assertEqual(layer.info.two_byte_header, header_len == 2) + self.assertEqual(layer.info.record_length, len(record) - header_len) + self.assertEqual(int(layer.info.msg_type), record[header_len]) + + def test_records_match_the_draft(self) -> None: + """Each record's own message type and body, decoded as a direct parse would.""" + from pcapkit.const.sslv2.handshake_type import HandshakeType + from pcapkit.protocols.application.sslv2 import SSLv2 + + wire = SEGMENTS['SERVER-HELLO + SERVER-VERIFY (3-octet header) + SERVER-FINISHED'] + layers, _ = _layers(SSLv2(wire, len(wire))) + self.assertEqual([layer.info.msg_type for layer in layers], + [HandshakeType.SERVER_HELLO, HandshakeType.SERVER_VERIFY, + HandshakeType.SERVER_FINISHED]) + self.assertEqual(layers[0].info.body.connection_id, b'\xbb' * 16) + self.assertEqual((layers[1].info.padding, layers[1].info.is_escape), (7, False)) + self.assertEqual([layer.info.body for layer in layers[1:]], + [bytes(range(16)), bytes(range(16, 32))]) + + wire = SEGMENTS['CLIENT-HELLO + CLIENT-MASTER-KEY'] + layers, _ = _layers(SSLv2(wire, len(wire))) + self.assertEqual(layers[0].info.body.cipher_specs, b'\x01\x00\x80') + self.assertEqual(layers[1].info.msg_type, HandshakeType.CLIENT_MASTER_KEY) + self.assertEqual(layers[1].info.body, CLIENT_MASTER_KEY[3:]) + + def test_parse_then_rebuild_is_byte_exact(self) -> None: + """Parsed directly and behind TLS's SSLv2 fallback on TCP 443.""" + from pcapkit.protocols.application.sslv2 import SSLv2 + from pcapkit.protocols.transport.tcp import TCP + + for name, wire in SEGMENTS.items(): + with self.subTest(segment=name): + parsed = SSLv2(wire, len(wire)) + rebuilt = SSLv2.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 + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + tcp = TCP(segment, len(segment)) + rebuilt_tcp = TCP.from_data(tcp.info) + self.assertEqual(str(tcp.protochain), f'TCP:{parsed.protochain}') + self.assertEqual(bytes(rebuilt_tcp).hex(), segment.hex()) + self.assertEqual(str(rebuilt_tcp.protochain), str(tcp.protochain)) + + def test_make_then_parse_is_byte_exact(self) -> None: + """Each record made from its parsed fields, carrying the next as ``payload``.""" + from pcapkit.protocols.application.sslv2 import SSLv2 + + for name, wire in SEGMENTS.items(): + with self.subTest(segment=name): + parsed = SSLv2(wire, len(wire)) + layers, _ = _layers(parsed) + maker = object.__new__(SSLv2) + schema = b'' # type: Any + for layer in reversed(layers): + info = layer.info + schema = maker.make(msg_type=info.msg_type, two_byte_header=info.two_byte_header, + is_escape=info.is_escape, padding=info.padding, + body=info.body, payload=schema) + made = bytes(schema) + self.assertEqual(made.hex(), wire.hex()) + self.assertEqual(str(SSLv2(made, len(made)).protochain), str(parsed.protochain)) + + def test_a_record_that_fails_to_decode_is_raw_with_the_rest(self) -> None: + """A RECORD-LENGTH of 0 ends the chain as ``Raw`` carrying ``read()``'s error.""" + from pcapkit.protocols.application.sslv2 import SSLv2 + from pcapkit.protocols.misc.raw import Raw + + for name, segment, chain in ( + ('second record', CLIENT_HELLO + ZERO_LENGTH + CLIENT_MASTER_KEY, 'SSLv2:Raw'), + ('third record', CLIENT_HELLO + CLIENT_MASTER_KEY + ZERO_LENGTH + CLIENT_MASTER_KEY, + 'SSLv2:SSLv2:Raw')): + with self.subTest(record=name): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = SSLv2(segment, len(segment)) + rebuilt = bytes(SSLv2.from_data(parsed.info)) + _, rest = _layers(parsed) + self.assertEqual(str(parsed.protochain), chain) + self.assertIs(type(rest), Raw) + self.assertEqual(rest.info.error, + 'SSLv2: record length (0) is too short to include a MSG-TYPE octet') + self.assertEqual(bytes(rest), ZERO_LENGTH + CLIENT_MASTER_KEY) + self.assertEqual(caught, []) + self.assertEqual(rebuilt, segment) + + def test_octets_opening_no_record_stay_raw_without_an_error(self) -> None: + from pcapkit.protocols.application.sslv2 import SSLv2 + from pcapkit.protocols.misc.raw import Raw + + for name, tail in (('a 2-octet header cut short', b'\x80'), + ('a 3-octet header cut short', b'\x00\x05'), + ('a record the segment ends inside', CLIENT_MASTER_KEY[:-1])): + with self.subTest(tail=name): + wire = CLIENT_HELLO + CLIENT_MASTER_KEY + tail + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + parsed = SSLv2(wire, len(wire)) + _, rest = _layers(parsed) + self.assertEqual(str(parsed.protochain), 'SSLv2:SSLv2:Raw') + self.assertIs(type(rest), Raw) + self.assertIsNone(rest.info.error) + self.assertEqual(bytes(rest), tail) + self.assertEqual(caught, []) + self.assertEqual(bytes(SSLv2.from_data(parsed.info)), wire) + + def test_a_parse_stopping_at_sslv2_keeps_the_rest_raw(self) -> None: + from pcapkit.protocols.application.sslv2 import SSLv2 + + wire = SEGMENTS['CLIENT-HELLO + CLIENT-MASTER-KEY'] + self.assertEqual(str(SSLv2(wire, len(wire)).protochain), 'SSLv2:SSLv2') + for stop in ({'protocol': 'SSLv2'}, {'protocol': 'SSL'}, {'layer': 'Application'}): + with self.subTest(stop=stop): + parsed = SSLv2(wire, len(wire), **stop) + self.assertEqual(str(parsed.protochain), 'SSLv2:Raw') + self.assertEqual(bytes(parsed.payload), CLIENT_MASTER_KEY) + + 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 + CLIENT_HELLO + CLIENT_MASTER_KEY * (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'] + ['SSLv2'] * 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) + + if __name__ == '__main__': unittest.main() From 40fec14f66d870f9fe4a84f00822b8fc74260ee7 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sun, 1 Nov 2026 11:40:30 -0500 Subject: [PATCH 2/2] fix(application): hand the parse stops to TLS's SSLv2 trial (#2567) - TLS._read_sslv2_fallback built its trial from kwargs that Protocol.__init__ had already stripped of layer, protocol and __context__, so on TCP 443 a stop at SSLv2, TLS, SSL or the Application layer still chained the later records (TCP:SSLv2:SSLv2). The trial now gets the context, and the stops: this dispatcher's own layer when it matched one, otherwise its limits. - sslv2.rst and the module docstring drop the "except behind the SSLv2 fallback" clause. - Tests: each stop form through TCP 443, plus a later record that fails to decode and one cut short, all byte-exact on rebuild; and a caller's context registry reaching every record after the first, which fails without the __context__ handoff. --- .../pcapkit/protocols/application/sslv2.rst | 4 +- pcapkit/protocols/application/sslv2.py | 4 +- pcapkit/protocols/application/tls.py | 5 +- .../protocols/application/test_sslv2_unit.py | 96 +++++++++++++++++++ 4 files changed, 102 insertions(+), 7 deletions(-) diff --git a/docs/source/pcapkit/protocols/application/sslv2.rst b/docs/source/pcapkit/protocols/application/sslv2.rst index 8b265d2427..34e76eaa91 100644 --- a/docs/source/pcapkit/protocols/application/sslv2.rst +++ b/docs/source/pcapkit/protocols/application/sslv2.rst @@ -56,9 +56,7 @@ a ``CLIENT-MASTER-KEY`` reads ``SSLv2:SSLv2``. The exceptions: * Octets that open no record -- a header cut short, or a record the segment ends inside -- stay raw, with no error. * A parse told to stop at SSLv2 (``protocol='SSLv2'`` or ``layer='Application'``) - keeps the rest of the segment raw, except behind the SSLv2 fallback of - :class:`~pcapkit.protocols.application.tls.TLS` on port 443, which is not - handed the stop. + 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. diff --git a/pcapkit/protocols/application/sslv2.py b/pcapkit/protocols/application/sslv2.py index 4e52ce6ac9..54ab497031 100644 --- a/pcapkit/protocols/application/sslv2.py +++ b/pcapkit/protocols/application/sslv2.py @@ -51,9 +51,7 @@ * Octets that open no record -- a header cut short, or a record the segment ends inside -- stay raw, with no error. * A parse told to stop at SSLv2 (``protocol='SSLv2'`` or ``layer='Application'``) - keeps the rest of the segment raw, except behind the SSLv2 fallback of - :class:`~pcapkit.protocols.application.tls.TLS` on port 443, which is not - handed the stop. + 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. diff --git a/pcapkit/protocols/application/tls.py b/pcapkit/protocols/application/tls.py index 1aac70756a..eabc0aa2a2 100644 --- a/pcapkit/protocols/application/tls.py +++ b/pcapkit/protocols/application/tls.py @@ -446,8 +446,11 @@ def _read_sslv2_fallback(self, length: 'int', **kwargs: 'Any') -> 'Optional[Any] :data:`None` if the payload is not a complete SSLv2 hello. """ + # NOTE: ``__init__`` consumed the stops, so they are handed on; a stop this + # dispatcher matched (``'TLS'``, ``'SSL'``) is passed as its layer (#2567). + stops = {'layer': self.__layer__} if self._sigterm else {'layer': self._exlayer, 'protocol': self._exproto} try: - obj = SSLv2(self._data, length, **kwargs) + obj = SSLv2(self._data, length, __context__=self._exctx, **stops, **kwargs) except ProtocolError: return None info = obj.info diff --git a/tests/protocols/application/test_sslv2_unit.py b/tests/protocols/application/test_sslv2_unit.py index 0a375dd2d9..e870c6f5a4 100644 --- a/tests/protocols/application/test_sslv2_unit.py +++ b/tests/protocols/application/test_sslv2_unit.py @@ -84,6 +84,19 @@ def _layers(proto: 'Any') -> 'tuple[list[Any], Any]': return layers, proto +def _behind_tcp_443(wire: bytes, **stop: 'Any') -> 'tuple[Any, Any]': + """``wire`` parsed behind TLS's SSLv2 fallback on TCP 443, and the layer ending its SSLv2 chain.""" + from pcapkit.protocols.transport.tcp import TCP + + segment = TCP_TO_443 + wire + tcp = TCP(segment, len(segment), **stop) + # NOTE: The first record is a TLS layer that adopted the SSLv2 parse, so walk by alias. + proto = tcp.payload + while proto.alias == 'SSLv2': + proto = proto.payload + return tcp, proto + + @unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') class TestSSLv2RoundTrip(unittest.TestCase): """Parse-then-rebuild pins for every record shape this package builds.""" @@ -874,6 +887,89 @@ def test_a_parse_stopping_at_sslv2_keeps_the_rest_raw(self) -> None: self.assertEqual(str(parsed.protochain), 'SSLv2:Raw') self.assertEqual(bytes(parsed.payload), CLIENT_MASTER_KEY) + def test_a_tcp_443_parse_stopping_at_sslv2_keeps_the_rest_raw(self) -> None: + """Behind TLS's SSLv2 fallback too, whether TLS or the SSLv2 trial matches the stop.""" + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.tcp import TCP + + wire = SEGMENTS['CLIENT-HELLO + CLIENT-MASTER-KEY'] + self.assertEqual(str(_behind_tcp_443(wire)[0].protochain), 'TCP:SSLv2:SSLv2') + for stop in ({'layer': 'Application'}, {'protocol': 'SSLv2'}, {'protocol': 'TLS'}, {'protocol': 'SSL'}): + with self.subTest(stop=stop): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + tcp, rest = _behind_tcp_443(wire, **stop) + rebuilt = bytes(TCP.from_data(tcp.info)) + self.assertEqual(str(tcp.protochain), 'TCP:SSLv2:Raw') + self.assertIs(type(rest), Raw) + self.assertIsNone(rest.info.error) + self.assertEqual(bytes(rest), CLIENT_MASTER_KEY) + self.assertEqual(caught, []) + self.assertEqual(rebuilt, TCP_TO_443 + wire) + + def test_a_tcp_443_record_that_fails_to_decode_is_raw_with_the_rest(self) -> None: + """Behind TLS's SSLv2 fallback, as in a direct parse.""" + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.tcp import TCP + + for name, wire, chain in ( + ('second record', CLIENT_HELLO + ZERO_LENGTH + CLIENT_MASTER_KEY, 'TCP:SSLv2:Raw'), + ('third record', CLIENT_HELLO + CLIENT_MASTER_KEY + ZERO_LENGTH + CLIENT_MASTER_KEY, + 'TCP:SSLv2:SSLv2:Raw')): + with self.subTest(record=name): + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + tcp, rest = _behind_tcp_443(wire) + rebuilt = bytes(TCP.from_data(tcp.info)) + self.assertEqual(str(tcp.protochain), chain) + self.assertIs(type(rest), Raw) + self.assertEqual(rest.info.error, + 'SSLv2: record length (0) is too short to include a MSG-TYPE octet') + self.assertEqual(bytes(rest), ZERO_LENGTH + CLIENT_MASTER_KEY) + self.assertEqual(caught, []) + self.assertEqual(rebuilt, TCP_TO_443 + wire) + + def test_a_tcp_443_record_cut_short_stays_raw_without_an_error(self) -> None: + """Behind TLS's SSLv2 fallback, as in a direct parse.""" + from pcapkit.protocols.misc.raw import Raw + from pcapkit.protocols.transport.tcp import TCP + + for name, tail in (('a 2-octet header cut short', b'\x80'), + ('a 3-octet header cut short', b'\x00\x05'), + ('a record the segment ends inside', CLIENT_MASTER_KEY[:-1])): + with self.subTest(tail=name): + wire = CLIENT_HELLO + CLIENT_MASTER_KEY + tail + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter('always') + tcp, rest = _behind_tcp_443(wire) + rebuilt = bytes(TCP.from_data(tcp.info)) + self.assertEqual(str(tcp.protochain), 'TCP:SSLv2:SSLv2:Raw') + self.assertIs(type(rest), Raw) + self.assertIsNone(rest.info.error) + self.assertEqual(bytes(rest), tail) + self.assertEqual(caught, []) + self.assertEqual(rebuilt, TCP_TO_443 + wire) + + def test_a_tcp_443_parse_hands_the_caller_context_to_every_record(self) -> None: + """Each record after the first behind TLS's SSLv2 fallback carries the caller's registry.""" + from pcapkit.corekit.context import ContextRegistry, ProtocolContext + + class SSLv2Context(ProtocolContext): + @classmethod + def protocol(cls) -> 'tuple[str, ...]': + return ('SSLv2',) + + context = SSLv2Context() + registry = ContextRegistry(context) + for name, wire in SEGMENTS.items(): + with self.subTest(segment=name): + tcp, _ = _behind_tcp_443(wire, __context__=registry) + layers, _ = _layers(tcp.payload.payload) + self.assertEqual(len(layers), len(_split_records(wire)) - 1) + for layer in layers: + self.assertEqual(dict(layer.context), dict(registry)) + self.assertIs(layer._get_context(SSLv2Context), context) # pylint: disable=protected-access + 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