fix(l2tpv2): correct the Type enum and L bit against RFC 2661 (#1175)

pcapkit/const/l2tp/type.py

@@ -20,9 +20,9 @@
 class Type(EnumRegistry, IntEnum):
     """[Type] L2TP Type"""
 
-    Control = 0
+    Data = 0
 
-    Data = 1
+    Control = 1
 
     @classmethod
     def _missing_(cls, value: 'int') -> 'Type':

pcapkit/protocols/data/link/l2tp.py

@@ -58,6 +58,8 @@ class L2TP(Protocol):
     if TYPE_CHECKING:
         #: Header length.
         hdr_len: 'int'
+        #: Offset pad (:data:`None` unless ``flags.offset`` is set).
+        padding: 'Optional[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

@@ -224,10 +224,10 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_L2TP':
 
         l2tp.__update__([
             ('hdr_len', hdr_len),
+            ('padding', schema.padding if flags.offset else None),
         ])
         if _size:
             self._read_fileng(_size)
-            # l2tp['padding'] = self._read_fileng(_size)
 
         length = schema.length if flags.len else (length or len(self))
         # L2TP carries no next-protocol field -- the payload is a PPP frame,
@@ -246,11 +246,13 @@ def make(self,
              type_reversed: 'bool' = False,
              priority: 'bool' = False,
              total_length: 'Optional[int]' = None,
+             length_flag: 'Optional[bool]' = None,
              tunnel_id: 'int' = 0,
              session_id: 'int' = 0,
              ns: 'Optional[int]' = None,
              nr: 'Optional[int]' = None,
              offset: 'Optional[int]' = None,
+             padding: 'Optional[bytes]' = None,
              payload: 'bytes | ProtocolBase | Schema' = b'',
              **kwargs: 'Any') -> 'Schema_L2TP':  # pylint: disable=unused-argument
         """Make (construct) packet data.
@@ -262,28 +264,52 @@ def make(self,
             type_namespace: Namespace of type.
             type_reversed: Reversed namespace of type.
             priority: Priority flag.
-            total_length: Length of the datagram, header included; the Length
-                field and its ``L`` flag are omitted when this is :data:`None`.
+            total_length: Length of the datagram, header included, written as
+                given; the real header-plus-payload length when :data:`None`
+                and the Length field is present.
+            length_flag: Whether the Length field and its ``L`` flag are
+                present. When :data:`None`, they are present if
+                ``total_length`` is given, or for a control message, which
+                :rfc:`2661` §3.1 requires to set ``L``.
             tunnel_id: Tunnel ID.
             session_id: Session ID.
             ns: Sequence number.
             nr: Acknowledgement number.
             offset: Offset size.
+            padding: Offset pad, zero-filled or truncated to ``offset`` octets;
+                all zeros when :data:`None`.
             payload: Payload data.
             **kwargs: Arbitrary keyword arguments.
 
         Returns:
             Constructed packet data.
 
+        Raises:
+            ProtocolError: If ``total_length`` is given with ``length_flag``
+                set to :data:`False`.
+
         """
         type_ = self._make_index(type, type_default, namespace=type_namespace,
                                  reversed=type_reversed, pack=False)
+        seq = ns is not None and nr is not None
+
+        if length_flag is None:
+            length_flag = total_length is not None or type_ == Enum_Type.Control
+        elif not length_flag and total_length is not None:
+            raise ProtocolError(f'{self.alias}: total_length given without the Length field')
+
+        # NOTE: An explicit ``total_length`` is not checked against the octets
+        # at hand, so that a ``from_data`` rebuild of a capture truncated by
+        # snaplen keeps the length the sender declared, as UDP does
+        # (:issue:`1155`).
+        if length_flag and total_length is None:
+            total_length = 6 + 2 * (1 + 2 * seq + (offset is not None)) + (offset or 0) + len(payload)
 
         return Schema_L2TP(
             flags={
                 'type': type_,
-                'len': total_length is not None,
-                'seq': ns is not None and nr is not None,
+                'len': length_flag,
+                'seq': seq,
                 'offset': offset is not None,
                 'prio': priority,
                 'version': version,
@@ -294,6 +320,7 @@ def make(self,
             ns=ns,
             nr=nr,
             offset=offset,
+            padding=b'' if padding is None else padding,
             payload=payload,
         )
 
@@ -370,10 +397,15 @@ def _make_data(cls, data: 'Data_L2TP') -> 'dict[str, Any]':  # type: ignore[over
             # under that key it never reached :meth:`make`, and the rebuild
             # dropped both the Length field and the ``L`` flag (:issue:`1172`).
             'total_length': data.length,
+            # NOTE: Passed so that a control message captured without ``L``
+            # rebuilds as captured, rather than gaining the Length field
+            # :meth:`make` would give it by default.
+            'length_flag': data.flags.len,
             'tunnel_id': data.tunnelid,
             'session_id': data.sessionid,
             'ns': data.ns,
             'nr': data.nr,
             'offset': data.offset,
+            'padding': data.get('padding'),
             'payload': cls._make_payload(data),
         }

pcapkit/protocols/schema/link/l2tp.py

@@ -85,4 +85,5 @@ class L2TP(Schema):
     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]', payload: 'bytes | ProtocolBase | Schema') -> 'None': ...
+                     offset: 'Optional[int]', padding: 'bytes',
+                     payload: 'bytes | ProtocolBase | Schema') -> 'None': ...

pcapkit/vendor/l2tp/type.py

