From 8d9644175b9f980613080020750085a93f5fa2d5 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Sat, 10 Oct 2026 01:21:49 -0400 Subject: [PATCH] refactor(toolkit): make tcp_segment and its result type public in toolkit.pcap (#1570) toolkit/pcapng.py imported the private _tcp_segment from toolkit/pcap.py. Under the #1519 rulings, as #1525 did for test_start_line, the shared helper becomes public in the module both callers already import. - pcap.py: _tcp_segment is now tcp_segment, and its result type _TCPSegment is now TCPSegment; both are in __all__, bodies unchanged. - pcapng.py: imports and calls tcp_segment. - docs: builtin.rst documents both under PCAP Tools in an Auxiliary Functions subsection, as 3rdparty.rst lays out its pages; block2frame moves into one under PCAP-NG Tools. - tests: test_pcap_tcp_segment_public_1570_unit.py pins __all__, the shared object, the NamedTuple fields, the helper on PCAP and PCAP-NG frames, and that all four TCP adapters call it. Neither name is re-exported from pcapkit.toolkit. --- docs/source/pcapkit/toolkit/builtin.rst | 16 +- pcapkit/toolkit/pcap.py | 28 ++- pcapkit/toolkit/pcapng.py | 6 +- .../test_pcap_tcp_segment_public_1570_unit.py | 164 ++++++++++++++++++ 4 files changed, 201 insertions(+), 13 deletions(-) create mode 100644 tests/toolkit/test_pcap_tcp_segment_public_1570_unit.py diff --git a/docs/source/pcapkit/toolkit/builtin.rst b/docs/source/pcapkit/toolkit/builtin.rst index 9976ba19bb..57e3520eb5 100644 --- a/docs/source/pcapkit/toolkit/builtin.rst +++ b/docs/source/pcapkit/toolkit/builtin.rst @@ -10,7 +10,9 @@ PCAP Tools :mod:`pcapkit.toolkit.pcap` contains the adapters for the PCAP format. The reassembly and flow tracing adapters return the data their :mod:`pcapkit.foundation` counterpart consumes, or :data:`None` if the frame -cannot be used for it. +cannot be used for it. The TCP adapters here and in +:mod:`pcapkit.toolkit.pcapng` read the segment with +:func:`~pcapkit.toolkit.pcap.tcp_segment`. .. autofunction:: pcapkit.toolkit.pcap.ipv4_reassembly @@ -20,6 +22,15 @@ cannot be used for it. .. autofunction:: pcapkit.toolkit.pcap.tcp_traceflow +Auxiliary Functions +------------------- + +.. autofunction:: pcapkit.toolkit.pcap.tcp_segment + +.. autoclass:: pcapkit.toolkit.pcap.TCPSegment + :members: + :show-inheritance: + PCAP-NG Tools ============= @@ -38,4 +49,7 @@ cannot be used for it. .. autofunction:: pcapkit.toolkit.pcapng.tcp_traceflow +Auxiliary Functions +------------------- + .. autofunction:: pcapkit.toolkit.pcapng.block2frame diff --git a/pcapkit/toolkit/pcap.py b/pcapkit/toolkit/pcap.py index e1c999593d..427276cbd8 100644 --- a/pcapkit/toolkit/pcap.py +++ b/pcapkit/toolkit/pcap.py @@ -7,7 +7,9 @@ :mod:`pcapkit.toolkit.pcap` contains the adapters for the PCAP format. The reassembly and flow tracing adapters return the data their :mod:`pcapkit.foundation` counterpart consumes, or :data:`None` if the frame -cannot be used for it. +cannot be used for it. The TCP adapters here and in +:mod:`pcapkit.toolkit.pcapng` read the segment with +:func:`~pcapkit.toolkit.pcap.tcp_segment`. """ from typing import TYPE_CHECKING, NamedTuple, cast @@ -33,11 +35,16 @@ from pcapkit.protocols.protocol import ProtocolBase from pcapkit.protocols.transport.tcp import TCP -__all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow'] +__all__ = ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', 'tcp_traceflow', + 'tcp_segment', 'TCPSegment'] -class _TCPSegment(NamedTuple): - """The fields of a TCP segment that the reassembly and flow tracing records take.""" +class TCPSegment(NamedTuple): + """The fields of a TCP segment that the reassembly and flow tracing records take. + + :func:`tcp_segment` returns one. + + """ #: Info of the IP layer carrying the segment. ip: 'Data_IPv4 | Data_IPv6' @@ -61,9 +68,12 @@ class _TCPSegment(NamedTuple): payload: 'bytes' -def _tcp_segment(frame: 'ProtocolBase') -> 'Optional[_TCPSegment]': +def tcp_segment(frame: 'ProtocolBase') -> 'Optional[TCPSegment]': """The TCP segment ``frame`` carries, or :data:`None` if it carries none. + The TCP reassembly and flow tracing adapters of this module and of + :mod:`pcapkit.toolkit.pcapng` all read the segment with this. + Args: frame: PCAP frame or PCAP-NG block. @@ -88,7 +98,7 @@ def _tcp_segment(frame: 'ProtocolBase') -> 'Optional[_TCPSegment]': if 'TCP' in frame: tcp = cast('TCP', frame['TCP']) tcp_info = tcp.info - return _TCPSegment( + return TCPSegment( ip=cast('IPv4 | IPv6', frame['IP']).info, srcport=tcp_info.srcport.port, dstport=tcp_info.dstport.port, @@ -126,7 +136,7 @@ def _tcp_segment(frame: 'ProtocolBase') -> 'Optional[_TCPSegment]': if hdr_len < 20: return None flags = octets[13] - return _TCPSegment( + return TCPSegment( ip=ip_info, srcport=int.from_bytes(octets[0:2], 'big'), dstport=int.from_bytes(octets[2:4], 'big'), @@ -270,7 +280,7 @@ def tcp_reassembly(frame: 'Frame') -> 'TCP_Packet | None': :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if (segment := _tcp_segment(frame)) is not None: + if (segment := tcp_segment(frame)) is not None: raw_len = len(segment.payload) data = TCP_Packet( bufid=( @@ -317,7 +327,7 @@ def tcp_traceflow(frame: 'Frame', *, data_link: 'LinkType') -> 'TF_TCP_Packet | :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if (segment := _tcp_segment(frame)) is not None: + if (segment := tcp_segment(frame)) is not None: data = TF_TCP_Packet( # type: ignore[type-var] protocol=data_link, # data link type from global header index=frame.info.number, # frame number diff --git a/pcapkit/toolkit/pcapng.py b/pcapkit/toolkit/pcapng.py index 117b3000ee..6aedcfeb22 100644 --- a/pcapkit/toolkit/pcapng.py +++ b/pcapkit/toolkit/pcapng.py @@ -18,7 +18,7 @@ from pcapkit.foundation.traceflow.data.tcp import Packet as TF_TCP_Packet from pcapkit.protocols.data.misc.pcap.frame import Frame as Data_Frame from pcapkit.protocols.data.misc.pcap.frame import FrameInfo as Data_FrameInfo -from pcapkit.toolkit.pcap import _tcp_segment +from pcapkit.toolkit.pcap import tcp_segment if TYPE_CHECKING: from ipaddress import IPv4Address, IPv6Address @@ -164,7 +164,7 @@ def tcp_reassembly(frame: 'PCAPNG') -> 'TCP_Packet | None': :class:`pcapkit.foundation.reassembly.tcp.TCP` """ - if (segment := _tcp_segment(frame)) is not None: + if (segment := tcp_segment(frame)) is not None: raw_len = len(segment.payload) frame_info = cast('Packet', frame.info) @@ -213,7 +213,7 @@ def tcp_traceflow(frame: 'PCAPNG', *, nanosecond: 'bool' = False) -> 'TF_TCP_Pac :class:`pcapkit.foundation.traceflow.tcp.TCP` """ - if (segment := _tcp_segment(frame)) is not None: + if (segment := tcp_segment(frame)) is not None: frame_info = cast('Packet', frame.info) data = TF_TCP_Packet( # type: ignore[type-var] protocol=frame.linktype, # data link type from global header diff --git a/tests/toolkit/test_pcap_tcp_segment_public_1570_unit.py b/tests/toolkit/test_pcap_tcp_segment_public_1570_unit.py new file mode 100644 index 0000000000..bd08a33bbb --- /dev/null +++ b/tests/toolkit/test_pcap_tcp_segment_public_1570_unit.py @@ -0,0 +1,164 @@ +# -*- coding: utf-8 -*- +"""``tcp_segment`` and ``TCPSegment`` are public in :mod:`pcapkit.toolkit.pcap`. C.f. #1570. + +The helper the TCP adapters of :mod:`pcapkit.toolkit.pcap` and +:mod:`pcapkit.toolkit.pcapng` share was the private ``pcap._tcp_segment``, which +``pcapng`` imported across modules, and its result type was the private +``pcap._TCPSegment``. Under the rulings on #1519 a shared helper becomes public in +the module both callers already import, as ``test_start_line`` did in #1525, so the +two keep their names without the underscore and stay in ``pcap``, which ``pcapng`` +already imports. The adapters' end-to-end behaviour is covered by +``tests/toolkit/test_pcap_tcp_segment_unit.py``; these tests pin the public surface +and that both toolkits read the segment through it. + +Every frame is built here and read back through the default engine, so the module +reads no sample capture and belongs to the unit tier. + +""" +from __future__ import annotations + +import importlib.util +import io +import struct +import unittest +import warnings +from typing import TYPE_CHECKING +from unittest import mock + +from tests._support import close_extractor, reimport_once_per_class + +if TYPE_CHECKING: + from typing import Any + +RUNTIME_DEPS = ('tbtrim', 'aenum', 'chardet', 'dictdumper') +HAS_RUNTIME = all(importlib.util.find_spec(name) is not None for name in RUNTIME_DEPS) + +#: An Ethernet frame of IPv4 carrying a 20-octet TCP header, port 1234 to 80, +#: ``seq`` 1000, ``ack`` 2000, SYN and ACK, and the payload ``data``. +FRAME = (b'\xbb' * 6 + b'\xaa' * 6 + b'\x08\x00' + + struct.pack('!BBHHHBBH4s4s', 0x45, 0, 44, 1, 0, 64, 6, 0, + bytes([192, 0, 2, 1]), bytes([198, 51, 100, 1])) + + struct.pack('!HHIIBBHHH', 1234, 80, 1000, 2000, 5 << 4, 0x12, 65535, 0, 0) + b'data') + +#: An Ethernet frame of ARP, which carries no IP layer and so no TCP segment. +ARP = b'\xff' * 6 + b'\xaa' * 6 + b'\x08\x06' + bytes.fromhex('0001080006040001') + bytes(20) + +#: The fields of :class:`~pcapkit.toolkit.pcap.TCPSegment`, in order. +FIELDS = ('ip', 'srcport', 'dstport', 'seq', 'ack', 'syn', 'fin', 'rst', 'header', 'payload') + + +def pcap(octets: 'bytes') -> 'bytes': + """A little-endian microsecond Ethernet PCAP holding the one record.""" + return (struct.pack(' 'bytes': + body += b'\x00' * (-len(body) % 4) + return struct.pack(' 'bytes': + """A little-endian PCAP-NG holding the one record, on an Ethernet interface.""" + ticks = 1500000000 * 10 ** 6 + 774 + return (_block(0x0A0D0D0A, struct.pack('> 32, ticks & 0xFFFFFFFF, len(octets), len(octets)) + + octets)) + + +class _Base(unittest.TestCase): + + def setUp(self) -> None: + reimport_once_per_class(self) + + def _frame(self, capture: 'bytes') -> 'Any': + """The one frame of ``capture``, read through the default engine.""" + from pcapkit import extract + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + extractor = extract(fin=io.BytesIO(capture), nofile=True, store=True) + self.addCleanup(close_extractor, extractor) + return extractor.frame[-1] + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestPublicSurface(_Base): + + def test_all(self) -> None: + from pcapkit.toolkit import pcap + + self.assertEqual(pcap.__all__, ['ipv4_reassembly', 'ipv6_reassembly', 'tcp_reassembly', + 'tcp_traceflow', 'tcp_segment', 'TCPSegment']) + + def test_callers_share_the_public_helper(self) -> None: + from pcapkit.toolkit import pcap, pcapng + + self.assertIs(pcapng.tcp_segment, pcap.tcp_segment) + for module, name in ((pcap, '_tcp_segment'), (pcap, '_TCPSegment'), (pcapng, '_tcp_segment')): + with self.subTest(module=module.__name__, name=name): + self.assertFalse(hasattr(module, name)) + + def test_result_type_is_the_public_named_tuple(self) -> None: + from pcapkit.toolkit.pcap import TCPSegment + + self.assertTrue(issubclass(TCPSegment, tuple)) + self.assertEqual(TCPSegment._fields, FIELDS) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestTCPSegment(_Base): + + def test_reads_a_parsed_segment_from_either_container(self) -> None: + from pcapkit.toolkit.pcap import TCPSegment, tcp_segment + + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + with self.subTest(container=container): + segment = tcp_segment(self._frame(write(FRAME))) + self.assertIsInstance(segment, TCPSegment) + assert segment is not None + self.assertEqual((str(segment.ip.src), str(segment.ip.dst)), ('192.0.2.1', '198.51.100.1')) + self.assertEqual((segment.srcport, segment.dstport, segment.seq, segment.ack), + (1234, 80, 1000, 2000)) + self.assertEqual((segment.syn, segment.fin, segment.rst), (True, False, False)) + self.assertEqual((segment.header, segment.payload), (FRAME[34:54], b'data')) + + def test_no_segment_without_an_ip_layer(self) -> None: + from pcapkit.toolkit.pcap import tcp_segment + + for container, write in (('pcap', pcap), ('pcapng', pcapng)): + with self.subTest(container=container): + self.assertIsNone(tcp_segment(self._frame(write(ARP)))) + + +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class TestAdaptersCallTheHelper(_Base): + """Each TCP adapter reads the segment through the module-level ``tcp_segment``.""" + + def test_both_toolkits_call_tcp_segment(self) -> None: + from pcapkit.const.reg.linktype import LinkType + from pcapkit.toolkit import pcap as toolkit_pcap + from pcapkit.toolkit import pcapng as toolkit_pcapng + + pcap_frame = self._frame(pcap(FRAME)) + pcapng_frame = self._frame(pcapng(FRAME)) + cases = ( + ('pcap.tcp_reassembly', toolkit_pcap, pcap_frame, toolkit_pcap.tcp_reassembly, {}), + ('pcap.tcp_traceflow', toolkit_pcap, pcap_frame, toolkit_pcap.tcp_traceflow, + {'data_link': LinkType.ETHERNET}), + ('pcapng.tcp_reassembly', toolkit_pcapng, pcapng_frame, toolkit_pcapng.tcp_reassembly, {}), + ('pcapng.tcp_traceflow', toolkit_pcapng, pcapng_frame, toolkit_pcapng.tcp_traceflow, {}), + ) + for name, module, frame, adapter, kwargs in cases: + with self.subTest(adapter=name): + with mock.patch.object(module, 'tcp_segment', wraps=toolkit_pcap.tcp_segment) as helper: + record = adapter(frame, **kwargs) + helper.assert_called_once_with(frame) + self.assertIsNotNone(record) + assert record is not None + self.assertEqual((record.header, bytes(record.payload)), (FRAME[34:54], b'data')) + + +if __name__ == '__main__': + unittest.main()