lint_rules_parser.py

#!/usr/bin/env python

import logging
import os
import xml.etree.ElementTree
from dataclasses import dataclass
from enum import Enum, auto
from typing import List, MutableMapping, Optional, Tuple, Union

from lark import Lark
from lark.visitors import Discard, Transformer, v_args

try:
    from matter_idl.lint.type_definitions import AttributeRequirement
except ImportError:
    import sys
    sys.path.append(os.path.join(os.path.abspath(
        os.path.dirname(__file__)), "..", ".."))
    from matter_idl.lint.type_definitions import AttributeRequirement

from matter_idl.lint.type_definitions import (ClusterAttributeDeny, ClusterCommandRequirement, ClusterRequirement,
                                              ClusterValidationRule, RequiredAttributesRule, RequiredCommandsRule)


class ElementNotFoundError(Exception):
    def __init__(self, name):
        super().__init__(f"Could not find {name}")


def parseNumberString(n: str) -> int:
    if n.startswith('0x'):
        return int(n[2:], 16)
    else:
        return int(n)


@dataclass
class RequiredAttribute:
    name: str
    code: int


@dataclass
class RequiredCommand:
    name: str
    code: int


@dataclass
class DecodedCluster:
    name: str
    code: int
    required_attributes: List[RequiredAttribute]
    required_commands: List[RequiredCommand]


class ClusterActionEnum(Enum):
    REQUIRE = auto()
    REJECT = auto()


@dataclass
class ServerClusterRequirement:
    action: ClusterActionEnum
    id: Union[str, int]


def DecodeClusterFromXml(element: xml.etree.ElementTree.Element):
    if element.tag != 'cluster':
        logging.error("Not a cluster element: %r" % element)
        return None

    # cluster elements contain among other children
    #  - name (general name for this cluster)
    #  - code (unique identifier, may be hex or numeric)
    #  - attribute with side, code and optional attributes
    try:
        name = element.find('name')
        if name is None or not name.text:
            raise ElementNotFoundError('name')

        name = name.text.replace(' ', '')
        required_attributes = []
        required_commands = []

        for attr in element.findall('attribute'):
            if attr.attrib['side'] != 'server':
                continue

            if 'optional' in attr.attrib and attr.attrib['optional'] == 'true':
                continue

            # when introducing access controls, the content of attributes may either be:
            # <attribute ...>myName</attribute>
            # or
            # <attribute ...><description>myName</description><access .../>...</attribute>
            attr_name = attr.text
            description = attr.find('description')
            if description is not None:
                attr_name = description.text

            required_attributes.append(
                RequiredAttribute(
                    name=attr_name,
                    code=parseNumberString(attr.attrib['code'])
                ))

        for cmd in element.findall('command'):
            if cmd.attrib['source'] != 'client':
                continue

            if 'optional' in cmd.attrib and cmd.attrib['optional'] == 'true':
                continue

            required_commands.append(RequiredCommand(
                name=cmd.attrib["name"], code=parseNumberString(cmd.attrib['code'])))

        code = element.find('code')
        if code is None:
            raise Exception("Failed to find cluster code")

        return DecodedCluster(
            name=name,
            code=parseNumberString(code.text),
            required_attributes=required_attributes,
            required_commands=required_commands
        )
    except Exception:
        logging.exception("Failed to decode cluster %r" % element)
        return None


def ClustersInXmlFile(path: str):
    logging.info("Loading XML from %s" % path)

    # root is expected to be just a "configurator" object
    configurator = xml.etree.ElementTree.parse(path).getroot()
    for child in configurator:
        if child.tag != 'cluster':
            continue
        yield child


class LintRulesContext:
    """Represents a context for loadint lint rules.

    Handles:
       - loading referenced files (matter xml definitions)
       - adding linter rules as data is parsed
       - Looking up identifiers for various rules
    """

    def __init__(self):
        self._required_attributes_rule = RequiredAttributesRule(
            "Required attributes")
        self._cluster_validation_rule = ClusterValidationRule(
            "Cluster validation")
        self._required_commands_rule = RequiredCommandsRule(
            "Required commands")

        # Map cluster names to the underlying code
        self._cluster_codes: MutableMapping[str, int] = {}

    def GetLinterRules(self):
        return [self._required_attributes_rule, self._required_commands_rule, self._cluster_validation_rule]

    def RequireAttribute(self, r: AttributeRequirement):
        self._required_attributes_rule.RequireAttribute(r)

    def Deny(self, what: ClusterAttributeDeny):
        self._required_attributes_rule.Deny(what)

    def FindClusterCode(self, name: str) -> Optional[Tuple[str, int]]:
        if name not in self._cluster_codes:
            # Name may be a number. If this can be parsed as a number, accept it anyway
            try:
                return "ID_%s" % name, parseNumberString(name)
            except ValueError:
                logging.error("UNKNOWN cluster name %s" % name)
                logging.error("Known names: %s" %
                              (",".join(self._cluster_codes.keys()), ))
                return None
        else:
            return name, self._cluster_codes[name]

    def RequireClusterInEndpoint(self, name: str, code: int):
        """Mark that a specific cluster is always required in the given endpoint
        """
        cluster_info = self.FindClusterCode(name)
        if not cluster_info:
            return

        name, cluster_code = cluster_info

        self._cluster_validation_rule.RequireClusterInEndpoint(ClusterRequirement(
            endpoint_id=code,
            cluster_code=cluster_code,
            cluster_name=name,
        ))

    def RejectClusterInEndpoint(self, name: str, code: int):
        """Mark that a specific cluster is always rejected in the given endpoint
        """
        cluster_info = self.FindClusterCode(name)
        if not cluster_info:
            return

        name, cluster_code = cluster_info

        self._cluster_validation_rule.RejectClusterInEndpoint(ClusterRequirement(
            endpoint_id=code,
            cluster_code=cluster_code,
            cluster_name=name,
        ))

    def LoadXml(self, path: str):
        """Load XML data from the given path and add it to
           internal processing. Adds attribute requirement rules
           as needed.
        """
        for cluster in ClustersInXmlFile(path):
            decoded = DecodeClusterFromXml(cluster)

            if not decoded:
                continue

            self._cluster_codes[decoded.name] = decoded.code

            for attr in decoded.required_attributes:
                self._required_attributes_rule.RequireAttribute(AttributeRequirement(
                    code=attr.code, name=attr.name, filter_cluster=decoded.code))

            for cmd in decoded.required_commands:
                self._required_commands_rule.RequireCommand(
                    ClusterCommandRequirement(
                        cluster_code=decoded.code,
                        command_code=cmd.code,
                        command_name=cmd.name
                    ))


