Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-08-18 11:43:34
__init__.py	file	502 B	0644	2026-08-13 00:00:00	edit
doc_generator.py	file	15,158 B	0644	2026-08-13 00:00:00	edit
nlspec.py	file	20,669 B	0644	2026-08-13 00:00:00	edit
ynl.py	file	43,640 B	0644	2026-08-13 00:00:00	edit

editing: ynl.py

# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause

from collections import namedtuple
from enum import Enum
import functools
import os
import random
import socket
import struct
from struct import Struct
import sys
import ipaddress
import uuid
import queue
import selectors
import time

from .nlspec import SpecFamily

#
# Generic Netlink code which should really be in some library, but I can't quickly find one.
#

class Netlink:
    # Netlink socket
    SOL_NETLINK = 270

NETLINK_ADD_MEMBERSHIP = 1
    NETLINK_CAP_ACK = 10
    NETLINK_EXT_ACK = 11
    NETLINK_GET_STRICT_CHK = 12

# Netlink message
    NLMSG_ERROR = 2
    NLMSG_DONE = 3

NLM_F_REQUEST = 1
    NLM_F_ACK = 4
    NLM_F_ROOT = 0x100
    NLM_F_MATCH = 0x200

NLM_F_REPLACE = 0x100
    NLM_F_EXCL = 0x200
    NLM_F_CREATE = 0x400
    NLM_F_APPEND = 0x800

NLM_F_CAPPED = 0x100
    NLM_F_ACK_TLVS = 0x200

NLM_F_DUMP = NLM_F_ROOT | NLM_F_MATCH

NLA_F_NESTED = 0x8000
    NLA_F_NET_BYTEORDER = 0x4000

NLA_TYPE_MASK = NLA_F_NESTED | NLA_F_NET_BYTEORDER

# Genetlink defines
    NETLINK_GENERIC = 16

GENL_ID_CTRL = 0x10

# nlctrl
    CTRL_CMD_GETFAMILY = 3

CTRL_ATTR_FAMILY_ID = 1
    CTRL_ATTR_FAMILY_NAME = 2
    CTRL_ATTR_MAXATTR = 5
    CTRL_ATTR_MCAST_GROUPS = 7

CTRL_ATTR_MCAST_GRP_NAME = 1
    CTRL_ATTR_MCAST_GRP_ID = 2

# Extack types
    NLMSGERR_ATTR_MSG = 1
    NLMSGERR_ATTR_OFFS = 2
    NLMSGERR_ATTR_COOKIE = 3
    NLMSGERR_ATTR_POLICY = 4
    NLMSGERR_ATTR_MISS_TYPE = 5
    NLMSGERR_ATTR_MISS_NEST = 6

# Policy types
    NL_POLICY_TYPE_ATTR_TYPE = 1
    NL_POLICY_TYPE_ATTR_MIN_VALUE_S = 2
    NL_POLICY_TYPE_ATTR_MAX_VALUE_S = 3
    NL_POLICY_TYPE_ATTR_MIN_VALUE_U = 4
    NL_POLICY_TYPE_ATTR_MAX_VALUE_U = 5
    NL_POLICY_TYPE_ATTR_MIN_LENGTH = 6
    NL_POLICY_TYPE_ATTR_MAX_LENGTH = 7
    NL_POLICY_TYPE_ATTR_POLICY_IDX = 8
    NL_POLICY_TYPE_ATTR_POLICY_MAXTYPE = 9
    NL_POLICY_TYPE_ATTR_BITFIELD32_MASK = 10
    NL_POLICY_TYPE_ATTR_PAD = 11
    NL_POLICY_TYPE_ATTR_MASK = 12

AttrType = Enum('AttrType', ['flag', 'u8', 'u16', 'u32', 'u64',
                                  's8', 's16', 's32', 's64',
                                  'binary', 'string', 'nul-string',
                                  'nested', 'nested-array',
                                  'bitfield32', 'sint', 'uint'])

class NlError(Exception):
  def __init__(self, nl_msg):
    self.nl_msg = nl_msg
    self.error = -nl_msg.error

def __str__(self):
    return f"Netlink error: {os.strerror(self.error)}\n{self.nl_msg}"

class ConfigError(Exception):
    pass

class NlAttr:
    ScalarFormat = namedtuple('ScalarFormat', ['native', 'big', 'little'])
    type_formats = {
        'u8' : ScalarFormat(Struct('B'), Struct("B"),  Struct("B")),
        's8' : ScalarFormat(Struct('b'), Struct("b"),  Struct("b")),
        'u16': ScalarFormat(Struct('H'), Struct(">H"), Struct("<H")),
        's16': ScalarFormat(Struct('h'), Struct(">h"), Struct("<h")),
        'u32': ScalarFormat(Struct('I'), Struct(">I"), Struct("<I")),
        's32': ScalarFormat(Struct('i'), Struct(">i"), Struct("<i")),
        'u64': ScalarFormat(Struct('Q'), Struct(">Q"), Struct("<Q")),
        's64': ScalarFormat(Struct('q'), Struct(">q"), Struct("<q"))
    }

