fix(link)!: L2TPv2 drops the reserved flag bits on rebuild, and make() drops a lone ns or nr

pcapkit/protocols/data/link/l2tp.py

@@ -29,9 +29,14 @@ class Flags(Data):
     offset: 'bool'
     #: Priority.
     prio: 'bool'
+    #: Reserved bits 2, 3, 5 and 8-11 of the flags word, in their on-wire
+    #: positions (i.e. masked by ``0x34F0``). :rfc:`2661` §3.1 has them "set
+    #: to 0 on outgoing messages and ignored on incoming messages"; they are
+    #: carried verbatim so that a rebuild reproduces the word as it arrived.
+    reserved: 'int'
 
     if TYPE_CHECKING:
-        def __init__(self, type: 'Type', len: 'bool', seq: 'bool', offset: 'bool', prio: 'bool') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements
+        def __init__(self, type: 'Type', len: 'bool', seq: 'bool', offset: 'bool', prio: 'bool', reserved: 'int') -> 'None': ...  # pylint: disable=unused-argument,super-init-not-called,redefined-builtin,multiple-statements,line-too-long
 
 
 @info_final

pcapkit/protocols/link/l2tpv2.py

@@ -206,6 +206,7 @@ def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_L2TP':
             seq=bool(_flag['seq']),
             offset=bool(_flag['offset']),
             prio=bool(_flag['prio']),
+            reserved=(_flag['reserved_1'] << 12) | (_flag['reserved_2'] << 10) | (_flag['reserved_3'] << 4),
         )
 
         _size = schema.offset if flags.offset else 0
