Page MenuHomeFreeBSD
Differential D31084 Diff 91913 tests/sys/net/routing/rtsock.py

Changeset View

Standalone View

View Options

tests/sys/net/routing/rtsock.py

  • This file was added.
#!/usr/local/bin/python3
from ctypes import *
import socket
import os
import sys
import unittest
from typing import List, Dict, Optional
def roundup2(val: int, num: int) -> int:
if val % num:
return (val | (num - 1)) + 1
else:
return val
class RtConst():
RTM_VERSION = 5
AF_INET = socket.AF_INET
AF_INET6 = socket.AF_INET6
AF_LINK = socket.AF_LINK
RTA_DST = 0x1
RTA_GATEWAY = 0x2
RTA_NETMASK = 0x4
RTA_GENMASK = 0x8
RTA_IFP = 0x10
RTA_IFA = 0x20
RTA_AUTHOR = 0x40
RTA_BRD = 0x80
RTM_ADD = 1
RTM_DELETE = 2
RTM_CHANGE = 3
RTM_GET = 4
RTF_UP = 0x1
RTF_GATEWAY = 0x2
RTF_HOST = 0x4
RTF_REJECT = 0x8
RTF_DYNAMIC = 0x10
RTF_MODIFIED = 0x20
RTF_DONE = 0x40
RTF_XRESOLVE = 0x200
RTF_LLINFO = 0x400
RTF_LLDATA = 0x400
RTF_STATIC = 0x800
RTF_BLACKHOLE = 0x1000
RTF_PROTO2 = 0x4000
RTF_PROTO1 = 0x8000
RTF_PROTO3 = 0x40000
RTF_FIXEDMTU = 0x80000
RTF_PINNED = 0x100000
RTF_LOCAL = 0x200000
RTF_BROADCAST = 0x400000
RTF_MULTICAST = 0x800000
RTF_STICKY = 0x10000000
RTF_RNH_LOCKED = 0x40000000
RTF_GWFLAG_COMPAT = 0x80000000
@staticmethod
def get_props(prefix: str) -> List[str]:
return [n for n in dir(RtConst) if n.startswith(prefix)]
@staticmethod
def get_name(prefix: str, value: int) -> str:
props = RtConst.get_props(prefix)
for prop in props:
if getattr(RtConst, prop) == value:
return prop
return "U:" . str(value)
@staticmethod
def get_bitmask_map(prefix: str, value: int) -> Dict[int, str]:
r1 = {}
props = RtConst.get_props(prefix)
propmap = {getattr(RtConst, prop): prop for prop in props}
v = 1
ret = {}
while value:
if v & value:
if v in propmap:
ret[v] = propmap[v]
else:
ret[v] = hex(v)
value -= v
v *= 2
return ret
@staticmethod
def get_bitmask_str(prefix: str, value: int) -> str:
bmap = RtConst.get_bitmask_map(prefix, value)
return ",".join([v for k, v in bmap.items()])
class RtMetrics(Structure):
_fields_ = [
("rmx_locks", c_ulong),
("rmx_mtu", c_ulong),
("rmx_hopcount", c_ulong),
("rmx_expire", c_ulong),
("rmx_recvpipe", c_ulong),
("rmx_sendpipe", c_ulong),
("rmx_ssthresh", c_ulong),
("rmx_rtt", c_ulong),
("rmx_rttvar", c_ulong),
("rmx_pksent", c_ulong),
("rmx_weight", c_ulong),
("rmx_nhidx", c_ulong),
("rmx_filler", c_ulong * 2),
]
class RtMsgHdr(Structure):
_fields_ = [
("rtm_msglen", c_ushort),
("rtm_version", c_byte),
("rtm_type", c_byte),
("rtm_index", c_ushort),
("_rtm_spare1", c_ushort),
("rtm_flags", c_int),
("rtm_addrs", c_int),
("rtm_pid", c_int),
("rtm_seq", c_int),
("rtm_errno", c_int),
("rtm_fmask", c_int),
("rtm_inits", c_ulong),
("rtm_rmx", RtMetrics),
]
class SockaddrIn(Structure):
_fields_ = [
("sin_len", c_byte),
("sin_family", c_byte),
("sin_port", c_ushort),
("sin_addr", c_uint32),
("sin_zero", c_char * 8),
]
class SockaddrIn6(Structure):
_fields_ = [
("sin6_len", c_byte),
("sin6_family", c_byte),
("sin6_port", c_ushort),
("sin6_flowinfo", c_uint32),
("sin6_addr", c_byte * 16),
("sin6_scope_id", c_uint32),
]
class SockaddrDl(Structure):
_fields_ = [
("sdl_len", c_byte),
("sdl_family", c_byte),
("sdl_index", c_ushort),
("sdl_type", c_byte),
("sdl_nlen", c_byte),
("sdl_alen", c_byte),
("sdl_slen", c_byte),
("sdl_data", c_byte * 8),
]
class SaHelper(object):
@staticmethod
def ip_sa(ip: str) -> bytes:
addr_int = int.from_bytes(socket.inet_pton(2, ip), sys.byteorder)
sin = SockaddrIn(sizeof(SockaddrIn), socket.AF_INET, 0, addr_int)
return bytes(sin)
@staticmethod
def ip6_sa(ip6: str, scopeid: int):
addr_bytes = socket.inet_pton(socket.AF_INET6, ip6)
sin6 = SockaddrIn6(sizeof(SockaddrIn6), socket.AF_INET6, 0, 0, addr_bytes, 0)
Lint: PEP8 E501
Inline Actions

