path: root/tools/net/ynl/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 yaml
import ipaddress
import uuid

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):