def __init__(self, raw, offset):
        self._len, self._type = struct.unpack("HH", raw[offset : offset + 4])
        self.type = self._type & ~Netlink.NLA_TYPE_MASK
        self.is_nest = self._type & Netlink.NLA_F_NESTED
        self.payload_len = self._len
        self.full_len = (self.payload_len + 3) & ~3
        self.raw = raw[offset + 4 : offset + self.payload_len]

@classmethod
    def get_format(cls, attr_type, byte_order=None):
        format = cls.type_formats[attr_type]
        if byte_order:
            return format.big if byte_order == "big-endian" \
                else format.little
        return format.native

def as_scalar(self, attr_type, byte_order=None):
        format = self.get_format(attr_type, byte_order)
        return format.unpack(self.raw)[0]

def as_auto_scalar(self, attr_type, byte_order=None):
        if len(self.raw) != 4 and len(self.raw) != 8:
            raise Exception(f"Auto-scalar len payload be 4 or 8 bytes, got {len(self.raw)}")
        real_type = attr_type[0] + str(len(self.raw) * 8)
        format = self.get_format(real_type, byte_order)
        return format.unpack(self.raw)[0]

def as_strz(self):
        return self.raw.decode('ascii')[:-1]

def as_bin(self):
        return self.raw

def as_c_array(self, type):
        format = self.get_format(type)
        return [ x[0] for x in format.iter_unpack(self.raw) ]

def __repr__(self):
        return f"[type:{self.type} len:{self._len}] {self.raw}"

class NlAttrs:
    def __init__(self, msg, offset=0):
        self.attrs = []

while offset < len(msg):
            attr = NlAttr(msg, offset)
            offset += attr.full_len
            self.attrs.append(attr)

def __iter__(self):
        yield from self.attrs

def __repr__(self):
        msg = ''
        for a in self.attrs:
            if msg:
                msg += '\n'
            msg += repr(a)
        return msg

class NlMsg:
    def __init__(self, msg, offset, attr_space=None):
        self.hdr = msg[offset : offset + 16]

self.nl_len, self.nl_type, self.nl_flags, self.nl_seq, self.nl_portid = \
            struct.unpack("IHHII", self.hdr)

self.raw = msg[offset + 16 : offset + self.nl_len]

self.error = 0
        self.done = 0

extack_off = None
        if self.nl_type == Netlink.NLMSG_ERROR:
            self.error = struct.unpack("i", self.raw[0:4])[0]
            self.done = 1
            extack_off = 20
        elif self.nl_type == Netlink.NLMSG_DONE:
            self.error = struct.unpack("i", self.raw[0:4])[0]
            self.done = 1
            extack_off = 4

self.extack = None
        if self.nl_flags & Netlink.NLM_F_ACK_TLVS and extack_off:
            self.extack = dict()
            extack_attrs = NlAttrs(self.raw[extack_off:])
            for extack in extack_attrs:
                if extack.type == Netlink.NLMSGERR_ATTR_MSG:
                    self.extack['msg'] = extack.as_strz()
                elif extack.type == Netlink.NLMSGERR_ATTR_MISS_TYPE:
                    self.extack['miss-type'] = extack.as_scalar('u32')
                elif extack.type == Netlink.NLMSGERR_ATTR_MISS_NEST:
                    self.extack['miss-nest'] = extack.as_scalar('u32')
                elif extack.type == Netlink.NLMSGERR_ATTR_OFFS:
                    self.extack['bad-attr-offs'] = extack.as_scalar('u32')
                elif extack.type == Netlink.NLMSGERR_ATTR_POLICY:
                    self.extack['policy'] = self._decode_policy(extack.raw)
                else:
                    if 'unknown' not in self.extack:
                        self.extack['unknown'] = []
                    self.extack['unknown'].append(extack)

if attr_space:
                self.annotate_extack(attr_space)

def _decode_policy(self, raw):
        policy = {}
        for attr in NlAttrs(raw):
            if attr.type == Netlink.NL_POLICY_TYPE_ATTR_TYPE:
                type = attr.as_scalar('u32')
                policy['type'] = Netlink.AttrType(type).name
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MIN_VALUE_S:
                policy['min-value'] = attr.as_scalar('s64')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MAX_VALUE_S:
                policy['max-value'] = attr.as_scalar('s64')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MIN_VALUE_U:
                policy['min-value'] = attr.as_scalar('u64')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MAX_VALUE_U:
                policy['max-value'] = attr.as_scalar('u64')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MIN_LENGTH:
                policy['min-length'] = attr.as_scalar('u32')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MAX_LENGTH:
                policy['max-length'] = attr.as_scalar('u32')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_BITFIELD32_MASK:
                policy['bitfield32-mask'] = attr.as_scalar('u32')
            elif attr.type == Netlink.NL_POLICY_TYPE_ATTR_MASK:
                policy['mask'] = attr.as_scalar('u64')
        return policy

def annotate_extack(self, attr_space):
        """ Make extack more human friendly with attribute information """

# We don't have the ability to parse nests yet, so only do global
        if 'miss-type' in self.extack and 'miss-nest' not in self.extack:
            miss_type = self.extack['miss-type']
            if miss_type in attr_space.attrs_by_val:
                spec = attr_space.attrs_by_val[miss_type]
                self.extack['miss-type'] = spec['name']
                if 'doc' in spec:
                    self.extack['miss-type-doc'] = spec['doc']

def cmd(self):
        return self.nl_type