line too long (85 > 79 characters)

Lint: PEP8 E501: line too long (85 > 79 characters)
return bytes(sin6)
@staticmethod
def link_sa(ifindex: Optional[int]=0, iftype: Optional[int]=0):
sa = SockaddrDl(sizeof(SockaddrDl), socket.AF_LINK, c_ushort(ifindex), iftype)
Lint: PEP8 E501
Inline Actions

line too long (86 > 79 characters)

Lint: PEP8 E501: line too long (86 > 79 characters)
return bytes(sa)
class BaseRtsockMessage(object):
def __init__(self, rtm_type):
self.rtm_type = rtm_type
self.ut = unittest.TestCase()
self.sa = SaHelper()
def assertEqual(self, a, b, msg=None):
self.ut.assertEqual(a, b, msg)
def assertNotEqual(self, a, b, msg=None):
self.ut.assertNotEqual(a, b, msg)
class RtsockRtMessage(BaseRtsockMessage):
messages = [RtConst.RTM_ADD, RtConst.RTM_DELETE, RtConst.RTM_CHANGE, RtConst.RTM_GET]
Lint: PEP8 E501
Inline Actions

line too long (89 > 79 characters)

Lint: PEP8 E501: line too long (89 > 79 characters)
def __init__(self, rtm_type, rtm_seq=1, dst_sa=None, mask_sa=None):
super().__init__(rtm_type)
self.rtm_flags = 0
self.rtm_seq = rtm_seq
self._attrs = {}
self.rtm_errno = 0
self.rtm_pid = 0
self._orig_data = None
if dst_sa:
self.add_sa_attr(RtConst.RTA_DST, dst_sa)
if mask_sa:
self.add_sa_attr(RtConst.RTA_NETMASK, mask_sa)
def add_sa_attr(self, attr_type, attr_bytes):
self._attrs[attr_type] = attr_bytes
def add_ip_attr(self, attr_type, ip: str):
self.add_sa_attr(attr_type, self.sa.ip_sa(ip))
def add_ip6_attr(self, attr_type, ip6: str, scopeid: int):
self.add_sa_attr(attr_type, self.sa.ip6_sa(ip6, scopeid))
def add_link_attr(self, attr_type, ifindex: Optional[int]=0):
self.add_sa_attr(attr_type, self.sa.link_sa(ifindex))
def print_sa_inet(self, sa: bytes):
if len(sa) < 8:
raise Exception("IPv4 sa size too small: {}".format(sa))
addr = socket.inet_ntop(socket.AF_INET, sa[4:8])
return "{}".format(addr)
def print_sa_inet6(self, sa: bytes):
if len(sa) < sizeof(SockaddrIn6):
raise Exception("IPv6 sa size too small: {}".format(sa))
addr = socket.inet_ntop(socket.AF_INET6, sa[8:24])
scopeid = struct.unpack(">I", sa[24:28])
return "{} scopeid {}".format(addr, scopeid)
def print_sa_link(self, sa: bytes, hd: Optional[bool] = True):
if len(sa) < sizeof(SockaddrDl):
raise Exception("LINK sa size too small: {}".format(sa))
sdl = SockaddrDl.from_buffer_copy(sa)
if sdl.sdl_index:
ifindex = "link#{} ".format(sdl.sdl_index)
else:
ifindex = ""
if sdl.sdl_nlen:
iface_offset = 8
if sdl.sdl_nlen + iface_offset > len(sa):
raise Exception("LINK sa sdl_nlen {} > total len {}".format(sdl.sdl_nlen, len(sa)))
Lint: PEP8 E501
Inline Actions

