From c308af6b7172028bf81e736884d35cd76677edf0 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 14:12:09 -0400 Subject: [PATCH] fix(toolkit): scapy packet2dict dumps nested packets as {} and mutates its input packet2dict now converts field values that are scapy packets, including those inside lists and tuples (IP/IPv6/MH options, DNS records, SCTP params, TCP-AO values), and builds each layer's dict from a copy of packet.fields, so the given packet is left unchanged and no default values are added to dissected layers. Closes #1259 Closes #1260 --- pcapkit/toolkit/scapy.py | 20 ++- tests/toolkit/test_scapy_packet2dict_unit.py | 148 +++++++++++++++++++ 2 files changed, 165 insertions(+), 3 deletions(-) create mode 100644 tests/toolkit/test_scapy_packet2dict_unit.py diff --git a/pcapkit/toolkit/scapy.py b/pcapkit/toolkit/scapy.py index d32c720aa0..3b45d816db 100644 --- a/pcapkit/toolkit/scapy.py +++ b/pcapkit/toolkit/scapy.py @@ -100,7 +100,11 @@ def packet2dict(packet: 'Packet') -> 'dict[str, Any]': packet: Scapy packet. Returns: - A :obj:`dict` mapping of packet data. + A :obj:`dict` mapping of packet data. Each layer carries only the fields + set on it (for a dissected packet, those read from the wire, with no + defaults filled in). Field values that are themselves + Scapy packets (e.g. IP options, DNS records), including those inside + lists, are converted recursively. The ``packet`` itself is not modified. Raises: ModuleNotFound: If `Scapy`_ is not installed. @@ -108,10 +112,20 @@ def packet2dict(packet: 'Packet') -> 'dict[str, Any]': """ if scapy is None: raise ModuleNotFound("No module named 'scapy'", name='scapy') - from scapy.packet import NoPayload + from scapy.packet import NoPayload, Packet + + def convert(value: 'Any') -> 'Any': + if isinstance(value, Packet): + return wrapper(value) + if isinstance(value, list): + return [convert(item) for item in value] + if isinstance(value, tuple): + return tuple(convert(item) for item in value) + return value def wrapper(packet: 'Packet') -> 'dict[str, Any]': - dict_ = packet.fields + # copy, so the payload names are not written into ``packet.fields`` + dict_ = {key: convert(val) for key, val in packet.fields.items()} payload = packet.payload if not isinstance(payload, NoPayload): dict_[payload.name] = wrapper(payload) diff --git a/tests/toolkit/test_scapy_packet2dict_unit.py b/tests/toolkit/test_scapy_packet2dict_unit.py new file mode 100644 index 0000000000..e116f636f5 --- /dev/null +++ b/tests/toolkit/test_scapy_packet2dict_unit.py @@ -0,0 +1,148 @@ +"""Regression tests for :func:`pcapkit.toolkit.scapy.packet2dict` (#1259, #1260).""" + +from __future__ import annotations + +import importlib.util +import unittest + +from tests._support import reimport_once_per_class + +HAS_SCAPY = importlib.util.find_spec('scapy') is not None +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + + +@unittest.skipUnless(HAS_RUNTIME and HAS_SCAPY, 'runtime dependencies not installed') +class ScapyPacket2DictTests(unittest.TestCase): + def setUp(self) -> None: + reimport_once_per_class(self) + + def _walk(self, value): + """Yield every value nested in ``value``.""" + yield value + if isinstance(value, dict): + for item in value.values(): + yield from self._walk(item) + elif isinstance(value, (list, tuple)): + for item in value: + yield from self._walk(item) + + def assertNoPacketObjects(self, converted) -> None: + from scapy.packet import Packet + + leftover = [type(v).__name__ for v in self._walk(converted) if isinstance(v, Packet)] + self.assertEqual(leftover, []) + + def test_ip_options_list_is_converted(self) -> None: + from scapy.layers.inet import IP, IPOption_NOP, IPOption_RR + from scapy.layers.l2 import Ether + + from pcapkit.toolkit.scapy import packet2dict + + built = Ether() / IP(options=[IPOption_NOP(), IPOption_RR(routers=['192.0.2.1'])]) + packet = Ether(bytes(built)) + converted = packet2dict(packet) + + self.assertNoPacketObjects(converted) + options = converted['Ethernet']['IP']['options'] + self.assertEqual(options[0]['option'], 1) + self.assertEqual(options[1]['routers'], ['192.0.2.1']) + + def test_hand_built_nested_packet_carries_only_its_set_fields(self) -> None: + from scapy.layers.inet import IP, IPOption_NOP, IPOption_RR + from scapy.layers.l2 import Ether + + from pcapkit.toolkit.scapy import packet2dict + + converted = packet2dict(Ether() / IP(options=[IPOption_NOP(), IPOption_RR(routers=['192.0.2.1'])])) + self.assertNoPacketObjects(converted) + self.assertEqual(converted['Ethernet']['IP']['options'], + [{}, {'routers': ['192.0.2.1']}]) + + def _assert_layer_keys_match_fields(self, packet) -> None: + from pcapkit.toolkit.scapy import packet2dict + + converted = packet2dict(packet) + node = converted[packet.name] + for layer in packet.iterpayloads(): + with self.subTest(layer=layer.name): + payload = layer.payload + extra = {payload.name} if payload else set() + self.assertEqual(set(node), set(layer.fields) | extra) + if payload: + node = node[payload.name] + + def test_dissected_icmp_echo_has_only_wire_fields(self) -> None: + from scapy.layers.inet import ICMP, IP + from scapy.layers.l2 import Ether + from scapy.packet import Raw + + built = Ether() / IP() / ICMP(type=8, id=1, seq=2) / Raw(b'ping') + self._assert_layer_keys_match_fields(Ether(bytes(built))) + + def test_fixture_icmp_echo_has_only_wire_fields(self) -> None: + from scapy.layers.inet import ICMP + from scapy.utils import rdpcap + + from tests._support import sample_path + + try: + path = sample_path('little_endian.pcap') + except FileNotFoundError as exc: + self.skipTest(str(exc)) + echoes = [p for p in rdpcap(path) if ICMP in p and p[ICMP].type in (0, 8)] + self.assertTrue(echoes) + self._assert_layer_keys_match_fields(echoes[0]) + + def test_dns_records_are_converted(self) -> None: + from scapy.layers.dns import DNS, DNSQR, DNSRR + from scapy.layers.inet import IP, UDP + from scapy.layers.l2 import Ether + + from pcapkit.toolkit.scapy import packet2dict + + built = (Ether() / IP() / UDP(sport=53, dport=53) / + DNS(qd=DNSQR(qname='example.com'), an=DNSRR(rrname='example.com', rdata='192.0.2.1'))) + converted = packet2dict(Ether(bytes(built))) + + self.assertNoPacketObjects(converted) + dns = converted['Ethernet']['IP']['UDP']['DNS'] + self.assertEqual(dns['qd'][0]['qname'], b'example.com.') + self.assertEqual(dns['an'][0]['rdata'], '192.0.2.1') + + def test_packet_inside_a_tuple_is_converted(self) -> None: + from scapy.layers.inet import IP, TCP, TCPAOValue + from scapy.layers.l2 import Ether + + from pcapkit.toolkit.scapy import packet2dict + + built = Ether() / IP() / TCP(options=[('AO', TCPAOValue(keyid=1, rnextkeyid=2, mac=b'\x01\x02\x03\x04'))]) + converted = packet2dict(Ether(bytes(built))) + + self.assertNoPacketObjects(converted) + name, value = converted['Ethernet']['IP']['TCP']['options'][0] + self.assertEqual(name, 'AO') + self.assertEqual(value['keyid'], 1) + self.assertEqual(value['mac'], b'\x01\x02\x03\x04') + + def test_input_packet_is_not_mutated(self) -> None: + from scapy.layers.inet import IP + from scapy.layers.l2 import Ether + from scapy.packet import Raw + + from pcapkit.toolkit.scapy import packet2dict + + for packet in (Ether() / IP() / Raw(b'x'), + Ether(bytes(Ether(dst='bb:bb:bb:bb:bb:bb') / IP() / Raw(b'x')))): + with self.subTest(packet=packet.summary()): + before = {layer.name: dict(layer.fields) for layer in packet.iterpayloads()} + converted = packet2dict(packet) + after = {layer.name: dict(layer.fields) for layer in packet.iterpayloads()} + + self.assertEqual(after, before) + self.assertIn('Raw', converted['Ethernet']['IP']) + self.assertEqual(bytes(packet.copy()), bytes(packet)) + + +if __name__ == '__main__': + unittest.main()