diff --git a/pcapkit/protocols/internet/ah.py b/pcapkit/protocols/internet/ah.py index c4ca82ebdc..7480755fcc 100644 --- a/pcapkit/protocols/internet/ah.py +++ b/pcapkit/protocols/internet/ah.py @@ -30,6 +30,7 @@ from pcapkit.protocols.data.internet.ah import AH as Data_AH from pcapkit.protocols.internet.ipsec import IPsec from pcapkit.protocols.internet.ipv6_ext import IPv6_Ext +from pcapkit.protocols.misc.null import NoPayload from pcapkit.protocols.schema.internet.ah import AH as Schema_AH from pcapkit.protocols.schema.internet.ah import AH_HEADER_LEN from pcapkit.utilities.exceptions import UnsupportedCall @@ -190,6 +191,9 @@ def read(self, length: 'Optional[int]' = None, *, version: 'Literal[4, 6]' = 4, ) if extension: + # NOTE: As an IPv6 extension header the upper layer belongs to the enclosing + # IPv6 header, so hold NoPayload here so that membership does not raise (#2484). + self._next = NoPayload() # pylint: disable=attribute-defined-outside-init return ah return self._decode_next_layer(ah, schema.next, length - ah.length) diff --git a/tests/protocols/internet/test_ah_unit.py b/tests/protocols/internet/test_ah_unit.py index ebfda58776..419f76d05e 100644 --- a/tests/protocols/internet/test_ah_unit.py +++ b/tests/protocols/internet/test_ah_unit.py @@ -1,7 +1,9 @@ from __future__ import annotations import importlib.util +import io import unittest +import warnings from unittest import mock from tests._support import reimport_once_per_class @@ -185,5 +187,85 @@ def test_docstring_cites_the_extension_header_registry(self) -> None: self.assertIn('qualifies it for a base besides', doc) +@unittest.skipUnless(HAS_RUNTIME, 'runtime dependencies not installed') +class AHMembershipTests(unittest.TestCase): + """``in`` on a terminal AH layer parsed from an IPv6 extension chain (#2484). + + The AH layer is taken from ``IPv6.extension_headers`` in extension mode. + ``in`` on it does not raise, and is False for a selector that does not match AH. + Frames are built in memory; no capture is read. + """ + + def setUp(self) -> None: + reimport_once_per_class(self) + + @staticmethod + def _ipv6(first: int, body: bytes) -> bytes: + import struct # pylint: disable=import-outside-toplevel + + return struct.pack('!IHBB', 6 << 28, len(body), first, 64) + bytes(32) + body + + @staticmethod + def _ah(next_: int) -> bytes: + import struct # pylint: disable=import-outside-toplevel + + # 12-octet ICV: AH is 24 octets, so the Payload Len field is 4 + return struct.pack('!BBHII', next_, 4, 0, 0x1234, 1) + bytes(12) + + @staticmethod + def _esp() -> bytes: + import struct # pylint: disable=import-outside-toplevel + + return struct.pack('!II', 0x1234, 1) + bytes(16) + + def _parse_ipv6(self, data: bytes): # type: ignore[no-untyped-def] + from pcapkit.protocols.internet.ipv6 import IPv6 + + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + return IPv6(io.BytesIO(data), len(data)) + + def test_terminal_ah_membership_is_false_for_other_selectors(self) -> None: + from pcapkit.protocols.internet.ah import AH + + ipv6 = self._parse_ipv6(self._ipv6(51, self._ah(59))) + (ah,) = [ext for ext in ipv6.extension_headers.values() if isinstance(ext, AH)] + + self.assertIn('AH', ah) + self.assertNotIn('ESP', ah) + self.assertNotIn('TCP', ah) + + def test_ah_followed_by_esp_membership_does_not_raise(self) -> None: + from pcapkit.protocols.internet.ah import AH + + ipv6 = self._parse_ipv6(self._ipv6(51, self._ah(50) + self._esp())) + ah = next(ext for ext in ipv6.extension_headers.values() if isinstance(ext, AH)) + + self.assertNotIn('ESP', ah) + + def test_esp_extension_layer_membership_is_unchanged(self) -> None: + from pcapkit.protocols.internet.esp import ESP + + ipv6 = self._parse_ipv6(self._ipv6(50, self._esp())) + esp = next(ext for ext in ipv6.extension_headers.values() if isinstance(ext, ESP)) + + self.assertIn('ESP', esp) + self.assertNotIn('AH', esp) + + def test_ipv4_ah_membership_is_unchanged(self) -> None: + import struct # pylint: disable=import-outside-toplevel + + from pcapkit.protocols.internet.ipv4 import IPv4 + + body = self._ah(59) + data = struct.pack('!BBHHHBBH', 0x45, 0, 20 + len(body), 0, 0, 64, 51, 0) + bytes(8) + body + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + ipv4 = IPv4(io.BytesIO(data), len(data)) + + self.assertNotIn('ESP', ipv4.payload) + self.assertIn('AH', ipv4.payload) + + if __name__ == '__main__': unittest.main()