def __repr__(self):
        msg = f"nl_len = {self.nl_len} ({len(self.raw)}) nl_flags = 0x{self.nl_flags:x} nl_type = {self.nl_type}"
        if self.error:
            msg += '\n\terror: ' + str(self.error)
        if self.extack:
            msg += '\n\textack: ' + repr(self.extack)
        return msg

class NlMsgs:
    def __init__(self, data):
        self.msgs = []

offset = 0
        while offset < len(data):
            msg = NlMsg(data, offset)
            offset += msg.nl_len
            self.msgs.append(msg)

def __iter__(self):
        yield from self.msgs

genl_family_name_to_id = None

def _genl_msg(nl_type, nl_flags, genl_cmd, genl_version, seq=None):
    # we prepend length in _genl_msg_finalize()
    if seq is None:
        seq = random.randint(1, 1024)
    nlmsg = struct.pack("HHII", nl_type, nl_flags, seq, 0)
    genlmsg = struct.pack("BBH", genl_cmd, genl_version, 0)
    return nlmsg + genlmsg

def _genl_msg_finalize(msg):
    return struct.pack("I", len(msg) + 4) + msg

def _genl_load_families():
    with socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, Netlink.NETLINK_GENERIC) as sock:
        sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_CAP_ACK, 1)

msg = _genl_msg(Netlink.GENL_ID_CTRL,
                        Netlink.NLM_F_REQUEST | Netlink.NLM_F_ACK | Netlink.NLM_F_DUMP,
                        Netlink.CTRL_CMD_GETFAMILY, 1)
        msg = _genl_msg_finalize(msg)

sock.send(msg, 0)

global genl_family_name_to_id
        genl_family_name_to_id = dict()

while True:
            reply = sock.recv(128 * 1024)
            nms = NlMsgs(reply)
            for nl_msg in nms:
                if nl_msg.error:
                    print("Netlink error:", nl_msg.error)
                    return
                if nl_msg.done:
                    return

gm = GenlMsg(nl_msg)
                fam = dict()
                for attr in NlAttrs(gm.raw):
                    if attr.type == Netlink.CTRL_ATTR_FAMILY_ID:
                        fam['id'] = attr.as_scalar('u16')
                    elif attr.type == Netlink.CTRL_ATTR_FAMILY_NAME:
                        fam['name'] = attr.as_strz()
                    elif attr.type == Netlink.CTRL_ATTR_MAXATTR:
                        fam['maxattr'] = attr.as_scalar('u32')
                    elif attr.type == Netlink.CTRL_ATTR_MCAST_GROUPS:
                        fam['mcast'] = dict()
                        for entry in NlAttrs(attr.raw):
                            mcast_name = None
                            mcast_id = None
                            for entry_attr in NlAttrs(entry.raw):
                                if entry_attr.type == Netlink.CTRL_ATTR_MCAST_GRP_NAME:
                                    mcast_name = entry_attr.as_strz()
                                elif entry_attr.type == Netlink.CTRL_ATTR_MCAST_GRP_ID:
                                    mcast_id = entry_attr.as_scalar('u32')
                            if mcast_name and mcast_id is not None:
                                fam['mcast'][mcast_name] = mcast_id
                if 'name' in fam and 'id' in fam:
                    genl_family_name_to_id[fam['name']] = fam

class GenlMsg:
    def __init__(self, nl_msg):
        self.nl = nl_msg
        self.genl_cmd, self.genl_version, _ = struct.unpack_from("BBH", nl_msg.raw, 0)
        self.raw = nl_msg.raw[4:]

def cmd(self):
        return self.genl_cmd

def __repr__(self):
        msg = repr(self.nl)
        msg += f"\tgenl_cmd = {self.genl_cmd} genl_ver = {self.genl_version}\n"
        for a in self.raw_attrs:
            msg += '\t\t' + repr(a) + '\n'
        return msg

class NetlinkProtocol:
    def __init__(self, family_name, proto_num):
        self.family_name = family_name
        self.proto_num = proto_num

def _message(self, nl_type, nl_flags, seq=None):
        if seq is None:
            seq = random.randint(1, 1024)
        nlmsg = struct.pack("HHII", nl_type, nl_flags, seq, 0)
        return nlmsg

def message(self, flags, command, version, seq=None):
        return self._message(command, flags, seq)

def _decode(self, nl_msg):
        return nl_msg

def decode(self, ynl, nl_msg, op):
        msg = self._decode(nl_msg)
        if op is None:
            op = ynl.rsp_by_value[msg.cmd()]
        fixed_header_size = ynl._struct_size(op.fixed_header)
        msg.raw_attrs = NlAttrs(msg.raw, fixed_header_size)
        return msg

def get_mcast_id(self, mcast_name, mcast_groups):
        if mcast_name not in mcast_groups:
            raise Exception(f'Multicast group "{mcast_name}" not present in the spec')
        return mcast_groups[mcast_name].value

def msghdr_size(self):
        return 16

class GenlProtocol(NetlinkProtocol):
    def __init__(self, family_name):
        super().__init__(family_name, Netlink.NETLINK_GENERIC)

global genl_family_name_to_id
        if genl_family_name_to_id is None:
            _genl_load_families()

self.genl_family = genl_family_name_to_id[family_name]
        self.family_id = genl_family_name_to_id[family_name]['id']