line too long (99 > 79 characters)

Lint: PEP8 E501: line too long (99 > 79 characters)
ifname = "ifname:{} ".format(bytes.decode(sa[iface_offset:iface_offset + sdl.sdl_nlen]))
Lint: PEP8 E501
Inline Actions

line too long (100 > 79 characters)

Lint: PEP8 E501: line too long (100 > 79 characters)
else:
ifname = ""
return "{}{}".format(ifindex, ifname)
def print_sa_unknown(self, sa: bytes):
return "unknown_type:{}".format(sa[1])
def print_sa(self, sa: bytes, hd: Optional[bool] = False):
if sa[0] != len(sa):
raise Exception("sa size {} != buffer size {}".format(sa[0], len(sa)))
Lint: PEP8 E501
Inline Actions

line too long (82 > 79 characters)

Lint: PEP8 E501: line too long (82 > 79 characters)
if len(sa) < 2:
raise Exception("sa type {} too short: {}".format(RtConst.get_name("AF_", sa[1]), len(sa)))
Lint: PEP8 E501
Inline Actions

line too long (103 > 79 characters)

Lint: PEP8 E501: line too long (103 > 79 characters)
if sa[1] == socket.AF_INET:
text = self.print_sa_inet(sa)
elif sa[1] == socket.AF_INET6:
text = self.print_sa_inet6(sa)
elif sa[1] == socket.AF_LINK:
text = self.print_sa_link(sa)
else:
text = self.print_sa_unknown(sa)
if hd:
dump = " [{}]".format(sa)
else:
dump = ""
return "{}{}".format(text, dump)
def print_message(self):
# RTM_GET: Report Metrics: len 272, pid: 87839, seq 1, errno 0, flags:<UP,GATEWAY,DONE,STATIC>
Lint: PEP8 E501
Inline Actions

line too long (102 > 79 characters)

Lint: PEP8 E501: line too long (102 > 79 characters)
if self._orig_data:
rtm_len = len(self._orig_data)
else:
rtm_len = len(bytes(self))
print("{}: len {}, pid: {}, seq {}, errno {}, flags: <{}>".format(
RtConst.get_name("RTM_", self.rtm_type),
rtm_len,
self.rtm_pid,
self.rtm_seq,
self.rtm_errno,
RtConst.get_bitmask_str("RTF_", self.rtm_flags)
))
rtm_addrs = sum(list(self._attrs.keys()))
print("Addrs: <{}>".format(RtConst.get_bitmask_str("RTA_", rtm_addrs)))
for attr in sorted(self._attrs.keys()):
sa_data = self.print_sa(self._attrs[attr])
print(" {}: {}".format(RtConst.get_name("RTA_", attr), sa_data))
@staticmethod
def verify_sa_inet(sa_data):
if len(sa_data) < 8:
raise Exception("IPv4 sa size too small: {}".format(sa_data))
if sa_data[0] > len(sa_data):
raise Exception("IPv4 sin_len too big: {} vs sa size {}: {}".format(sa_data[0], len(sa_data), sa_data))
Lint: PEP8 E501
Inline Actions

line too long (115 > 79 characters)

Lint: PEP8 E501: line too long (115 > 79 characters)
sin = SockaddrIn.from_buffer_copy(sa_data)
self.assertEqual(sin.sin_port, 0)
assert sin.sin_zero == [0] * 8
def compare_sa(self, sa_type, sa_data):
if len(sa_data) < 4:
raise Exception("sa_len for type {} too short: {}".format())
our_sa = self._attrs[sa_type]
self.assertEqual(len(sa_data), len(our_sa))
self.assertEqual(our_sa, sa_data)
def verify(self, rtm_type: int, rtm_sa):
assert self.rtm_type == rtm_type
assert self.rtm_errno == 0
hdr = RtMsgHdr.from_buffer_copy(self._orig_data)
assert hdr._rtm_spare1 == 0
for sa_type, sa_data in rtm_sa.items():
if sa_type not in self._attrs:
raise Exception("SA type {} not present".format(RtConst.get_name("RTA_", sa_type)))
Lint: PEP8 E501
Inline Actions

