Date: Sat, 1 Apr 2023 19:36:55 GMT From: "Alexander V. Chernikov" <melifaro@FreeBSD.org> To: src-committers@FreeBSD.org, dev-commits-src-all@FreeBSD.org, dev-commits-src-main@FreeBSD.org Subject: git: fee65b7e21ee - main - tests: split netlink.py into multiple files to impove maintainability. Message-ID: <202304011936.331JatO2013549@gitrepo.freebsd.org>
next in thread | raw e-mail | index | archive | help
The branch main has been updated by melifaro: URL: https://cgit.FreeBSD.org/src/commit/?id=fee65b7e21eeeb625dcaba8cf2b9e0cf5af83498 commit fee65b7e21eeeb625dcaba8cf2b9e0cf5af83498 Author: Alexander V. Chernikov <melifaro@FreeBSD.org> AuthorDate: 2023-03-31 09:55:36 +0000 Commit: Alexander V. Chernikov <melifaro@FreeBSD.org> CommitDate: 2023-04-01 19:31:38 +0000 tests: split netlink.py into multiple files to impove maintainability. This diff does not contain any functional changes. Its sole purpose is splitting netlink.py into smaller chunks. The new code simplifies the upcoming generic netlink support introduction. MFC after: 2 weeks Differential Revision: https://reviews.freebsd.org/D39365 --- tests/atf_python/sys/Makefile | 2 +- tests/atf_python/sys/net/Makefile | 2 +- tests/atf_python/sys/net/netlink.py | 1728 ------------------------- tests/atf_python/sys/netlink/Makefile | 11 + tests/atf_python/sys/netlink/__init__.py | 0 tests/atf_python/sys/netlink/attrs.py | 287 ++++ tests/atf_python/sys/netlink/base_headers.py | 65 + tests/atf_python/sys/netlink/message.py | 201 +++ tests/atf_python/sys/netlink/netlink.py | 368 ++++++ tests/atf_python/sys/netlink/netlink_route.py | 727 +++++++++++ tests/atf_python/sys/netlink/utils.py | 74 ++ tests/sys/netlink/test_nl_core.py | 4 +- tests/sys/netlink/test_rtnl_iface.py | 32 +- tests/sys/netlink/test_rtnl_ifaddr.py | 19 +- tests/sys/netlink/test_rtnl_neigh.py | 12 +- tests/sys/netlink/test_rtnl_route.py | 22 +- 16 files changed, 1778 insertions(+), 1776 deletions(-) diff --git a/tests/atf_python/sys/Makefile b/tests/atf_python/sys/Makefile index ff4cf17b85d2..540c3803ada5 100644 --- a/tests/atf_python/sys/Makefile +++ b/tests/atf_python/sys/Makefile @@ -3,7 +3,7 @@ .PATH: ${.CURDIR} FILES= __init__.py -SUBDIR= net +SUBDIR= net netlink .include <bsd.own.mk> FILESDIR= ${TESTSBASE}/atf_python/sys diff --git a/tests/atf_python/sys/net/Makefile b/tests/atf_python/sys/net/Makefile index 63efefd27142..05b1d8afe863 100644 --- a/tests/atf_python/sys/net/Makefile +++ b/tests/atf_python/sys/net/Makefile @@ -2,7 +2,7 @@ .PATH: ${.CURDIR} -FILES= __init__.py netlink.py rtsock.py tools.py vnet.py +FILES= __init__.py rtsock.py tools.py vnet.py .include <bsd.own.mk> FILESDIR= ${TESTSBASE}/atf_python/sys/net diff --git a/tests/atf_python/sys/net/netlink.py b/tests/atf_python/sys/net/netlink.py deleted file mode 100644 index 053eb432e915..000000000000 --- a/tests/atf_python/sys/net/netlink.py +++ /dev/null @@ -1,1728 +0,0 @@ -#!/usr/local/bin/python3 -import os -import socket -import struct -import sys -import unittest -from ctypes import c_int -from ctypes import c_ubyte -from ctypes import c_uint -from ctypes import c_ushort -from ctypes import sizeof -from ctypes import Structure -from enum import auto -from enum import Enum -from typing import Any -from typing import Dict -from typing import List -from typing import NamedTuple - - -def roundup2(val: int, num: int) -> int: - if val % num: - return (val | (num - 1)) + 1 - else: - return val - - -def align4(val: int) -> int: - return roundup2(val, 4) - - -def enum_or_int(val) -> int: - if isinstance(val, Enum): - return val.value - return val - - -class SockaddrNl(Structure): - _fields_ = [ - ("nl_len", c_ubyte), - ("nl_family", c_ubyte), - ("nl_pad", c_ushort), - ("nl_pid", c_uint), - ("nl_groups", c_uint), - ] - - -class Nlmsghdr(Structure): - _fields_ = [ - ("nlmsg_len", c_uint), - ("nlmsg_type", c_ushort), - ("nlmsg_flags", c_ushort), - ("nlmsg_seq", c_uint), - ("nlmsg_pid", c_uint), - ] - - -class Nlmsgdone(Structure): - _fields_ = [ - ("error", c_int), - ] - - -class Nlmsgerr(Structure): - _fields_ = [ - ("error", c_int), - ("msg", Nlmsghdr), - ] - - -class NlErrattrType(Enum): - NLMSGERR_ATTR_UNUSED = 0 - NLMSGERR_ATTR_MSG = auto() - NLMSGERR_ATTR_OFFS = auto() - NLMSGERR_ATTR_COOKIE = auto() - NLMSGERR_ATTR_POLICY = auto() - - -class RtattrType(Enum): - RTA_UNSPEC = 0 - RTA_DST = 1 - RTA_SRC = 2 - RTA_IIF = 3 - RTA_OIF = 4 - RTA_GATEWAY = 5 - RTA_PRIORITY = 6 - RTA_PREFSRC = 7 - RTA_METRICS = 8 - RTA_MULTIPATH = 9 - # RTA_PROTOINFO = 10 - RTA_KNH_ID = 10 - RTA_FLOW = 11 - RTA_CACHEINFO = 12 - RTA_SESSION = 13 - # RTA_MP_ALGO = 14 - RTA_RTFLAGS = 14 - RTA_TABLE = 15 - RTA_MARK = 16 - RTA_MFC_STATS = 17 - RTA_VIA = 18 - RTA_NEWDST = 19 - RTA_PREF = 20 - RTA_ENCAP_TYPE = 21 - RTA_ENCAP = 22 - RTA_EXPIRES = 23 - RTA_PAD = 24 - RTA_UID = 25 - RTA_TTL_PROPAGATE = 26 - RTA_IP_PROTO = 27 - RTA_SPORT = 28 - RTA_DPORT = 29 - RTA_NH_ID = 30 - - -class NlMsgType(Enum): - NLMSG_NOOP = 1 - NLMSG_ERROR = 2 - NLMSG_DONE = 3 - NLMSG_OVERRUN = 4 - - -class NlRtMsgType(Enum): - RTM_NEWLINK = 16 - RTM_DELLINK = 17 - RTM_GETLINK = 18 - RTM_SETLINK = 19 - RTM_NEWADDR = 20 - RTM_DELADDR = 21 - RTM_GETADDR = 22 - RTM_NEWROUTE = 24 - RTM_DELROUTE = 25 - RTM_GETROUTE = 26 - RTM_NEWNEIGH = 28 - RTM_DELNEIGH = 29 - RTM_GETNEIGH = 30 - RTM_NEWRULE = 32 - RTM_DELRULE = 33 - RTM_GETRULE = 34 - RTM_NEWQDISC = 36 - RTM_DELQDISC = 37 - RTM_GETQDISC = 38 - RTM_NEWTCLASS = 40 - RTM_DELTCLASS = 41 - RTM_GETTCLASS = 42 - RTM_NEWTFILTER = 44 - RTM_DELTFILTER = 45 - RTM_GETTFILTER = 46 - RTM_NEWACTION = 48 - RTM_DELACTION = 49 - RTM_GETACTION = 50 - RTM_NEWPREFIX = 52 - RTM_GETMULTICAST = 58 - RTM_GETANYCAST = 62 - RTM_NEWNEIGHTBL = 64 - RTM_GETNEIGHTBL = 66 - RTM_SETNEIGHTBL = 67 - RTM_NEWNDUSEROPT = 68 - RTM_NEWADDRLABEL = 72 - RTM_DELADDRLABEL = 73 - RTM_GETADDRLABEL = 74 - RTM_GETDCB = 78 - RTM_SETDCB = 79 - RTM_NEWNETCONF = 80 - RTM_GETNETCONF = 82 - RTM_NEWMDB = 84 - RTM_DELMDB = 85 - RTM_GETMDB = 86 - RTM_NEWNSID = 88 - RTM_DELNSID = 89 - RTM_GETNSID = 90 - RTM_NEWSTATS = 92 - RTM_GETSTATS = 94 - - -class RtAttr(Structure): - _fields_ = [ - ("rta_len", c_ushort), - ("rta_type", c_ushort), - ] - - -class RtMsgHdr(Structure): - _fields_ = [ - ("rtm_family", c_ubyte), - ("rtm_dst_len", c_ubyte), - ("rtm_src_len", c_ubyte), - ("rtm_tos", c_ubyte), - ("rtm_table", c_ubyte), - ("rtm_protocol", c_ubyte), - ("rtm_scope", c_ubyte), - ("rtm_type", c_ubyte), - ("rtm_flags", c_uint), - ] - - -class RtMsgFlags(Enum): - RTM_F_NOTIFY = 0x100 - RTM_F_CLONED = 0x200 - RTM_F_EQUALIZE = 0x400 - RTM_F_PREFIX = 0x800 - RTM_F_LOOKUP_TABLE = 0x1000 - RTM_F_FIB_MATCH = 0x2000 - RTM_F_OFFLOAD = 0x4000 - RTM_F_TRAP = 0x8000 - RTM_F_OFFLOAD_FAILED = 0x20000000 - - -class AddressFamilyLinux(Enum): - AF_INET = socket.AF_INET - AF_INET6 = socket.AF_INET6 - AF_NETLINK = 16 - - -class AddressFamilyBsd(Enum): - AF_INET = socket.AF_INET - AF_INET6 = socket.AF_INET6 - AF_NETLINK = 38 - - -class NlmBaseFlags(Enum): - NLM_F_REQUEST = 0x01 - NLM_F_MULTI = 0x02 - NLM_F_ACK = 0x04 - NLM_F_ECHO = 0x08 - NLM_F_DUMP_INTR = 0x10 - NLM_F_DUMP_FILTERED = 0x20 - - -# XXX: in python3.8 it is possible to -# class NlmGetFlags(Enum, NlmBaseFlags): - - -class NlmGetFlags(Enum): - NLM_F_ROOT = 0x100 - NLM_F_MATCH = 0x200 - NLM_F_ATOMIC = 0x400 - - -class NlmNewFlags(Enum): - NLM_F_REPLACE = 0x100 - NLM_F_EXCL = 0x200 - NLM_F_CREATE = 0x400 - NLM_F_APPEND = 0x800 - - -class NlmDeleteFlags(Enum): - NLM_F_NONREC = 0x100 - - -class NlmAckFlags(Enum): - NLM_F_CAPPED = 0x100 - NLM_F_ACK_TLVS = 0x200 - - -class RtScope(Enum): - RT_SCOPE_UNIVERSE = 0 - RT_SCOPE_SITE = 200 - RT_SCOPE_LINK = 253 - RT_SCOPE_HOST = 254 - RT_SCOPE_NOWHERE = 255 - - -class RtType(Enum): - RTN_UNSPEC = 0 - RTN_UNICAST = auto() - RTN_LOCAL = auto() - RTN_BROADCAST = auto() - RTN_ANYCAST = auto() - RTN_MULTICAST = auto() - RTN_BLACKHOLE = auto() - RTN_UNREACHABLE = auto() - RTN_PROHIBIT = auto() - RTN_THROW = auto() - RTN_NAT = auto() - RTN_XRESOLVE = auto() - - -class RtProto(Enum): - RTPROT_UNSPEC = 0 - RTPROT_REDIRECT = 1 - RTPROT_KERNEL = 2 - RTPROT_BOOT = 3 - RTPROT_STATIC = 4 - RTPROT_GATED = 8 - RTPROT_RA = 9 - RTPROT_MRT = 10 - RTPROT_ZEBRA = 11 - RTPROT_BIRD = 12 - RTPROT_DNROUTED = 13 - RTPROT_XORP = 14 - RTPROT_NTK = 15 - RTPROT_DHCP = 16 - RTPROT_MROUTED = 17 - RTPROT_KEEPALIVED = 18 - RTPROT_BABEL = 42 - RTPROT_OPENR = 99 - RTPROT_BGP = 186 - RTPROT_ISIS = 187 - RTPROT_OSPF = 188 - RTPROT_RIP = 189 - RTPROT_EIGRP = 192 - - -class NlRtaxType(Enum): - RTAX_UNSPEC = 0 - RTAX_LOCK = auto() - RTAX_MTU = auto() - RTAX_WINDOW = auto() - RTAX_RTT = auto() - RTAX_RTTVAR = auto() - RTAX_SSTHRESH = auto() - RTAX_CWND = auto() - RTAX_ADVMSS = auto() - RTAX_REORDERING = auto() - RTAX_HOPLIMIT = auto() - RTAX_INITCWND = auto() - RTAX_FEATURES = auto() - RTAX_RTO_MIN = auto() - RTAX_INITRWND = auto() - RTAX_QUICKACK = auto() - RTAX_CC_ALGO = auto() - RTAX_FASTOPEN_NO_COOKIE = auto() - - -class RtFlagsBSD(Enum): - 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 - - -class NlRtGroup(Enum): - RTNLGRP_NONE = 0 - RTNLGRP_LINK = auto() - RTNLGRP_NOTIFY = auto() - RTNLGRP_NEIGH = auto() - RTNLGRP_TC = auto() - RTNLGRP_IPV4_IFADDR = auto() - RTNLGRP_IPV4_MROUTE = auto() - RTNLGRP_IPV4_ROUTE = auto() - RTNLGRP_IPV4_RULE = auto() - RTNLGRP_IPV6_IFADDR = auto() - RTNLGRP_IPV6_MROUTE = auto() - RTNLGRP_IPV6_ROUTE = auto() - RTNLGRP_IPV6_IFINFO = auto() - RTNLGRP_DECnet_IFADDR = auto() - RTNLGRP_NOP2 = auto() - RTNLGRP_DECnet_ROUTE = auto() - RTNLGRP_DECnet_RULE = auto() - RTNLGRP_NOP4 = auto() - RTNLGRP_IPV6_PREFIX = auto() - RTNLGRP_IPV6_RULE = auto() - RTNLGRP_ND_USEROPT = auto() - RTNLGRP_PHONET_IFADDR = auto() - RTNLGRP_PHONET_ROUTE = auto() - RTNLGRP_DCB = auto() - RTNLGRP_IPV4_NETCONF = auto() - RTNLGRP_IPV6_NETCONF = auto() - RTNLGRP_MDB = auto() - RTNLGRP_MPLS_ROUTE = auto() - RTNLGRP_NSID = auto() - RTNLGRP_MPLS_NETCONF = auto() - RTNLGRP_IPV4_MROUTE_R = auto() - RTNLGRP_IPV6_MROUTE_R = auto() - RTNLGRP_NEXTHOP = auto() - RTNLGRP_BRVLAN = auto() - - -class IfinfoMsg(Structure): - _fields_ = [ - ("ifi_family", c_ubyte), - ("__ifi_pad", c_ubyte), - ("ifi_type", c_ushort), - ("ifi_index", c_int), - ("ifi_flags", c_uint), - ("ifi_change", c_uint), - ] - - -class IflattrType(Enum): - IFLA_UNSPEC = 0 - IFLA_ADDRESS = auto() - IFLA_BROADCAST = auto() - IFLA_IFNAME = auto() - IFLA_MTU = auto() - IFLA_LINK = auto() - IFLA_QDISC = auto() - IFLA_STATS = auto() - IFLA_COST = auto() - IFLA_PRIORITY = auto() - IFLA_MASTER = auto() - IFLA_WIRELESS = auto() - IFLA_PROTINFO = auto() - IFLA_TXQLEN = auto() - IFLA_MAP = auto() - IFLA_WEIGHT = auto() - IFLA_OPERSTATE = auto() - IFLA_LINKMODE = auto() - IFLA_LINKINFO = auto() - IFLA_NET_NS_PID = auto() - IFLA_IFALIAS = auto() - IFLA_NUM_VF = auto() - IFLA_VFINFO_LIST = auto() - IFLA_STATS64 = auto() - IFLA_VF_PORTS = auto() - IFLA_PORT_SELF = auto() - IFLA_AF_SPEC = auto() - IFLA_GROUP = auto() - IFLA_NET_NS_FD = auto() - IFLA_EXT_MASK = auto() - IFLA_PROMISCUITY = auto() - IFLA_NUM_TX_QUEUES = auto() - IFLA_NUM_RX_QUEUES = auto() - IFLA_CARRIER = auto() - IFLA_PHYS_PORT_ID = auto() - IFLA_CARRIER_CHANGES = auto() - IFLA_PHYS_SWITCH_ID = auto() - IFLA_LINK_NETNSID = auto() - IFLA_PHYS_PORT_NAME = auto() - IFLA_PROTO_DOWN = auto() - IFLA_GSO_MAX_SEGS = auto() - IFLA_GSO_MAX_SIZE = auto() - IFLA_PAD = auto() - IFLA_XDP = auto() - IFLA_EVENT = auto() - IFLA_NEW_NETNSID = auto() - IFLA_IF_NETNSID = auto() - IFLA_CARRIER_UP_COUNT = auto() - IFLA_CARRIER_DOWN_COUNT = auto() - IFLA_NEW_IFINDEX = auto() - IFLA_MIN_MTU = auto() - IFLA_MAX_MTU = auto() - IFLA_PROP_LIST = auto() - IFLA_ALT_IFNAME = auto() - IFLA_PERM_ADDRESS = auto() - IFLA_PROTO_DOWN_REASON = auto() - - -class IflinkInfo(Enum): - IFLA_INFO_UNSPEC = 0 - IFLA_INFO_KIND = auto() - IFLA_INFO_DATA = auto() - IFLA_INFO_XSTATS = auto() - IFLA_INFO_SLAVE_KIND = auto() - IFLA_INFO_SLAVE_DATA = auto() - - -class IfLinkInfoDataVlan(Enum): - IFLA_VLAN_UNSPEC = 0 - IFLA_VLAN_ID = auto() - IFLA_VLAN_FLAGS = auto() - IFLA_VLAN_EGRESS_QOS = auto() - IFLA_VLAN_INGRESS_QOS = auto() - IFLA_VLAN_PROTOCOL = auto() - - -class IfaddrMsg(Structure): - _fields_ = [ - ("ifa_family", c_ubyte), - ("ifa_prefixlen", c_ubyte), - ("ifa_flags", c_ubyte), - ("ifa_scope", c_ubyte), - ("ifa_index", c_uint), - ] - - -class IfattrType(Enum): - IFA_UNSPEC = 0 - IFA_ADDRESS = auto() - IFA_LOCAL = auto() - IFA_LABEL = auto() - IFA_BROADCAST = auto() - IFA_ANYCAST = auto() - IFA_CACHEINFO = auto() - IFA_MULTICAST = auto() - IFA_FLAGS = auto() - IFA_RT_PRIORITY = auto() - IFA_TARGET_NETNSID = auto() - - -class NdMsg(Structure): - _fields_ = [ - ("ndm_family", c_ubyte), - ("ndm_pad1", c_ubyte), - ("ndm_pad2", c_ubyte), - ("ndm_ifindex", c_uint), - ("ndm_state", c_ushort), - ("ndm_flags", c_ubyte), - ("ndm_type", c_ubyte), - ] - - -class NdAttrType(Enum): - NDA_UNSPEC = 0 - NDA_DST = 1 - NDA_LLADDR = 2 - NDA_CACHEINFO = 3 - NDA_PROBES = 4 - NDA_VLAN = 5 - NDA_PORT = 6 - NDA_VNI = 7 - NDA_IFINDEX = 8 - NDA_MASTER = 9 - NDA_LINK_NETNSID = 10 - NDA_SRC_VNI = 11 - NDA_PROTOCOL = 12 - NDA_NH_ID = 13 - NDA_FDB_EXT_ATTRS = 14 - NDA_FLAGS_EXT = 15 - NDA_NDM_STATE_MASK = 16 - NDA_NDM_FLAGS_MASK = 17 - - -class GenlMsgHdr(Structure): - _fields_ = [ - ("cmd", c_ubyte), - ("version", c_ubyte), - ("reserved", c_ushort), - ] - - -class NlConst: - AF_NETLINK = 38 - NETLINK_ROUTE = 0 - NETLINK_GENERIC = 16 - - -class NlHelper: - def __init__(self): - self._pmap = {} - self._af_cls = self.get_af_cls() - self._seq_counter = 1 - self.pid = os.getpid() - - def get_seq(self): - ret = self._seq_counter - self._seq_counter += 1 - return ret - - def get_af_cls(self): - if sys.platform.startswith("freebsd"): - cls = AddressFamilyBsd - else: - cls = AddressFamilyLinux - return cls - - @staticmethod - def build_propmap(cls): - ret = {} - for prop in dir(cls): - if not prop.startswith("_"): - ret[getattr(cls, prop).value] = prop - return ret - - def get_propmap(self, cls): - if cls not in self._pmap: - self._pmap[cls] = self.build_propmap(cls) - return self._pmap[cls] - - def get_name_propmap(self, cls): - ret = {} - for prop in dir(cls): - if not prop.startswith("_"): - ret[prop] = getattr(cls, prop).value - return ret - - def get_attr_byval(self, cls, attr_val): - propmap = self.get_propmap(cls) - return propmap.get(attr_val) - - def get_nlmsg_name(self, val): - for cls in [NlRtMsgType, NlMsgType]: - v = self.get_attr_byval(cls, val) - if v is not None: - return v - return "msg#{}".format(val) - - def get_af_name(self, family): - v = self.get_attr_byval(self._af_cls, family) - if v is not None: - return v - return "af#{}".format(family) - - def get_af_value(self, family_str: str) -> int: - propmap = self.get_name_propmap(self._af_cls) - return propmap.get(family_str) - - def get_rta_name(self, val): - return self.get_attr_byval(RtattrType, val) - - @staticmethod - def get_bitmask_map(propmap, val): - v = 1 - ret = {} - while val: - if v & val: - if v in propmap: - ret[v] = propmap[v] - else: - ret[v] = hex(v) - val -= v - v *= 2 - return ret - - def get_bitmask_str(self, cls, val): - bmap = self.get_bitmask_map(self.get_propmap(cls), val) - return ",".join([v for k, v in bmap.items()]) - - @staticmethod - def get_bitmask_str_uncached(cls, val): - pmap = NlHelper.build_propmap(cls) - bmap = NlHelper.get_bitmask_map(pmap, val) - return ",".join([v for k, v in bmap.items()]) - - def get_nlm_flags_str(self, msg_str: str, reply: bool, val): - if reply: - return self.get_bitmask_str(NlmAckFlags, val) - if msg_str.startswith("RTM_GET"): - return self.get_bitmask_str(NlmGetFlags, val) - elif msg_str.startswith("RTM_DEL"): - return self.get_bitmask_str(NlmDeleteFlags, val) - elif msg_str.startswith("RTM_NEW"): - return self.get_bitmask_str(NlmNewFlags, val) - else: - return self.get_bitmask_str(NlmBaseFlags, val) - - -class NlAttr(object): - def __init__(self, nla_type, data): - if isinstance(nla_type, Enum): - self._nla_type = nla_type.value - self._enum = nla_type - else: - self._nla_type = nla_type - self._enum = None - self.nla_list = [] - self._data = data - - @property - def nla_type(self): - return self._nla_type & 0x3F - - @property - def nla_len(self): - return len(self._data) + 4 - - def add_nla(self, nla): - self.nla_list.append(nla) - - def print_attr(self, prepend=""): - if self._enum is not None: - type_str = self._enum.name - else: - type_str = "nla#{}".format(self.nla_type) - print( - "{}len={} type={}({}){}".format( - prepend, self.nla_len, type_str, self.nla_type, self._print_attr_value() - ) - ) - - @staticmethod - def _validate(data): - if len(data) < 4: - raise ValueError("attribute too short") - nla_len, nla_type = struct.unpack("@HH", data[:4]) - if nla_len > len(data): - raise ValueError("attribute length too big") - if nla_len < 4: - raise ValueError("attribute length too short") - - @classmethod - def _parse(cls, data): - nla_len, nla_type = struct.unpack("@HH", data[:4]) - return cls(nla_type, data[4:]) - - @classmethod - def from_bytes(cls, data, attr_type_enum=None): - cls._validate(data) - attr = cls._parse(data) - attr._enum = attr_type_enum - return attr - - def _to_bytes(self, data: bytes): - ret = data - if align4(len(ret)) != len(ret): - ret = data + bytes(align4(len(ret)) - len(ret)) - return struct.pack("@HH", len(data) + 4, self._nla_type) + ret - - def __bytes__(self): - return self._to_bytes(self._data) - - def _print_attr_value(self): - return " " + " ".join(["x{:02X}".format(b) for b in self._data]) - - -class NlAttrNested(NlAttr): - def __init__(self, nla_type, val): - super().__init__(nla_type, b"") - self.nla_list = val - - @property - def nla_len(self): - return align4(len(b"".join([bytes(nla) for nla in self.nla_list]))) + 4 - - def print_attr(self, prepend=""): - if self._enum is not None: - type_str = self._enum.name - else: - type_str = "nla#{}".format(self.nla_type) - print( - "{}len={} type={}({}) {{".format( - prepend, self.nla_len, type_str, self.nla_type - ) - ) - for nla in self.nla_list: - nla.print_attr(prepend + " ") - print("{}}}".format(prepend)) - - def __bytes__(self): - return self._to_bytes(b"".join([bytes(nla) for nla in self.nla_list])) - - -class NlAttrU32(NlAttr): - def __init__(self, nla_type, val): - self.u32 = enum_or_int(val) - super().__init__(nla_type, b"") - - @property - def nla_len(self): - return 8 - - def _print_attr_value(self): - return " val={}".format(self.u32) - - @staticmethod - def _validate(data): - assert len(data) == 8 - nla_len, nla_type = struct.unpack("@HH", data[:4]) - assert nla_len == 8 - - @classmethod - def _parse(cls, data): - nla_len, nla_type, val = struct.unpack("@HHI", data) - return cls(nla_type, val) - - def __bytes__(self): - return self._to_bytes(struct.pack("@I", self.u32)) - - -class NlAttrU16(NlAttr): - def __init__(self, nla_type, val): - self.u16 = enum_or_int(val) - super().__init__(nla_type, b"") - - @property - def nla_len(self): - return 6 - - def _print_attr_value(self): - return " val={}".format(self.u16) - - @staticmethod - def _validate(data): - assert len(data) == 6 - nla_len, nla_type = struct.unpack("@HH", data[:4]) - assert nla_len == 6 - - @classmethod - def _parse(cls, data): - nla_len, nla_type, val = struct.unpack("@HHH", data) - return cls(nla_type, val) - - def __bytes__(self): - return self._to_bytes(struct.pack("@H", self.u16)) - - -class NlAttrU8(NlAttr): - def __init__(self, nla_type, val): - self.u8 = enum_or_int(val) - super().__init__(nla_type, b"") - - @property - def nla_len(self): - return 5 - - def _print_attr_value(self): - return " val={}".format(self.u8) - - @staticmethod - def _validate(data): - assert len(data) == 5 - nla_len, nla_type = struct.unpack("@HH", data[:4]) - assert nla_len == 5 - - @classmethod - def _parse(cls, data): - nla_len, nla_type, val = struct.unpack("@HHB", data) - return cls(nla_type, val) - - def __bytes__(self): - return self._to_bytes(struct.pack("@B", self.u8)) - - -class NlAttrRtFlags(NlAttrU32): - def _print_attr_value(self): - s = NlHelper.get_bitmask_str_uncached(RtFlagsBSD, self.u32) - return " rtflags={}".format(s) - - -class NlAttrIp(NlAttr): - def __init__(self, nla_type, addr: str): - super().__init__(nla_type, b"") - self.addr = addr - if ":" in self.addr: - self.family = socket.AF_INET6 - else: - self.family = socket.AF_INET - - @staticmethod - def _validate(data): - nla_len, nla_type = struct.unpack("@HH", data[:4]) - data_len = nla_len - 4 - if data_len != 4 and data_len != 16: - raise ValueError( - "Error validating attr {}: nla_len is not valid".format( # noqa: E501 - nla_type - ) - ) - - @property - def nla_len(self): - if self.family == socket.AF_INET6: - return 20 - else: - return 8 - return align4(len(self._data)) + 4 - - @classmethod - def _parse(cls, data): - nla_len, nla_type = struct.unpack("@HH", data[:4]) - data_len = len(data) - 4 - if data_len == 4: - addr = socket.inet_ntop(socket.AF_INET, data[4:8]) - else: - addr = socket.inet_ntop(socket.AF_INET6, data[4:20]) - return cls(nla_type, addr) - - def __bytes__(self): - return self._to_bytes(socket.inet_pton(self.family, self.addr)) - - def _print_attr_value(self): - return " addr={}".format(self.addr) - - -class NlAttrIfindex(NlAttrU32): - def _print_attr_value(self): - try: - ifname = socket.if_indextoname(self.u32) - return " iface={}(#{})".format(ifname, self.u32) - except OSError: - pass - return " iface=if#{}".format(self.u32) - - -class NlAttrMac(NlAttr): - def _print_attr_value(self): - return ["{:02}".format(int(d)) for d in data[4:]].join(":") - - -class NlAttrTable(NlAttrU32): - def _print_attr_value(self): - return " rtable={}".format(self.u32) - - -class NlAttrNhId(NlAttrU32): - def _print_attr_value(self): - return " nh_id={}".format(self.u32) - - -class NlAttrKNhId(NlAttrU32): - def _print_attr_value(self): - return " knh_id={}".format(self.u32) - - -class NlAttrMac(NlAttr): - def _print_attr_value(self): - return ' mac="' + ":".join(["{:02X}".format(b) for b in self._data]) + '"' - - -class NlAttrIfStats(NlAttr): - def _print_attr_value(self): - return " stats={...}" - - -class NlAttrVia(NlAttr): - def __init__(self, nla_type, family, addr: str): - super().__init__(nla_type, b"") - self.addr = addr *** 2736 LINES SKIPPED ***
Want to link to this message? Use this URL: <https://mail-archive.FreeBSD.org/cgi/mid.cgi?202304011936.331JatO2013549>