def message(self, flags, command, version, seq=None):
        nlmsg = self._message(self.family_id, flags, seq)
        genlmsg = struct.pack("BBH", command, version, 0)
        return nlmsg + genlmsg

def _decode(self, nl_msg):
        return GenlMsg(nl_msg)

def get_mcast_id(self, mcast_name, mcast_groups):
        if mcast_name not in self.genl_family['mcast']:
            raise Exception(f'Multicast group "{mcast_name}" not present in the family')
        return self.genl_family['mcast'][mcast_name]

def msghdr_size(self):
        return super().msghdr_size() + 4

class SpaceAttrs:
    SpecValuesPair = namedtuple('SpecValuesPair', ['spec', 'values'])

def __init__(self, attr_space, attrs, outer = None):
        outer_scopes = outer.scopes if outer else []
        inner_scope = self.SpecValuesPair(attr_space, attrs)
        self.scopes = [inner_scope] + outer_scopes

def lookup(self, name):
        for scope in self.scopes:
            if name in scope.spec:
                if name in scope.values:
                    return scope.values[name]
                spec_name = scope.spec.yaml['name']
                raise Exception(
                    f"No value for '{name}' in attribute space '{spec_name}'")
        raise Exception(f"Attribute '{name}' not defined in any attribute-set")

#
# YNL implementation details.
#

class YnlFamily(SpecFamily):
    def __init__(self, def_path, schema=None, process_unknown=False,
                 recv_size=0):
        super().__init__(def_path, schema)

self.include_raw = False
        self.process_unknown = process_unknown

try:
            if self.proto == "netlink-raw":
                self.nlproto = NetlinkProtocol(self.yaml['name'],
                                               self.yaml['protonum'])
            else:
                self.nlproto = GenlProtocol(self.yaml['name'])
        except KeyError:
            raise Exception(f"Family '{self.yaml['name']}' not supported by the kernel")

self._recv_dbg = False
        # Note that netlink will use conservative (min) message size for
        # the first dump recv() on the socket, our setting will only matter
        # from the second recv() on.
        self._recv_size = recv_size if recv_size else 131072
        # Netlink will always allocate at least PAGE_SIZE - sizeof(skb_shinfo)
        # for a message, so smaller receive sizes will lead to truncation.
        # Note that the min size for other families may be larger than 4k!
        if self._recv_size < 4000:
            raise ConfigError()

self.sock = socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, self.nlproto.proto_num)
        self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_CAP_ACK, 1)
        self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_EXT_ACK, 1)
        self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_GET_STRICT_CHK, 1)

self.async_msg_ids = set()
        self.async_msg_queue = queue.Queue()

for msg in self.msgs.values():
            if msg.is_async:
                self.async_msg_ids.add(msg.rsp_value)

for op_name, op in self.ops.items():
            bound_f = functools.partial(self._op, op_name)
            setattr(self, op.ident_name, bound_f)

def ntf_subscribe(self, mcast_name):
        mcast_id = self.nlproto.get_mcast_id(mcast_name, self.mcast_groups)
        self.sock.bind((0, 0))
        self.sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_ADD_MEMBERSHIP,
                             mcast_id)

def set_recv_dbg(self, enabled):
        self._recv_dbg = enabled

def _recv_dbg_print(self, reply, nl_msgs):
        if not self._recv_dbg:
            return
        print("Recv: read", len(reply), "bytes,",
              len(nl_msgs.msgs), "messages", file=sys.stderr)
        for nl_msg in nl_msgs:
            print("  ", nl_msg, file=sys.stderr)

def _encode_enum(self, attr_spec, value):
        enum = self.consts[attr_spec['enum']]
        if enum.type == 'flags' or attr_spec.get('enum-as-flags', False):
            scalar = 0
            if isinstance(value, str):
                value = [value]
            for single_value in value:
                scalar += enum.entries[single_value].user_value(as_flags = True)
            return scalar
        else:
            return enum.entries[value].user_value()

def _get_scalar(self, attr_spec, value):
        try:
            return int(value)
        except (ValueError, TypeError) as e:
            if 'enum' in attr_spec:
                return self._encode_enum(attr_spec, value)
            if attr_spec.display_hint:
                return self._from_string(value, attr_spec)
            raise e

def _add_attr(self, space, name, value, search_attrs):
        try:
            attr = self.attr_sets[space][name]
        except KeyError:
            raise Exception(f"Space '{space}' has no attribute '{name}'")
        nl_type = attr.value

if attr.is_multi and isinstance(value, list):
            attr_payload = b''
            for subvalue in value:
                attr_payload += self._add_attr(space, name, subvalue, search_attrs)
            return attr_payload

