path: root/tools/net/ynl/pyynl/lib/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):
        msg = "Netlink error: "

        extack = self.nl_msg.extack.copy() if self.nl_msg.extack else {}
        if 'msg' in extack:
            msg += extack['msg'] + ': '
            del extack['msg']
        msg += os.strerror(self.error)
        if extack:
            msg += ' ' + str(extack)
        return 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(