Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 4 additions & 0 deletions pcapkit/protocols/internet/ah.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand Down
82 changes: 82 additions & 0 deletions tests/protocols/internet/test_ah_unit.py
Original file line number Diff line number Diff line change
@@ -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
Expand Down Expand Up @@ -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()
Loading