if attr["type"] == 'nest':
            nl_type |= Netlink.NLA_F_NESTED
            sub_space = attr['nested-attributes']
            attr_payload = self._add_nest_attrs(value, sub_space, search_attrs)
        elif attr['type'] == 'indexed-array' and attr['sub-type'] == 'nest':
            nl_type |= Netlink.NLA_F_NESTED
            sub_space = attr['nested-attributes']
            attr_payload = self._encode_indexed_array(value, sub_space,
                                                      search_attrs)
        elif attr["type"] == 'flag':
            if not value:
                # If value is absent or false then skip attribute creation.
                return b''
            attr_payload = b''
        elif attr["type"] == 'string':
            attr_payload = str(value).encode('ascii') + b'\x00'
        elif attr["type"] == 'binary':
            if value is None:
                attr_payload = b''
            elif isinstance(value, bytes):
                attr_payload = value
            elif isinstance(value, str):
                if attr.display_hint:
                    attr_payload = self._from_string(value, attr)
                else:
                    attr_payload = bytes.fromhex(value)
            elif isinstance(value, dict) and attr.struct_name:
                attr_payload = self._encode_struct(attr.struct_name, value)
            elif isinstance(value, list) and attr.sub_type in NlAttr.type_formats:
                format = NlAttr.get_format(attr.sub_type)
                attr_payload = b''.join([format.pack(x) for x in value])
            else:
                raise Exception(f'Unknown type for binary attribute, value: {value}')
        elif attr['type'] in NlAttr.type_formats or attr.is_auto_scalar:
            scalar = self._get_scalar(attr, value)
            if attr.is_auto_scalar:
                attr_type = attr["type"][0] + ('32' if scalar.bit_length() <= 32 else '64')
            else:
                attr_type = attr["type"]
            format = NlAttr.get_format(attr_type, attr.byte_order)
            attr_payload = format.pack(scalar)
        elif attr['type'] in "bitfield32":
            scalar_value = self._get_scalar(attr, value["value"])
            scalar_selector = self._get_scalar(attr, value["selector"])
            attr_payload = struct.pack("II", scalar_value, scalar_selector)
        elif attr['type'] == 'sub-message':
            msg_format, _ = self._resolve_selector(attr, search_attrs)
            attr_payload = b''
            if msg_format.fixed_header:
                attr_payload += self._encode_struct(msg_format.fixed_header, value)
            if msg_format.attr_set:
                if msg_format.attr_set in self.attr_sets:
                    nl_type |= Netlink.NLA_F_NESTED
                    sub_attrs = SpaceAttrs(msg_format.attr_set, value, search_attrs)
                    for subname, subvalue in value.items():
                        attr_payload += self._add_attr(msg_format.attr_set,
                                                       subname, subvalue, sub_attrs)
                else:
                    raise Exception(f"Unknown attribute-set '{msg_format.attr_set}'")
        else:
            raise Exception(f'Unknown type at {space} {name} {value} {attr["type"]}')

return self._add_attr_raw(nl_type, attr_payload)

def _add_attr_raw(self, nl_type, attr_payload):
        pad = b'\x00' * ((4 - len(attr_payload) % 4) % 4)
        return struct.pack('HH', len(attr_payload) + 4, nl_type) + attr_payload + pad

def _add_nest_attrs(self, value, sub_space, search_attrs):
        sub_attrs = SpaceAttrs(self.attr_sets[sub_space], value, search_attrs)
        attr_payload = b''
        for subname, subvalue in value.items():
            attr_payload += self._add_attr(sub_space, subname, subvalue,
                                           sub_attrs)
        return attr_payload

def _encode_indexed_array(self, vals, sub_space, search_attrs):
        attr_payload = b''
        for i, val in enumerate(vals):
            idx = i | Netlink.NLA_F_NESTED
            val_payload = self._add_nest_attrs(val, sub_space, search_attrs)
            attr_payload += self._add_attr_raw(idx, val_payload)
        return attr_payload

def _get_enum_or_unknown(self, enum, raw):
        try:
            name = enum.entries_by_val[raw].name
        except KeyError as error:
            if self.process_unknown:
                name = f"Unknown({raw})"
            else:
                raise error
        return name

def _decode_enum(self, raw, attr_spec):
        enum = self.consts[attr_spec['enum']]
        if enum.type == 'flags' or attr_spec.get('enum-as-flags', False):
            i = 0
            value = set()
            while raw:
                if raw & 1:
                    value.add(self._get_enum_or_unknown(enum, i))
                raw >>= 1
                i += 1
        else:
            value = self._get_enum_or_unknown(enum, raw)
        return value

def _decode_binary(self, attr, attr_spec):
        if attr_spec.struct_name:
            decoded = self._decode_struct(attr.raw, attr_spec.struct_name)
        elif attr_spec.sub_type:
            decoded = attr.as_c_array(attr_spec.sub_type)
            if 'enum' in attr_spec:
                decoded = [ self._decode_enum(x, attr_spec) for x in decoded ]
            elif attr_spec.display_hint:
                decoded = [ self._formatted_string(x, attr_spec.display_hint)
                            for x in decoded ]
        else:
            decoded = attr.as_bin()
            if attr_spec.display_hint:
                decoded = self._formatted_string(decoded, attr_spec.display_hint)
        return decoded

def _decode_array_attr(self, attr, attr_spec):
        decoded = []
        offset = 0
        while offset < len(attr.raw):
            item = NlAttr(attr.raw, offset)
            offset += item.full_len

