fix(ipx,ah): IPX from_data raises struct.error; AH reserved field written back as zero

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

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,

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.
 

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': ...

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')

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
+<pcapkit.protocols.internet.ipx.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 <pcapkit.protocols.internet.ipx.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()

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: