| #!/usr/bin/env python3 |
| # ex: set filetype=python: |
| |
| """Define and implement the Abstract Syntax Tree for the XDR language.""" |
| |
| import sys |
| from typing import List |
| from dataclasses import dataclass, KW_ONLY |
| |
| from lark import ast_utils, Transformer |
| from lark.tree import Meta |
| |
| this_module = sys.modules[__name__] |
| |
| big_endian = [] |
| excluded_apis = [] |
| header_name = "none" |
| public_apis = [] |
| structs = set() |
| pass_by_reference = set() |
| |
| constants = {} |
| |
| |
| def xdr_quadlen(val: str) -> int: |
| """Return integer XDR width of an XDR type""" |
| if val in constants: |
| octets = constants[val] |
| else: |
| octets = int(val) |
| return int((octets + 3) / 4) |
| |
| |
| symbolic_widths = { |
| "void": ["XDR_void"], |
| "bool": ["XDR_bool"], |
| "short": ["XDR_short"], |
| "unsigned_short": ["XDR_unsigned_short"], |
| "int": ["XDR_int"], |
| "unsigned_int": ["XDR_unsigned_int"], |
| "long": ["XDR_long"], |
| "unsigned_long": ["XDR_unsigned_long"], |
| "hyper": ["XDR_hyper"], |
| "unsigned_hyper": ["XDR_unsigned_hyper"], |
| } |
| |
| # Numeric XDR widths are tracked in a dictionary that is keyed |
| # by type_name because sometimes a caller has nothing more than |
| # the type_name to use to figure out the numeric width. |
| max_widths = { |
| "void": 0, |
| "bool": 1, |
| "short": 1, |
| "unsigned_short": 1, |
| "int": 1, |
| "unsigned_int": 1, |
| "long": 1, |
| "unsigned_long": 1, |
| "hyper": 2, |
| "unsigned_hyper": 2, |
| } |
| |
| |
| @dataclass |
| class _XdrAst(ast_utils.Ast): |
| """Base class for the XDR abstract syntax tree""" |
| |
| # Source position of the construct's declared identifier, when |
| # the transformer records one, so semantic diagnostics can point |
| # at the exact declaration. The KW_ONLY marker makes the fields |
| # keyword-only, so they never disturb the positional child |
| # ordering lark uses to build each node; 0 means the position was |
| # not recorded. |
| _: KW_ONLY |
| line: int = 0 |
| column: int = 0 |
| |
| |
| @dataclass |
| class _XdrIdentifier(_XdrAst): |
| """Corresponds to 'identifier' in the XDR language grammar""" |
| |
| symbol: str |
| |
| |
| @dataclass |
| class _XdrValue(_XdrAst): |
| """Corresponds to 'value' in the XDR language grammar""" |
| |
| value: str |
| |
| |
| @dataclass |
| class _XdrConstantValue(_XdrAst): |
| """Corresponds to 'constant' in the XDR language grammar""" |
| |
| value: int |
| |
| |
| @dataclass |
| class _XdrTypeSpecifier(_XdrAst): |
| """Corresponds to 'type_specifier' in the XDR language grammar""" |
| |
| type_name: str |
| c_classifier: str = "" |
| |
| |
| @dataclass |
| class _XdrDefinedType(_XdrTypeSpecifier): |
| """Corresponds to a type defined by the input specification""" |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return [get_header_name().upper() + "_" + self.type_name + "_sz"] |
| |
| def __post_init__(self): |
| if self.type_name in structs: |
| self.c_classifier = "struct " |
| symbolic_widths[self.type_name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrBuiltInType(_XdrTypeSpecifier): |
| """Corresponds to a built-in XDR type""" |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return symbolic_widths[self.type_name] |
| |
| |
| @dataclass |
| class _XdrDeclaration(_XdrAst): |
| """Base class of XDR type declarations""" |
| |
| |
| @dataclass |
| class _XdrFixedLengthOpaque(_XdrDeclaration): |
| """A fixed-length opaque declaration""" |
| |
| name: str |
| size: str |
| template: str = "fixed_length_opaque" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return xdr_quadlen(self.size) |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return ["XDR_QUADLEN(" + self.size + ")"] |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrVariableLengthOpaque(_XdrDeclaration): |
| """A variable-length opaque declaration""" |
| |
| name: str |
| maxsize: str |
| template: str = "variable_length_opaque" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 1 + xdr_quadlen(self.maxsize) |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| widths = ["XDR_unsigned_int"] |
| if self.maxsize != "0": |
| widths.append("XDR_QUADLEN(" + self.maxsize + ")") |
| return widths |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrString(_XdrDeclaration): |
| """A (NUL-terminated) variable-length string declaration""" |
| |
| name: str |
| maxsize: str |
| template: str = "string" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 1 + xdr_quadlen(self.maxsize) |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| widths = ["XDR_unsigned_int"] |
| if self.maxsize != "0": |
| widths.append("XDR_QUADLEN(" + self.maxsize + ")") |
| return widths |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrFixedLengthArray(_XdrDeclaration): |
| """A fixed-length array declaration""" |
| |
| name: str |
| spec: _XdrTypeSpecifier |
| size: str |
| template: str = "fixed_length_array" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return xdr_quadlen(self.size) * max_widths[self.spec.type_name] |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| item_width = " + ".join(symbolic_widths[self.spec.type_name]) |
| return ["(" + self.size + " * (" + item_width + "))"] |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrVariableLengthArray(_XdrDeclaration): |
| """A variable-length array declaration""" |
| |
| name: str |
| spec: _XdrTypeSpecifier |
| maxsize: str |
| template: str = "variable_length_array" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 1 + (xdr_quadlen(self.maxsize) * max_widths[self.spec.type_name]) |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| widths = ["XDR_unsigned_int"] |
| if self.maxsize != "0": |
| item_width = " + ".join(symbolic_widths[self.spec.type_name]) |
| widths.append("(" + self.maxsize + " * (" + item_width + "))") |
| return widths |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrOptionalData(_XdrDeclaration): |
| """An 'optional_data' declaration""" |
| |
| name: str |
| spec: _XdrTypeSpecifier |
| template: str = "optional_data" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 1 |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return ["XDR_bool"] |
| |
| def __post_init__(self): |
| structs.add(self.name) |
| pass_by_reference.add(self.name) |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrBasic(_XdrDeclaration): |
| """A 'basic' declaration""" |
| |
| name: str |
| spec: _XdrTypeSpecifier |
| template: str = "basic" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return max_widths[self.spec.type_name] |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return symbolic_widths[self.spec.type_name] |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrVoid(_XdrDeclaration): |
| """A void declaration""" |
| |
| name: str = "void" |
| template: str = "void" |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 0 |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return [] |
| |
| |
| @dataclass |
| class _XdrConstant(_XdrAst): |
| """Corresponds to 'constant_def' in the grammar""" |
| |
| name: str |
| value: str |
| |
| def __post_init__(self): |
| if self.value not in constants: |
| constants[self.name] = int(self.value, 0) |
| |
| |
| @dataclass |
| class _XdrEnumerator(_XdrAst): |
| """An 'identifier = value' enumerator""" |
| |
| name: str |
| value: str |
| |
| def __post_init__(self): |
| if self.value not in constants: |
| constants[self.name] = int(self.value, 0) |
| |
| |
| @dataclass |
| class _XdrEnum(_XdrAst): |
| """An XDR enum definition""" |
| |
| name: str |
| enumerators: List[_XdrEnumerator] |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return 1 |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return ["XDR_int"] |
| |
| def __post_init__(self): |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrStruct(_XdrAst): |
| """An XDR struct definition""" |
| |
| name: str |
| fields: List[_XdrDeclaration] |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| width = 0 |
| for field in self.fields: |
| width += field.max_width() |
| return width |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| widths = [] |
| for field in self.fields: |
| widths += field.symbolic_width() |
| return widths |
| |
| def __post_init__(self): |
| structs.add(self.name) |
| pass_by_reference.add(self.name) |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrPointer(_XdrAst): |
| """An XDR pointer definition""" |
| |
| name: str |
| fields: List[_XdrDeclaration] |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| width = 1 |
| for field in self.fields[0:-1]: |
| width += field.max_width() |
| return width |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| widths = [] |
| widths += ["XDR_bool"] |
| for field in self.fields[0:-1]: |
| widths += field.symbolic_width() |
| return widths |
| |
| def __post_init__(self): |
| structs.add(self.name) |
| pass_by_reference.add(self.name) |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrTypedef(_XdrAst): |
| """An XDR typedef""" |
| |
| declaration: _XdrDeclaration |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| return self.declaration.max_width() |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| return self.declaration.symbolic_width() |
| |
| def __post_init__(self): |
| if isinstance(self.declaration, _XdrBasic): |
| new_type = self.declaration |
| if isinstance(new_type.spec, _XdrDefinedType): |
| if new_type.spec.type_name in pass_by_reference: |
| pass_by_reference.add(new_type.name) |
| max_widths[new_type.name] = self.max_width() |
| symbolic_widths[new_type.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _XdrCaseSpec(_XdrAst): |
| """One case in an XDR union""" |
| |
| values: List[str] |
| arm: _XdrDeclaration |
| template: str = "case_spec" |
| |
| |
| @dataclass |
| class _XdrDefaultSpec(_XdrAst): |
| """Default case in an XDR union""" |
| |
| arm: _XdrDeclaration |
| template: str = "default_spec" |
| |
| |
| @dataclass |
| class _XdrUnion(_XdrAst): |
| """An XDR union""" |
| |
| name: str |
| discriminant: _XdrDeclaration |
| cases: List[_XdrCaseSpec] |
| default: _XdrDeclaration |
| |
| def max_width(self) -> int: |
| """Return width of type in XDR_UNITS""" |
| max_width = 0 |
| for case in self.cases: |
| if case.arm.max_width() > max_width: |
| max_width = case.arm.max_width() |
| if self.default: |
| if self.default.arm.max_width() > max_width: |
| max_width = self.default.arm.max_width() |
| return 1 + max_width |
| |
| def symbolic_width(self) -> List: |
| """Return list containing XDR width of type's components""" |
| max_width = 0 |
| for case in self.cases: |
| if case.arm.max_width() > max_width: |
| max_width = case.arm.max_width() |
| width = case.arm.symbolic_width() |
| if self.default: |
| if self.default.arm.max_width() > max_width: |
| max_width = self.default.arm.max_width() |
| width = self.default.arm.symbolic_width() |
| return symbolic_widths[self.discriminant.name] + width |
| |
| def __post_init__(self): |
| structs.add(self.name) |
| pass_by_reference.add(self.name) |
| max_widths[self.name] = self.max_width() |
| symbolic_widths[self.name] = self.symbolic_width() |
| |
| |
| @dataclass |
| class _RpcProcedure(_XdrAst): |
| """RPC procedure definition""" |
| |
| name: str |
| number: int |
| argument: _XdrTypeSpecifier |
| result: _XdrTypeSpecifier |
| |
| |
| @dataclass |
| class _RpcVersion(_XdrAst): |
| """RPC version definition""" |
| |
| name: str |
| number: int |
| procedures: List[_RpcProcedure] |
| |
| |
| @dataclass |
| class _RpcProgram(_XdrAst): |
| """RPC program definition""" |
| |
| name: str |
| number: int |
| versions: List[_RpcVersion] |
| |
| |
| @dataclass |
| class _Pragma(_XdrAst): |
| """Empty class for pragma directives""" |
| |
| |
| @dataclass |
| class _XdrPassthru(_XdrAst): |
| """Passthrough line to emit verbatim in output""" |
| |
| content: str |
| |
| |
| @dataclass |
| class Definition(_XdrAst, ast_utils.WithMeta): |
| """Corresponds to 'definition' in the grammar""" |
| |
| meta: Meta |
| value: _XdrAst |
| |
| |
| @dataclass |
| class Specification(_XdrAst, ast_utils.AsList): |
| """Corresponds to 'specification' in the grammar""" |
| |
| definitions: List[Definition] |
| |
| |
| class ParseToAst(Transformer): |
| """Functions that transform productions into AST nodes""" |
| |
| def identifier(self, children): |
| """Instantiate one _XdrIdentifier object""" |
| token = children[0] |
| return _XdrIdentifier(token.value, line=token.line, column=token.column) |
| |
| def value(self, children): |
| """Instantiate one _XdrValue object""" |
| if isinstance(children[0], _XdrIdentifier): |
| return _XdrValue(children[0].symbol) |
| return _XdrValue(children[0].children[0].value) |
| |
| def constant(self, children): |
| """Instantiate one _XdrConstantValue object""" |
| match children[0].data: |
| case "decimal_constant": |
| value = int(children[0].children[0].value, base=10) |
| case "hexadecimal_constant": |
| value = int(children[0].children[0].value, base=16) |
| case "octal_constant": |
| value = int(children[0].children[0].value, base=8) |
| return _XdrConstantValue(value) |
| |
| def type_specifier(self, children): |
| """Instantiate one _XdrTypeSpecifier object""" |
| if isinstance(children[0], _XdrIdentifier): |
| name = children[0].symbol |
| return _XdrDefinedType(type_name=name) |
| |
| name = children[0].data.value |
| return _XdrBuiltInType(type_name=name) |
| |
| def constant_def(self, children): |
| """Instantiate one _XdrConstant object""" |
| ident = children[0] |
| value = children[1].value |
| return _XdrConstant(ident.symbol, value, line=ident.line, column=ident.column) |
| |
| def enum(self, children): |
| """Instantiate one _XdrEnum object""" |
| name_ident = children[0] |
| |
| i = 0 |
| enumerators = [] |
| body = children[1] |
| while i < len(body.children): |
| ident = body.children[i] |
| value = body.children[i + 1].value |
| enumerators.append( |
| _XdrEnumerator( |
| ident.symbol, value, line=ident.line, column=ident.column |
| ) |
| ) |
| i = i + 2 |
| |
| return _XdrEnum( |
| name_ident.symbol, |
| enumerators, |
| line=name_ident.line, |
| column=name_ident.column, |
| ) |
| |
| def fixed_length_opaque(self, children): |
| """Instantiate one _XdrFixedLengthOpaque declaration object""" |
| ident = children[0] |
| size = children[1].value |
| |
| return _XdrFixedLengthOpaque( |
| ident.symbol, size, line=ident.line, column=ident.column |
| ) |
| |
| def variable_length_opaque(self, children): |
| """Instantiate one _XdrVariableLengthOpaque declaration object""" |
| ident = children[0] |
| if children[1] is not None: |
| maxsize = children[1].value |
| else: |
| maxsize = "0" |
| |
| return _XdrVariableLengthOpaque( |
| ident.symbol, maxsize, line=ident.line, column=ident.column |
| ) |
| |
| def string(self, children): |
| """Instantiate one _XdrString declaration object""" |
| ident = children[0] |
| if children[1] is not None: |
| maxsize = children[1].value |
| else: |
| maxsize = "0" |
| |
| return _XdrString(ident.symbol, maxsize, line=ident.line, column=ident.column) |
| |
| def fixed_length_array(self, children): |
| """Instantiate one _XdrFixedLengthArray declaration object""" |
| spec = children[0] |
| ident = children[1] |
| size = children[2].value |
| |
| return _XdrFixedLengthArray( |
| ident.symbol, spec, size, line=ident.line, column=ident.column |
| ) |
| |
| def variable_length_array(self, children): |
| """Instantiate one _XdrVariableLengthArray declaration object""" |
| spec = children[0] |
| ident = children[1] |
| if children[2] is not None: |
| maxsize = children[2].value |
| else: |
| maxsize = "0" |
| |
| return _XdrVariableLengthArray( |
| ident.symbol, spec, maxsize, line=ident.line, column=ident.column |
| ) |
| |
| def optional_data(self, children): |
| """Instantiate one _XdrOptionalData declaration object""" |
| spec = children[0] |
| ident = children[1] |
| |
| return _XdrOptionalData( |
| ident.symbol, spec, line=ident.line, column=ident.column |
| ) |
| |
| def basic(self, children): |
| """Instantiate one _XdrBasic object""" |
| spec = children[0] |
| ident = children[1] |
| |
| return _XdrBasic(ident.symbol, spec, line=ident.line, column=ident.column) |
| |
| def void(self, children): |
| """Instantiate one _XdrVoid declaration object""" |
| |
| return _XdrVoid() |
| |
| def struct(self, children): |
| """Instantiate one _XdrStruct object""" |
| ident = children[0] |
| name = ident.symbol |
| fields = children[1].children |
| pos = {"line": ident.line, "column": ident.column} |
| |
| last_field = fields[-1] |
| if ( |
| isinstance(last_field, _XdrOptionalData) |
| and name == last_field.spec.type_name |
| ): |
| return _XdrPointer(name, fields, **pos) |
| |
| return _XdrStruct(name, fields, **pos) |
| |
| def typedef(self, children): |
| """Instantiate one _XdrTypedef object""" |
| new_type = children[0] |
| |
| return _XdrTypedef(new_type) |
| |
| def case_spec(self, children): |
| """Instantiate one _XdrCaseSpec object""" |
| values = [] |
| for item in children[0:-1]: |
| values.append(item.value) |
| arm = children[-1] |
| |
| return _XdrCaseSpec(values, arm) |
| |
| def default_spec(self, children): |
| """Instantiate one _XdrDefaultSpec object""" |
| arm = children[0] |
| |
| return _XdrDefaultSpec(arm) |
| |
| def union(self, children): |
| """Instantiate one _XdrUnion object""" |
| ident = children[0] |
| |
| body = children[1] |
| discriminant = body.children[0].children[0] |
| cases = body.children[1:-1] |
| default = body.children[-1] |
| |
| return _XdrUnion( |
| ident.symbol, |
| discriminant, |
| cases, |
| default, |
| line=ident.line, |
| column=ident.column, |
| ) |
| |
| def procedure_def(self, children): |
| """Instantiate one _RpcProcedure object""" |
| result = children[0] |
| ident = children[1] |
| argument = children[2] |
| number = children[3].value |
| |
| return _RpcProcedure( |
| ident.symbol, |
| number, |
| argument, |
| result, |
| line=ident.line, |
| column=ident.column, |
| ) |
| |
| def version_def(self, children): |
| """Instantiate one _RpcVersion object""" |
| ident = children[0] |
| number = children[-1].value |
| procedures = children[1:-1] |
| |
| return _RpcVersion( |
| ident.symbol, number, procedures, line=ident.line, column=ident.column |
| ) |
| |
| def program_def(self, children): |
| """Instantiate one _RpcProgram object""" |
| ident = children[0] |
| number = children[-1].value |
| versions = children[1:-1] |
| |
| return _RpcProgram( |
| ident.symbol, number, versions, line=ident.line, column=ident.column |
| ) |
| |
| def pragma_def(self, children): |
| """Instantiate one _Pragma object""" |
| directive = children[0].children[0].data |
| match directive: |
| case "big_endian_directive": |
| big_endian.append(children[1].symbol) |
| case "exclude_directive": |
| excluded_apis.append(children[1].symbol) |
| case "header_directive": |
| global header_name |
| header_name = children[1].symbol |
| case "public_directive": |
| public_apis.append(children[1].symbol) |
| case _: |
| raise NotImplementedError("Directive not supported") |
| return _Pragma() |
| |
| def passthru_def(self, children): |
| """Instantiate one _XdrPassthru object""" |
| token = children[0] |
| content = token.value[1:] |
| return _XdrPassthru(content) |
| |
| |
| transformer = ast_utils.create_transformer(this_module, ParseToAst()) |
| |
| |
| def _merge_consecutive_passthru(definitions: List[Definition]) -> List[Definition]: |
| """Merge consecutive passthru definitions into single nodes""" |
| result = [] |
| i = 0 |
| while i < len(definitions): |
| if isinstance(definitions[i].value, _XdrPassthru): |
| lines = [definitions[i].value.content] |
| meta = definitions[i].meta |
| j = i + 1 |
| while j < len(definitions) and isinstance( |
| definitions[j].value, _XdrPassthru |
| ): |
| lines.append(definitions[j].value.content) |
| j += 1 |
| merged = _XdrPassthru("\n".join(lines)) |
| result.append(Definition(meta, merged)) |
| i = j |
| else: |
| result.append(definitions[i]) |
| i += 1 |
| return result |
| |
| |
| def _meta_line(meta) -> int: |
| """Return the 1-based source line for a node's meta, or 0 if unknown""" |
| try: |
| return meta.line |
| except AttributeError: |
| return 0 |
| |
| |
| class XdrSemanticError(Exception): |
| """A specification that parses but violates an XDR semantic rule. |
| |
| Detection lives in the language-independent front end because a |
| duplicate name is malformed XDR regardless of the output language. |
| """ |
| |
| def __init__(self, message: str, meta): |
| super().__init__(message) |
| self.message = message |
| self.line = _meta_line(meta) |
| self.column = getattr(meta, "column", 0) |
| |
| |
| def _introduced_names(value): |
| """Yield (name, node) for each identifier a definition introduces.""" |
| if isinstance(value, (_XdrStruct, _XdrUnion, _XdrPointer)): |
| yield value.name, value |
| elif isinstance(value, _XdrEnum): |
| yield value.name, value |
| for enumerator in value.enumerators: |
| yield enumerator.name, enumerator |
| elif isinstance(value, _XdrTypedef): |
| yield value.declaration.name, value.declaration |
| elif isinstance(value, _XdrConstant): |
| yield value.name, value |
| elif isinstance(value, _RpcProgram): |
| yield value.name, value |
| |
| |
| def _check_rpc_scope_names(program: "_RpcProgram") -> None: |
| """Enforce RFC 5531 Section 12.3 scoping within an RPC program. |
| |
| A version name and number are unique within the program and a |
| procedure name and number are unique within its version. |
| """ |
| version_names = set() |
| version_numbers = set() |
| for version in program.versions: |
| if version.name in version_names: |
| raise XdrSemanticError( |
| f"duplicate version name '{version.name}'" |
| f" in program '{program.name}'", |
| version, |
| ) |
| version_names.add(version.name) |
| if version.number in version_numbers: |
| raise XdrSemanticError( |
| f"duplicate version number {version.number}" |
| f" in program '{program.name}'", |
| version, |
| ) |
| version_numbers.add(version.number) |
| procedure_names = set() |
| procedure_numbers = set() |
| for procedure in version.procedures: |
| if procedure.name in procedure_names: |
| raise XdrSemanticError( |
| f"duplicate procedure name '{procedure.name}'" |
| f" in version '{version.name}'", |
| procedure, |
| ) |
| procedure_names.add(procedure.name) |
| if procedure.number in procedure_numbers: |
| raise XdrSemanticError( |
| f"duplicate procedure number {procedure.number}" |
| f" in version '{version.name}'", |
| procedure, |
| ) |
| procedure_numbers.add(procedure.number) |
| |
| |
| def check_duplicate_definitions(root: "Specification") -> None: |
| """Reject a spec that declares an identifier more than once. |
| |
| RFC 4506 Section 6.4 places constant and type identifiers in a |
| single name space that must be unique within a specification. |
| RFC 5531 Section 12.3 adds RPC program names to that name space |
| and scopes version names and numbers to their program and |
| procedure names and numbers to their version. |
| """ |
| seen = {} |
| for definition in root.definitions: |
| for name, node in _introduced_names(definition.value): |
| where = node if node.line else definition.meta |
| first = seen.get(name) |
| if first is not None: |
| raise XdrSemanticError( |
| f"duplicate identifier '{name}'" |
| f" (first declared at line {_meta_line(first)})", |
| where, |
| ) |
| seen[name] = where |
| if isinstance(definition.value, _RpcProgram): |
| _check_rpc_scope_names(definition.value) |
| |
| |
| # RFC 5531 (Section 9) encodes program, version, and procedure numbers |
| # as unsigned 32-bit integers, so each must fall within [0, 2**32 - 1]. |
| _RPC_NUMBER_MAX = 2**32 - 1 |
| |
| |
| def _check_rpc_number(kind: str, number: int, scope: str, meta) -> None: |
| """Reject one RPC number that is negative or wider than 32 bits.""" |
| if number < 0: |
| raise XdrSemanticError( |
| f"negative {kind} number {number} {scope}", |
| meta, |
| ) |
| if number > _RPC_NUMBER_MAX: |
| raise XdrSemanticError( |
| f"{kind} number {number} {scope} exceeds {_RPC_NUMBER_MAX}", |
| meta, |
| ) |
| |
| |
| def check_rpc_number_range(root: "Specification") -> None: |
| """Reject an out-of-range program, version, or procedure number. |
| |
| RFC 5531 assigns only unsigned constants to program, version, and |
| procedure numbers (Section 12.3) and encodes each as an unsigned |
| 32-bit integer (Section 9). RFC 4506 Section 6.2 permits a signed |
| decimal constant for XDR constants in general and sets no ceiling on |
| magnitude, so the grammar accepts an out-of-range value; the range |
| is enforced here instead. The parser retains no per-version or |
| per-procedure source location, so a violation is reported against the |
| program definition. |
| """ |
| for definition in root.definitions: |
| program = definition.value |
| if not isinstance(program, _RpcProgram): |
| continue |
| _check_rpc_number( |
| "program", |
| program.number, |
| f"in program '{program.name}'", |
| definition.meta, |
| ) |
| for version in program.versions: |
| _check_rpc_number( |
| "version", |
| version.number, |
| f"in program '{program.name}'", |
| definition.meta, |
| ) |
| for procedure in version.procedures: |
| _check_rpc_number( |
| "procedure", |
| procedure.number, |
| f"in version '{version.name}'", |
| definition.meta, |
| ) |
| |
| |
| def transform_parse_tree(parse_tree): |
| """Transform productions into an abstract syntax tree""" |
| ast = transformer.transform(parse_tree) |
| ast.definitions = _merge_consecutive_passthru(ast.definitions) |
| check_duplicate_definitions(ast) |
| check_rpc_number_range(ast) |
| return ast |
| |
| |
| def get_header_name() -> str: |
| """Return header name set by pragma header directive""" |
| return header_name |