if attr_spec["sub-type"] == 'nest':
                subattrs = self._decode(NlAttrs(item.raw), attr_spec['nested-attributes'])
                decoded.append({ item.type: subattrs })
            elif attr_spec["sub-type"] == 'binary':
                subattr = item.as_bin()
                if attr_spec.display_hint:
                    subattr = self._formatted_string(subattr, attr_spec.display_hint)
                decoded.append(subattr)
            elif attr_spec["sub-type"] in NlAttr.type_formats:
                subattr = item.as_scalar(attr_spec['sub-type'], attr_spec.byte_order)
                if 'enum' in attr_spec:
                    subattr = self._decode_enum(subattr, attr_spec)
                elif attr_spec.display_hint:
                    subattr = self._formatted_string(subattr, attr_spec.display_hint)
                decoded.append(subattr)
            else:
                raise Exception(f'Unknown {attr_spec["sub-type"]} with name {attr_spec["name"]}')
        return decoded

def _decode_nest_type_value(self, attr, attr_spec):
        decoded = {}
        value = attr
        for name in attr_spec['type-value']:
            value = NlAttr(value.raw, 0)
            decoded[name] = value.type
        subattrs = self._decode(NlAttrs(value.raw), attr_spec['nested-attributes'])
        decoded.update(subattrs)
        return decoded

def _decode_unknown(self, attr):
        if attr.is_nest:
            return self._decode(NlAttrs(attr.raw), None)
        else:
            return attr.as_bin()

def _rsp_add(self, rsp, name, is_multi, decoded):
        if is_multi is None:
            if name in rsp and type(rsp[name]) is not list:
                rsp[name] = [rsp[name]]
                is_multi = True
            else:
                is_multi = False

if not is_multi:
            rsp[name] = decoded
        elif name in rsp:
            rsp[name].append(decoded)
        else:
            rsp[name] = [decoded]

def _resolve_selector(self, attr_spec, search_attrs):
        sub_msg = attr_spec.sub_message
        if sub_msg not in self.sub_msgs:
            raise Exception(f"No sub-message spec named {sub_msg} for {attr_spec.name}")
        sub_msg_spec = self.sub_msgs[sub_msg]

selector = attr_spec.selector
        value = search_attrs.lookup(selector)
        if value not in sub_msg_spec.formats:
            raise Exception(f"No message format for '{value}' in sub-message spec '{sub_msg}'")

spec = sub_msg_spec.formats[value]
        return spec, value

def _decode_sub_msg(self, attr, attr_spec, search_attrs):
        msg_format, _ = self._resolve_selector(attr_spec, search_attrs)
        decoded = {}
        offset = 0
        if msg_format.fixed_header:
            decoded.update(self._decode_struct(attr.raw, msg_format.fixed_header))
            offset = self._struct_size(msg_format.fixed_header)
        if msg_format.attr_set:
            if msg_format.attr_set in self.attr_sets:
                subdict = self._decode(NlAttrs(attr.raw, offset), msg_format.attr_set)
                decoded.update(subdict)
            else:
                raise Exception(f"Unknown attribute-set '{msg_format.attr_set}' when decoding '{attr_spec.name}'")
        return decoded

def _decode(self, attrs, space, outer_attrs = None):
        rsp = dict()
        if space:
            attr_space = self.attr_sets[space]
            search_attrs = SpaceAttrs(attr_space, rsp, outer_attrs)

for attr in attrs:
            try:
                attr_spec = attr_space.attrs_by_val[attr.type]
            except (KeyError, UnboundLocalError):
                if not self.process_unknown:
                    raise Exception(f"Space '{space}' has no attribute with value '{attr.type}'")
                attr_name = f"UnknownAttr({attr.type})"
                self._rsp_add(rsp, attr_name, None, self._decode_unknown(attr))
                continue

try:
                if attr_spec["type"] == 'nest':
                    subdict = self._decode(NlAttrs(attr.raw), attr_spec['nested-attributes'], search_attrs)
                    decoded = subdict
                elif attr_spec["type"] == 'string':
                    decoded = attr.as_strz()
                elif attr_spec["type"] == 'binary':
                    decoded = self._decode_binary(attr, attr_spec)
                elif attr_spec["type"] == 'flag':
                    decoded = True
                elif attr_spec.is_auto_scalar:
                    decoded = attr.as_auto_scalar(attr_spec['type'], attr_spec.byte_order)
                    if 'enum' in attr_spec:
                        decoded = self._decode_enum(decoded, attr_spec)
                elif attr_spec["type"] in NlAttr.type_formats:
                    decoded = attr.as_scalar(attr_spec['type'], attr_spec.byte_order)
                    if 'enum' in attr_spec:
                        decoded = self._decode_enum(decoded, attr_spec)
                    elif attr_spec.display_hint:
                        decoded = self._formatted_string(decoded, attr_spec.display_hint)
                elif attr_spec["type"] == 'indexed-array':
                    decoded = self._decode_array_attr(attr, attr_spec)
                elif attr_spec["type"] == 'bitfield32':
                    value, selector = struct.unpack("II", attr.raw)
                    if 'enum' in attr_spec:
                        value = self._decode_enum(value, attr_spec)
                        selector = self._decode_enum(selector, attr_spec)
                    decoded = {"value": value, "selector": selector}
                elif attr_spec["type"] == 'sub-message':
                    decoded = self._decode_sub_msg(attr, attr_spec, search_attrs)
                elif attr_spec["type"] == 'nest-type-value':
                    decoded = self._decode_nest_type_value(attr, attr_spec)
                else:
                    if not self.process_unknown:
                        raise Exception(f'Unknown {attr_spec["type"]} with name {attr_spec["name"]}')
                    decoded = self._decode_unknown(attr)

