| Server IP : 88.223.91.163 / Your IP : 216.73.216.122 Web Server : LiteSpeed System : Linux id-dci-web1986.main-hosting.eu 5.14.0-611.26.1.el9_7.x86_64 #1 SMP PREEMPT_DYNAMIC Thu Jan 29 05:24:47 EST 2026 x86_64 User : u686484674 ( 686484674) PHP Version : 8.0.30 Disable Function : system, exec, shell_exec, passthru, mysql_list_dbs, ini_alter, dl, symlink, link, chgrp, leak, popen, apache_child_terminate, virtual, mb_send_mail MySQL : OFF | cURL : ON | WGET : ON | Perl : OFF | Python : OFF | Sudo : OFF | Pkexec : OFF Directory : /proc/self/root/opt/./alt/python311/lib/python3.11/site-packages/pyroute2/protocols/ |
Upload File : |
import struct
from socket import AF_INET, inet_ntop, inet_pton
from typing import (
Any,
Callable,
ClassVar,
Literal,
NamedTuple,
Optional,
Union,
)
from pyroute2.common import basestring, hexdump
#
# IEEE = 802.3 Ethernet magic constants. The frame sizes omit
# the preamble and FCS/CRC (frame check sequence).
#
ETH_ALEN = 6 # Octets in one ethernet addr
ETH_HLEN = 14 # Total octets in header.
ETH_ZLEN = 60 # Min. octets in frame sans FCS
ETH_DATA_LEN = 1500 # Max. octets in payload
ETH_FRAME_LEN = 1514 # Max. octets in frame sans FCS
ETH_FCS_LEN = 4 # Octets in the FCS
#
# These are the defined Ethernet Protocol ID's.
#
ETH_P_LOOP = 0x0060 # Ethernet Loopback packet
ETH_P_PUP = 0x0200 # Xerox PUP packet
ETH_P_PUPAT = 0x0201 # Xerox PUP Addr Trans packet
ETH_P_IP = 0x0800 # Internet Protocol packet
ETH_P_X25 = 0x0805 # CCITT X.25
ETH_P_ARP = 0x0806 # Address Resolution packet
ETH_P_BPQ = 0x08FF # G8BPQ AX.25 Ethernet Packet
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
ETH_P_IEEEPUP = 0x0A00 # Xerox IEEE802.3 PUP packet
ETH_P_IEEEPUPAT = 0x0A01 # Xerox IEEE802.3 PUP Addr Trans packet
ETH_P_DEC = 0x6000 # DEC Assigned proto
ETH_P_DNA_DL = 0x6001 # DEC DNA Dump/Load
ETH_P_DNA_RC = 0x6002 # DEC DNA Remote Console
ETH_P_DNA_RT = 0x6003 # DEC DNA Routing
ETH_P_LAT = 0x6004 # DEC LAT
ETH_P_DIAG = 0x6005 # DEC Diagnostics
ETH_P_CUST = 0x6006 # DEC Customer use
ETH_P_SCA = 0x6007 # DEC Systems Comms Arch
ETH_P_TEB = 0x6558 # Trans Ether Bridging
ETH_P_RARP = 0x8035 # Reverse Addr Res packet
ETH_P_ATALK = 0x809B # Appletalk DDP
ETH_P_AARP = 0x80F3 # Appletalk AARP
ETH_P_8021Q = 0x8100 # = 802.1Q VLAN Extended Header
ETH_P_IPX = 0x8137 # IPX over DIX
ETH_P_IPV6 = 0x86DD # IPv6 over bluebook
ETH_P_PAUSE = 0x8808 # IEEE Pause frames. See = 802.3 = 31B
ETH_P_SLOW = 0x8809 # Slow Protocol. See = 802.3ad = 43B
ETH_P_WCCP = 0x883E # Web-cache coordination protocol
# defined in draft-wilson-wrec-wccp-v2-00.txt
ETH_P_PPP_DISC = 0x8863 # PPPoE discovery messages
ETH_P_PPP_SES = 0x8864 # PPPoE session messages
ETH_P_MPLS_UC = 0x8847 # MPLS Unicast traffic
ETH_P_MPLS_MC = 0x8848 # MPLS Multicast traffic
ETH_P_ATMMPOA = 0x884C # MultiProtocol Over ATM
ETH_P_LINK_CTL = 0x886C # HPNA, wlan link local tunnel
ETH_P_ATMFATE = 0x8884 # Frame-based ATM Transport over Ethernet
ETH_P_PAE = 0x888E # Port Access Entity (IEEE = 802.1X)
ETH_P_AOE = 0x88A2 # ATA over Ethernet
ETH_P_8021AD = 0x88A8 # = 802.1ad Service VLAN
ETH_P_802_EX1 = 0x88B5 # = 802.1 Local Experimental = 1.
ETH_P_TIPC = 0x88CA # TIPC
ETH_P_8021AH = 0x88E7 # = 802.1ah Backbone Service Tag
ETH_P_1588 = 0x88F7 # IEEE = 1588 Timesync
ETH_P_FCOE = 0x8906 # Fibre Channel over Ethernet
ETH_P_TDLS = 0x890D # TDLS
ETH_P_FIP = 0x8914 # FCoE Initialization Protocol
ETH_P_QINQ1 = 0x9100 # deprecated QinQ VLAN
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
ETH_P_QINQ2 = 0x9200 # deprecated QinQ VLAN
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
ETH_P_QINQ3 = 0x9300 # deprecated QinQ VLAN
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
ETH_P_EDSA = 0xDADA # Ethertype DSA
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
ETH_P_AF_IUCV = 0xFBFB # IBM af_iucv
# ^^^ [ NOT AN OFFICIALLY REGISTERED ID ]
#
# Non DIX types. Won't clash for = 1500 types.
#
ETH_P_802_3 = 0x0001 # Dummy type for = 802.3 frames
ETH_P_AX25 = 0x0002 # Dummy protocol id for AX.25
ETH_P_ALL = 0x0003 # Every packet (be careful!!!)
ETH_P_802_2 = 0x0004 # = 802.2 frames
ETH_P_SNAP = 0x0005 # Internal only
ETH_P_DDCMP = 0x0006 # DEC DDCMP: Internal only
ETH_P_WAN_PPP = 0x0007 # Dummy type for WAN PPP frames*/
ETH_P_PPP_MP = 0x0008 # Dummy type for PPP MP frames
ETH_P_LOCALTALK = 0x0009 # Localtalk pseudo type
ETH_P_CAN = 0x000C # Controller Area Network
ETH_P_PPPTALK = 0x0010 # Dummy type for Atalk over PPP*/
ETH_P_TR_802_2 = 0x0011 # = 802.2 frames
ETH_P_MOBITEX = 0x0015 # Mobitex (kaz@cafe.net)
ETH_P_CONTROL = 0x0016 # Card specific control frames
ETH_P_IRDA = 0x0017 # Linux-IrDA
ETH_P_ECONET = 0x0018 # Acorn Econet
ETH_P_HDLC = 0x0019 # HDLC frames
ETH_P_ARCNET = 0x001A # = 1A for ArcNet :-)
ETH_P_DSA = 0x001B # Distributed Switch Arch.
ETH_P_TRAILER = 0x001C # Trailer switch tagging
ETH_P_PHONET = 0x00F5 # Nokia Phonet frames
ETH_P_IEEE802154 = 0x00F6 # IEEE802.15.4 frame
ETH_P_CAIF = 0x00F7 # ST-Ericsson CAIF protocol
def _decode_mac(value: Union[bytes, tuple[int, ...]]) -> str:
if isinstance(value, tuple):
value = bytes(value)
if not len(value) == 6:
raise ValueError('Invalid mac address, must be 6 bytes long')
return value.hex(sep=':')
def _encode_mac(value: str) -> list[int]:
return [int(i, 16) for i in value.split(':')]
class Policy(NamedTuple):
'''Used in option/msg subclasses.
The format is either a struct format or 'string'.
encode is called with the python value to serialize it.
decode is called with the result(s) of the unpacking to decode values.
'''
format: str
# FIXME: there are so many Any here it's nearly useless,
# encode/decode callbacks should have a clean definition instead
encode: Callable[[Any], Any] = lambda x: x
decode: Union[Callable[[Any], Any], Callable[[tuple[Any, ...]], Any]] = (
lambda x: x
)
class msg(dict):
fields: ClassVar[
tuple[Union[tuple[str, str, Union[int, bytes]], tuple[str, str]], ...]
] = ()
types: ClassVar[dict[str, Union[str, Policy]]] = {
'uint8': 'B',
'uint16': 'H',
'uint32': 'I',
'be16': '>H',
'be32': '>I',
'ip4addr': Policy(
format='4s',
decode=lambda x: inet_ntop(AF_INET, x),
encode=lambda x: [inet_pton(AF_INET, x)],
),
'l2addr': Policy(format='6B', decode=_decode_mac, encode=_encode_mac),
'l2paddr': Policy(
format='6B10s',
decode=lambda x: _decode_mac(x[:6]),
encode=lambda x: _encode_mac(x) + [10 * b'\x00'],
),
}
def __init__(
self,
content: Optional[dict] = None,
buf: bytes = b'',
offset: int = 0,
value: Any = None,
) -> None:
content = content or {}
dict.__init__(self, content)
self.buf = buf
self.offset = offset
self.value = value
self._fields_names: tuple[str, ...] = ()
self._register_fields()
def _register_fields(self) -> None:
self._fields_names = tuple([x[0] for x in self.fields])
def _get_routine(
self, mode: Literal['encode', 'decode'], fmt: str
) -> tuple[str, Callable[[Any], Any]]:
real_fmt = self.types.get(fmt, fmt)
if isinstance(real_fmt, Policy):
return (real_fmt.format, getattr(real_fmt, mode, lambda x: x))
else:
return (real_fmt, lambda x: x)
def reset(self) -> None:
self.buf = b''
def decode(self) -> 'msg':
for field in self.fields:
name, sfmt = field[:2]
fmt, routine = self._get_routine('decode', sfmt)
size = struct.calcsize(fmt)
try:
value = struct.unpack_from(fmt, self.buf, self.offset)
except struct.error as err:
raise ValueError(
f'Cannot decode {type(self).__name__} {name}: {err}'
)
if len(value) == 1:
value = value[0]
if isinstance(value, bytes) and sfmt[-1] == 's':
value = value.lstrip(b'\x00')
try:
value = value.decode('utf-8')
except UnicodeDecodeError:
value = hexdump(value)
self[name] = routine(value)
self.offset += size
return self
def encode(self):
self._register_fields()
for field in self.fields:
name, fmt = field[:2]
default = b'\x00' if len(field) <= 2 else field[2]
fmt, routine = self._get_routine('encode', fmt)
# special case: string
if fmt == 'string':
self.buf += routine(self[name])[0]
else:
size = struct.calcsize(fmt)
if self[name] is None:
if not isinstance(default, basestring):
self.buf += struct.pack(fmt, default)
else:
self.buf += default * (size // len(default))
else:
value = routine(self[name])
if not isinstance(value, (set, tuple, list)):
value = [value]
self.buf += struct.pack(fmt, *value)
return self
def __getitem__(self, key):
try:
return dict.__getitem__(self, key)
except KeyError:
if key in self._fields_names:
return None
raise
class ethmsg(msg):
fields = (('dst', 'l2addr'), ('src', 'l2addr'), ('type', 'be16'))
class ip6msg(msg):
fields = (
('version', 'uint8', 6 << 4),
('_flow0', 'uint8'),
('_flow1', 'uint8'),
('_flow2', 'uint8'),
('plen', 'uin16'),
('next_header', 'uint8'),
('hop_limit', 'uint8'),
('src', 'ip6addr'),
('dst', 'ip6addr'),
)
class ip4msg(msg):
fields = (
('verlen', 'uint8', 0x45),
('dsf', 'uint8'),
('len', 'be16'),
('id', 'be16'),
('flags', 'uint16'),
('ttl', 'uint8', 128),
('proto', 'uint8'),
('csum', 'be16'),
('src', 'ip4addr'),
('dst', 'ip4addr'),
)
class udp4_pseudo_header(msg):
fields = (
('src', 'ip4addr'),
('dst', 'ip4addr'),
('pad', 'uint8'),
('proto', 'uint8', 17),
('len', 'be16'),
)
class udpmsg(msg):
fields = (
('sport', 'be16'),
('dport', 'be16'),
('len', 'be16'),
('csum', 'be16'),
)