fix(l2tpv2,ospf)!: keep the octets past the Length field as captured (#1455)

pcapkit/protocols/application/ospf.py

@@ -185,6 +185,15 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_OSPF':
             ospf.__update__([
                 ('auth', cast('bytes', schema.auth_data)),
             ])
+
+        # NOTE: Octets captured past the Packet Length belong to this layer,
+        # since its own length field is what leaves them out of the payload --
+        # the message digest of cryptographic authentication among them -- and
+        # the rebuild keeps them as captured (:issue:`1455`).
+        if schema.trailer:
+            ospf.__update__([
+                ('trailer', schema.trailer),
+            ])
         # OSPF carries no next-protocol field -- the body is LSAs and packet-type
         # specific fields, which pcapkit does not dissect -- so dispatch on the
         # -1 sentinel, as ARP does, rather than on a code read off the wire;
@@ -207,6 +216,7 @@ def make(self,
              auth_type_reversed: 'bool' = False,
              auth_data: 'bytes | Schema_CryptographicAuthentication | Data_CryptographicAuthentication' = b'\x00\x00\x00\x00\x00\x00\x00\x00',
              payload: 'bytes | ProtocolBase | Schema' = b'',
+             trailer: 'bytes' = b'',
              **kwargs: 'Any') -> 'Schema_OSPF':
         """Make (construct) packet data.
 
@@ -219,9 +229,7 @@ def make(self,
             packet_length: Packet length (header included). If not given, it is
                 computed as 24 plus the length of ``payload``. Give it
                 explicitly when ``payload`` carries octets outside the OSPF
-                packet, such as the message digest appended under
-                cryptographic authentication (:rfc:`2328#appendix-D.4.3`),
-                which Packet Length does not count.
+                packet, or pass those as ``trailer`` instead.
             router_id: Router ID.
             area_id: Area ID.
             checksum: Checksum.
@@ -231,6 +239,9 @@ def make(self,
             auth_type_reversed: Reverse namespace for ``auth_type``.
             auth_data: Authentication data.
             payload: Payload data.
+            trailer: Octets after the packet, outside the Packet Length, such
+                as the message digest appended under cryptographic
+                authentication (:rfc:`2328#appendix-D.4.3`); written as is.
             **kwargs: Arbitrary keyword arguments.
 
         Returns:
@@ -273,6 +284,7 @@ def make(self,
             auth_type=auth_type_,  # type: ignore[arg-type]
             auth_data=data,
             payload=payload,
+            trailer=trailer,
         )
 
     ##########################################################################
@@ -319,7 +331,9 @@ def _make_data(cls, data: 'Data_OSPF') -> 'dict[str, Any]':  # type: ignore[over
             'checksum': data.chksum,
             'auth_type': data.autype,
             'auth_data': data.auth,
-            'payload': cls._make_payload(data)
+            'payload': cls._make_payload(data),
+            # NOTE: And the octets captured past the Packet Length (:issue:`1455`).
+            'trailer': data.get('trailer', b''),
         }
 
     def _read_id_numbers(self, id: 'bytes') -> 'IPv4Address':

pcapkit/protocols/data/application/ospf.py

@@ -55,6 +55,9 @@ class OSPF(Protocol):
     if TYPE_CHECKING:
         #: Authentication.
         auth: 'bytes | CryptographicAuthentication'
+        #: Octets captured past the Packet Length, present only when there are
+        #: any.
+        trailer: 'bytes'
 
         def __init__(self, version: 'int', type: 'Packet', len: 'int', router_id: 'IPv4Address',
                      area_id: 'IPv4Address', chksum: 'bytes', autype: 'Authentication') -> 'None': ...  # pylint: disable=super-init-not-called,unused-argument,line-too-long,multiple-statements,redefined-builtin

pcapkit/protocols/data/link/l2tp.py

@@ -65,6 +65,9 @@ class L2TP(Protocol):
         hdr_len: 'int'
         #: Offset pad (:data:`None` unless ``flags.offset`` is set).
         padding: 'Optional[bytes]'
+        #: Octets captured past the Length field, present only when there are
+        #: any.
+        trailer: 'bytes'
 
         def __init__(self, flags: 'Flags', version: 'int', length: 'Optional[int]', tunnelid: 'int', sessionid: 'int',
                      ns: 'Optional[int]', nr: 'Optional[int]', offset: 'Optional[int]') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements,line-too-long

pcapkit/protocols/link/l2tpv2.py

@@ -230,6 +230,14 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_L2TP':
         if _size:
             self._read_fileng(_size)
 
+        # NOTE: Octets captured past the Length field belong to this layer,
+        # since its own length field is what leaves them out of the payload,
+        # and the rebuild keeps them as captured (:issue:`1455`).
+        if schema.trailer:
+            l2tp.__update__([
+                ('trailer', schema.trailer),
+            ])
+
         length = schema.length if flags.len else (length or len(self))
         # L2TP carries no next-protocol field -- the payload is a PPP frame,
         # which pcapkit does not dissect -- so dispatch on the -1 sentinel, as
@@ -256,6 +264,7 @@ def make(self,
              offset: 'Optional[int]' = None,
              padding: 'Optional[bytes]' = None,
              payload: 'bytes | ProtocolBase | Schema' = b'',
+             trailer: 'bytes' = b'',
              **kwargs: 'Any') -> 'Schema_L2TP':  # pylint: disable=unused-argument
         """Make (construct) packet data.
 
@@ -283,6 +292,8 @@ def make(self,
             padding: Offset pad, zero-filled or truncated to ``offset`` octets;
                 all zeros when :data:`None`.
             payload: Payload data.
+            trailer: Octets after the datagram, outside the Length field;
+                written as is.
             **kwargs: Arbitrary keyword arguments.
 
         Returns:
@@ -338,6 +349,7 @@ def make(self,
             offset=offset,
             padding=b'' if padding is None else padding,
             payload=payload,
+            trailer=trailer,
         )
 
     ##########################################################################
@@ -425,4 +437,6 @@ def _make_data(cls, data: 'Data_L2TP') -> 'dict[str, Any]':  # type: ignore[over
             'offset': data.offset,
             'padding': data.get('padding'),
             'payload': cls._make_payload(data),
+            # NOTE: And the octets captured past the Length field (:issue:`1455`).
+            'trailer': data.get('trailer', b''),
         }

pcapkit/protocols/schema/application/ospf.py

@@ -79,12 +79,17 @@ class OSPF(Schema):
         selector=ospf_auth_data_selector,
     )
     #: Payload.