self._rsp_add(rsp, attr_spec["name"], attr_spec.is_multi, decoded)
            except:
                print(f"Error decoding '{attr_spec.name}' from '{space}'")
                raise

return rsp

def _decode_extack_path(self, attrs, attr_set, offset, target, search_attrs):
        for attr in attrs:
            try:
                attr_spec = attr_set.attrs_by_val[attr.type]
            except KeyError:
                raise Exception(f"Space '{attr_set.name}' has no attribute with value '{attr.type}'")
            if offset > target:
                break
            if offset == target:
                return '.' + attr_spec.name

if offset + attr.full_len <= target:
                offset += attr.full_len
                continue

pathname = attr_spec.name
            if attr_spec['type'] == 'nest':
                sub_attrs = self.attr_sets[attr_spec['nested-attributes']]
                search_attrs = SpaceAttrs(sub_attrs, search_attrs.lookup(attr_spec['name']))
            elif attr_spec['type'] == 'sub-message':
                msg_format, value = self._resolve_selector(attr_spec, search_attrs)
                if msg_format is None:
                    raise Exception(f"Can't resolve sub-message of {attr_spec['name']} for extack")
                sub_attrs = self.attr_sets[msg_format.attr_set]
                pathname += f"({value})"
            else:
                raise Exception(f"Can't dive into {attr.type} ({attr_spec['name']}) for extack")
            offset += 4
            subpath = self._decode_extack_path(NlAttrs(attr.raw), sub_attrs,
                                               offset, target, search_attrs)
            if subpath is None:
                return None
            return '.' + pathname + subpath

return None

def _decode_extack(self, request, op, extack, vals):
        if 'bad-attr-offs' not in extack:
            return

msg = self.nlproto.decode(self, NlMsg(request, 0, op.attr_set), op)
        offset = self.nlproto.msghdr_size() + self._struct_size(op.fixed_header)
        search_attrs = SpaceAttrs(op.attr_set, vals)
        path = self._decode_extack_path(msg.raw_attrs, op.attr_set, offset,
                                        extack['bad-attr-offs'], search_attrs)
        if path:
            del extack['bad-attr-offs']
            extack['bad-attr'] = path

def _struct_size(self, name):
        if name:
            members = self.consts[name].members
            size = 0
            for m in members:
                if m.type in ['pad', 'binary']:
                    if m.struct:
                        size += self._struct_size(m.struct)
                    else:
                        size += m.len
                else:
                    format = NlAttr.get_format(m.type, m.byte_order)
                    size += format.size
            return size
        else:
            return 0

def _decode_struct(self, data, name):
        members = self.consts[name].members
        attrs = dict()
        offset = 0
        for m in members:
            value = None
            if m.type == 'pad':
                offset += m.len
            elif m.type == 'binary':
                if m.struct:
                    len = self._struct_size(m.struct)
                    value = self._decode_struct(data[offset : offset + len],
                                                m.struct)
                    offset += len
                else:
                    value = data[offset : offset + m.len]
                    offset += m.len
            else:
                format = NlAttr.get_format(m.type, m.byte_order)
                [ value ] = format.unpack_from(data, offset)
                offset += format.size
            if value is not None:
                if m.enum:
                    value = self._decode_enum(value, m)
                elif m.display_hint:
                    value = self._formatted_string(value, m.display_hint)
                attrs[m.name] = value
        return attrs

def _encode_struct(self, name, vals):
        members = self.consts[name].members
        attr_payload = b''
        for m in members:
            value = vals.pop(m.name) if m.name in vals else None
            if m.type == 'pad':
                attr_payload += bytearray(m.len)
            elif m.type == 'binary':
                if m.struct:
                    if value is None:
                        value = dict()
                    attr_payload += self._encode_struct(m.struct, value)
                else:
                    if value is None:
                        attr_payload += bytearray(m.len)
                    else:
                        attr_payload += bytes.fromhex(value)
            else:
                if value is None:
                    value = 0
                format = NlAttr.get_format(m.type, m.byte_order)
                attr_payload += format.pack(value)
        return attr_payload

def _formatted_string(self, raw, display_hint):
        if display_hint == 'mac':
            formatted = ':'.join('%02x' % b for b in raw)
        elif display_hint == 'hex':
            if isinstance(raw, int):
                formatted = hex(raw)
            else:
                formatted = bytes.hex(raw, ' ')
        elif display_hint in [ 'ipv4', 'ipv6', 'ipv4-or-v6' ]:
            formatted = format(ipaddress.ip_address(raw))
        elif display_hint == 'uuid':
            formatted = str(uuid.UUID(bytes=raw))
        else:
            formatted = raw
        return formatted

def _from_string(self, string, attr_spec):
        if attr_spec.display_hint in ['ipv4', 'ipv6', 'ipv4-or-v6']:
            ip = ipaddress.ip_address(string)
            if attr_spec['type'] == 'binary':
                raw = ip.packed
            else:
                raw = int(ip)
        elif attr_spec.display_hint == 'hex':
            if attr_spec['type'] == 'binary':
                raw = bytes.fromhex(string)
            else:
                raw = int(string, 16)
        else:
            raise Exception(f"Display hint '{attr_spec.display_hint}' not implemented"
                            f" when parsing '{attr_spec['name']}'")
        return raw

