From b185ad43a4064b7ab29660279977f1a8b4330e78 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 14:57:25 -0400 Subject: [PATCH] fix(ipx,ah): IPX from_data raises struct.error; AH reserved field written back as zero - IPX: rebuild the 12 address octets from the parsed network, node and socket values; `make` takes `total_length=None` and writes a given value as is, so `from_data` keeps the parsed packet length. - AH: the reserved field is a UInt16Field, carried as `AH.reserved` in the data model and as `make(reserved=0)` (RFC 4302 section 2.3). - Update the IPX and AH `_make_data` unit tests, which pinned the old pass-through and the missing reserved key. Closes #1315 Closes #1327 --- pcapkit/protocols/data/internet/ah.py | 4 +- pcapkit/protocols/internet/ah.py | 6 + pcapkit/protocols/internet/ipx.py | 28 ++++- pcapkit/protocols/schema/internet/ah.py | 11 +- tests/protocols/internet/test_ah_unit.py | 2 + .../internet/test_ipx_ah_roundtrip_unit.py | 104 ++++++++++++++++++ tests/protocols/internet/test_ipx_unit.py | 14 ++- 7 files changed, 156 insertions(+), 13 deletions(-) create mode 100644 tests/protocols/internet/test_ipx_ah_roundtrip_unit.py diff --git a/pcapkit/protocols/data/internet/ah.py b/pcapkit/protocols/data/internet/ah.py index e7e4e708f4..54308cf1f0 100644 --- a/pcapkit/protocols/data/internet/ah.py +++ b/pcapkit/protocols/data/internet/ah.py @@ -20,6 +20,8 @@ class AH(Protocol): next: 'TransType' #: Payload length. length: 'int' + #: Reserved. + reserved: 'int' #: Security parameters index. spi: 'int' #: Sequence number field. @@ -28,4 +30,4 @@ class AH(Protocol): icv: 'bytes' if TYPE_CHECKING: - def __init__(self, next: 'TransType', length: 'int', spi: 'int', seq: 'int', icv: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin + def __init__(self, next: 'TransType', length: 'int', reserved: 'int', spi: 'int', seq: 'int', icv: 'bytes') -> 'None': ... # pylint: disable=unused-argument,super-init-not-called,multiple-statements,redefined-builtin,line-too-long diff --git a/pcapkit/protocols/internet/ah.py b/pcapkit/protocols/internet/ah.py index 06466953a1..908a9fbd67 100644 --- a/pcapkit/protocols/internet/ah.py +++ b/pcapkit/protocols/internet/ah.py @@ -177,6 +177,7 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, ah = Data_AH( next=schema.next, length=(schema.len + 2) * 4, + reserved=schema.reserved, spi=schema.spi, seq=schema.seq, icv=schema.icv, @@ -191,6 +192,7 @@ def make(self, next_default: 'Optional[int]' = None, next_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long next_reversed: 'bool' = False, + reserved: 'int' = 0, spi: 'int' = 0, seq: 'int' = 0, icv: 'bytes' = b'', @@ -203,6 +205,8 @@ def make(self, next_default: Default value of next header type. next_namespace: Namespace of next header type. next_reversed: If the namespace is reversed. + reserved: Reserved field; zero on send per :rfc:`4302#section-2.3`, + but written as given so that a parsed header rebuilds unchanged. spi: Security Parameters Index. seq: Sequence Number Field. icv: Integrity Check Value-ICV. @@ -220,6 +224,7 @@ def make(self, return Schema_AH( next=next_value, # type: ignore[arg-type] len=length, + reserved=reserved, spi=spi, seq=seq, icv=icv, @@ -302,6 +307,7 @@ def _make_data(cls, data: 'Data_AH') -> 'dict[str, Any]': # type: ignore[overri """ return { 'next': data.next, + 'reserved': data.reserved, 'spi': data.spi, 'seq': data.seq, 'icv': data.icv, diff --git a/pcapkit/protocols/internet/ipx.py b/pcapkit/protocols/internet/ipx.py index 0be03af494..176a9e642b 100644 --- a/pcapkit/protocols/internet/ipx.py +++ b/pcapkit/protocols/internet/ipx.py @@ -118,6 +118,7 @@ def make(self, type_default: 'Optional[int]' = None, type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None, # pylint: disable=line-too-long type_reversed: 'bool' = False, + total_length: 'Optional[int]' = None, dst: 'bytes' = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', src: 'bytes' = b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00', payload: 'bytes | ProtocolBase | Schema' = b'', @@ -131,6 +132,8 @@ def make(self, type_default: Default value for undefined packet type. type_namespace: Namespace for packet type. type_reversed: Reverse namespace for packet type. + total_length: Packet length (header includes); computed as ``30`` plus + the payload length when not given. dst: Destination Address. src: Source Address. payload: Payload data. @@ -142,10 +145,12 @@ def make(self, """ type_val = self._make_index(type, type_default, namespace=type_namespace, reversed=type_reversed, pack=False) + if total_length is None: + total_length = 30 + len(payload) return Schema_IPX( chksum=chksum, - len=30 + len(payload), + len=total_length, count=count, type=type_val, # type: ignore[arg-type] dst=dst, @@ -192,11 +197,28 @@ def _make_data(cls, data: 'Data_IPX') -> 'dict[str, Any]': # type: ignore[overr 'chksum': data.chksum, 'count': data.count, 'type': data.type, - 'dst': data.dst, - 'src': data.src, + 'total_length': data.len, + 'dst': cls._make_ipx_address(data.dst), + 'src': cls._make_ipx_address(data.src), 'payload': cls._make_payload(data), } + @staticmethod + def _make_ipx_address(addr: 'Data_Address') -> 'bytes': + """Make IPX address field. + + Args: + addr: Parsed IPX address field. + + Returns: + The 12-octet IPX address: network number, node number and + socket number. + + """ + return (bytes.fromhex(addr.network.replace(':', '')) + + bytes.fromhex(addr.node.replace('-', '')) + + int(addr.socket).to_bytes(2, 'big', signed=False)) + def _read_ipx_address(self, addr: 'bytes') -> 'Data_Address': """Read IPX address field. diff --git a/pcapkit/protocols/schema/internet/ah.py b/pcapkit/protocols/schema/internet/ah.py index 5bc31837d3..012e063cb8 100644 --- a/pcapkit/protocols/schema/internet/ah.py +++ b/pcapkit/protocols/schema/internet/ah.py @@ -6,8 +6,8 @@ from pcapkit.const.reg.transtype import TransType as Enum_TransType from pcapkit.corekit.fields.misc import PayloadField -from pcapkit.corekit.fields.numbers import EnumField, UInt8Field, UInt32Field -from pcapkit.corekit.fields.strings import BytesField, PaddingField +from pcapkit.corekit.fields.numbers import EnumField, UInt8Field, UInt16Field, UInt32Field +from pcapkit.corekit.fields.strings import BytesField from pcapkit.protocols.schema.schema import Schema, schema_final __all__ = ['AH'] @@ -25,7 +25,7 @@ class AH(Schema): #: Payload length. len: 'int' = UInt8Field() #: Reserved. - reserved: 'bytes' = PaddingField(length=2) + reserved: 'int' = UInt16Field() #: Security parameters index. spi: 'int' = UInt32Field() #: Sequence number field. @@ -36,5 +36,6 @@ class AH(Schema): payload: 'bytes' = PayloadField() if TYPE_CHECKING: - def __init__(self, next: 'Enum_TransType', len: 'int', spi: 'int', seq: 'int', - icv: 'bytes', payload: 'bytes | ProtocolBase | Schema') -> 'None': ... + def __init__(self, next: 'Enum_TransType', len: 'int', reserved: 'int', spi: 'int', + seq: 'int', icv: 'bytes', + payload: 'bytes | ProtocolBase | Schema') -> 'None': ... diff --git a/tests/protocols/internet/test_ah_unit.py b/tests/protocols/internet/test_ah_unit.py index a87e682b97..655eee61e7 100644 --- a/tests/protocols/internet/test_ah_unit.py +++ b/tests/protocols/internet/test_ah_unit.py @@ -25,6 +25,7 @@ def test_ah_index_length_and_make_data(self) -> None: data = DummyDict( next=TransType.TCP, + reserved=0xdead, spi=0x12345678, seq=7, icv=b'auth', @@ -36,6 +37,7 @@ def test_ah_index_length_and_make_data(self) -> None: self.assertEqual(proto.__length_hint__(), 20) values = AH._make_data(data) self.assertEqual(values['next'], TransType.TCP) + self.assertEqual(values['reserved'], 0xdead) self.assertEqual(values['spi'], 0x12345678) self.assertEqual(values['seq'], 7) self.assertEqual(values['icv'], b'auth') diff --git a/tests/protocols/internet/test_ipx_ah_roundtrip_unit.py b/tests/protocols/internet/test_ipx_ah_roundtrip_unit.py new file mode 100644 index 0000000000..fec4962a67 --- /dev/null +++ b/tests/protocols/internet/test_ipx_ah_roundtrip_unit.py @@ -0,0 +1,104 @@ +# -*- coding: utf-8 -*- +"""IPX and AH headers rebuild byte for byte through ``from_data``. + +GitHub issue #1315: :meth:`IPX._make_data +` handed the parsed +:class:`~pcapkit.protocols.data.internet.ipx.Address` objects to a schema that +packs 12 raw octets, so every rebuild raised :exc:`struct.error`; and +:meth:`IPX.make ` recomputed the +packet length instead of keeping the parsed one. + +GitHub issue #1327: the AH reserved field (:rfc:`4302#section-2.3`) was a +padding field, so a non-zero value was written back as zero. + +Every case builds its own octets in memory and reads no capture. + +Classes are imported inside each test, after +:func:`~tests._support.reimport_once_per_class`, not at module load, so that +they belong to the :mod:`pcapkit` import that is live when the test runs. + +""" + +import io +import unittest +import warnings + +from tests._support import reimport_once_per_class + +#: Checksum, length, hop count, type, destination and source, then payload. +IPX_PACKETS = { + 'all-zero': bytes.fromhex('0000' '001e') + bytes(26), + 'addresses': (bytes.fromhex('ffff' '001e' '05' '11') + + bytes.fromhex('0a0b0c0d' '010203040506' '4004') + + bytes.fromhex('deadbeef' '112233445566' 'abcd')), + 'payload': bytes.fromhex('ffff' '0020' '05' '11') + bytes(24) + b'hi', + 'length-over-data': bytes.fromhex('ffff' '0040' '05' '11') + bytes(24) + b'hi', + 'length-under-header': bytes.fromhex('ffff' '0010' '05' '11') + bytes(24), +} + +#: Next header, payload length, reserved, SPI, sequence number and ICV. +AH_HEADERS = { + 'reserved-nonzero': bytes.fromhex('0604' 'dead') + bytes(8) + bytes(range(12)), + 'reserved-zero': bytes.fromhex('3304' '0000' '00000100' '00000002') + bytes(range(12)), +} + + +class TestIPXRoundTrip(unittest.TestCase): + """Pin the IPX rebuild of addresses and packet length.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.ipx import IPX + + for name, packet in IPX_PACKETS.items(): + with self.subTest(packet=name), warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = IPX(io.BytesIO(packet), len(packet), extension=True) + self.assertEqual(IPX.from_data(parsed.info).data, packet) + + def test_make_computes_length_when_not_given(self) -> None: + from pcapkit.protocols.internet.ipx import IPX + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = IPX(payload=b'hi') + self.assertEqual(proto.data[2:4], bytes.fromhex('0020')) + + +class TestAHRoundTrip(unittest.TestCase): + """Pin the AH rebuild of the reserved field.""" + + def setUp(self) -> None: + reimport_once_per_class(self) + + def test_from_data_rebuilds_byte_for_byte(self) -> None: + from pcapkit.protocols.internet.ah import AH + + for name, header in AH_HEADERS.items(): + with self.subTest(header=name), warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = AH(io.BytesIO(header), len(header), extension=True) + self.assertEqual(AH.from_data(parsed.info).data, header) + + def test_reserved_is_parsed(self) -> None: + from pcapkit.protocols.internet.ah import AH + + header = AH_HEADERS['reserved-nonzero'] + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + parsed = AH(io.BytesIO(header), len(header), extension=True) + self.assertEqual(parsed.info.reserved, 0xdead) + + def test_make_reserved_defaults_to_zero(self) -> None: + from pcapkit.protocols.internet.ah import AH + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + proto = AH(icv=bytes(12)) + self.assertEqual(proto.data[2:4], bytes(2)) + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/protocols/internet/test_ipx_unit.py b/tests/protocols/internet/test_ipx_unit.py index 86dfdf2513..82d9254090 100644 --- a/tests/protocols/internet/test_ipx_unit.py +++ b/tests/protocols/internet/test_ipx_unit.py @@ -20,15 +20,20 @@ def setUp(self) -> None: reimport_once_per_class(self) def test_ipx_index_length_and_make_data(self) -> None: + from pcapkit.const.ipx.socket import Socket from pcapkit.const.reg.transtype import TransType + from pcapkit.protocols.data.internet.ipx import Address from pcapkit.protocols.internet.ipx import IPX data = DummyDict( chksum=b'\xff\xff', + len=32, count=42, type=4, - dst='dst-node', - src='src-node', + dst=Address(network='00:00:00:01', node='00-01-02-03-04-05', + socket=Socket.get(0x0456), addr='unused'), + src=Address(network='0a:0b:0c:0d', node='aa-bb-cc-dd-ee-ff', + socket=Socket.get(0x4004), addr='unused'), __next_type__=None, ) proto = object.__new__(IPX) @@ -39,8 +44,9 @@ def test_ipx_index_length_and_make_data(self) -> None: self.assertEqual(values['chksum'], b'\xff\xff') self.assertEqual(values['count'], 42) self.assertEqual(values['type'], 4) - self.assertEqual(values['dst'], 'dst-node') - self.assertEqual(values['src'], 'src-node') + self.assertEqual(values['total_length'], 32) + self.assertEqual(values['dst'], bytes.fromhex('00000001' '000102030405' '0456')) + self.assertEqual(values['src'], bytes.fromhex('0a0b0c0d' 'aabbccddeeff' '4004')) self.assertIn('payload', values) def test_ipx_make_builds_schema_with_addresses(self) -> None: