diff --git a/pcapkit/protocols/application/http.py b/pcapkit/protocols/application/http.py index 5f4b546f1d..10f4c5685f 100644 --- a/pcapkit/protocols/application/http.py +++ b/pcapkit/protocols/application/http.py @@ -16,10 +16,11 @@ """ import contextlib import struct -from typing import TYPE_CHECKING, Generic +from typing import TYPE_CHECKING, Generic, cast from pcapkit.corekit.protochain import ProtoChain from pcapkit.protocols.application.application import Application +from pcapkit.protocols.data.data import Data from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.protocol import _PT, _ST from pcapkit.utilities.exceptions import ProtocolError @@ -27,7 +28,7 @@ if TYPE_CHECKING: from typing import Any, Optional - from typing_extensions import Literal + from typing_extensions import Literal, Self __all__ = ['HTTP'] @@ -296,6 +297,33 @@ def pack(self, **kwargs: 'Any') -> 'bytes': # Utilities. ########################################################################## + @classmethod + def from_data(cls, data: '_PT | dict[str, Any]', **kwargs: 'Any') -> 'Self': + """Create protocol instance from data. + + Args: + data: Protocol data. A :obj:`dict`, as :meth:`Info.to_dict + ` returns it, is rebuilt + into the versioned data model its keys belong to -- ``receipt`` + is HTTP/1.*'s and ``sid`` HTTP/2's -- since this class's own data + model carries no version. + **kwargs: As :meth:`ProtocolBase.from_data + `. + + Returns: + Protocol instance. + + """ + from pcapkit.protocols.data.application.httpv1 import HTTP as Data_HTTPv1 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + from pcapkit.protocols.data.application.httpv2 import HTTP as Data_HTTPv2 # isort: skip # pylint: disable=line-too-long,import-outside-toplevel + + if not isinstance(data, Data) and not issubclass(cls.__data__, (Data_HTTPv1, Data_HTTPv2)): + if 'receipt' in data: + data = cast('_PT', Data_HTTPv1.from_dict(data)) + elif 'sid' in data: + data = cast('_PT', Data_HTTPv2.from_dict(data)) + return super().from_data(data, **kwargs) + @classmethod def _make_data(cls, data: '_PT') -> 'dict[str, Any]': # type: ignore[override] """Create key-value pairs from ``data`` for protocol construction. diff --git a/pcapkit/protocols/internet/ipv6.py b/pcapkit/protocols/internet/ipv6.py index 7d6eeb073d..9241e41713 100644 --- a/pcapkit/protocols/internet/ipv6.py +++ b/pcapkit/protocols/internet/ipv6.py @@ -359,12 +359,18 @@ def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]': # type: ignore[over Key-value pairs for protocol construction. """ - upper = cls._make_payload(data) + exthdr = data.get('__exthdr__') # type: Optional[tuple[tuple[Type[ProtocolBase], Any], ...]] # pylint: disable=line-too-long + exclude = () # type: tuple[str, ...] + if exthdr is None: + # NOTE: ``data`` was rebuilt from :meth:`Info.to_dict`, which leaves + # ``__exthdr__`` out, so the chain is recovered from the keys the + # extension headers were written under; those are no payload. + exclude, exthdr = cls._lookup_exthdr(data) + upper = cls._make_payload(data, exclude=exclude) payload = upper # type: bytes | ProtocolBase # NOTE: The extension headers sit between this header and the upper # layer payload, so they are rebuilt in wire order ahead of it. - exthdr = data.get('__exthdr__') if exthdr: payload = b''.join(proto.from_data(info).data for proto, info in exthdr) + upper.data @@ -385,6 +391,49 @@ def _make_data(cls, data: 'Data_IPv6') -> 'dict[str, Any]': # type: ignore[over 'trailer': data.get('trailer', b''), } + @classmethod + def _lookup_exthdr(cls, data: 'Data_IPv6') -> 'tuple[tuple[str, ...], tuple[tuple[Type[ProtocolBase], Any], ...]]': # pylint: disable=line-too-long + """Recover the extension header chain of ``data`` without ``__exthdr__``. + + :meth:`_decode_next_layer` writes each extension header's info under a + key of its own, in wire order, ahead of the upper layer. Walking the + ``next`` field from :attr:`data.next ` names the parser + of each in turn, exactly as the walk that parsed them did. + + Args: + data: protocol data, without ``__exthdr__`` + + Returns: + The keys the extension headers are under, and the chain in the + shape of ``__exthdr__``, i.e. the parser class and info of each. + + Note: + A header repeated in the chain is written under the key of its + predecessor, so only the last of them is left in ``data``. + + """ + names = [] # type: list[str] + chain = [] # type: list[tuple[Type[ProtocolBase], Any]] + proto = data.get('next') + for key in cls._lookup_nested(data): + try: + ex_proto = Enum_ExtensionHeader(proto) + except ValueError: + break + + # NOTE: As in :meth:`_decode_next_layer`, a code with no parser + # carrying ``next`` ends the chain, and so does a fragment header. + info = data[key] + if 'next' not in info: + break + names.append(key) + chain.append((cls._lookup_next_layer(cls.__proto__, int(ex_proto)), info)) + + proto = info['next'] + if ex_proto == Enum_ExtensionHeader.IPv6_Frag: + break + return tuple(names), tuple(chain) + def _read_ip_hextet(self) -> 'tuple[int, int, int]': """Read the first four octets of IPv6. diff --git a/pcapkit/protocols/protocol.py b/pcapkit/protocols/protocol.py index 09741b290e..9ce4902d41 100644 --- a/pcapkit/protocols/protocol.py +++ b/pcapkit/protocols/protocol.py @@ -14,6 +14,7 @@ import abc import ast import collections +import collections.abc import contextlib import difflib import enum @@ -1907,7 +1908,7 @@ def _make_data(cls, data: 'Data') -> 'dict[str, Any]': return data.to_dict() @classmethod - def _make_payload(cls, data: 'Data') -> 'ProtocolBase': + def _make_payload(cls, data: 'Data', exclude: 'tuple[str, ...]' = ()) -> 'ProtocolBase': """Create payload from ``data`` for protocol construction. This method uses ``__next_type__`` and ``__next_name__`` to @@ -1916,13 +1917,28 @@ def _make_payload(cls, data: 'Data') -> 'ProtocolBase': instance will be returned. Otherwise, the payload will be constructed by :meth:`Protocol.from_data `. + :meth:`Info.to_dict ` leaves + both keys out, so ``data`` rebuilt from such a :obj:`dict` carries + neither. The payload is then found from the structure of ``data`` + instead, by :meth:`_lookup_payload`. + Args: data: protocol data + exclude: keys of ``data`` that are not the payload, though they may + look like it, passed on to :meth:`_lookup_payload` Returns: Payload for protocol construction. """ + if '__next_type__' not in data and '__next_name__' not in data: + proto, name = cls._lookup_payload(data, exclude) + if proto is None or name is None: + from pcapkit.protocols.misc.null import \ + NoPayload # pylint: disable=import-outside-toplevel + return NoPayload() + return proto.from_data(data[name]) + proto = cast('Optional[Type[Protocol]]', data.get('__next_type__')) if proto is None or not (isinstance(proto, type) and issubclass(proto, ProtocolBase)): from pcapkit.protocols.misc.null import \ @@ -1937,6 +1953,82 @@ def _make_payload(cls, data: 'Data') -> 'ProtocolBase': return proto.from_data(data[name]) + @classmethod + def _lookup_payload(cls, data: 'Data', exclude: 'tuple[str, ...]' = ()) -> 'tuple[Optional[Type[ProtocolBase]], Optional[str]]': # pylint: disable=line-too-long + """Find the payload of ``data`` that carries no ``__next_type__``. + + :meth:`_decode_next_layer` writes the payload's info under the payload's + :attr:`info_name`, after every field of the data model, so the payload is + the *last* key that the data model does not declare and whose value is a + mapping. Its protocol is the one of :attr:`__proto__` -- or its fallback + -- whose :attr:`info_name` the key is. That is the lowercased class name + by default, but a protocol may override it: every version of + :class:`~pcapkit.protocols.link.l2tp.L2TP` is ``l2tp``. + + Args: + data: protocol data + exclude: keys of ``data`` to pass over, such as the IPv6 extension + headers, which are written like a payload but are not one + + Returns: + The payload protocol and the key of its info in ``data``, or + :data:`None` for both if ``data`` has no payload this protocol can + dispatch to. + + """ + keys = [key for key in cls._lookup_nested(data) if key not in exclude] + if not keys: + return None, None + name = keys[-1] + + registry = cls.__proto__ + candidates = list(registry.values()) + if registry.default_factory is not None: + candidates.append(registry.default_factory()) + for entry in candidates: + proto = entry.klass if isinstance(entry, ModuleDescriptor) else entry + if not (isinstance(proto, type) and issubclass(proto, ProtocolBase)): + continue + + # NOTE: :attr:`info_name` is an instance property, so it is read off + # a bare instance; ``__new__`` only sets up the per-instance caches. + # An abstract class cannot be instantiated, and is never a payload. + try: + label = proto.__new__(proto).info_name + except TypeError: + continue + if label == name: + return proto, name + return None, None + + @staticmethod + def _lookup_nested(data: 'Data') -> 'list[str]': + """Keys of ``data`` that its data model does not declare and that hold a mapping. + + These are where the next layer protocols wrote their info, in the order + they wrote it -- the payload, and the IPv6 extension headers. + + Args: + data: protocol data + + Returns: + The keys, in the order of ``data``. + + """ + # NOTE: :func:`inspect.get_annotations` is new in Python 3.10; below it + # a class's own annotations are in its ``__dict__``, where from 3.14 on + # they are evaluated lazily and may not be. + get_annotations = getattr(inspect, 'get_annotations', None) + + declared = set() # type: set[str] + for klass in type(data).mro(): + if get_annotations is None: + declared.update(klass.__dict__.get('__annotations__', {})) + else: + declared.update(get_annotations(klass)) + return [key for key in data + if key not in declared and isinstance(data[key], collections.abc.Mapping)] + @staticmethod def _lookup_registry(registry: 'DefaultDict[Any, _VT]', code: 'Any') -> '_VT': """Look up a dispatch registry entry without recording a miss. @@ -2041,7 +2133,8 @@ def _decode_next_layer(self, dict_: '_PT', proto: 'int', length: 'Optional[int]' We added a new key ``__next_type__`` to ``dict_`` to store the next layer protocol type, and a new key ``__next_name__`` to store the next layer protocol name. These two keys will **NOT** - be included when :meth:`Info.to_dict ` is called. + be included when :meth:`Info.to_dict ` is called; + :meth:`_lookup_payload` finds the payload of such a :obj:`dict` instead. """ next_ = cast('ProtocolBase', self._import_next_layer(proto, length, packet=packet)) # type: ignore[misc,call-arg,redundant-cast] diff --git a/tests/protocols/test_capture_roundtrip_runtime.py b/tests/protocols/test_capture_roundtrip_runtime.py index 4fde75d81f..cb7f85b00c 100644 --- a/tests/protocols/test_capture_roundtrip_runtime.py +++ b/tests/protocols/test_capture_roundtrip_runtime.py @@ -63,17 +63,6 @@ class Gap(NamedTuple): #: IPv6 extension-header rebuild (#1446). LAYER_GAPS: 'dict[str, Gap]' = {} -_DICT_DROPS_PAYLOAD = ( - 'from_data(info.to_dict()) rebuilds the header only: to_dict() leaves out the ' - '__next_type__ and __next_name__ keys _make_payload dispatches on ' - '(pcapkit/protocols/protocol.py:1926), so the payload becomes NoPayload. ' - 'IPv6 also loses its extension headers, which _make_data reads from the ' - 'equally dropped __exthdr__ key (pcapkit/protocols/internet/ipv6.py:367). ' - 'Repro: Ethernet.from_data(eth.info.to_dict()).data is 14 of 1870 octets ' - 'for ipv4.pcap frame 1.' -) - - class DictGap(NamedTuple): """A protocol whose dict-form rebuild does not close.""" @@ -89,17 +78,8 @@ class DictGap(NamedTuple): #: Every protocol, by class name, whose rebuild from ``info.to_dict()`` fails. -DICT_GAPS = { - **{name: DictGap(1447, 'TRUNCATED', 'payload', _DICT_DROPS_PAYLOAD) - for name in ('Frame', 'Ethernet', 'ARP', 'IPv4', 'IPv6', 'TCP', 'UDP')}, - 'HTTP': DictGap( - 1447, 'ProtocolError', 'always', - "HTTP.from_data(info.to_dict()) raises ProtocolError('invalid HTTP data: " - "Raw'): the dict is rebuilt into a generic data model that " - 'HTTP._make_data does not accept. Repro: the first HTTP layer of ' - 'http.pcap.', - ), -} +#: Empty since the payload is found from the dict's structure (#1447). +DICT_GAPS: 'dict[str, DictGap]' = {} def captures() -> 'list[str]': diff --git a/tests/protocols/test_to_dict_payload_roundtrip_1447_unit.py b/tests/protocols/test_to_dict_payload_roundtrip_1447_unit.py new file mode 100644 index 0000000000..0ce7c0463f --- /dev/null +++ b/tests/protocols/test_to_dict_payload_roundtrip_1447_unit.py @@ -0,0 +1,189 @@ +# -*- coding: utf-8 -*- +"""``from_data(info.to_dict())`` rebuilds the payload, byte for byte. + +GitHub issue #1447: :meth:`Info.to_dict ` +leaves out ``__next_type__`` and ``__next_name__``, which +:meth:`~pcapkit.protocols.protocol.ProtocolBase._make_payload` dispatched on, so +a layer rebuilt from its dict lost its payload: an Ethernet frame came back as +its 14-octet header. IPv6 lost its extension headers the same way, with +``__exthdr__``, and HTTP raised ``ProtocolError('invalid HTTP data: Raw')``. + +The dict itself is left as it was, since the dumps are written from it; the +payload is found from the dict's structure instead. + +Every case builds its own octets and reads no capture. Classes are imported +inside each test, after :func:`~tests._support.reimport_once_per_class`. + +""" + +import io +import struct +import unittest +import warnings + +from tests._support import reimport_once_per_class + +MAC = bytes.fromhex('0123456789ab' 'fedcba987654') +SRC4, DST4 = bytes([192, 0, 2, 1]), bytes([192, 0, 2, 2]) +SRC6 = bytes.fromhex('20010db8' + '00' * 11 + '01') +DST6 = bytes.fromhex('20010db8' + '00' * 11 + '02') + + +def ipv4(proto: int, payload: bytes) -> bytes: + """An IPv4 header carrying ``payload`` as protocol ``proto``.""" + return struct.pack('!BBHHHBBH4s4s', 0x45, 0, 20 + len(payload), 0, 0x4000, 64, proto, 0, + SRC4, DST4) + payload + + +def tcp(dport: int, payload: bytes) -> bytes: + """A TCP header with no options, to ``dport``.""" + return struct.pack('!HHIIBBHHH', 49152, dport, 1, 1, 0x50, 0x18, 0xffff, 0, 0) + payload + + +def udp(payload: bytes, port: int = 40000) -> bytes: + """A UDP header from ``port`` to itself; the default is unregistered, so the payload is Raw.""" + return struct.pack('!HHHH', port, port, 8 + len(payload), 0) + payload + + +HTTP1 = b'GET / HTTP/1.1\r\nHost: example\r\n\r\n' +#: The HTTP/2 connection preface and an empty SETTINGS frame. +HTTP2 = b'PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n' + bytes.fromhex('000000' '04' '00' '00000000') +#: An L2TPv2 control header (T and L set): length 12, tunnel 1, session 2. +#: Its info is under ``l2tp``, the ``info_name`` every L2TP version shares, +#: not under its lowercased class name. +L2TPV2 = bytes.fromhex('c802' '000c' '0001' '0002' '0000' '0000') +ARP = bytes.fromhex('0001' '0800' '06' '04' '0001') + MAC[6:] + SRC4 + bytes(6) + DST4 +#: Hop-by-Hop Options, then Destination Options, then UDP; each a lone PadN. +EXTHDR = bytes.fromhex('3c' '00' '0104' '00000000') + bytes.fromhex('11' '00' '0104' '00000000') +IPV6_PAYLOAD = EXTHDR + udp(b'ping') +IPV6 = struct.pack('!IHBB16s16s', 6 << 28, len(IPV6_PAYLOAD), 0, 64, SRC6, DST6) + IPV6_PAYLOAD + +#: Link-layer frames, each carried in a PCAP record. +FRAMES = { + 'ipv4-tcp-http1': MAC + b'\x08\x00' + ipv4(6, tcp(80, HTTP1)), + 'ipv4-tcp-http2': MAC + b'\x08\x00' + ipv4(6, tcp(80, HTTP2)), + 'ipv4-udp': MAC + b'\x08\x00' + ipv4(17, udp(b'pong')), + 'ipv4-udp-l2tpv2': MAC + b'\x08\x00' + ipv4(17, udp(L2TPV2, port=1701)), + 'arp-padded': MAC + b'\x08\x06' + ARP + bytes(18), + 'ipv6-exthdr-udp': MAC + b'\x86\xdd' + IPV6, +} + +#: Little-endian global header: v2.4, Ethernet. +GLOBAL = bytes.fromhex('d4c3b2a1' '0200' '0400' '00000000' '00000000' 'ffff0000' '01000000') + +#: Every layer type #1447 names, each of which these frames must exercise, and +#: L2TPv2, whose ``info_name`` is not its lowercased class name. +LAYERS = {'Frame', 'Ethernet', 'ARP', 'IPv4', 'IPv6', 'TCP', 'UDP', 'HTTP', 'L2TPv2'} + + +class TestToDictPayloadRoundTrip(unittest.TestCase): + """Pin the rebuild of a layer from its dict.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + quiet = warnings.catch_warnings() + quiet.__enter__() # pylint: disable=unnecessary-dunder-call + self.addCleanup(quiet.__exit__, None, None, None) + warnings.simplefilter('ignore') + + def frame(self, name: str): # type: ignore[no-untyped-def] + """Parse ``FRAMES[name]`` as the first record of a PCAP file.""" + from pcapkit.protocols.misc.pcap.frame import Frame + from pcapkit.protocols.misc.pcap.header import Header + + packet = FRAMES[name] + record = struct.pack(' None: + seen = set() + for name in FRAMES: + for layer in self.layers(name): + seen.add(type(layer).__name__) + with self.subTest(frame=name, layer=type(layer).__name__): + self.assertEqual(self.rebuild(layer).data, layer.data) + self.assertLessEqual(LAYERS, seen) + + def test_rebuilt_payload_has_the_parsed_type(self) -> None: + for name in FRAMES: + for layer in self.layers(name): + with self.subTest(frame=name, layer=type(layer).__name__): + self.assertIs(type(self.rebuild(layer).payload), type(layer.payload)) + + def test_ipv6_extension_headers_rebuild_from_its_dict(self) -> None: + from pcapkit.protocols.internet.ipv6 import IPv6 + + ipv6 = self.frame('ipv6-exthdr-udp')[IPv6] + dict_ = ipv6.info.to_dict() + self.assertNotIn('__exthdr__', dict_) + + rebuilt = IPv6.from_data(dict_) + self.assertEqual(rebuilt.data, ipv6.data) + self.assertEqual( + [type(ext).__name__ for ext in rebuilt._exthdr.values()], # pylint: disable=protected-access + ['HOPOPT', 'IPv6_Opts'], + ) + + def test_payload_is_found_by_info_name_not_class_name(self) -> None: + from pcapkit.protocols.link.l2tpv2 import L2TPv2 + from pcapkit.protocols.transport.udp import UDP + + udp_ = self.frame('ipv4-udp-l2tpv2')[UDP] + self.assertIsInstance(udp_.payload, L2TPv2) + self.assertIn('l2tp', udp_.info.to_dict()) + + rebuilt = UDP.from_data(udp_.info.to_dict()) + self.assertIsInstance(rebuilt.payload, L2TPv2) + self.assertEqual(rebuilt.data, udp_.data) + self.assertEqual(len(rebuilt.data), 8 + len(L2TPV2)) + + def test_http_dict_keeps_its_version(self) -> None: + from pcapkit.protocols.application.http import HTTP + + for name, alias in (('ipv4-tcp-http1', 'HTTP/1.1'), ('ipv4-tcp-http2', 'HTTP/2')): + with self.subTest(frame=name): + http = self.frame(name)[HTTP] + self.assertEqual(http.alias, alias) + self.assertEqual(HTTP.from_data(http.info.to_dict()).alias, alias) + + def test_dict_without_a_payload_key_rebuilds_the_header_alone(self) -> None: + from pcapkit.protocols.link.ethernet import Ethernet + from pcapkit.protocols.misc.null import NoPayload + + eth = Ethernet(FRAMES['ipv4-udp']) + dict_ = eth.info.to_dict() + del dict_['ipv4'] + + rebuilt = Ethernet.from_data(dict_) + self.assertIsInstance(rebuilt.payload, NoPayload) + self.assertEqual(rebuilt.data, FRAMES['ipv4-udp'][:14]) + + def test_to_dict_still_leaves_the_dispatch_keys_out(self) -> None: + # The dumps are written from ``to_dict``, so the fix must not add to it. + for name in FRAMES: + for layer in self.layers(name): + with self.subTest(frame=name, layer=type(layer).__name__): + dict_ = layer.info.to_dict() + for key in ('__next_type__', '__next_name__', '__exthdr__'): + self.assertNotIn(key, dict_) + + +if __name__ == '__main__': + unittest.main()