-    payload: 'bytes' = PayloadField()
+    payload: 'bytes' = PayloadField(length=lambda pkt: pkt['length'] - 24)
+    #: Octets captured past the Packet Length, such as the message digest
+    #: that cryptographic authentication appends (:rfc:`2328#appendix-D.4.3`),
+    #: c.f. :attr:`IPv4.trailer <pcapkit.protocols.schema.internet.ipv4.IPv4.trailer>`.
+    trailer: 'bytes' = PayloadField(default=b'')
 
     if TYPE_CHECKING:
         def __init__(self, version: 'int', type: 'Enum_Packet', length: 'int',
                      router_id: 'IPv4Address | bytes | str | int',
                      area_id: 'IPv4Address | bytes | str | int',
                      checksum: 'bytes', auth_type: 'Enum_Authentication',
                      auth_data: 'bytes | CryptographicAuthentication',
-                     payload: 'bytes | ProtocolBase | Schema') -> 'None': ...
+                     payload: 'bytes | ProtocolBase | Schema',
+                     trailer: 'bytes' = b'') -> 'None': ...

pcapkit/protocols/schema/link/l2tp.py

@@ -13,7 +13,7 @@
 __all__ = ['L2TP']
 
 if TYPE_CHECKING:
-    from typing import Optional
+    from typing import Any, Optional
 
     from pcapkit.protocols.protocol import ProtocolBase
 
@@ -43,6 +43,24 @@ class FlagsType(TypedDict):
         version: Literal[2]
 
 
+def l2tp_payload_length(pkt: 'dict[str, Any]') -> 'int':
+    """Length of the :attr:`L2TP.payload` field.
+
+    Args:
+        pkt: Packet data.
+
+    Returns:
+        The Length field less the header, when the ``L`` flag is set; otherwise
+        ``-1``, so that the payload takes the rest of the datagram.
+
+    """
+    flags = pkt['flags']
+    if not flags['len']:
+        return -1
+    size = pkt['offset'] if flags['offset'] else 0
+    return pkt['length'] - (6 + 2 * (1 + 2 * flags['seq'] + flags['offset']) + size)
+
+
 @schema_final
 class L2TP(Schema):
     """Header schema for L2TP packet."""
