src/Processor.py - master - Flipper/Marginal

src/Processor.py
568 строк20 KB
Alex
Marginal v1.0.0 RELEASE
11 янв 2026, 18:02
11 янв 2026, 18:025e1eff8
import math
import ast
import json
import builtins
from statistics import mean, stdev, variance, median

from os import path
from utils import getBasePath, getBakedPath


class Processor:
    """
    Safe mathematical expression processor.

    Validates and evaluates user-defined calculation expressions using
    Python's AST to prevent unsafe operations while supporting common
    mathematical, statistical, and iterable constructs.
    """

    def __init__(self, backbone):
        """
        Initialize the processor with access to the application logger.

        Args:
            backbone: Application backbone instance.
        """
        self.logger = backbone.logger

        # Try to load whitelist from file, fall back to defaults if needed
        self._loadWhitelist()

        # Validate loaded whitelist contents
        self._validateWhitelist()

    def _createDefaultWhitelist(self, filepath, defaultSafeFunctions=None, defaultAllowedNodes=None):
        """
        Create a default whitelist.json file with provided or hardcoded values.

        Args:
            filepath (str): Path where to create the whitelist file.
            defaultSafeFunctions (dict, optional): Safe functions to use if provided.
            defaultAllowedNodes (list, optional): Allowed nodes to use if provided.
        """
        # Use provided defaults or fall back to hardcoded values
        if defaultSafeFunctions is not None:
            safeFunctions = defaultSafeFunctions
        else:
            # Hardcoded safeFunctions in JSON format
            safeFunctions = {
                "mean": "statistics.mean",
                "stdev": "statistics.stdev",
                "variance": "statistics.variance",
                "median": "statistics.median",
                "sqrt": "math.sqrt",
                "pow": "builtins.pow",
                "abs": "builtins.abs",
                "log": "math.log",
                "log10": "math.log10",
                "sin": "math.sin",
                "cos": "math.cos",
                "tan": "math.tan",
                "floor": "math.floor",
                "ceil": "math.ceil",
                "max": "builtins.max",
                "min": "builtins.min",
                "sum": "builtins.sum",
                "len": "builtins.len",
                "tuple": "builtins.tuple",
                "dict": "builtins.dict",
                "range": "builtins.range",
                "zip": "builtins.zip"
            }

        if defaultAllowedNodes is not None:
            allowedNodes = defaultAllowedNodes
        else:
            # Hardcoded allowedNodes in JSON format
            allowedNodes = [
                "Expression",
                "BinOp",
                "UnaryOp",
                "Add",
                "Sub",
                "Mult",
                "Div",
                "Pow",
                "Mod",
                "USub",
                "UAdd",
                "Call",
                "Name",
                "Load",
                "Constant",
                "comprehension",
                "ListComp",
                "List",
                "Tuple",
                "GeneratorExp",
                "IfExp",
                "Compare",
                "Subscript",
                "Index",
                "BoolOp",
                "And",
                "Or"
            ]

        defaultWhitelist = {
            "safeFunctions": safeFunctions,
            "allowedNodes": allowedNodes
        }

        try:
            # Create directory if it doesn't exist
            directory = path.dirname(filepath)
            if directory and not path.exists(directory):
                from os import makedirs
                makedirs(directory, exist_ok=True)

            with open(filepath, 'w') as f:
                json.dump(defaultWhitelist, f, indent=2)

            if self.logger:
                self.logger.log(
                    "Processor",
                    f"Created default whitelist file at: {filepath}",
                    False
                )
        except Exception as e:
            if self.logger:
                self.logger.log(
                    "Processor",
                    f"Failed to create whitelist file: {e}",
                    True
                )

    def _loadWhitelist(self):
        """
        Load safe functions and allowed AST nodes from whitelist.json.
        Falls back to default whitelist if user's file fails, then to
        hard-coded defaults if all fails.
        """
        # Default paths for whitelist files
        userWhitelistPath = path.join(getBasePath(), r"config\whitelist.json")
        defaultWhitelistPath = path.join(getBakedPath(), r"local\app_defaultProcessorWhitelist.json")

        # Hard-coded defaults (original implementation)
        hardcodedSafeFunctions = {
            "mean": mean,
            "stdev": stdev,
            "variance": variance,
            "median": median,
            "sqrt": math.sqrt,
            "pow": pow,
            "abs": abs,
            "log": math.log,
            "log10": math.log10,
            "sin": math.sin,
            "cos": math.cos,
            "tan": math.tan,
            "floor": math.floor,
            "ceil": math.ceil,
            "max": max,
            "min": min,
            "sum": sum,
            "len": len,
            "tuple": tuple,
            "dict": dict,
            "range": range,
            "zip": zip,
        }

        hardcodedAllowedNodes = (
            ast.Expression,
            ast.BinOp,
            ast.UnaryOp,
            ast.Add,
            ast.Sub,
            ast.Mult,
            ast.Div,
            ast.Pow,
            ast.Mod,
            ast.USub,
            ast.UAdd,
            ast.Call,
            ast.Name,
            ast.Load,
            ast.Constant,
            ast.comprehension,
            ast.ListComp,
            ast.List,
            ast.Tuple,
            ast.GeneratorExp,
            ast.IfExp,
            ast.Compare,
            ast.Subscript,
            ast.Index,
            ast.BoolOp,
            ast.And,
            ast.Or,
        )

        # Mapping for converting string node names to actual AST classes
        astNodeMapping = {name: getattr(ast, name) for name in dir(ast)
                          if isinstance(getattr(ast, name), type)}

        # Mapping for resolving function names to actual functions
        functionSources = {
            "builtins": builtins,
            "math": math,
            "statistics": __import__('statistics'),
        }

        # First, try to load the default whitelist
        defaultWhitelistData = None
        try:
            with open(defaultWhitelistPath, 'r') as f:
                defaultWhitelistData = json.load(f)
        except (FileNotFoundError, json.JSONDecodeError, PermissionError) as e:
            if self.logger:
                self.logger.log(
                    "Processor",
                    f"Default whitelist not available: {e}",
                    False
                )
            defaultWhitelistData = None

        # Check if user whitelist exists, create it if not using default values
        if not path.exists(userWhitelistPath):
            if self.logger:
                self.logger.log(
                    "Processor",
                    f"User whitelist not found at {userWhitelistPath}. Creating with available defaults.",
                    False
                )
            # Use default whitelist data if available, otherwise use hardcoded
            self._createDefaultWhitelist(
                userWhitelistPath,
                defaultSafeFunctions=defaultWhitelistData.get("safeFunctions") if defaultWhitelistData else None,
                defaultAllowedNodes=defaultWhitelistData.get("allowedNodes") if defaultWhitelistData else None
            )

        # Try to load user's whitelist
        whitelistData = None
        source = "hardcoded defaults"

        try:
            with open(userWhitelistPath, 'r') as f:
                whitelistData = json.load(f)
            source = f"user file '{userWhitelistPath}'"

        except (FileNotFoundError, json.JSONDecodeError, PermissionError) as e:
            if self.logger:
                self.logger.log(
                    "Processor",
                    f"Failed to load user whitelist: {e}. Trying default whitelist.",
                    False
                )

            # Fall back to default whitelist
            if defaultWhitelistData is not None:
                whitelistData = defaultWhitelistData
                source = f"default file '{defaultWhitelistPath}'"
            else:
                if self.logger:
                    self.logger.log(
                        "Processor",
                        f"No default whitelist available. Using hard-coded defaults.",
                        False
                    )
                # Use hard-coded defaults
                whitelistData = None

        # Process loaded whitelist data or use hardcoded defaults
        if whitelistData:
            try:
                # Parse safeFunctions from JSON
                self.safeFunctions = self._parseSafeFunctions(
                    whitelistData.get("safeFunctions", {}),
                    functionSources
                )

                # Parse allowedNodes from JSON
                self.allowedNodes = self._parseAllowedNodes(
                    whitelistData.get("allowedNodes", []),
                    astNodeMapping
                )

                if self.logger:
                    self.logger.log(
                        "Processor",
                        f"Successfully loaded whitelist from {source}",
                        False
                    )

            except (KeyError, ValueError, AttributeError) as e:
                if self.logger:
                    self.logger.log(
                        "Processor",
                        f"Error parsing whitelist from {source}: {e}. Using hard-coded defaults.",
                        True
                    )
                # Fall back to hardcoded defaults
                self.safeFunctions = hardcodedSafeFunctions
                self.allowedNodes = hardcodedAllowedNodes
        else:
            # Use hardcoded defaults
            self.safeFunctions = hardcodedSafeFunctions
            self.allowedNodes = hardcodedAllowedNodes

    def _parseSafeFunctions(self, functionsDict, functionSources):
        """
        Parse safeFunctions dictionary from JSON data.

        Args:
            functionsDict (dict): Dictionary mapping function names to
                                 function specifications from JSON.
            functionSources (dict): Mapping of module names to module objects.

        Returns:
            dict -> Mapping of function names to actual function objects.

        Raises:
            ValueError: If function specification is invalid or function
                       cannot be found.
        """
        safeFunctions = {}

        for funcName, funcSpec in functionsDict.items():
            if isinstance(funcSpec, str):
                # String format: "module.function" or "builtin_function"
                if '.' in funcSpec:
                    # Module.function format
                    moduleName, actualFuncName = funcSpec.rsplit('.', 1)
                    if moduleName not in functionSources:
                        raise ValueError(f"Unknown module '{moduleName}' for function '{funcName}'")

                    module = functionSources[moduleName]
                    if not hasattr(module, actualFuncName):
                        raise ValueError(f"Function '{actualFuncName}' not found in module '{moduleName}'")

                    function = getattr(module, actualFuncName)
                    if not callable(function):
                        raise ValueError(f"'{actualFuncName}' in module '{moduleName}' is not callable")

                    safeFunctions[funcName] = function
                else:
                    # Assume it's a builtin
                    if not hasattr(builtins, funcSpec):
                        raise ValueError(f"Builtin function '{funcSpec}' not found")

                    function = getattr(builtins, funcSpec)
                    if not callable(function):
                        raise ValueError(f"'{funcSpec}' is not callable")

                    safeFunctions[funcName] = function

            elif isinstance(funcSpec, dict):
                # Dictionary format for more complex specifications
                # (future extensibility)
                raise ValueError("Dictionary format for function specifications not yet implemented")
            else:
                raise ValueError(f"Invalid function specification for '{funcName}'")

        return safeFunctions

    def _parseAllowedNodes(self, nodeNames, astNodeMapping):
        """
        Parse allowedNodes list from JSON data.

        Args:
            nodeNames (list): List of AST node type names from JSON.
            astNodeMapping (dict): Mapping of node names to AST classes.

        Returns:
            tuple -> Tuple of allowed AST node classes.

        Raises:
            ValueError: If node name is not a valid AST node type.
        """
        allowedNodes = []

        for nodeName in nodeNames:
            if nodeName not in astNodeMapping:
                raise ValueError(f"Invalid AST node type: '{nodeName}'")

            nodeClass = astNodeMapping[nodeName]
            allowedNodes.append(nodeClass)

        return tuple(allowedNodes)

    def _validateWhitelist(self):
        """
        Validate the loaded whitelist to ensure it contains necessary
        components and all functions are callable.
        """
        # Validate safeFunctions
        if not hasattr(self, 'safeFunctions') or not isinstance(self.safeFunctions, dict):
            raise ValueError("safeFunctions must be a dictionary")

        for funcName, func in self.safeFunctions.items():
            if not callable(func):
                raise ValueError(f"Function '{funcName}' in safeFunctions is not callable")

        # Validate allowedNodes
        if not hasattr(self, 'allowedNodes') or not isinstance(self.allowedNodes, tuple):
            raise ValueError("allowedNodes must be a tuple")

        for nodeClass in self.allowedNodes:
            if not isinstance(nodeClass, type):
                raise ValueError(f"Invalid node class in allowedNodes: {nodeClass}")

    def _validateExpression(self, expression, allowedNames):
        """
        Validate an expression AST to ensure only safe operations are used.

        Args:
            expression (str): Expression string to validate.
            allowedNames (set[str]): Variable names allowed in this context.

        Returns:
            ast.Expression -> Parsed and validated AST.

        Raises:
            ValueError: If disallowed syntax or names are detected.
        """
        tree = ast.parse(expression, mode="eval")

        def validateNode(node, allowedNames):
            if not isinstance(node, self.allowedNodes):
                raise ValueError(
                    f"Disallowed syntax: {type(node).__name__}"
                )

            if isinstance(node, ast.Name):
                if node.id not in allowedNames:
                    raise ValueError(
                        f"Unknown name '{node.id}'"
                    )

            elif isinstance(node, ast.Call):
                if not isinstance(node.func, ast.Name):
                    raise ValueError(
                        "Only direct function calls allowed"
                    )
                if node.func.id not in self.safeFunctions:
                    raise ValueError(
                        f"Function '{node.func.id}' not allowed"
                    )

                # Validate arguments using the current allowed scope
                for arg in node.args:
                    validateNode(arg, allowedNames)

            elif isinstance(node, (ast.ListComp, ast.GeneratorExp)):
                # Comprehensions introduce a local iteration variable scope
                newAllowed = set(allowedNames)

                for generator in node.generators:
                    if isinstance(generator.target, ast.Name):
                        newAllowed.add(generator.target.id)

                    # Validate iterable using the outer scope
                    validateNode(generator.iter, allowedNames)

                # Validate expression body using the extended scope
                validateNode(node.elt, newAllowed)

                # Validate conditional clauses inside the comprehension
                for generator in node.generators:
                    for ifCond in generator.ifs:
                        validateNode(ifCond, newAllowed)

            else:
                # Recursively validate all child nodes
                for child in ast.iter_child_nodes(node):
                    validateNode(child, allowedNames)

        validateNode(tree.body, set(allowedNames))
        return tree

    def process(self, profile, data, createLog):
        """
        Execute all calculations defined in a profile.

        Args:
            profile (dict): Calculation profile definition.
            data (dict): Input data values.
            createLog (bool): Enable per-calculation logging.

        Returns:
            dict -> Mapping of calculation names to computed values.

        Raises:
            Exception: Propagates validation or evaluation failures.
        """
        # Avoid mutating caller-provided data
        data = dict(data)

        # Initial evaluation context
        localVars = {**data, **self.safeFunctions}
        results = {}

        # Load constants into evaluation context
        for key, const in profile.get("constants", {}).items():
            value = const["value"]
            data[key] = value
            localVars[key] = value

        calculations = profile["calculations"]

        # Collect calculation names for forward reference validation
        allCalculationNames = {calc["name"] for calc in calculations}

        # First pass: validate all expressions (forward references allowed)
        for calc in calculations:
            name = calc["name"]
            expression = calc["operation"]

            try:
                allowedNames = set(localVars.keys()) | allCalculationNames
                self._validateExpression(expression, allowedNames)

            except Exception as e:
                if self.logger:
                    self.logger.log(
                        "Processor",
                        f"Validation failed for {name}: {e}",
                        True
                    )
                raise

        # Second pass: execute calculations sequentially
        for calc in calculations:
            name = calc["name"]
            expression = calc["operation"]
            precision = calc.get("precision")

            try:
                tree = ast.parse(expression, mode="eval")

                value = eval(
                    compile(tree, "<expression>", "eval"),
                    {"__builtins__": {}},
                    localVars
                )

                if precision is not None and isinstance(value, (int, float)):
                    value = round(value, precision)

                results[name] = value
                localVars[name] = value

                if createLog:
                    self.logger.logCalculation(
                        f"-> Evaluating {name}: {expression} = {value}"
                    )

            except Exception as e:
                if self.logger:
                    self.logger.log(
                        "Processor",
                        f"Calculation failed for {name}: {e}",
                        True
                    )
                raise

        return results