emirhanyildirim34555's picture
Publish reviewed research staging to main
2e4432d verified
Raw History Blame Contribute Delete
25.6 kB
"""ICL bytecode compiler and independent stack interpreter.
This backend shares scalar arithmetic, builtins, and resource accounting with
the tree interpreter; it does not call the tree interpreter or execute Python
source. Checkpoint hashes detect accidental modification, not adversarial
forgery: they are unkeyed checksums and confer no trust or authority.
"""
from __future__ import annotations
import hashlib
import json
import math
from copy import deepcopy
from .runtime import ICLError, RuntimeContext, builtin, checked_binary, checked_unary, DEFAULT_LIMITS
from .checker import valid_type, validate_value
BYTECODE_SCHEMA = "icl-bytecode/1"
CHECKPOINT_SCHEMA = "icl-vm-checkpoint/1"
OPCODES = {"PUSH", "LOAD", "LET", "STORE", "POP", "BINARY", "UNARY", "CALL",
"LIST", "TUPLE", "RECORD", "INDEX", "FIELD", "JUMP", "JUMP_FALSE",
"LOOP_RESET", "LOOP_GUARD", "ENTER_SCOPE", "EXIT_SCOPE", "RETURN"}
_DEFAULT_LIMITS = dict(DEFAULT_LIMITS)
def _fail(code, message, span=None):
code = {"E_BYTECODE": "BYTECODE_CORRUPT", "E_CHECKPOINT": "CHECKPOINT_CORRUPT",
"E_LIMIT": "MEMORY_LIMIT", "E_NUMERIC": "NONFINITE", "E_VALUE": "TYPE",
"E_TYPE": "TYPE", "E_ENTRY": "NAME", "E_NAME": "NAME", "E_ARITY": "TYPE",
"E_IMMUTABLE": "IMMUTABLE", "E_INDEX": "BOUNDS", "E_FIELD": "TYPE"}.get(code, code)
raise ICLError(code, message, span)
def _canonical(value):
try:
return json.dumps(value, sort_keys=True, separators=(",", ":"),
ensure_ascii=False, allow_nan=False)
except (TypeError, ValueError, RecursionError):
_fail("E_BYTECODE", "Value is not finite canonical JSON")
def _hash(value):
return hashlib.sha256(_canonical(value).encode("utf-8")).hexdigest()
def compile_bytecode(program):
"""Lower a checked ICL AST to a JSON-serializable instruction stream."""
functions = {}
for function in program.get("functions", []):
name = function["name"]
if name in functions:
_fail("E_BYTECODE", "Duplicate function: " + name)
code = []
loop_sequence = [0]
def emit(op, span=None, **kwargs):
index = len(code)
code.append({"op": op, **kwargs, "span": span or function.get("span")})
return index
def expr(node):
kind = node["kind"]
span = node.get("span")
if kind == "literal":
emit("PUSH", span, value=node["value"])
elif kind == "var":
emit("LOAD", span, name=node["name"])
elif kind == "binary":
# Preserve short-circuit evaluation without calling AST eval.
if node["op"] in ("&&", "and"):
expr(node["left"])
branch = emit("JUMP_FALSE", span, target=0)
expr(node["right"])
finish = emit("JUMP", span, target=0)
code[branch]["target"] = len(code)
emit("PUSH", span, value=False)
code[finish]["target"] = len(code)
elif node["op"] in ("||", "or"):
expr(node["left"])
branch = emit("JUMP_FALSE", span, target=0)
emit("PUSH", span, value=True)
finish = emit("JUMP", span, target=0)
code[branch]["target"] = len(code)
expr(node["right"])
code[finish]["target"] = len(code)
else:
expr(node["left"])
expr(node["right"])
emit("BINARY", span, operator=node["op"])
elif kind == "unary":
operand = node["expr"]
if node["op"] == "-" and operand["kind"] == "literal" and type(operand["value"]) is int and operand["value"] == 1 << 63:
emit("PUSH", span, value=-(1 << 63))
else:
expr(operand)
emit("UNARY", span, operator=node["op"])
elif kind == "call":
for argument in node["args"]:
expr(argument)
emit("CALL", span, name=node["name"], argc=len(node["args"]))
elif kind in ("list", "tuple"):
for item in node["items"]:
expr(item)
emit(kind.upper(), span, count=len(node["items"]))
elif kind == "record":
ordered_fields = sorted(node["fields"])
for name in ordered_fields:
expr(node["fields"][name])
emit("RECORD", span, fields=ordered_fields)
elif kind == "index":
expr(node["target"])
expr(node["index"])
emit("INDEX", span)
elif kind == "field":
expr(node["target"])
emit("FIELD", span, name=node["name"])
else:
_fail("E_BYTECODE", "Unsupported expression: " + str(kind), span)
def statements(nodes, scoped=False):
if scoped:
emit("ENTER_SCOPE")
for node in nodes:
kind, span = node["kind"], node.get("span")
if kind == "let":
expr(node["expr"])
emit("LET", span, name=node["name"], mutable=node.get("mutable", False))
elif kind == "assign":
expr(node["expr"])
emit("STORE", span, name=node["name"])
elif kind == "return":
expr(node["expr"]) if node.get("expr") is not None else emit("PUSH", span, value=None)
emit("RETURN", span)
elif kind == "expr":
expr(node["expr"])
emit("POP", span)
elif kind == "if":
expr(node["cond"])
branch = emit("JUMP_FALSE", span, target=0)
statements(node["then"], scoped=True)
finish = emit("JUMP", span, target=0)
code[branch]["target"] = len(code)
statements(node.get("else", []), scoped=True)
code[finish]["target"] = len(code)
elif kind == "while":
loop_sequence[0] += 1
key = str(loop_sequence[0])
emit("LOOP_RESET", span, key=key)
start = len(code)
expr(node["cond"])
branch = emit("JUMP_FALSE", span, target=0)
emit("LOOP_GUARD", span, key=key, limit=node["limit"])
statements(node["body"], scoped=True)
emit("JUMP", span, target=start)
code[branch]["target"] = len(code)
else:
_fail("E_BYTECODE", "Unsupported statement: " + str(kind), span)
if scoped:
emit("EXIT_SCOPE")
# Parameters occupy the call frame; the function body is a child lexical
# scope, matching the source rule that a body local may shadow a parameter.
statements(function["body"], scoped=True)
emit("PUSH", value=None)
emit("RETURN")
functions[name] = {"params": function["params"], "return_type": function["return_type"],
"code": code}
result = {"schema": BYTECODE_SCHEMA, "module": program.get("module", "main"), "functions": functions}
validate_bytecode(result)
return result
def _integer(value, lower=0, upper=10000000):
return type(value) is int and lower <= value <= upper
def _validate_bytecode_impl(bytecode):
"""Fail closed on malformed code before executing any instruction."""
if not isinstance(bytecode, dict) or bytecode.get("schema") != BYTECODE_SCHEMA:
_fail("E_BYTECODE", "Unsupported bytecode schema")
if len(_canonical(bytecode)) > 20000000:
_fail("E_BYTECODE", "Bytecode exceeds serialized size limit")
functions = bytecode.get("functions")
if not isinstance(functions, dict) or not functions or len(functions) > 10000:
_fail("E_BYTECODE", "Invalid function table")
for name, function in functions.items():
if not isinstance(name, str) or not name or not isinstance(function, dict):
_fail("E_BYTECODE", "Invalid function descriptor")
params, code = function.get("params"), function.get("code")
if not isinstance(params, list) or any(not isinstance(p, (list, tuple)) or len(p) != 2
or not isinstance(p[0], str) or not isinstance(p[1], str) for p in params):
_fail("E_BYTECODE", "Invalid parameter table")
if len({p[0] for p in params}) != len(params):
_fail("E_BYTECODE", "Duplicate parameter")
if any(not valid_type(p[1]) for p in params) or not isinstance(function.get("return_type"), str) or not valid_type(function["return_type"]):
_fail("E_BYTECODE", "Invalid function type signature")
if not isinstance(code, list) or not code or len(code) > 1000000:
_fail("E_BYTECODE", "Invalid instruction stream")
for instruction in code:
if not isinstance(instruction, dict) or not isinstance(instruction.get("op"), str) or instruction["op"] not in OPCODES:
_fail("E_BYTECODE", "Unknown or malformed opcode")
op = instruction["op"]
if op in {"LOAD", "LET", "STORE", "CALL", "FIELD"} and not isinstance(instruction.get("name"), str):
_fail("E_BYTECODE", "Invalid identifier operand")
if op == "PUSH" and "value" not in instruction:
_fail("E_BYTECODE", "Missing literal operand")
if op == "LET" and type(instruction.get("mutable")) is not bool:
_fail("E_BYTECODE", "Invalid mutability operand")
if op in {"LIST", "TUPLE"} and not _integer(instruction.get("count")):
_fail("E_BYTECODE", "Invalid collection size")
if op == "CALL" and not _integer(instruction.get("argc")):
_fail("E_BYTECODE", "Invalid argument count")
if op == "RECORD" and (not isinstance(instruction.get("fields"), list)
or any(not isinstance(k, str) for k in instruction["fields"])
or len(instruction["fields"]) != len(set(instruction["fields"]))):
_fail("E_BYTECODE", "Invalid record fields")
if op in {"JUMP", "JUMP_FALSE"} and not _integer(instruction.get("target"), 0, len(code) - 1):
_fail("E_BYTECODE", "Jump target outside instruction stream")
if op == "BINARY" and (not isinstance(instruction.get("operator"), str) or instruction["operator"] not in {"+", "-", "*", "/", "%", "==", "!=", "<", "<=", ">", ">=", "&&", "||"}):
_fail("E_BYTECODE", "Invalid binary operator")
if op == "UNARY" and (not isinstance(instruction.get("operator"), str) or instruction["operator"] not in {"-", "!"}):
_fail("E_BYTECODE", "Invalid unary operator")
if op in {"LOOP_RESET", "LOOP_GUARD"} and not isinstance(instruction.get("key"), str):
_fail("E_BYTECODE", "Invalid loop key")
if op == "LOOP_GUARD" and not _integer(instruction.get("limit"), 0):
_fail("E_BYTECODE", "Invalid loop limit")
return True
def validate_bytecode(bytecode):
"""Validate untrusted JSON data and expose only deterministic ICL errors."""
try:
return _validate_bytecode_impl(bytecode)
except ICLError:
raise
except (TypeError, KeyError, IndexError, ValueError, AttributeError, RecursionError):
raise ICLError("BYTECODE_CORRUPT", "Malformed bytecode descriptor") from None
def _normalized_limits(limits):
result = dict(_DEFAULT_LIMITS)
if limits:
for key, value in limits.items():
if key not in result or not _integer(value, 1, result[key]):
_fail("RESOURCE_POLICY", "Invalid resource limit: " + str(key))
result[key] = value
return result
def _validate_value(value, limits, depth=0, budget=None):
if budget is None:
budget = [limits["max_items"]]
budget[0] -= 1
if budget[0] < 0 or depth > 128:
_fail("E_LIMIT", "Value exceeds collection or nesting limit")
if type(value) is int and not -(1 << 63) <= value < (1 << 63):
_fail("OVERFLOW", "Int64 overflow")
if value is None or type(value) in (bool, int):
return
if type(value) is float:
if not math.isfinite(value):
_fail("E_NUMERIC", "Non-finite floating-point value")
return
if isinstance(value, str):
if len(value) > limits["max_output"]:
_fail("E_LIMIT", "String exceeds output limit")
return
if isinstance(value, (list, tuple)):
for item in value:
_validate_value(item, limits, depth + 1, budget)
return
if isinstance(value, dict) and all(isinstance(k, str) for k in value):
for item in value.values():
_validate_value(item, limits, depth + 1, budget)
return
_fail("E_VALUE", "Unsupported runtime value")
def _new_frame(name, args, functions, ctx):
params = functions[name]["params"]
if len(args) != len(params):
_fail("E_ARITY", "Wrong argument count for " + name)
for (_, typ), value in zip(params, args):
validate_value(value, typ, ctx)
scope = {"locals": {p[0]: deepcopy(v) for p, v in zip(params, args)},
"mutable": {p[0]: False for p in params}}
return {"function": name, "pc": 0, "scopes": [scope], "stack": [], "loops": {}}
def _execute_bytecode(bytecode, entry="main", args=None, limits=None, policy=None,
checkpoint=None, stop_after=None):
"""Execute, pause, or resume verified bytecode. No ambient I/O is available."""
validate_bytecode(bytecode)
functions = bytecode["functions"]
limits = _normalized_limits(limits)
# Authority always comes from this trusted host argument, including replay.
# A checkpoint is data with an unkeyed checksum, never an authorization token.
policy = {} if policy is None else policy
if not isinstance(policy, dict):
_fail("RESOURCE_POLICY", "Host policy must be an object")
if stop_after is not None and not _integer(stop_after, 0):
_fail("E_LIMIT", "stop_after must be a nonnegative integer")
if checkpoint is not None:
if not isinstance(checkpoint, dict) or checkpoint.get("schema") != CHECKPOINT_SCHEMA:
_fail("E_CHECKPOINT", "Unsupported checkpoint schema")
payload = {k: v for k, v in checkpoint.items() if k != "integrity"}
if checkpoint.get("integrity") != _hash(payload) or checkpoint.get("bytecode_hash") != _hash(bytecode):
_fail("E_CHECKPOINT", "Checkpoint integrity or bytecode identity mismatch")
if len(_canonical(checkpoint)) > 20000000:
_fail("E_CHECKPOINT", "Checkpoint too large")
previous_limits = _normalized_limits(checkpoint.get("limits"))
limits = {k: min(v, previous_limits[k]) for k, v in limits.items()}
prior_policy = checkpoint.get("policy")
if policy != prior_policy:
_fail("E_CHECKPOINT", "Replay requires an independently supplied identical host policy")
ctx = RuntimeContext(limits=limits, policy=policy)
if checkpoint is None:
if entry not in functions:
_fail("E_ENTRY", "Unknown entry function: " + entry)
args = [] if args is None else args
if not isinstance(args, list):
_fail("E_ARITY", "Entry arguments must be a list")
_validate_value(args, limits)
frames = [_new_frame(entry, args, functions, ctx)]
else:
frames = deepcopy(checkpoint.get("frames"))
steps, trace, effects = checkpoint.get("steps"), checkpoint.get("trace"), checkpoint.get("effects")
if not _integer(steps, 0, limits["max_steps"]) or not isinstance(trace, list) or not isinstance(effects, list):
_fail("E_CHECKPOINT", "Malformed checkpoint accounting")
if not isinstance(frames, list) or not frames or len(frames) > limits["max_stack"]:
_fail("E_CHECKPOINT", "Malformed checkpoint frame stack")
for frame in frames:
if not isinstance(frame, dict) or frame.get("function") not in functions:
_fail("E_CHECKPOINT", "Unknown checkpoint function")
if not _integer(frame.get("pc"), 0, len(functions[frame["function"]]["code"]) - 1):
_fail("E_CHECKPOINT", "Invalid checkpoint instruction pointer")
if not isinstance(frame.get("scopes"), list) or not frame["scopes"] or not isinstance(frame.get("stack"), list):
_fail("E_CHECKPOINT", "Invalid checkpoint values")
if len(frame["scopes"]) > limits["max_steps"] + 1:
_fail("E_CHECKPOINT", "Too many lexical scopes")
for scope in frame["scopes"]:
if not isinstance(scope, dict) or not isinstance(scope.get("locals"), dict) or not isinstance(scope.get("mutable"), dict) or set(scope["mutable"]) != set(scope["locals"]) or any(type(v) is not bool for v in scope["mutable"].values()):
_fail("E_CHECKPOINT", "Invalid checkpoint mutability table")
_validate_value(scope["locals"], limits)
if not isinstance(frame.get("loops"), dict) or any(not _integer(v) for v in frame["loops"].values()):
_fail("E_CHECKPOINT", "Invalid checkpoint loop counters")
_validate_value(frame["stack"], limits)
ctx.steps, ctx.trace, ctx.effects = steps, deepcopy(trace), deepcopy(effects)
if not _integer(checkpoint.get("prng"), 0, (1 << 31) - 1):
_fail("E_CHECKPOINT", "Invalid checkpoint PRNG state")
ctx.prng = checkpoint["prng"]
start_steps = ctx.steps
def result(value, paused=False):
output = {"value": value, "trace": ctx.trace, "effects": ctx.effects,
"steps": ctx.steps, "halted": not paused}
if paused:
payload = {"schema": CHECKPOINT_SCHEMA, "bytecode_hash": _hash(bytecode),
"frames": deepcopy(frames), "steps": ctx.steps, "trace": deepcopy(ctx.trace),
"effects": deepcopy(ctx.effects), "limits": limits, "policy": ctx.policy,
"prng": ctx.prng}
payload["integrity"] = _hash(payload)
output["checkpoint"] = payload
return output
while frames:
if stop_after is not None and ctx.steps - start_steps >= stop_after:
return result(None, True)
if len(frames) > limits["max_stack"]:
_fail("STACK_LIMIT", "VM call stack limit exceeded")
frame = frames[-1]
code = functions[frame["function"]]["code"]
if not 0 <= frame["pc"] < len(code):
_fail("E_BYTECODE", "Instruction pointer outside function")
instruction = code[frame["pc"]]
input_state = {"frames": frames, "prng": ctx.prng}
trace_index = len(ctx.trace)
effects_before = len(ctx.effects)
ctx.step("BC:" + instruction["op"], inputs=input_state)
frame["pc"] += 1
op, span, stack = instruction["op"], instruction.get("span"), frame["stack"]
def finish_trace(return_value=None, returning=False):
state = {"frames": frames, "prng": ctx.prng}
if returning:
state["return_value"] = return_value
event = ctx.trace[trace_index]
event["output_hash"] = _hash(state)
event["state_ref"] = event["output_hash"]
if len(ctx.effects) > effects_before:
effect = ctx.effects[-1]
event["effect_request"] = {"action": effect["action"], "payload_hash": effect["payload_hash"]}
event["effect_decision"] = effect["decision"]
def pop(count=1):
if len(stack) < count:
_fail("E_BYTECODE", "Operand stack underflow", span)
if count == 0:
return []
values = stack[-count:]
del stack[-count:]
return values
if op == "PUSH":
stack.append(deepcopy(instruction["value"]))
elif op == "LOAD":
name = instruction["name"]
scope = next((s for s in reversed(frame["scopes"]) if name in s["locals"]), None)
if scope is None:
_fail("E_NAME", "Undefined variable: " + name, span)
stack.append(deepcopy(scope["locals"][name]))
elif op == "LET":
name = instruction["name"]
scope = frame["scopes"][-1]
if name in scope["locals"]:
_fail("E_NAME", "Duplicate local declaration: " + name, span)
scope["locals"][name] = pop()[0]
scope["mutable"][name] = instruction["mutable"]
elif op == "STORE":
name = instruction["name"]
scope = next((s for s in reversed(frame["scopes"]) if name in s["locals"]), None)
if scope is None:
_fail("E_NAME", "Undefined variable: " + name, span)
if not scope["mutable"][name]:
_fail("E_IMMUTABLE", "Cannot assign immutable variable: " + name, span)
scope["locals"][name] = pop()[0]
elif op == "ENTER_SCOPE":
frame["scopes"].append({"locals": {}, "mutable": {}})
elif op == "EXIT_SCOPE":
if len(frame["scopes"]) <= 1:
_fail("E_BYTECODE", "Cannot exit function root scope", span)
frame["scopes"].pop()
elif op == "POP":
pop()
elif op == "BINARY":
a, b = pop(2)
stack.append(checked_binary(instruction["operator"], a, b))
elif op == "UNARY":
stack.append(checked_unary(instruction["operator"], pop()[0]))
elif op == "CALL":
arguments = pop(instruction["argc"])
name = instruction["name"]
if name in functions:
if len(frames) >= limits["max_stack"]:
_fail("STACK_LIMIT", "VM call stack limit exceeded", span)
frames.append(_new_frame(name, arguments, functions, ctx))
else:
stack.append(builtin(name, arguments, ctx))
elif op in {"LIST", "TUPLE"}:
values = pop(instruction["count"])
stack.append(values)
elif op == "RECORD":
stack.append(dict(zip(instruction["fields"], pop(len(instruction["fields"])))))
elif op == "INDEX":
target, index = pop(2)
if not isinstance(target, (list, tuple, str)) or type(index) is not int or not 0 <= index < len(target):
_fail("E_INDEX", "Invalid or out-of-range index", span)
stack.append(target[index])
elif op == "FIELD":
target = pop()[0]
if not isinstance(target, dict) or instruction["name"] not in target:
_fail("E_FIELD", "Missing record field: " + instruction["name"], span)
stack.append(target[instruction["name"]])
elif op == "JUMP":
frame["pc"] = instruction["target"]
elif op == "JUMP_FALSE":
condition = pop()[0]
if type(condition) is not bool:
_fail("E_TYPE", "Branch condition must be Bool", span)
if not condition:
frame["pc"] = instruction["target"]
elif op == "LOOP_RESET":
frame["loops"][instruction["key"]] = 0
elif op == "LOOP_GUARD":
key = instruction["key"]
if key not in frame["loops"]:
_fail("E_BYTECODE", "Loop guard without reset", span)
frame["loops"][key] += 1
if frame["loops"][key] > instruction["limit"]:
_fail("LOOP_LIMIT", "Bounded loop limit exceeded", span)
elif op == "RETURN":
value = pop()[0]
validate_value(value, functions[frame["function"]]["return_type"], ctx)
if stack:
_fail("E_BYTECODE", "Unbalanced operand stack at return", span)
frames.pop()
if not frames:
_validate_value(value, limits)
finish_trace(value, returning=True)
return result(value)
frames[-1]["stack"].append(value)
if len(stack) > limits["max_items"]:
_fail("E_LIMIT", "VM operand stack limit exceeded", span)
if stack:
_validate_value(stack[-1], limits)
ctx.check_value(stack[-1])
finish_trace()
_fail("E_BYTECODE", "VM terminated without return")
def execute_bytecode(bytecode, entry="main", args=None, limits=None, policy=None,
checkpoint=None, stop_after=None):
"""Public fail-closed boundary; malformed data cannot escape as host errors."""
try:
return _execute_bytecode(bytecode, entry, args, limits, policy, checkpoint, stop_after)
except ICLError:
raise
except (TypeError, KeyError, IndexError, ValueError, OverflowError, ZeroDivisionError, AttributeError) as error:
raise ICLError("TYPE", "Malformed bytecode operand or primitive argument: " + type(error).__name__) from None
except RecursionError:
raise ICLError("STACK_LIMIT", "Bytecode structure exceeds host recursion limit") from None