@@ -89,10 +107,16 @@ class L2TP(Schema):
         lambda packet: packet['flags']['offset'],
     )
     #: Payload of L2TP packet.
-    payload: 'bytes' = PayloadField()
+    payload: 'bytes' = PayloadField(length=l2tp_payload_length)
+    #: Octets captured past the Length field, c.f.
+    #: :attr:`IPv4.trailer <pcapkit.protocols.schema.internet.ipv4.IPv4.trailer>`.
+    #: Always empty when the ``L`` flag is clear, since the payload then takes
+    #: the rest of the datagram.
+    trailer: 'bytes' = PayloadField(default=b'')
 
     if TYPE_CHECKING:
         def __init__(self, flags: 'FlagsType', length: 'Optional[int]', tunnel_id: 'int',
                      session_id: 'int', ns: 'Optional[int]', nr: 'Optional[int]',
                      offset: 'Optional[int]', padding: 'bytes',
-                     payload: 'bytes | ProtocolBase | Schema') -> 'None': ...
+                     payload: 'bytes | ProtocolBase | Schema',
+                     trailer: 'bytes' = b'') -> 'None': ...

tests/protocols/link/test_l2tpv2_ospf_trailer_kept_unit.py

@@ -0,0 +1,159 @@
+# -*- coding: utf-8 -*-
+"""Octets captured past an L2TPv2 or OSPF packet's own length survive a rebuild.
+
+GitHub issue #1455: :class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` reads its
+Length field and :class:`~pcapkit.protocols.application.ospf.OSPF` its Packet
+Length, but the octets after either were kept by no ``info``, so
+:meth:`~pcapkit.protocols.protocol.ProtocolBase.from_data` dropped them: an
+L2TPv2 header of Length 8 followed by two octets rebuilt as 8 octets of its 10,
+and an OSPF header of Packet Length 24 followed by two as 24 of its 26.
+
+Each layer now owns them as ``trailer``, as IPv4, IPv6 (#1209) and IPX (#1433)
+do, and :meth:`make` writes a ``trailer`` outside the length it declares.
+
+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`,
+c.f. ``tests/protocols/link/test_ethernet_mac_roundtrip_unit.py``.
+
+"""
+
+import unittest
+import warnings
+
+from tests._support import reimport_once_per_class
+
+
+def _ospf(length: int) -> bytes:
+    """Return a 24-octet OSPFv2 Hello header whose Packet Length is ``length``."""
+    return b'\x02\x01' + length.to_bytes(2, 'big') + bytes(20)
+
+
+#: L2TPv2 datagrams with octets past the Length field: (octets, trailer).
+L2TP_TRAILER = {
+    'header only': (bytes.fromhex('4002000800010002') + b'pp', b'pp'),
+    'payload': (bytes.fromhex('4002000a00010002') + b'ab' + b'pp', b'pp'),
+    'offset pad': (bytes.fromhex('42020010000100020004') + bytes(4) + b'ab' + b'pp', b'pp'),
+    'sequence': (bytes.fromhex('4802000c0001000200050006') + b'\xee' * 3, b'\xee' * 3),
+}
+
+#: L2TPv2 datagrams that end at or before their own length, or carry none.
+L2TP_NO_TRAILER = {
+    'exact': bytes.fromhex('4002000a00010002') + b'ab',
+    'truncated': bytes.fromhex('4002001000010002') + b'ab',
+    'no Length field': bytes.fromhex('0002' '00010002') + b'abpp',
+}
+
+#: OSPF packets with octets past the Packet Length: (octets, trailer).
+OSPF_TRAILER = {
+    'header only': (_ospf(24) + b'zz', b'zz'),
+    'payload': (_ospf(28) + b'abcd' + b'zz', b'zz'),
+    # RFC 2328 D.4.3: the digest follows the packet and Packet Length omits it.
+    'crypto digest': (b'\x02\x01\x00\x1c' + bytes(10) + b'\x00\x02' + bytes.fromhex('0000011000000001')
+                      + b'abcd' + bytes(range(16)), bytes(range(16))),
+}
+
+#: OSPF packets that end at or before their own length.
+OSPF_NO_TRAILER = {
+    'exact': _ospf(28) + b'abcd',
+    'truncated': _ospf(40) + b'abcd',
+}
+
+
+class TestL2TPv2TrailerKept(unittest.TestCase):
+    """Pin the octets past an L2TPv2 Length field through ``from_data``."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_datagram_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        for name, (raw, trailer) in L2TP_TRAILER.items():
+            with self.subTest(datagram=name):
+                parsed = L2TPv2(raw, len(raw))
+                self.assertEqual(parsed.info.trailer, trailer)
+                self.assertEqual(bytes(parsed.__header__.payload),
+                                 raw[parsed.info.hdr_len:len(raw) - len(trailer)])
+                self.assertEqual(L2TPv2.from_data(parsed.info).data, raw)
+
+    def test_no_trailer_key_without_excess(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        for name, raw in L2TP_NO_TRAILER.items():
+            with self.subTest(datagram=name):
+                parsed = L2TPv2(raw, len(raw))
+                self.assertNotIn('trailer', parsed.info)
+                self.assertEqual(L2TPv2.from_data(parsed.info).data, raw)
+
+    def test_udp_carrier_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.transport.udp import UDP
+
+        l2tp, trailer = L2TP_TRAILER['payload']
+        raw = bytes.fromhex('06a506a5') + (8 + len(l2tp)).to_bytes(2, 'big') + b'\x00\x00' + l2tp
+        parsed = UDP(raw, len(raw))
+        self.assertEqual(parsed.info.l2tp.trailer, trailer)
+        self.assertEqual(UDP.from_data(parsed.info).data, raw)
+
+    def test_make_writes_trailer_outside_the_length(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        made = L2TPv2(length_flag=True, payload=b'ab', trailer=b'\x00\x01')
+        self.assertEqual(made.info.length, 10)
+        self.assertEqual(made.data[-4:], b'ab\x00\x01')
+        self.assertEqual(made.info.trailer, b'\x00\x01')
+        reparsed = L2TPv2(made.data, len(made.data))
+        self.assertEqual(reparsed.info.trailer, b'\x00\x01')
+        self.assertEqual(L2TPv2.from_data(reparsed.info).data, made.data)
+
+
+class TestOSPFTrailerKept(unittest.TestCase):
+    """Pin the octets past an OSPF Packet Length through ``from_data``."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_packet_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.application.ospf import OSPF
+
+        for name, (raw, trailer) in OSPF_TRAILER.items():
+            with self.subTest(packet=name):
+                parsed = OSPF(raw, len(raw))
+                self.assertEqual(parsed.info.trailer, trailer)
+                self.assertEqual(bytes(parsed.__header__.payload), raw[24:len(raw) - len(trailer)])
+                self.assertEqual(OSPF.from_data(parsed.info).data, raw)
+
+    def test_no_trailer_key_without_excess(self) -> None:
+        from pcapkit.protocols.application.ospf import OSPF
+
+        for name, raw in OSPF_NO_TRAILER.items():
+            with self.subTest(packet=name):
+                parsed = OSPF(raw, len(raw))
+                self.assertNotIn('trailer', parsed.info)
+                self.assertEqual(OSPF.from_data(parsed.info).data, raw)
+
+    def test_ipv4_carrier_rebuilds_byte_for_byte(self) -> None:
+        from pcapkit.protocols.internet.ipv4 import IPv4
+
+        ospf, trailer = OSPF_TRAILER['crypto digest']
+        raw = (b'\x45\x00' + (20 + len(ospf)).to_bytes(2, 'big') + bytes(4)
+               + b'\x01\x59' + bytes(2) + bytes(8) + ospf)
+        with warnings.catch_warnings():
+            warnings.simplefilter('ignore')
+            parsed = IPv4(raw, len(raw))
+        self.assertEqual(parsed.info.ospf.trailer, trailer)
+        self.assertEqual(IPv4.from_data(parsed.info).data, raw)
+
+    def test_make_writes_trailer_outside_the_length(self) -> None:
+        from pcapkit.protocols.application.ospf import OSPF
+
+        made = OSPF(payload=b'abcd', trailer=b'\x00\x01')
+        self.assertEqual(made.info.len, 28)
+        self.assertEqual(made.data[-6:], b'abcd\x00\x01')
+        self.assertEqual(made.info.trailer, b'\x00\x01')
+        reparsed = OSPF(made.data, len(made.data))
+        self.assertEqual(reparsed.info.trailer, b'\x00\x01')
+        self.assertEqual(OSPF.from_data(reparsed.info).data, made.data)
+
+
+if __name__ == '__main__':
+    unittest.main()