diff --git a/pcapkit/protocols/application/ospf.py b/pcapkit/protocols/application/ospf.py index aeef7c9c70..58c90e6f7f 100644 --- a/pcapkit/protocols/application/ospf.py +++ b/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': diff --git a/pcapkit/protocols/data/application/ospf.py b/pcapkit/protocols/data/application/ospf.py index c6f9cf5c4b..e1fe7806cb 100644 --- a/pcapkit/protocols/data/application/ospf.py +++ b/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 diff --git a/pcapkit/protocols/data/link/l2tp.py b/pcapkit/protocols/data/link/l2tp.py index 8f62f44e50..0246eb88fb 100644 --- a/pcapkit/protocols/data/link/l2tp.py +++ b/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 diff --git a/pcapkit/protocols/link/l2tpv2.py b/pcapkit/protocols/link/l2tpv2.py index f249ac2e5f..ecaee75975 100644 --- a/pcapkit/protocols/link/l2tpv2.py +++ b/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''), } diff --git a/pcapkit/protocols/schema/application/ospf.py b/pcapkit/protocols/schema/application/ospf.py index 0551f96d7c..c28108c74c 100644 --- a/pcapkit/protocols/schema/application/ospf.py +++ b/pcapkit/protocols/schema/application/ospf.py @@ -79,7 +79,11 @@ 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 `. + trailer: 'bytes' = PayloadField(default=b'') if TYPE_CHECKING: def __init__(self, version: 'int', type: 'Enum_Packet', length: 'int', @@ -87,4 +91,5 @@ def __init__(self, version: 'int', type: 'Enum_Packet', length: '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': ... diff --git a/pcapkit/protocols/schema/link/l2tp.py b/pcapkit/protocols/schema/link/l2tp.py index 38414d7e93..20b1327dd3 100644 --- a/pcapkit/protocols/schema/link/l2tp.py +++ b/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 `. + #: 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': ... diff --git a/tests/protocols/link/test_l2tpv2_ospf_trailer_kept_unit.py b/tests/protocols/link/test_l2tpv2_ospf_trailer_kept_unit.py new file mode 100644 index 0000000000..6d242faf75 --- /dev/null +++ b/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()