| # 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, as_flags=None): |
| if self.enum_set['type'] == 'flags' or as_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, as_flags=None): |
| mask = 0 |
| for e in self.entries.values(): |
| mask += e.user_value(as_flags) |
| 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 |
| is_multi bool, attr may repeat multiple times |
| struct_name string, name of struct definition |
| sub_type string, name of sub type |
| """ |
| 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) |
| self.struct_name = yaml.get('struct') |
| self.sub_type = yaml.get('sub-type') |
| self.byte_order = yaml.get('byte-order') |
| |
| |
| 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 SpecStructMember(SpecElement): |
| """Struct member attribute |
| |
| Represents a single struct member attribute. |
| |
| Attributes: |
| type string, type of the member attribute |
| """ |
| def __init__(self, family, yaml): |
| super().__init__(family, yaml) |
| self.type = yaml['type'] |
| |
| |
| 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.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 |
| 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.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 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) |
| 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) |
| msgs_by_value dict of all messages (indexed by name) |
| ops dict of all valid requests / responses |
| consts dict of all constants/enums |
| fixed_header string, optional name of family default fixed header struct |
| """ |
| def __init__(self, spec_path, schema_path=None): |
| with open(spec_path, "r") as stream: |
| prefix = '# SPDX-License-Identifier: ' |
| first = stream.readline().strip() |
| if not first.startswith(prefix): |
| raise Exception('SPDX license tag required in the spec') |
| self.license = first[len(prefix):] |
| |
| stream.seek(0) |
| 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_struct(self, elem): |
| return SpecStruct(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): |
| 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: |
| 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 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 |
| |
| 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 |