class LintRulesTransformer(Transformer):
    """
    A transformer capable to transform data parsed by Lark according to
    lint_rules_grammar.lark.
    """

    def __init__(self, file_name: str):
        self.context = LintRulesContext()
        self.file_name = file_name

    def positive_integer(self, tokens):
        """Numbers in the grammar are integers or hex numbers.
        """
        if len(tokens) != 1:
            raise Exception("Unexpected argument counts")

        return parseNumberString(tokens[0].value)

    @v_args(inline=True)
    def negative_integer(self, value):
        return -value

    @v_args(inline=True)
    def integer(self, value):
        return value

    def id(self, tokens):
        """An id is a string containing an identifier
        """
        if len(tokens) != 1:
            raise Exception("Unexpected argument counts")
        return tokens[0].value

    def ESCAPED_STRING(self, s):
        # handle escapes, skip the start and end quotes
        return s.value[1:-1].encode('utf-8').decode('unicode-escape')

    def start(self, instructions):
        # At this point processing is considered done, return all
        # linter rules that were found
        return self.context.GetLinterRules()

    def instruction(self, instruction):
        return Discard

    def all_endpoint_rule(self, rules: List[Union[AttributeRequirement, ClusterAttributeDeny]]):
        for rule in rules:
            if type(rule) is AttributeRequirement:
                self.context.RequireAttribute(rule)
            elif type(rule) is ClusterAttributeDeny:
                self.context.Deny(rule)
            else:
                raise Exception("Unkown endpoint requirement: %r" % rule)

        return Discard

    @v_args(inline=True)
    def load_xml(self, path):
        if not os.path.isabs(path):
            path = os.path.abspath(os.path.join(
                os.path.dirname(self.file_name), path))

        self.context.LoadXml(path)

    @v_args(inline=True)
    def required_global_attribute(self, name, code):
        return AttributeRequirement(code=code, name=name)

    @v_args(inline=True)
    def specific_endpoint_rule(self, code, *requirements):
        for requirement in requirements:
            if requirement.action == ClusterActionEnum.REQUIRE:
                self.context.RequireClusterInEndpoint(requirement.id, code)
            elif requirement.action == ClusterActionEnum.REJECT:
                self.context.RejectClusterInEndpoint(requirement.id, code)
            else:
                raise Exception("Unexpected requirement action %r" %
                                requirement.action)

        return Discard

    @v_args(inline=True)
    def required_server_cluster(self, id):
        return ServerClusterRequirement(ClusterActionEnum.REQUIRE, id)

    @v_args(inline=True)
    def rejected_server_cluster(self, id):
        return ServerClusterRequirement(ClusterActionEnum.REJECT, id)

    @v_args(inline=True)
    def denylist_cluster_attribute(self, cluster_id, attribute_id):
        return ClusterAttributeDeny(cluster_id, attribute_id)


class Parser:
    def __init__(self, parser, file_name: str):
        self.parser = parser
        self.file_name = file_name

    def parse(self):
        data = LintRulesTransformer(self.file_name).transform(
            self.parser.parse(open(self.file_name, "rt").read()))
        return data


def CreateParser(file_name: str):
    """
    Generates a parser that will process a ".matter" file into a IDL
    """
    return Parser(
        Lark.open('lint_rules_grammar.lark', rel_to=__file__, parser='lalr', propagate_positions=True, maybe_placeholders=True), file_name=file_name)


if __name__ == '__main__':
    # This Parser is generally not intended to be run as a stand-alone binary.
    # The ability to run is for debug and to print out the parsed AST.
    import click

    # Supported log levels, mapping string values required for argument
    # parsing into logging constants
    __LOG_LEVELS__ = {
        'debug': logging.DEBUG,
        'info': logging.INFO,
        'warn': logging.WARN,
        'fatal': logging.FATAL,
    }

    @click.command()
    @click.option(
        '--log-level',
        default='INFO',
        type=click.Choice(list(__LOG_LEVELS__.keys()), case_sensitive=False),
        help='Determines the verbosity of script output.')
    @click.argument('filename')
    def main(log_level, filename=None):
        logging.basicConfig(
            level=__LOG_LEVELS__[log_level],
            format='%(asctime)s %(levelname)-7s %(message)s',
        )

        logging.info("Starting to parse ...")
        data = CreateParser(filename).parse()
        logging.info("Parse completed")

        logging.info("Data:")
        logging.info("%r" % data)

    main(auto_envvar_prefix='CHIP')