@@ -245,6 +246,7 @@ def make(self,
              type_namespace: 'Optional[dict[str, int] | dict[int, str] | Type[StdlibEnum] | Type[AenumEnum]]' = None,  # pylint: disable=line-too-long
              type_reversed: 'bool' = False,
              priority: 'bool' = False,
+             reserved: 'int' = 0,
              total_length: 'Optional[int]' = None,
              length_flag: 'Optional[bool]' = None,
              tunnel_id: 'int' = 0,
@@ -264,6 +266,8 @@ def make(self,
             type_namespace: Namespace of type.
             type_reversed: Reversed namespace of type.
             priority: Priority flag.
+            reserved: Reserved bits of the flags word, in their on-wire
+                positions (bits 2, 3, 5 and 8-11, i.e. within ``0x34F0``).
             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.
@@ -286,12 +290,21 @@ def make(self,
 
         Raises:
             ProtocolError: If ``total_length`` is given with ``length_flag``
-                set to :data:`False`.
+                set to :data:`False`, if only one of ``ns`` and ``nr`` is
+                given, or if ``reserved`` sets a bit outside ``0x34F0``.
 
         """
         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
+        # NOTE: :rfc:`2661` §3.1 makes ``S`` mark the presence of *both* Ns and
+        # Nr, so a header cannot carry one without the other. Refuse a lone
+        # value rather than drop it, or invent the other on the wire.
+        if (ns is None) != (nr is None):
+            raise ProtocolError(f'{self.alias}: ns and nr must be given together')
+        seq = ns is not None
+
+        if reserved & ~0x34F0:
+            raise ProtocolError(f'{self.alias}: reserved bits outside 0x34F0: {reserved:#06x}')
 
         if length_flag is None:
             length_flag = total_length is not None or type_ == Enum_Type.Control
@@ -312,6 +325,9 @@ def make(self,
                 'seq': seq,
                 'offset': offset is not None,
                 'prio': priority,
+                'reserved_1': (reserved >> 12) & 0b11,
+                'reserved_2': (reserved >> 10) & 0b1,
+                'reserved_3': (reserved >> 4) & 0b1111,
                 'version': version,
             },
             length=total_length,
@@ -391,6 +407,7 @@ def _make_data(cls, data: 'Data_L2TP') -> 'dict[str, Any]':  # type: ignore[over
         return {
             'type': data.flags.type,
             'priority': data.flags.prio,
+            'reserved': data.flags.reserved,
             'version': data.version,
             # NOTE: Passed as ``total_length`` since ``length`` is the parse
             # length :class:`~pcapkit.protocols.protocol.ProtocolBase` takes, so

pcapkit/protocols/schema/link/l2tp.py

@@ -33,6 +33,12 @@ class FlagsType(TypedDict):
         offset: int
         #: Priority of L2TP packet.
         prio: int
+        #: Reserved bits 2 and 3 of L2TP packet.
+        reserved_1: int
+        #: Reserved bit 5 of L2TP packet.
+        reserved_2: int
+        #: Reserved bits 8 to 11 of L2TP packet.
+        reserved_3: int
         #: Version of L2TP packet.
         version: Literal[2]
 
@@ -45,9 +51,12 @@ class L2TP(Schema):
     flags: 'FlagsType' = BitField(length=2, namespace={
         'type': (0, 1),
         'len': (1, 1),
+        'reserved_1': (2, 2),
         'seq': (4, 1),
+        'reserved_2': (5, 1),
         'offset': (6, 1),
         'prio': (7, 1),
+        'reserved_3': (8, 4),
         'version': (12, 4),
     })
     #: Length of L2TP packet.

tests/protocols/link/test_l2tpv2_roundtrip_unit.py

@@ -0,0 +1,97 @@
+# -*- coding: utf-8 -*-
+"""L2TPv2 keeps its reserved flag bits, and never drops a lone Ns or Nr.
+
+GitHub issue #1207: the flags :class:`~pcapkit.corekit.fields.strings.BitField`
+declared no reserved subfields, so bits 2, 3, 5 and 8-11 were lost on a
+``from_data`` rebuild: ``200200010002ff03002145000000`` came back as
+``000200010002ff03002145000000``.
+
+GitHub issue #1208: :meth:`~pcapkit.protocols.link.l2tpv2.L2TPv2.make` set ``S``
+only when both ``ns`` and ``nr`` were given, so ``L2TPv2(ns=7)`` silently wrote
+neither. :rfc:`2661` §3.1 has ``S`` mark the presence of both fields together,
+so a lone value is now refused with
+:class:`~pcapkit.utilities.exceptions.ProtocolError`.
+
+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 unittest
+
+from tests._support import reimport_once_per_class
+
+#: A PPP frame header and the first IPv4 octet, as the L2TPv2 payload.
+PAYLOAD = bytes.fromhex('ff03002145')
+
+#: Each reserved bit group on its own, then all of them at once.
+RESERVED_MASKS = (0x2000, 0x1000, 0x0400, 0x00f0, 0x34f0)
+
+
+class TestL2TPv2RoundTrip(unittest.TestCase):
+    """Pin the reserved flag bits and the Ns/Nr pairing."""
+
+    def setUp(self) -> None:
+        reimport_once_per_class(self)
+
+    def test_issue_1207_reproduction(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        raw = bytes.fromhex('200200010002ff03002145000000')
+        parsed = L2TPv2(raw)
+        self.assertEqual(parsed.info.flags.reserved, 0x2000)
+        self.assertEqual(L2TPv2.from_data(parsed.info).data, raw)
+
+    def test_each_reserved_mask_round_trips(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        for mask in RESERVED_MASKS:
+            raw = (mask | 0x0002).to_bytes(2, 'big') + bytes.fromhex('00010002') + PAYLOAD
+            with self.subTest(mask=hex(mask)):
+                parsed = L2TPv2(raw)
+                self.assertEqual(parsed.info.flags.reserved, mask)
+                self.assertEqual(L2TPv2.from_data(parsed.info).data, raw)
+
+    def test_reserved_bits_round_trip_inside_udp(self) -> None:
+        from pcapkit.protocols.transport.udp import UDP
+
+        l2tp = bytes.fromhex('34f200010002') + PAYLOAD
+        raw = (1701).to_bytes(2, 'big') * 2 + (8 + len(l2tp)).to_bytes(2, 'big') + b'\x00\x00' + l2tp
+        parsed = UDP(raw)
+        self.assertEqual(parsed.payload.info.flags.reserved, 0x34f0)
+        self.assertEqual(UDP.from_data(parsed.info).data, raw)
+
+    def test_make_writes_reserved_and_refuses_other_bits(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        self.assertEqual(L2TPv2(type=0, reserved=0x0400).data[:2], bytes.fromhex('0402'))
+        self.assertEqual(L2TPv2(type=0).data[:2], bytes.fromhex('0002'))
+        for bad in (0x0001, 0x0800, 0x8000):
+            with self.subTest(reserved=hex(bad)):
+                with self.assertRaises(ProtocolError):
+                    L2TPv2(type=0, reserved=bad)
+
+    def test_issue_1208_lone_ns_or_nr_is_refused(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+        from pcapkit.utilities.exceptions import ProtocolError
+
+        for kwargs in ({'ns': 7}, {'nr': 7}, {'ns': 0}, {'nr': 0}):
+            with self.subTest(**kwargs):
+                with self.assertRaises(ProtocolError):
+                    L2TPv2(**kwargs)
+
+    def test_ns_and_nr_together_round_trip(self) -> None:
+        from pcapkit.protocols.link.l2tpv2 import L2TPv2
+
+        made = L2TPv2(type=0, ns=7, nr=0)
+        self.assertEqual(made.data, bytes.fromhex('08020000000000070000'))
+        self.assertEqual((made.info.ns, made.info.nr), (7, 0))
+        self.assertEqual(L2TPv2.from_data(made.info).data, made.data)
+
+
+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, len=True, prio=False),
+            flags=DummyData(type=True, len=True, prio=False, reserved=0),
             version=2,
             length=32,
             tunnelid=3,
@@ -635,7 +635,8 @@ def test_l2tp_properties_read_and_make_variants(self) -> None:
         reader = object.__new__(L2TPv2)
         reader.__header__ = SchemaL2TP(
             flags={'type': Type.Control, 'len': True, 'seq': True, 'offset': True,
-                   'prio': True, 'version': 2},
+                   'prio': True, 'reserved_1': 0, 'reserved_2': 0,
+                   'reserved_3': 0, 'version': 2},
             length=24,
             tunnel_id=1,
             session_id=2,
@@ -658,7 +659,8 @@ def test_l2tp_properties_read_and_make_variants(self) -> None:
         reader_no_flags._data = b'\x00' * 14
         reader_no_flags.__header__ = SchemaL2TP(
             flags={'type': Type.Data, 'len': False, 'seq': False, 'offset': False,
-                   'prio': False, 'version': 2},
+                   'prio': False, 'reserved_1': 0, 'reserved_2': 0,
+                   'reserved_3': 0, 'version': 2},
             length=None,
             tunnel_id=5,
             session_id=6,