diff options
Diffstat (limited to 'tools/net/ynl/lib/nlspec.py')
| -rw-r--r-- | tools/net/ynl/lib/nlspec.py | 414 |
1 files changed, 414 insertions, 0 deletions
diff --git a/tools/net/ynl/lib/nlspec.py b/tools/net/ynl/lib/nlspec.py new file mode 100644 index 000000000000..a34d088f6743 --- /dev/null +++ b/tools/net/ynl/lib/nlspec.py @@ -0,0 +1,414 @@ +# SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause + +import collections +import importlib +import os +import yaml + + +# To be loaded dynamically as needed +jsonschema = None + + +class SpecElement: + """Netlink spec element. + + Abstract element of the Netlink spec. Implements the dictionary interface + for access to the raw spec. Supports iterative resolution of dependencies + across elements and class inheritance levels. The elements of the spec + may refer to each other, and although loops should be very rare, having + to maintain correct ordering of instantiation is painful, so the resolve() + method should be used to perform parts of init which require access to + other parts of the spec. + + Attributes: + yaml raw spec as loaded from the spec file + family back reference to the full family + + name name of the entity as listed in the spec (optional) + ident_name name which can be safely used as identifier in code (optional) + """ + def __init__(self, family, yaml): + self.yaml = yaml + self.family = family + + if 'name' in self.yaml: + self.name = self.yaml['name'] + self.ident_name = self.name.replace('-', '_') + + self._super_resolved = False + family.add_unresolved(self) + + def __getitem__(self, key): + return self.yaml[key] + + def __contains__(self, key): + return key in self.yaml + + def get(self, key, default=None): + return self.yaml.get(key, default) + + def resolve_up(self, up): + if not self._super_resolved: + up.resolve() + self._super_resolved = True + + def resolve(self): + pass + + +class SpecEnumEntry(SpecElement): + """ Entry within an enum declared in the Netlink spec. + + Attributes: + doc documentation string + enum_set back reference to the enum + value numerical value of this enum (use accessors in most situations!) + + Methods: + raw_value raw value, i.e. the id in the enum, unlike user value which is a mask for flags + user_value user value, same as raw value for enums, for flags it's the mask + """ + def __init__(self, enum_set, yaml, prev, value_start): + if isinstance(yaml, str): + yaml = {'name': yaml} + super().__init__(enum_set.family, yaml) + + self.doc = yaml.get('doc', '') + self.enum_set = enum_set + + if 'value' in yaml: + self.value = yaml['value'] + elif prev: + self.value = prev.value + 1 + else: + self.value = value_start + + def has_doc(self): + return bool(self.doc) + + def raw_value(self): + return self.value + + def user_value(self): + if self.enum_set['type'] == 'flags': + return 1 << self.value + else: + return self.value + + +class SpecEnumSet(SpecElement): + """ Enum type + + Represents an enumeration (list of numerical constants) + as declared in the "definitions" section of the spec. + + Attributes: + type enum or flags + entries entries by name + entries_by_val entries by value + Methods: + get_mask for flags compute the mask of all defined values + """ + def __init__(self, family, yaml): + super().__init__(family, yaml) + + self.type = yaml['type'] + + prev_entry = None + value_start = self.yaml.get('value-start', 0) + self.entries = dict() + self.entries_by_val = dict() + for entry in self.yaml['entries']: + e = self.new_entry(entry, prev_entry, value_start) + self.entries[e.name] = e + self.entries_by_val[e.raw_value()] = e + prev_entry = e + + def new_entry(self, entry, prev_entry, value_start): + return SpecEnumEntry(self, entry, prev_entry, value_start) + + def has_doc(self): + if 'doc' in self.yaml: + return True + for entry in self.entries.values(): + if entry.has_doc(): + return True + return False + + def get_mask(self): + mask = 0 + idx = self.yaml.get('value-start', 0) + for _ in self.entries.values(): + mask |= 1 << idx + idx += 1 + return mask + + +class SpecAttr(SpecElement): + """ Single Netlink atttribute type + + Represents a single attribute type within an attr space. + + Attributes: + value numerical ID when serialized + attr_set Attribute Set containing this attr + """ + def __init__(self, family, attr_set, yaml, value): + super().__init__(family, yaml) + + self.value = value + self.attr_set = attr_set + self.is_multi = yaml.get('multi-attr', False) + + +class SpecAttrSet(SpecElement): + """ Netlink Attribute Set class. + + Represents a ID space of attributes within Netlink. + + Note that unlike other elements, which expose contents of the raw spec + via the dictionary interface Attribute Set exposes attributes by name. + + Attributes: + attrs ordered dict of all attributes (indexed by name) + attrs_by_val ordered dict of all attributes (indexed by value) + subset_of parent set if this is a subset, otherwise None + """ + def __init__(self, family, yaml): + super().__init__(family, yaml) + + self.subset_of = self.yaml.get('subset-of', None) + + self.attrs = collections.OrderedDict() + self.attrs_by_val = collections.OrderedDict() + + if self.subset_of is None: + val = 1 + for elem in self.yaml['attributes']: + if 'value' in elem: + val = elem['value'] + + attr = self.new_attr(elem, val) + self.attrs[attr.name] = attr + self.attrs_by_val[attr.value] = attr + val += 1 + else: + real_set = family.attr_sets[self.subset_of] + for elem in self.yaml['attributes']: + attr = real_set[elem['name']] + self.attrs[attr.name] = attr + self.attrs_by_val[attr.value] = attr + + def new_attr(self, elem, value): + return SpecAttr(self.family, self, elem, value) + + def __getitem__(self, key): + return self.attrs[key] + + def __contains__(self, key): + return key in self.attrs + + def __iter__(self): + yield from self.attrs + + def items(self): + return self.attrs.items() + + +class SpecOperation(SpecElement): + """Netlink Operation + + Information about a single Netlink operation. + + Attributes: + value numerical ID when serialized, None if req/rsp values differ + + req_value numerical ID when serialized, user -> kernel + rsp_value numerical ID when serialized, user <- kernel + is_call bool, whether the operation is a call + is_async bool, whether the operation is a notification + is_resv bool, whether the operation does not exist (it's just a reserved ID) + attr_set attribute set name + + yaml raw spec as loaded from the spec file + """ + def __init__(self, family, yaml, req_value, rsp_value): + super().__init__(family, yaml) + + self.value = req_value if req_value == rsp_value else None + self.req_value = req_value + self.rsp_value = rsp_value + + self.is_call = 'do' in yaml or 'dump' in yaml + self.is_async = 'notify' in yaml or 'event' in yaml + self.is_resv = not self.is_async and not self.is_call + + # Added by resolve: + self.attr_set = None + delattr(self, "attr_set") + + def resolve(self): + self.resolve_up(super()) + + if 'attribute-set' in self.yaml: + attr_set_name = self.yaml['attribute-set'] + elif 'notify' in self.yaml: + msg = self.family.msgs[self.yaml['notify']] + attr_set_name = msg['attribute-set'] + elif self.is_resv: + attr_set_name = '' + else: + raise Exception(f"Can't resolve attribute set for op '{self.name}'") + if attr_set_name: + self.attr_set = self.family.attr_sets[attr_set_name] + + +class SpecFamily(SpecElement): + """ Netlink Family Spec class. + + Netlink family information loaded from a spec (e.g. in YAML). + Takes care of unfolding implicit information which can be skipped + in the spec itself for brevity. + + The class can be used like a dictionary to access the raw spec + elements but that's usually a bad idea. + + Attributes: + proto protocol type (e.g. genetlink) + + attr_sets dict of attribute sets + msgs dict of all messages (index by name) + msgs_by_value dict of all messages (indexed by name) + ops dict of all valid requests / responses + consts dict of all constants/enums + """ + def __init__(self, spec_path, schema_path=None): + with open(spec_path, "r") as stream: + spec = yaml.safe_load(stream) + + self._resolution_list = [] + + super().__init__(self, spec) + + self.proto = self.yaml.get('protocol', 'genetlink') + + if schema_path is None: + schema_path = os.path.dirname(os.path.dirname(spec_path)) + f'/{self.proto}.yaml' + if schema_path: + global jsonschema + + with open(schema_path, "r") as stream: + schema = yaml.safe_load(stream) + + if jsonschema is None: + jsonschema = importlib.import_module("jsonschema") + + jsonschema.validate(self.yaml, schema) + + self.attr_sets = collections.OrderedDict() + self.msgs = collections.OrderedDict() + self.req_by_value = collections.OrderedDict() + self.rsp_by_value = collections.OrderedDict() + self.ops = collections.OrderedDict() + self.consts = collections.OrderedDict() + + last_exception = None + while len(self._resolution_list) > 0: + resolved = [] + unresolved = self._resolution_list + self._resolution_list = [] + + for elem in unresolved: + try: + elem.resolve() + except (KeyError, AttributeError) as e: + self._resolution_list.append(elem) + last_exception = e + continue + + resolved.append(elem) + + if len(resolved) == 0: + raise last_exception + + def new_enum(self, elem): + return SpecEnumSet(self, elem) + + def new_attr_set(self, elem): + return SpecAttrSet(self, elem) + + def new_operation(self, elem, req_val, rsp_val): + return SpecOperation(self, elem, req_val, rsp_val) + + def add_unresolved(self, elem): + self._resolution_list.append(elem) + + def _dictify_ops_unified(self): + val = 1 + for elem in self.yaml['operations']['list']: + if 'value' in elem: + val = elem['value'] + + op = self.new_operation(elem, val, val) + val += 1 + + self.msgs[op.name] = op + + def _dictify_ops_directional(self): + req_val = rsp_val = 1 + for elem in self.yaml['operations']['list']: + if 'notify' in elem: + if 'value' in elem: + rsp_val = elem['value'] + req_val_next = req_val + rsp_val_next = rsp_val + 1 + req_val = None + elif 'do' in elem or 'dump' in elem: + mode = elem['do'] if 'do' in elem else elem['dump'] + + v = mode.get('request', {}).get('value', None) + if v: + req_val = v + v = mode.get('reply', {}).get('value', None) + if v: + rsp_val = v + + rsp_inc = 1 if 'reply' in mode else 0 + req_val_next = req_val + 1 + rsp_val_next = rsp_val + rsp_inc + else: + raise Exception("Can't parse directional ops") + + op = self.new_operation(elem, req_val, rsp_val) + req_val = req_val_next + rsp_val = rsp_val_next + + self.msgs[op.name] = op + + def resolve(self): + self.resolve_up(super()) + + for elem in self.yaml['definitions']: + if elem['type'] == 'enum' or elem['type'] == 'flags': + self.consts[elem['name']] = self.new_enum(elem) + else: + self.consts[elem['name']] = elem + + for elem in self.yaml['attribute-sets']: + attr_set = self.new_attr_set(elem) + self.attr_sets[elem['name']] = attr_set + + msg_id_model = self.yaml['operations'].get('enum-model', 'unified') + if msg_id_model == 'unified': + self._dictify_ops_unified() + elif msg_id_model == 'directional': + self._dictify_ops_directional() + + for op in self.msgs.values(): + if op.req_value is not None: + self.req_by_value[op.req_value] = op + if op.rsp_value is not None: + self.rsp_by_value[op.rsp_value] = op + if not op.is_async and 'attribute-set' in op: + self.ops[op.name] = op |