def handle_ntf(self, decoded):
        msg = dict()
        if self.include_raw:
            msg['raw'] = decoded
        op = self.rsp_by_value[decoded.cmd()]
        attrs = self._decode(decoded.raw_attrs, op.attr_set.name)
        if op.fixed_header:
            attrs.update(self._decode_struct(decoded.raw, op.fixed_header))

msg['name'] = op['name']
        msg['msg'] = attrs
        self.async_msg_queue.put(msg)

def check_ntf(self):
        while True:
            try:
                reply = self.sock.recv(self._recv_size, socket.MSG_DONTWAIT)
            except BlockingIOError:
                return

nms = NlMsgs(reply)
            self._recv_dbg_print(reply, nms)
            for nl_msg in nms:
                if nl_msg.error:
                    print("Netlink error in ntf!?", os.strerror(-nl_msg.error))
                    print(nl_msg)
                    continue
                if nl_msg.done:
                    print("Netlink done while checking for ntf!?")
                    continue

decoded = self.nlproto.decode(self, nl_msg, None)
                if decoded.cmd() not in self.async_msg_ids:
                    print("Unexpected msg id while checking for ntf", decoded)
                    continue

self.handle_ntf(decoded)

def poll_ntf(self, duration=None):
        start_time = time.time()
        selector = selectors.DefaultSelector()
        selector.register(self.sock, selectors.EVENT_READ)

while True:
            try:
                yield self.async_msg_queue.get_nowait()
            except queue.Empty:
                if duration is not None:
                    timeout = start_time + duration - time.time()
                    if timeout <= 0:
                        return
                else:
                    timeout = None
                events = selector.select(timeout)
                if events:
                    self.check_ntf()

def operation_do_attributes(self, name):
      """
      For a given operation name, find and return a supported
      set of attributes (as a dict).
      """
      op = self.find_operation(name)
      if not op:
        return None

return op['do']['request']['attributes'].copy()

def _encode_message(self, op, vals, flags, req_seq):
        nl_flags = Netlink.NLM_F_REQUEST | Netlink.NLM_F_ACK
        for flag in flags or []:
            nl_flags |= flag

msg = self.nlproto.message(nl_flags, op.req_value, 1, req_seq)
        if op.fixed_header:
            msg += self._encode_struct(op.fixed_header, vals)
        search_attrs = SpaceAttrs(op.attr_set, vals)
        for name, value in vals.items():
            msg += self._add_attr(op.attr_set.name, name, value, search_attrs)
        msg = _genl_msg_finalize(msg)
        return msg

def _ops(self, ops):
        reqs_by_seq = {}
        req_seq = random.randint(1024, 65535)
        payload = b''
        for (method, vals, flags) in ops:
            op = self.ops[method]
            msg = self._encode_message(op, vals, flags, req_seq)
            reqs_by_seq[req_seq] = (op, vals, msg, flags)
            payload += msg
            req_seq += 1

self.sock.send(payload, 0)

done = False
        rsp = []
        op_rsp = []
        while not done:
            reply = self.sock.recv(self._recv_size)
            nms = NlMsgs(reply)
            self._recv_dbg_print(reply, nms)
            for nl_msg in nms:
                if nl_msg.nl_seq in reqs_by_seq:
                    (op, vals, req_msg, req_flags) = reqs_by_seq[nl_msg.nl_seq]
                    if nl_msg.extack:
                        nl_msg.annotate_extack(op.attr_set)
                        self._decode_extack(req_msg, op, nl_msg.extack, vals)
                else:
                    op = None
                    req_flags = []

if nl_msg.error:
                    raise NlError(nl_msg)
                if nl_msg.done:
                    if nl_msg.extack:
                        print("Netlink warning:")
                        print(nl_msg)

if Netlink.NLM_F_DUMP in req_flags:
                        rsp.append(op_rsp)
                    elif not op_rsp:
                        rsp.append(None)
                    elif len(op_rsp) == 1:
                        rsp.append(op_rsp[0])
                    else:
                        rsp.append(op_rsp)
                    op_rsp = []

del reqs_by_seq[nl_msg.nl_seq]
                    done = len(reqs_by_seq) == 0
                    break

decoded = self.nlproto.decode(self, nl_msg, op)

# Check if this is a reply to our request
                if nl_msg.nl_seq not in reqs_by_seq or decoded.cmd() != op.rsp_value:
                    if decoded.cmd() in self.async_msg_ids:
                        self.handle_ntf(decoded)
                        continue
                    else:
                        print('Unexpected message: ' + repr(decoded))
                        continue

rsp_msg = self._decode(decoded.raw_attrs, op.attr_set.name)
                if op.fixed_header:
                    rsp_msg.update(self._decode_struct(decoded.raw, op.fixed_header))
                op_rsp.append(rsp_msg)

return rsp

def _op(self, method, vals, flags=None, dump=False):
        req_flags = flags or []
        if dump:
            req_flags.append(Netlink.NLM_F_DUMP)

ops = [(method, vals, req_flags)]
        return self._ops(ops)[0]

def do(self, method, vals, flags=None):
        return self._op(method, vals, flags)

def dump(self, method, vals):
        return self._op(method, vals, dump=True)

def do_multi(self, ops):
        return self._ops(ops)

cancel