| # SPDX-License-Identifier: GPL-2.0 OR BSD-3-Clause | |
| # | |
| # pylint: disable=missing-function-docstring, too-many-instance-attributes, too-many-branches | |
| """ | |
| The nlspec is a python library for parsing and using YNL netlink | |
| specifications. | |
| """ | |
| import collections | |
| import importlib | |
| import os | |
| import yaml as pyyaml | |
| from .specdir import find_spec, SYS_SCHEMA_DIR | |
| class SpecException(Exception): | |
| """Netlink spec exception. | |
| """ | |
| 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, as_flags=None): | |
| if self.enum_set['type'] == 'flags' or as_flags: | |
| return 1 << self.value | |
| 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 = {} | |
| self.entries_by_val = {} | |
| 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 | |
| return self.has_entry_doc() | |
| def has_entry_doc(self): | |
| for entry in self.entries.values(): | |
| if entry.has_doc(): | |
| return True | |
| return False | |
| def get_mask(self, as_flags=None): | |
| mask = 0 | |
| for e in self.entries.values(): | |
| mask += e.user_value(as_flags) | |
| return mask | |
| class SpecAttr(SpecElement): | |
| """ Single Netlink attribute type | |
| Represents a single attribute type within an attr space. | |
| Attributes: | |
| type string, attribute type | |
| value numerical ID when serialized | |
| attr_set Attribute Set containing this attr | |
| is_multi bool, attr may repeat multiple times | |
| struct_name string, name of struct definition | |
| sub_type string, name of sub type | |
| len integer, optional byte length of binary types | |
| display_hint string, hint to help choose format specifier | |
| when displaying the value | |
| sub_message string, name of sub message type | |
| selector string, name of attribute used to select | |
| sub-message type | |
| is_auto_scalar bool, attr is a variable-size scalar | |
| """ | |
| def __init__(self, family, attr_set, yaml, value): | |
| super().__init__(family, yaml) | |
| self.type = yaml['type'] | |
| self.value = value | |
| self.attr_set = attr_set | |
| self.is_multi = yaml.get('multi-attr', False) | |
| self.struct_name = yaml.get('struct') | |
| self.sub_type = yaml.get('sub-type') | |
| self.byte_order = yaml.get('byte-order') | |
| self.len = yaml.get('len') | |
| self.display_hint = yaml.get('display-hint') | |
| self.sub_message = yaml.get('sub-message') | |
| self.selector = yaml.get('selector') | |
| self.is_auto_scalar = self.type in ("sint", "uint") | |
| 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']: | |
| real_attr = real_set[elem['name']] | |
| combined_elem = real_attr.yaml | elem | |
| attr = self.new_attr(combined_elem, real_attr.value) | |
| 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 SpecStructMember(SpecElement): | |
| """Struct member attribute | |
| Represents a single struct member attribute. | |
| Attributes: | |
| type string, type of the member attribute | |
| byte_order string or None for native byte order | |
| enum string, name of the enum definition | |
| len integer, optional byte length of binary types | |
| display_hint string, hint to help choose format specifier | |
| when displaying the value | |
| struct string, name of nested struct type | |
| """ | |
| def __init__(self, family, yaml): | |
| super().__init__(family, yaml) | |
| self.type = yaml['type'] | |
| self.byte_order = yaml.get('byte-order') | |
| self.enum = yaml.get('enum') | |
| self.len = yaml.get('len') | |
| self.display_hint = yaml.get('display-hint') | |
| self.struct = yaml.get('struct') | |
| class SpecStruct(SpecElement): | |
| """Netlink struct type | |
| Represents a C struct definition. | |
| Attributes: | |
| members ordered list of struct members | |
| """ | |
| def __init__(self, family, yaml): | |
| super().__init__(family, yaml) | |
| self.members = [] | |
| for member in yaml.get('members', []): | |
| self.members.append(self.new_member(family, member)) | |
| def new_member(self, family, elem): | |
| return SpecStructMember(family, elem) | |
| def __iter__(self): | |
| yield from self.members | |
| def items(self): | |
| return self.members | |
| class SpecSubMessage(SpecElement): | |
| """ Netlink sub-message definition | |
| Represents a set of sub-message formats for polymorphic nlattrs | |
| that contain type-specific sub messages. | |
| Attributes: | |
| name string, name of sub-message definition | |
| formats dict of sub-message formats indexed by match value | |
| """ | |
| def __init__(self, family, yaml): | |
| super().__init__(family, yaml) | |
| self.formats = collections.OrderedDict() | |
| for elem in self.yaml['formats']: | |
| msg_format = self.new_format(family, elem) | |
| self.formats[msg_format.value] = msg_format | |
| def new_format(self, family, msg_format): | |
| return SpecSubMessageFormat(family, msg_format) | |
| class SpecSubMessageFormat(SpecElement): | |
| """ Netlink sub-message format definition | |
| Represents a single format for a sub-message. | |
| Attributes: | |
| value attribute value to match against type selector | |
| fixed_header string, name of fixed header, or None | |
| attr_set string, name of attribute set, or None | |
| """ | |
| def __init__(self, family, yaml): | |
| super().__init__(family, yaml) | |
| self.value = yaml.get('value') | |
| self.fixed_header = yaml.get('fixed-header') | |
| self.attr_set = yaml.get('attribute-set') | |
| 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 | |
| modes supported operation modes (do, dump, event etc.) | |
| 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 | |
| fixed_header string, optional name of fixed header struct | |
| 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.modes = yaml.keys() & {'do', 'dump', 'event', 'notify'} | |
| 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 | |
| self.fixed_header = self.yaml.get('fixed-header', family.fixed_header) | |
| # 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 SpecException(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 SpecMcastGroup(SpecElement): | |
| """Netlink Multicast Group | |
| Information about a multicast group. | |
| Value is only used for classic netlink families that use the | |
| netlink-raw schema. Genetlink families use dynamic ID allocation | |
| where the ids of multicast groups get resolved at runtime. Value | |
| will be None for genetlink families. | |
| Attributes: | |
| name name of the mulitcast group | |
| value integer id of this multicast group for netlink-raw or None | |
| yaml raw spec as loaded from the spec file | |
| """ | |
| def __init__(self, family, yaml): | |
| super().__init__(family, yaml) | |
| self.value = self.yaml.get('value') | |
| 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) | |
| msg_id_model enum-model for operations (unified, directional etc.) | |
| license spec license (loaded from an SPDX tag on the spec) | |
| attr_sets dict of attribute sets | |
| msgs dict of all messages (index by name) | |
| sub_msgs dict of all sub messages (index by name) | |
| ops dict of all valid requests / responses | |
| ntfs dict of all async events | |
| consts dict of all constants/enums | |
| fixed_header string, optional name of family default fixed header struct | |
| mcast_groups dict of all multicast groups (index by name) | |
| kernel_family dict of kernel family attributes | |
| """ | |
| # To be loaded dynamically as needed | |
| jsonschema = None | |
| try: | |
| _yaml_loader = pyyaml.CSafeLoader | |
| except AttributeError: | |
| _yaml_loader = pyyaml.SafeLoader | |
| def __init__(self, spec_path=None, schema_path=None, exclude_ops=None, | |
| family=None): | |
| # schema_path selects how the spec is validated: | |
| # None -- no preference: validate against the default schema, | |
| # but trust (skip) installed specs selected by family= | |
| # True -- always validate against the default schema | |
| # path -- validate against this schema | |
| # '' -- do not validate | |
| if (spec_path is None) == (family is None): | |
| raise ValueError("Specify exactly one of spec path or family name") | |
| if family is not None: | |
| spec_path = find_spec(family) | |
| # Installed specs are assumed correct, so skip schema validation | |
| # to save cycles unless the caller asked to validate. | |
| if schema_path is None and spec_path.startswith(SYS_SCHEMA_DIR): | |
| schema_path = '' | |
| with open(spec_path, "r", encoding='utf-8') as stream: | |
| prefix = '# SPDX-License-Identifier: ' | |
| first = stream.readline().strip() | |
| if not first.startswith(prefix): | |
| raise SpecException('SPDX license tag required in the spec') | |
| self.license = first[len(prefix):] | |
| stream.seek(0) | |
| spec = pyyaml.load(stream, Loader=self._yaml_loader) | |
| self.fixed_header = None | |
| self._resolution_list = [] | |
| super().__init__(self, spec) | |
| self._exclude_ops = exclude_ops if exclude_ops else [] | |
| self.proto = self.yaml.get('protocol', 'genetlink') | |
| self.msg_id_model = self.yaml['operations'].get('enum-model', 'unified') | |
| if schema_path is None or schema_path is True: | |
| schema_path = os.path.dirname(os.path.dirname(spec_path)) + f'/{self.proto}.yaml' | |
| if schema_path: | |
| with open(schema_path, "r", encoding='utf-8') as stream: | |
| schema = pyyaml.load(stream, Loader=self._yaml_loader) | |
| if SpecFamily.jsonschema is None: | |
| SpecFamily.jsonschema = importlib.import_module("jsonschema") | |
| SpecFamily.jsonschema.validate(self.yaml, schema) | |
| self.attr_sets = collections.OrderedDict() | |
| self.sub_msgs = collections.OrderedDict() | |
| self.msgs = collections.OrderedDict() | |
| self.req_by_value = collections.OrderedDict() | |
| self.rsp_by_value = collections.OrderedDict() | |
| self.ops = collections.OrderedDict() | |
| self.ntfs = collections.OrderedDict() | |
| self.consts = collections.OrderedDict() | |
| self.mcast_groups = collections.OrderedDict() | |
| self.kernel_family = collections.OrderedDict(self.yaml.get('kernel-family', {})) | |
| 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_struct(self, elem): | |
| return SpecStruct(self, elem) | |
| def new_sub_message(self, elem): | |
| return SpecSubMessage(self, elem) | |
| def new_operation(self, elem, req_val, rsp_val): | |
| return SpecOperation(self, elem, req_val, rsp_val) | |
| def new_mcast_group(self, elem): | |
| return SpecMcastGroup(self, elem) | |
| def add_unresolved(self, elem): | |
| self._resolution_list.append(elem) | |
| def _dictify_ops_unified(self): | |
| self.fixed_header = self.yaml['operations'].get('fixed-header') | |
| 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): | |
| self.fixed_header = self.yaml['operations'].get('fixed-header') | |
| req_val = rsp_val = 1 | |
| for elem in self.yaml['operations']['list']: | |
| if 'notify' in elem or 'event' 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 SpecException("Can't parse directional ops") | |
| if req_val == req_val_next: | |
| req_val = None | |
| if rsp_val == rsp_val_next: | |
| rsp_val = None | |
| skip = False | |
| for exclude in self._exclude_ops: | |
| skip |= bool(exclude.match(elem['name'])) | |
| if not skip: | |
| op = self.new_operation(elem, req_val, rsp_val) | |
| self.msgs[op.name] = op | |
| req_val = req_val_next | |
| rsp_val = rsp_val_next | |
| def find_operation(self, name): | |
| """ | |
| For a given operation name, find and return operation spec. | |
| """ | |
| for op in self.yaml['operations']['list']: | |
| if name == op['name']: | |
| return op | |
| return None | |
| def resolve(self): | |
| self.resolve_up(super()) | |
| definitions = self.yaml.get('definitions', []) | |
| for elem in definitions: | |
| if elem['type'] == 'enum' or elem['type'] == 'flags': | |
| self.consts[elem['name']] = self.new_enum(elem) | |
| elif elem['type'] == 'struct': | |
| self.consts[elem['name']] = self.new_struct(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 | |
| for elem in self.yaml.get('sub-messages', []): | |
| sub_message = self.new_sub_message(elem) | |
| self.sub_msgs[sub_message.name] = sub_message | |
| if self.msg_id_model == 'unified': | |
| self._dictify_ops_unified() | |
| elif self.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 | |
| elif op.is_async: | |
| self.ntfs[op.name] = op | |
| mcgs = self.yaml.get('mcast-groups') | |
| if mcgs: | |
| for elem in mcgs['list']: | |
| mcg = self.new_mcast_group(elem) | |
| self.mcast_groups[elem['name']] = mcg |