# -*- coding: utf-8 -*-
"""L2TPv2 - Layer Two Tunnelling Protocol version 2
======================================================
.. module:: pcapkit.protocols.link.l2tpv2
:mod:`pcapkit.protocols.link.l2tpv2` contains
:class:`~pcapkit.protocols.link.l2tpv2.L2TPv2` only, which implements extractor
for the Layer Two Tunnelling Protocol version 2 (L2TPv2) [*]_ as specified by
:rfc:`2661` -- a 16-bit tunnel ID and a 16-bit session ID, with the version
nibble reading ``2``. Its structure is described as below:
.. table::
======= ===== ===================== ==========================================
Octets Bits Name Description
======= ===== ===================== ==========================================
0 0 ``l2tp.flags`` Flags and Version Info
------- ----- --------------------- ------------------------------------------
0 0 ``l2tp.flags.type`` Type (control / data)
------- ----- --------------------- ------------------------------------------
0 1 ``l2tp.flags.len`` Length
------- ----- --------------------- ------------------------------------------
0 2 Reserved (must be zero)
------- ----- --------------------- ------------------------------------------
0 4 ``l2tp.flags.seq`` Sequence
------- ----- --------------------- ------------------------------------------
0 5 Reserved (must be zero)
------- ----- --------------------- ------------------------------------------
0 6 ``l2tp.flags.offset`` Offset
------- ----- --------------------- ------------------------------------------
0 7 ``l2tp.flags.prio`` Priority
------- ----- --------------------- ------------------------------------------
1 8 Reserved (must be zero)
------- ----- --------------------- ------------------------------------------
1 12 ``l2tp.version`` Version (``2``)
------- ----- --------------------- ------------------------------------------
2 16 ``l2tp.length`` Length (optional by ``len``)
------- ----- --------------------- ------------------------------------------
4 32 ``l2tp.tunnelid`` Tunnel ID
------- ----- --------------------- ------------------------------------------
6 48 ``l2tp.sessionid`` Session ID
------- ----- --------------------- ------------------------------------------
8 64 ``l2tp.ns`` Sequence Number (optional by ``seq``)
------- ----- --------------------- ------------------------------------------
10 80 ``l2tp.nr`` Next Sequence Number (optional by ``seq``)
------- ----- --------------------- ------------------------------------------
12 96 ``l2tp.offset`` Offset Size (optional by ``offset``)
======= ===== ===================== ==========================================
.. [*] https://en.wikipedia.org/wiki/Layer_2_Tunneling_Protocol
"""
from typing import TYPE_CHECKING
from pcapkit.const.l2tp.type import Type as Enum_Type
from pcapkit.protocols.data.link.l2tp import L2TP as Data_L2TP
from pcapkit.protocols.data.link.l2tp import Flags as Data_Flags
from pcapkit.protocols.link.l2tp import L2TP
from pcapkit.protocols.schema.link.l2tp import L2TP as Schema_L2TP
from pcapkit.utilities.exceptions import ProtocolError, UnsupportedCall
if TYPE_CHECKING:
from enum import IntEnum as StdlibEnum
from typing import Any, NoReturn, Optional, Type
from aenum import IntEnum as AenumEnum
from typing_extensions import Literal
from pcapkit.protocols.protocol import ProtocolBase
from pcapkit.protocols.schema.schema import Schema
__all__ = ['L2TPv2']
# NOTE: ``schema`` and ``data`` are stated here rather than inherited from
# :class:`~pcapkit.protocols.link.l2tp.L2TP`, which declares none: the base is
# version-agnostic and the header schema is not. They must be passed explicitly
# in any case, since
# :meth:`ProtocolBase.__init_subclass__ <pcapkit.protocols.protocol.ProtocolBase.__init_subclass__>`
# resolves an omitted schema by looking the *subclass name* up in
# :mod:`pcapkit.protocols.schema` -- there is no ``L2TPv2`` there -- and assigns
# unconditionally, so omitting them would silently bind ``Schema_Raw``/``Data_Raw``.
[docs]
class L2TPv2(L2TP[Data_L2TP, Schema_L2TP],
schema=Schema_L2TP, data=Data_L2TP):
"""This class implements Layer Two Tunnelling Protocol version 2.
The protocol is dispatched from :attr:`UDP.__proto__
<pcapkit.protocols.transport.udp.UDP.__proto__>` at port 1701.
Only the :rfc:`2661` framing is read: :meth:`read` refuses a datagram whose
version nibble is not ``2``, so a v3 datagram arriving on port 1701
(:rfc:`3931` §4.1.2 shares the port) degrades to
:class:`~pcapkit.protocols.misc.raw.Raw` instead of being reported as v2.
.. versionadded:: 1.5.0
Takes over the version-2 parsing that
:class:`~pcapkit.protocols.link.l2tp.L2TP` carried, and refuses a version
nibble other than ``2`` (:issue:`548`).
Note:
IANA protocol number 115 (``L2TP``) is deliberately left unbound. It
references :rfc:`3931`, i.e. **L2TPv3**, whose session and control
message headers are a different shape -- so the binding waits on an
``L2TPv3`` class rather than on this one. Binding *this* class there does
not work (:issue:`548`): over IP the v3 session header carries no
version nibble at all, so this class cannot recognise that the datagram
is not its own. See :mod:`pcapkit.protocols.link.l2tp` for the
versions not implemented.
The class subclasses :class:`~pcapkit.protocols.link.link.Link` and so
reports ``layer == 'Link'`` even though it is carried inside UDP: a
tunnelling protocol is placed by what it carries. See
:doc:`/contributing/conventions/protocol-layer-placement`.
"""
##########################################################################
# Properties.
##########################################################################
@property
def name(self) -> 'Literal["Layer 2 Tunnelling Protocol version 2"]':
"""Name of current protocol."""
return 'Layer 2 Tunnelling Protocol version 2'
@property
def alias(self) -> 'Literal["L2TPv2"]':
"""Acronym of current protocol."""
return 'L2TPv2'
@property
def version(self) -> 'Literal[2]':
"""Version of current protocol.
Always ``2``: :rfc:`2661` fixes the version nibble, and a datagram
carrying any other value is a different protocol reached through a
different class. c.f. :class:`~pcapkit.protocols.link.l2tp.L2TP`.
This is enforced rather than merely asserted -- :meth:`read` refuses a
datagram whose nibble is not ``2``, so this hard-coded answer cannot
disagree with
:attr:`info.version <pcapkit.protocols.data.link.l2tp.L2TP.version>`
on the same octets.
"""
return 2
@property
def length(self) -> 'int':
"""Header length of current protocol."""
return self._info.hdr_len
@property
def type(self) -> 'Literal["control", "data"]':
"""L2TP type."""
return self._info.flags.type
##########################################################################
# Methods.
##########################################################################
[docs]
def read(self, length: 'Optional[int]' = None, **kwargs: 'Any') -> 'Data_L2TP': # pylint: disable=unused-argument
"""Read Layer Two Tunnelling Protocol.
Structure of L2TP header [:rfc:`2661`]:
.. code-block:: text
0 1 2 3
0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1 2 3 4 5 6 7 8 9 0 1
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
|T|L|x|x|S|x|O|P|x|x|x|x| Ver | Length (opt) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Tunnel ID | Session ID |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Ns (opt) | Nr (opt) |
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
| Offset Size (opt) | Offset pad... (opt)
+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+-+
Args:
length: Length of packet data.
**kwargs: Arbitrary keyword arguments.
Returns:
Parsed packet data.
"""
schema = self.__header__
_flag = schema.flags
# NOTE: :rfc:`2661` §3.1 fixes ``Ver`` at 2 and reserves 1 "to permit
# detection of L2F packets should they arrive intermixed with L2TP
# packets", while :rfc:`3931` uses 3 -- so a datagram carrying any other
# nibble is a different protocol, exactly as ``version`` documents.
# Refuse it rather than parse it: every field after this word has a
# different meaning (or no meaning) in another version, so continuing
# reports a tunnel and session ID assembled out of octets that are
# neither: an :rfc:`3931` §4.1.1 L2TPv3-over-IP datagram would otherwise
# yield ``version=4``, ``tunnelid=0x5678`` and ``sessionid=0xff03``, read
# out of the top half of a Session ID and the first two octets of the PPP
# frame behind it (:issue:`548`). This is also what keeps the hard-coded
# ``Literal[2]`` of ``version`` true, instead of disagreeing with
# ``info.version`` on the same datagram.
if _flag['version'] != 2:
raise ProtocolError(f'{self.alias}: invalid version: {_flag["version"]}')
flags = Data_Flags(
type=Enum_Type(_flag['type']),
len=bool(_flag['len']),
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
hdr_len = 6 + 2 * (flags.len + 2 * flags.seq + flags.offset) + _size
l2tp = Data_L2TP(
flags=flags,
version=_flag['version'],
length=schema.length if flags.len else None,
tunnelid=schema.tunnel_id,
sessionid=schema.session_id,
ns=schema.ns if flags.seq else None,
nr=schema.nr if flags.seq else None,
offset=_size if flags.offset else None,
)
l2tp.__update__([
('hdr_len', hdr_len),
('padding', schema.padding if flags.offset else None),
])
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
# ARP does, rather than on a code read off the wire. Passing the
# remaining length here would dispatch on it as if it were an EtherType,
# resolving to Raw only because a length rarely collides with a
# registered one.
return self._decode_next_layer(l2tp, -1, length - hdr_len)
[docs]
def make(self,
version: 'Literal[2]' = 2,
type: 'Enum_Type | StdlibEnum | AenumEnum | str | int' = Enum_Type.Data,
type_default: 'Optional[int]' = None,
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,
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'',
trailer: 'bytes' = b'',
**kwargs: 'Any') -> 'Schema_L2TP': # pylint: disable=unused-argument
"""Make (construct) packet data.
Args:
version: L2TP version.
type: L2TP type.
type_default: Default value of type.
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.
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.
trailer: Octets after the datagram, outside the Length field;
written as is.
**kwargs: Arbitrary keyword arguments.
Returns:
Constructed packet data.
Raises:
ProtocolError: If ``total_length`` is given with ``length_flag``
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)
# 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
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': length_flag,
'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,
tunnel_id=tunnel_id,
session_id=session_id,
ns=ns,
nr=nr,
offset=offset,
padding=b'' if padding is None else padding,
payload=payload,
trailer=trailer,
)
##########################################################################
# Methods.
##########################################################################
[docs]
@classmethod
def id(cls) -> 'tuple[Literal["L2TP"], Literal["L2TPv2"]]':
"""Index ID of the protocol.
Returns:
Index ID of the protocol -- the canonical family name first, then
this version's own label, as
:meth:`HTTPv1.id <pcapkit.protocols.application.httpv1.HTTP.id>`
does. Element zero is what callers treat as canonical, so both
``protocol='L2TP'`` and ``protocol='L2TPv2'`` select this class.
"""
return ('L2TP', 'L2TPv2')
##########################################################################
# Data models.
##########################################################################
def __length_hint__(self) -> 'Literal[16]':
"""Return an estimated length for the object."""
return 16
[docs]
@classmethod
def __index__(cls) -> 'NoReturn': # pylint: disable=invalid-index-returned
"""Numeral registry index of the protocol.
Raises:
UnsupportedCall: This protocol has no registry entry.
Note:
L2TPv2 is reached through **UDP port 1701**, and a port is not an
``__index__`` value anywhere in this package -- every non-raising
``__index__`` returns a :class:`~pcapkit.const.reg.transtype.TransType`,
:class:`~pcapkit.const.reg.ethertype.EtherType` or
:class:`~pcapkit.const.reg.linktype.LinkType`, bar the PCAP frame and
PCAP-NG block instances that return their own index number, and
:meth:`Application.__index__
<pcapkit.protocols.application.application.Application.__index__>`
raises for exactly this reason. So there is no index to return here.
``L2TPv3`` will be the first class in this family with one, IANA
protocol number 115.
"""
raise UnsupportedCall(f'{cls.__name__!r} object cannot be interpreted as an integer')
##########################################################################
# Utilities.
##########################################################################
[docs]
@classmethod
def _make_data(cls, data: 'Data_L2TP') -> 'dict[str, Any]': # type: ignore[override]
"""Create key-value pairs from ``data`` for protocol construction.
Args:
data: protocol data
Returns:
Key-value pairs for protocol construction.
"""
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
# 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),
# NOTE: And the octets captured past the Length field (:issue:`1455`).
'trailer': data.get('trailer', b''),
}