@@ -20,10 +20,11 @@
 
 __all__ = ['Type']
 
-#: L2TP packet types.
+#: L2TP packet types, i.e. the ``T`` bit: :rfc:`2661` §3.1 sets it "to 0 for
+#: a data message and 1 for a control message".
 DATA = {
-    0:  'Control',
-    1:  'Data',
+    0:  'Data',
+    1:  'Control',
 }  # type: dict[int, str]
 
 

tests/protocols/link/test_l2tpv2_type_unit.py

@@ -0,0 +1,103 @@
+# -*- coding: utf-8 -*-
+"""L2TPv2 reads and writes the ``T`` bit, and sets ``L``, as :rfc:`2661` says.
+
+GitHub issue #1175: :class:`~pcapkit.const.l2tp.type.Type` had ``Control = 0``
+and ``Data = 1``, the reverse of :rfc:`2661` §3.1, which sets ``T`` "to 0 for a
+data message and 1 for a control message". So the control ZLB ACK
+``c802000c0001000000010002`` parsed as ``Data``, and
+:meth:`~pcapkit.protocols.link.l2tpv2.L2TPv2.make`'s default ``Type.Data``
+wrote a control header. ``make()`` also wrote ``L=0`` for a control message
+given no ``total_length``, though §3.1 requires ``L=1`` there, and a
+``from_data`` rebuild zeroed a non-zero offset pad.
+
+Every case builds its own octets in memory and reads no capture.
+
+:class:`L2TPv2` is imported inside each test, after
+:func:`~tests._support.reimport_once_per_class`, not at module load, for the
+reason :mod:`tests.protocols.link.test_ethernet_mac_roundtrip_unit` gives.
+
+"""
+
+import io
+import unittest
+
+from tests._support import reimport_once_per_class
+
+#: A control ZLB ACK: ``T``, ``L`` and ``S`` set, tunnel 1, Ns 1, Nr 2.
+ZLB_ACK = bytes.fromhex('c802000c0001000000010002')
+
+#: A data message with ``O`` set and a three-octet, non-zero offset pad.
+PADDED = bytes.fromhex('0202' '0001' '0002' '0003' '010203' 'ff03')
+
+
+class TestL2TPv2Type(unittest.TestCase):
+    """Pin the ``T`` bit values and the ``L`` bit of a control message."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_enum_values_follow_rfc_2661(self) -> None:
+        from pcapkit.const.l2tp.type import Type
+
+        self.assertEqual(Type.Data, 0)
+        self.assertEqual(Type.Control, 1)
+        self.assertIs(Type(0), Type.Data)
+        self.assertIs(Type(1), Type.Control)
+
+    def test_zlb_ack_parses_as_control(self) -> None:
+        from pcapkit.const.l2tp.type import Type
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        parsed = L2TPv2(io.BytesIO(ZLB_ACK), len(ZLB_ACK))
+        self.assertIs(parsed.info.flags.type, Type.Control)
+
+    def test_default_make_is_a_data_message(self) -> None:
+        from pcapkit.const.l2tp.type import Type
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        made = L2TPv2(tunnel_id=1, session_id=2)
+        self.assertEqual(made.data, bytes.fromhex('0002' '0001' '0002'))
+        self.assertIs(made.info.flags.type, Type.Data)
+
+    def test_make_control_sets_l_and_the_real_length(self) -> None:
+        from pcapkit.const.l2tp.type import Type
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        self.assertEqual(L2TPv2(type=Type.Control, tunnel_id=1, ns=1, nr=2).data, ZLB_ACK)
+
+        made = L2TPv2(type=Type.Control, tunnel_id=1, session_id=2, payload=b'abcd')
+        self.assertEqual(made.data, bytes.fromhex('c002' '000c' '0001' '0002') + b'abcd')
+        self.assertTrue(made.info.flags.len)
+        self.assertEqual(made.info.length, 12)
+
+    def test_make_writes_an_explicit_total_length_as_given(self) -> None:
+        from pcapkit.const.l2tp.type import Type
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        made = L2TPv2(type=Type.Control, tunnel_id=1, ns=1, nr=2, total_length=99)
+        self.assertEqual(made.info.length, 99)
+
+    def test_make_rejects_total_length_without_the_length_field(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        with self.assertRaises(ProtocolError):
+            L2TPv2(total_length=12, length_flag=False)
+
+    def test_control_without_l_rebuilds_as_captured(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        raw = bytes.fromhex('8802' '0001' '0000' '0000' '0001')  # T and S, no L
+        parsed = L2TPv2(io.BytesIO(raw), len(raw))
+        self.assertEqual(L2TPv2.from_data(parsed.info).data, raw)
+
+    def test_offset_pad_survives_from_data(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        parsed = L2TPv2(io.BytesIO(PADDED), len(PADDED))
+        self.assertEqual(parsed.info.padding, b'\x01\x02\x03')
+        self.assertEqual(L2TPv2.from_data(parsed.info).data, PADDED)
+
+
+if __name__ == '__main__':
+    unittest.main()

tests/protocols/link/test_link_unit.py

@@ -348,7 +348,7 @@ def test_l2tp_index_is_unsupported_and_make_data_preserves_flags(self) -> None:
         from pcapkit.utilities.exceptions import UnsupportedCall
 
         data = DummyData(
-            flags=DummyData(type=True, prio=False),
+            flags=DummyData(type=True, len=True, prio=False),
             version=2,
             length=32,
             tunnelid=3,