From 53f2d0fc15d83b700f29f46718f5c42520185339 Mon Sep 17 00:00:00 2001 From: Jarry Shaw Date: Wed, 7 Oct 2026 00:19:38 -0400 Subject: [PATCH] fix(l2tpv2): correct the Type enum and L bit against RFC 2661 (#1175) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - vendor/l2tp/type.py: Data = 0, Control = 1, per RFC 2661 §3.1; the const is regenerated from it. BREAKING: Enum_Type values swap. - L2TPv2.make(): a control message sets L and defaults total_length to the real header-plus-payload length; a new length_flag keyword, fed by _make_data, keeps a captured control message without L byte-exact. An explicit total_length is written as given, as UDP does (#1155). - The offset pad is recorded as info.padding and passed back to make(), so from_data no longer zeroes it. Closes #1175 --- pcapkit/const/l2tp/type.py | 4 +- pcapkit/protocols/data/link/l2tp.py | 2 + pcapkit/protocols/link/l2tpv2.py | 42 ++++++- pcapkit/protocols/schema/link/l2tp.py | 3 +- pcapkit/vendor/l2tp/type.py | 7 +- tests/protocols/link/test_l2tpv2_type_unit.py | 103 ++++++++++++++++++ tests/protocols/link/test_link_unit.py | 2 +- 7 files changed, 151 insertions(+), 12 deletions(-) create mode 100644 tests/protocols/link/test_l2tpv2_type_unit.py diff --git a/pcapkit/const/l2tp/type.py b/pcapkit/const/l2tp/type.py index e6e3dcbb01..c2a7880e63 100644 --- a/pcapkit/const/l2tp/type.py +++ b/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': diff --git a/pcapkit/protocols/data/link/l2tp.py b/pcapkit/protocols/data/link/l2tp.py index ca74033599..2b9beb19d2 100644 --- a/pcapkit/protocols/data/link/l2tp.py +++ b/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 diff --git a/pcapkit/protocols/link/l2tpv2.py b/pcapkit/protocols/link/l2tpv2.py index 5ab40853e0..a161e71c3f 100644 --- a/pcapkit/protocols/link/l2tpv2.py +++ b/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), } diff --git a/pcapkit/protocols/schema/link/l2tp.py b/pcapkit/protocols/schema/link/l2tp.py index e391d030e9..f610b3c017 100644 --- a/pcapkit/protocols/schema/link/l2tp.py +++ b/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': ... diff --git a/pcapkit/vendor/l2tp/type.py b/pcapkit/vendor/l2tp/type.py index 0d4dda6e34..ff920116de 100644 --- a/pcapkit/vendor/l2tp/type.py +++ b/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] diff --git a/tests/protocols/link/test_l2tpv2_type_unit.py b/tests/protocols/link/test_l2tpv2_type_unit.py new file mode 100644 index 0000000000..c09a115357 --- /dev/null +++ b/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() diff --git a/tests/protocols/link/test_link_unit.py b/tests/protocols/link/test_link_unit.py index d541a476c0..d618ad5103 100644 --- a/tests/protocols/link/test_link_unit.py +++ b/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,