summaryrefslogtreecommitdiff
path: root/tools/net/ynl/lib/ynl.py
diff options
context:
space:
mode:
Diffstat (limited to 'tools/net/ynl/lib/ynl.py')
-rw-r--r--tools/net/ynl/lib/ynl.py137
1 files changed, 76 insertions, 61 deletions
diff --git a/tools/net/ynl/lib/ynl.py b/tools/net/ynl/lib/ynl.py
index 3144f33196be..1b3a36fbb1c3 100644
--- a/tools/net/ynl/lib/ynl.py
+++ b/tools/net/ynl/lib/ynl.py
@@ -1,11 +1,15 @@
# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause
+from collections import namedtuple
import functools
import os
import random
import socket
import struct
+from struct import Struct
import yaml
+import ipaddress
+import uuid
from .nlspec import SpecFamily
@@ -76,10 +80,17 @@ class NlError(Exception):
class NlAttr:
- type_formats = { 'u8' : ('B', 1), 's8' : ('b', 1),
- 'u16': ('H', 2), 's16': ('h', 2),
- 'u32': ('I', 4), 's32': ('i', 4),
- 'u64': ('Q', 8), 's64': ('q', 8) }
+ 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])
@@ -88,25 +99,31 @@ class NlAttr:
self.full_len = (self.payload_len + 3) & ~3
self.raw = raw[offset + 4:offset + self.payload_len]
- def format_byte_order(byte_order):
+ @classmethod
+ def get_format(cls, attr_type, byte_order=None):
+ format = cls.type_formats[attr_type]
if byte_order:
- return ">" if byte_order == "big-endian" else "<"
- return ""
-
- def as_u8(self):
- return struct.unpack("B", self.raw)[0]
-
- def as_u16(self, byte_order=None):
- endian = NlAttr.format_byte_order(byte_order)
- return struct.unpack(f"{endian}H", self.raw)[0]
-
- def as_u32(self, byte_order=None):
- endian = NlAttr.format_byte_order(byte_order)
- return struct.unpack(f"{endian}I", self.raw)[0]
+ return format.big if byte_order == "big-endian" \
+ else format.little
+ return format.native
+
+ @classmethod
+ def formatted_string(cls, raw, display_hint):
+ if display_hint == 'mac':
+ formatted = ':'.join('%02x' % b for b in raw)
+ elif display_hint == 'hex':
+ formatted = bytes.hex(raw, ' ')
+ elif display_hint in [ 'ipv4', 'ipv6' ]:
+ formatted = format(ipaddress.ip_address(raw))
+ elif display_hint == 'uuid':
+ formatted = str(uuid.UUID(bytes=raw))
+ else:
+ formatted = raw
+ return formatted
- def as_u64(self, byte_order=None):
- endian = NlAttr.format_byte_order(byte_order)
- return struct.unpack(f"{endian}Q", self.raw)[0]
+ 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_strz(self):
return self.raw.decode('ascii')[:-1]
@@ -115,18 +132,24 @@ class NlAttr:
return self.raw
def as_c_array(self, type):
- format, _ = self.type_formats[type]
- return list({ x[0] for x in struct.iter_unpack(format, self.raw) })
+ format = self.get_format(type)
+ return [ x[0] for x in format.iter_unpack(self.raw) ]
def as_struct(self, members):
value = dict()
offset = 0
for m in members:
# TODO: handle non-scalar members
- format, size = self.type_formats[m.type]
- decoded = struct.unpack_from(format, self.raw, offset)
- offset += size
- value[m.name] = decoded[0]
+ if m.type == 'binary':
+ decoded = self.raw[offset:offset+m['len']]
+ offset += m['len']
+ elif m.type in NlAttr.type_formats:
+ format = self.get_format(m.type, m.byte_order)
+ [ decoded ] = format.unpack_from(self.raw, offset)
+ offset += format.size
+ if m.display_hint:
+ decoded = self.formatted_string(decoded, m.display_hint)
+ value[m.name] = decoded
return value
def __repr__(self):
@@ -184,11 +207,11 @@ class NlMsg:
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_u32()
+ self.extack['miss-type'] = extack.as_scalar('u32')
elif extack.type == Netlink.NLMSGERR_ATTR_MISS_NEST:
- self.extack['miss-nest'] = extack.as_u32()
+ self.extack['miss-nest'] = extack.as_scalar('u32')
elif extack.type == Netlink.NLMSGERR_ATTR_OFFS:
- self.extack['bad-attr-offs'] = extack.as_u32()
+ self.extack['bad-attr-offs'] = extack.as_scalar('u32')
else:
if 'unknown' not in self.extack:
self.extack['unknown'] = []
@@ -272,11 +295,11 @@ def _genl_load_families():
fam = dict()
for attr in gm.raw_attrs:
if attr.type == Netlink.CTRL_ATTR_FAMILY_ID:
- fam['id'] = attr.as_u16()
+ 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_u32()
+ fam['maxattr'] = attr.as_scalar('u32')
elif attr.type == Netlink.CTRL_ATTR_MCAST_GROUPS:
fam['mcast'] = dict()
for entry in NlAttrs(attr.raw):
@@ -286,7 +309,7 @@ def _genl_load_families():
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_u32()
+ 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:
@@ -304,9 +327,9 @@ class GenlMsg:
self.fixed_header_attrs = dict()
for m in fixed_header_members:
- format, size = NlAttr.type_formats[m.type]
- decoded = struct.unpack_from(format, nl_msg.raw, offset)
- offset += size
+ format = NlAttr.get_format(m.type, m.byte_order)
+ decoded = format.unpack_from(nl_msg.raw, offset)
+ offset += format.size
self.fixed_header_attrs[m.name] = decoded[0]
self.raw = nl_msg.raw[offset:]
@@ -381,21 +404,13 @@ class YnlFamily(SpecFamily):
attr_payload += self._add_attr(attr['nested-attributes'], subname, subvalue)
elif attr["type"] == 'flag':
attr_payload = b''
- elif attr["type"] == 'u8':
- attr_payload = struct.pack("B", int(value))
- elif attr["type"] == 'u16':
- endian = NlAttr.format_byte_order(attr.byte_order)
- attr_payload = struct.pack(f"{endian}H", int(value))
- elif attr["type"] == 'u32':
- endian = NlAttr.format_byte_order(attr.byte_order)
- attr_payload = struct.pack(f"{endian}I", int(value))
- elif attr["type"] == 'u64':
- endian = NlAttr.format_byte_order(attr.byte_order)
- attr_payload = struct.pack(f"{endian}Q", int(value))
elif attr["type"] == 'string':
attr_payload = str(value).encode('ascii') + b'\x00'
elif attr["type"] == 'binary':
- attr_payload = value
+ attr_payload = bytes.fromhex(value)
+ elif attr['type'] in NlAttr.type_formats:
+ format = NlAttr.get_format(attr['type'], attr.byte_order)
+ attr_payload = format.pack(int(value))
else:
raise Exception(f'Unknown type at {space} {name} {value} {attr["type"]}')
@@ -419,11 +434,17 @@ class YnlFamily(SpecFamily):
def _decode_binary(self, attr, attr_spec):
if attr_spec.struct_name:
- decoded = attr.as_struct(self.consts[attr_spec.struct_name])
+ members = self.consts[attr_spec.struct_name]
+ decoded = attr.as_struct(members)
+ for m in members:
+ if m.enum:
+ self._decode_enum(decoded, m)
elif attr_spec.sub_type:
decoded = attr.as_c_array(attr_spec.sub_type)
else:
decoded = attr.as_bin()
+ if attr_spec.display_hint:
+ decoded = NlAttr.formatted_string(decoded, attr_spec.display_hint)
return decoded
def _decode(self, attrs, space):
@@ -434,22 +455,16 @@ class YnlFamily(SpecFamily):
if attr_spec["type"] == 'nest':
subdict = self._decode(NlAttrs(attr.raw), attr_spec['nested-attributes'])
decoded = subdict
- elif attr_spec['type'] == 'u8':
- decoded = attr.as_u8()
- elif attr_spec['type'] == 'u16':
- decoded = attr.as_u16(attr_spec.byte_order)
- elif attr_spec['type'] == 'u32':
- decoded = attr.as_u32(attr_spec.byte_order)
- elif attr_spec['type'] == 'u64':
- decoded = attr.as_u64(attr_spec.byte_order)
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["type"] in NlAttr.type_formats:
+ decoded = attr.as_scalar(attr_spec['type'], attr_spec.byte_order)
else:
- raise Exception(f'Unknown {attr.type} {attr_spec["name"]} {attr_spec["type"]}')
+ raise Exception(f'Unknown {attr_spec["type"]} with name {attr_spec["name"]}')
if not attr_spec.is_multi:
rsp[attr_spec['name']] = decoded
@@ -554,9 +569,9 @@ class YnlFamily(SpecFamily):
if op.fixed_header:
fixed_header_members = self.consts[op.fixed_header].members
for m in fixed_header_members:
- value = vals.pop(m.name)
- format, _ = NlAttr.type_formats[m.type]
- msg += struct.pack(format, value)
+ value = vals.pop(m.name) if m.name in vals else 0
+ format = NlAttr.get_format(m.type, m.byte_order)
+ msg += format.pack(value)
for name, value in vals.items():
msg += self._add_attr(op.attr_set.name, name, value)
msg = _genl_msg_finalize(msg)