line too long (99 > 79 characters)

Lint: PEP8 E501: line too long (99 > 79 characters)
self.compare_sa(sa_type, sa_data)
@classmethod
def from_bytes(cls, data: bytes):
if len(data) < sizeof(RtMsgHdr):
raise Exception("messages size {} is less than expected {}".format(len(data), sizeof(RtMsgHdr)))
Lint: PEP8 E501
Inline Actions

line too long (108 > 79 characters)

Lint: PEP8 E501: line too long (108 > 79 characters)
hdr = RtMsgHdr.from_buffer_copy(data)
self = cls(hdr.rtm_type)
self.rtm_flags = hdr.rtm_flags
self.rtm_seq = hdr.rtm_seq
self.rtm_errno = hdr.rtm_errno
self.rtm_pid = hdr.rtm_pid
self.rtm_len = len(data)
self._orig_data = data
off = sizeof(RtMsgHdr)
v = 1
addrs_mask = hdr.rtm_addrs
while addrs_mask:
if addrs_mask & v:
addrs_mask -= v
if off + data[off] > len(data):
raise Exception("SA sizeof for {} > total message length: {}+{} > {}".format(
Lint: PEP8 E501
Inline Actions

line too long (97 > 79 characters)

Lint: PEP8 E501: line too long (97 > 79 characters)
RtConst.get_name("RTA_", v), off, data[off], len(data)))
Lint: PEP8 E501
Inline Actions

line too long (80 > 79 characters)

Lint: PEP8 E501: line too long (80 > 79 characters)
self._attrs[v] = data[off:off + data[off]]
off += roundup2(data[off], 4)
v *= 2
return self
def __bytes__(self):
sz = sizeof(RtMsgHdr)
addrs_mask = 0
for k, v in self._attrs.items():
sz += roundup2(len(v), 4)
addrs_mask += k
hdr = RtMsgHdr(
rtm_msglen=sz,
rtm_version=RtConst.RTM_VERSION,
rtm_type=self.rtm_type,
rtm_flags=self.rtm_flags,
rtm_seq=self.rtm_seq,
rtm_addrs=addrs_mask,
)
buf = bytearray(sz)
buf[0:sizeof(RtMsgHdr)] = hdr
off = sizeof(RtMsgHdr)
for attr in sorted(self._attrs.keys()):
v = self._attrs[attr]
sa_len = len(v)
buf[off:off + sa_len] = v
off += roundup2(len(v), 4)
return bytes(buf)
class Rtsock():
def __init__(self):
self.rtsock_fd = self._setup_rtsock()
self.rtm_seq = 1
self.msgmap = self.build_msgmap()
def build_msgmap(self):
classes = [RtsockRtMessage]
xmap = {}
for cls in classes:
for message in cls.messages:
xmap[message] = cls
return xmap
def get_seq(self):
ret = self.rtm_seq
self.rtm_seq += 1
return ret
def _setup_rtsock(self) -> int:
s = socket.socket(socket.AF_ROUTE, socket.SOCK_RAW, socket.AF_UNSPEC)
s.setsockopt(socket.SOL_SOCKET, socket.SO_USELOOPBACK, 1)
return s
def write_message(self, msg):
print("vvvvvvvv OUT vvvvvvvv")
msg.print_message()
msg_bytes = bytes(msg)
try:
ret = os.write(self.rtsock_fd.fileno(), bytes(msg))
except Exception as e:
print("write({}) -> {}".format(len(msg_bytes), e))
def parse_message(self, data: bytes):
if len(data) < 4:
raise Exception("Short read from rtsock: {} bytes".format(len(data)))
Lint: PEP8 E501
Inline Actions

line too long (81 > 79 characters)

Lint: PEP8 E501: line too long (81 > 79 characters)
rtm_type = data[4]
if rtm_type not in self.msgmap:
return None
def write_data(self, data: bytes):
self.rtsock_fd.send(data)
def read_data(self, seq: Optional[int] = None) -> bytes:
while True:
data = self.rtsock_fd.recv(4096)
if seq is None:
break
if len(data) > sizeof(RtMsgHdr):
hdr = RtMsgHdr.from_buffer_copy(data)
if hdr.rtm_seq == seq:
break
return data
def read_message(self) -> bytes:
data = self.read_data()
return self.parse_message(data)