125 lines
3.5 KiB
Python
125 lines
3.5 KiB
Python
"""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="<safety-predicate>", 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
|