diff --git a/pcapkit/protocols/data/link/l2tp.py b/pcapkit/protocols/data/link/l2tp.py index 2b9beb19d2..8f62f44e50 100644 --- a/pcapkit/protocols/data/link/l2tp.py +++ b/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 diff --git a/pcapkit/protocols/link/l2tpv2.py b/pcapkit/protocols/link/l2tpv2.py index a161e71c3f..f249ac2e5f 100644 --- a/pcapkit/protocols/link/l2tpv2.py +++ b/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 diff --git a/pcapkit/protocols/schema/link/l2tp.py b/pcapkit/protocols/schema/link/l2tp.py index f610b3c017..38414d7e93 100644 --- a/pcapkit/protocols/schema/link/l2tp.py +++ b/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. diff --git a/tests/protocols/link/test_l2tpv2_roundtrip_unit.py b/tests/protocols/link/test_l2tpv2_roundtrip_unit.py new file mode 100644 index 0000000000..62b4c3ecd2 --- /dev/null +++ b/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() diff --git a/tests/protocols/link/test_link_unit.py b/tests/protocols/link/test_link_unit.py index d618ad5103..e9f5eb69f5 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, 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,