diff --git a/pcapkit/toolkit/dpkt.py b/pcapkit/toolkit/dpkt.py index a0e9a8d61e..830fba6810 100644 --- a/pcapkit/toolkit/dpkt.py +++ b/pcapkit/toolkit/dpkt.py @@ -742,6 +742,32 @@ def ipv6_reassembly(packet: 'Packet', timestamp: 'float', *, return None +def _tcp_flags(tcp: 'TCP') -> 'tuple[bool, bool, bool]': + """Read the RST, SYN and FIN flags of a DPKT TCP segment. + + Args: + tcp: DPKT TCP packet. + + Returns: + Whether RST, SYN and FIN are set, in that order. + + Note: + Each flag is masked out of :attr:`dpkt.tcp.TCP.flags` by its ``TH_*`` + value, so it is read from the flags octet alone, as the default engine + reads it. `DPKT`_ 1.9.8 makes that field nine bits wide, its top bit + the NS bit (:rfc:`3540`): read by its position among the field's binary + digits, every flag was one bit low with NS set, a SYN reading as a FIN + (:issue:`1618`). + + .. _DPKT: https://dpkt.readthedocs.io + + """ + from dpkt.tcp import TH_FIN, TH_RST, TH_SYN # isort:skip + + flags = tcp.flags + return bool(flags & TH_RST), bool(flags & TH_SYN), bool(flags & TH_FIN) + + def tcp_reassembly(packet: 'Packet', timestamp: 'float', *, count: 'int' = -1) -> 'TCP_Packet | None': """Make data for TCP reassembly. @@ -778,7 +804,7 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *, """ ip, tcp = _tcp_carrier(packet) if ip is not None and tcp is not None: - flags = bin(tcp.flags)[2:].zfill(8) + rst, syn, fin = _tcp_flags(tcp) wire, _ = _layer2bytes(packet, tcp) raw_len = len(tcp.data) # payload length, header excludes @@ -792,9 +818,9 @@ def tcp_reassembly(packet: 'Packet', timestamp: 'float', *, num=count, # original packet range number ack=tcp.ack, # acknowledgement dsn=tcp.seq, # data sequence number - rst=bool(int(flags[5])), # reset connection flag - syn=bool(int(flags[6])), # synchronise flag - fin=bool(int(flags[7])), # finish flag + rst=rst, # reset connection flag + syn=syn, # synchronise flag + fin=fin, # finish flag header=wire[:tcp.off * 4], # raw bytes type header payload=bytearray(bytes(tcp.data)), # raw bytearray type payload first=tcp.seq, # first sequence number of payload @@ -844,16 +870,16 @@ def tcp_traceflow(packet: 'Packet', timestamp: 'float', *, """ ip, tcp = _tcp_carrier(packet) if ip is not None and tcp is not None: - flags = bin(tcp.flags)[2:].zfill(8) + rst, syn, fin = _tcp_flags(tcp) wire, _ = _layer2bytes(packet, tcp) data = TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from global header index=count, # frame number frame=packet2frame(packet, timestamp, data_link=data_link), # extracted packet - syn=bool(int(flags[6])), # TCP synchronise (SYN) flag - fin=bool(int(flags[7])), # TCP finish (FIN) flag - rst=bool(int(flags[5])), # TCP reset (RST) flag + syn=syn, # TCP synchronise (SYN) flag + fin=fin, # TCP finish (FIN) flag + rst=rst, # TCP reset (RST) flag src=ipaddress.ip_address(ip.src), # source IP dst=ipaddress.ip_address(ip.dst), # destination IP srcport=tcp.sport, # TCP source port diff --git a/tests/toolkit/test_dpkt_tcp_ns_flags_1618_unit.py b/tests/toolkit/test_dpkt_tcp_ns_flags_1618_unit.py new file mode 100644 index 0000000000..2f61981240 --- /dev/null +++ b/tests/toolkit/test_dpkt_tcp_ns_flags_1618_unit.py @@ -0,0 +1,106 @@ +# -*- coding: utf-8 -*- +"""The dpkt engine reads RST, SYN and FIN as the default engine does, whatever the NS bit. C.f. #1618. + +`dpkt`_ 1.9.8 makes :attr:`dpkt.tcp.TCP.flags` nine bits wide, its top bit the +NS bit (:rfc:`3540`). :func:`~pcapkit.toolkit.dpkt.tcp_reassembly` and +:func:`~pcapkit.toolkit.dpkt.tcp_traceflow` read each flag by its position among +that field's binary digits, padded to eight, so with NS set every flag was read +one bit low: a SYN as a FIN, an RST as a SYN, a PSH as an RST. They now mask +each flag out of the flags octet, as the default engine reads it. + +Every capture is built here, so the module is in the unit tier. + +.. _dpkt: https://dpkt.readthedocs.io + +""" +from __future__ import annotations + +import importlib.util +import struct +import unittest + +from tests._support import reimport_once_per_class +from tests.foundation import _roundtrip as wire +from tests.toolkit.test_dpkt_scapy_ipv4_header_checks_1607_unit import read +from tests.toolkit.test_tunnelled_tcp_flow_key_1581_unit import has + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: The low nibble of the data offset octet: three reserved bits, then NS. +NS = 0x01 +NIBBLES = range(16) + + +def segment(number: int, nibble: int, octet: int) -> bytes: + """A TCP header of 20 octets on a port pair of its own, its offset octet's low nibble and flags octet as given.""" + return struct.pack('!HHIIBBHHH', 1000 + number, 80, 100, 0, 0x50 | nibble, octet, 8192, 0, 0) + + +def expected(octet: int) -> 'dict[str, bool]': + """RST, SYN and FIN as the flags octet ``octet`` sets them (:rfc:`9293#section-3.1`).""" + return {'rst': bool(octet & 0x04), 'syn': bool(octet & 0x02), 'fin': bool(octet & 0x01)} + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class DPKTTCPNSFlagsTests(unittest.TestCase): + """C.f. #1618.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + if not has('dpkt'): + self.skipTest('dpkt not installed') + + def test_adapters_read_each_flag_from_the_flags_octet(self) -> None: + # Every flags octet behind every NS and reserved bit, through both + # adapters; the header handed on keeps the captured NS bit. + import dpkt + + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import dpkt as toolkit + + wrong = [] + for nibble in NIBBLES: + for octet in range(256): + header = segment(0, nibble, octet) + frame = wire.ethernet(wire.ipv4(header, proto=6, df=True), 0x0800) + packet = dpkt.ethernet.Ethernet(frame) + toolkit.attach_buffer(packet, frame) + want = expected(octet) + for name, data in ( + ('tcp_reassembly', toolkit.tcp_reassembly(packet, 1.0)), + ('tcp_traceflow', toolkit.tcp_traceflow(packet, 1.0, data_link=LinkType.ETHERNET)), + ): + got = {flag: data[flag] for flag in want} + if got != want or data['header'] != header: + wrong.append((name, hex(nibble), hex(octet), got, want)) + self.assertEqual(wrong, []) + + def test_a_syn_with_ns_set_is_read_as_a_syn(self) -> None: + # The case #1618 reports: one SYN, NS set, read as a FIN. + seen = read(self, 'dpkt', [wire.ethernet(wire.ipv4(segment(1, NS, 0x02), proto=6, df=True), 0x0800)]) + for aspect in ('tcp', 'trace'): + with self.subTest(aspect=aspect): + self.assertEqual({flag: seen[aspect][1][flag] for flag in ('rst', 'syn', 'fin')}, + {'rst': False, 'syn': True, 'fin': False}) + + def test_extraction_agrees_with_the_default_engine(self) -> None: + # Every input to TCP reassembly and flow tracing, every field of it, and + # what comes out, for every flags octet with NS and the reserved bits + # clear and set. + cases = [(nibble, octet) for nibble in (0x0, NS, 0xE, 0xF) for octet in range(256)] + frames = [wire.ethernet(wire.ipv4(segment(number, nibble, octet), proto=6, df=True), 0x0800) + for number, (nibble, octet) in enumerate(cases)] + base = read(self, 'default', frames) + self.assertEqual(sorted(base['tcp']), list(range(1, len(cases) + 1))) + self.assertEqual([{flag: base['tcp'][number][flag] for flag in ('rst', 'syn', 'fin')} + for number in range(1, len(cases) + 1)], + [expected(octet) for _, octet in cases]) + seen = read(self, 'dpkt', frames) + for aspect in ('tcp', 'trace', 'ids', 'flows'): + with self.subTest(aspect=aspect): + self.assertEqual(seen[aspect], base[aspect]) + + +if __name__ == '__main__': + unittest.main()