"""Runtime safety contract evaluator. Provides a small, auditable evaluator that can safely evaluate boolean predicates expressed over a LocalProblem's state variables. This is intentionally minimal and does not allow function calls, attribute access, or other unsafe operations. It raises ValueError for disallowed expressions. The evaluator accepts expressions like: "battery > 20 and distance_to_goal < 5" and looks up names from a provided state mapping. """ from __future__ import annotations import ast from typing import Any, Dict ALLOWED_NODES = { ast.Expression, ast.BoolOp, ast.UnaryOp, ast.BinOp, ast.Compare, ast.Name, ast.Load, ast.Constant, ast.And, ast.Or, ast.Not, ast.Add, ast.Sub, ast.Mult, ast.Div, ast.Mod, ast.Pow, ast.Eq, ast.NotEq, ast.Lt, ast.LtE, ast.Gt, ast.GtE, ast.Is, ast.IsNot, ast.In, ast.NotIn, ast.UnaryOp, ast.USub, ast.UAdd, } class UnsafeExpression(ValueError): pass def _ensure_safe(node: ast.AST) -> None: """Recursively ensure AST contains only allowed nodes. Raises UnsafeExpression on disallowed nodes. """ node_type = type(node) if node_type not in ALLOWED_NODES: raise UnsafeExpression(f"Disallowed AST node: {node_type.__name__}") # Recurse into supported child fields for child in ast.iter_child_nodes(node): _ensure_safe(child) def evaluate_predicate(expr: str, state: Dict[str, Any]) -> bool: """Evaluate a boolean predicate string against a state mapping. Only names present in `state` are allowed identifiers. Any unknown name raises UnsafeExpression. Examples: evaluate_predicate("battery > 20 and dist < 5", {"battery": 50, "dist": 3}) """ try: tree = ast.parse(expr, mode="eval") except SyntaxError as e: raise UnsafeExpression(f"Syntax error in predicate: {e}") _ensure_safe(tree) # Prepare a restricted eval environment: only allow names from state class NameResolver(ast.NodeTransformer): def visit_Name(self, node: ast.Name) -> ast.AST: if node.id not in state: raise UnsafeExpression(f"Unknown name in predicate: {node.id}") # Replace the Name node with a Constant node containing the value value = state[node.id] return ast.copy_location(ast.Constant(value=value), node) resolved = NameResolver().visit(tree) ast.fix_missing_locations(resolved) compiled = compile(resolved, filename="", mode="eval") result = eval(compiled, {"__builtins__": {}}, {}) if not isinstance(result, (bool, int, float)): # We expect a boolean-like result; numbers are allowed (0/1) raise UnsafeExpression("Predicate did not evaluate to a boolean or numeric value") return bool(result) def check_contracts(contracts: Dict[str, Dict[str, Any]], state: Dict[str, Any]): """Evaluate a dict of contracts against state. contracts: mapping contract_id -> {"predicate": str, "fail_safe": str} Returns a list of (contract_id, passed: bool, fail_safe) tuples. """ results = [] for cid, c in contracts.items(): pred = c.get("predicate") or c.get("predicate_str") or "False" fail_safe = c.get("fail_safe", "halt") try: passed = evaluate_predicate(pred, state) except UnsafeExpression: # Treat unsafe/malformed predicate as violation passed = False results.append((cid, passed, fail_safe)) return results