59 lines
1.9 KiB
Python
59 lines
1.9 KiB
Python
"""Wire protocol and pure-logic helpers shared between the debug agent and MCP server.
|
|
No ctypes, no Windows dependency -- runs and tests on any platform."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import operator
|
|
from typing import Callable, Optional
|
|
|
|
_OPS = {
|
|
"==": operator.eq,
|
|
"!=": operator.ne,
|
|
"<": operator.lt,
|
|
">": operator.gt,
|
|
"<=": operator.le,
|
|
">=": operator.ge,
|
|
}
|
|
|
|
|
|
def _parse_memory_expr(expr: str) -> int:
|
|
"""Parse "0x47eb48+0x18" or "0x47eb48" into an absolute address."""
|
|
if "+" in expr:
|
|
base_str, offset_str = expr.split("+", 1)
|
|
return int(base_str.strip(), 16) + int(offset_str.strip(), 16)
|
|
return int(expr.strip(), 16)
|
|
|
|
|
|
def evaluate_condition(
|
|
condition: dict,
|
|
registers: dict,
|
|
read_memory: Optional[Callable[[int, int], bytes]],
|
|
) -> bool:
|
|
"""Evaluate a conditional-breakpoint expression against a register snapshot and/or
|
|
memory reader. `condition` is either {"register": name, "op": ..., "value": hexstr}
|
|
or {"memory": expr, "size": n, "op": ..., "value": hexstr}."""
|
|
op_fn = _OPS[condition["op"]]
|
|
expected = int(condition["value"], 16)
|
|
|
|
if "register" in condition:
|
|
actual = registers[condition["register"]]
|
|
elif "memory" in condition:
|
|
addr = _parse_memory_expr(condition["memory"])
|
|
size = condition["size"]
|
|
raw = read_memory(addr, size)
|
|
actual = int.from_bytes(raw, "little")
|
|
else:
|
|
raise ValueError(f"condition must have 'register' or 'memory': {condition}")
|
|
|
|
return op_fn(actual, expected)
|
|
|
|
|
|
def encode_message(op: str, params: dict) -> bytes:
|
|
"""Encode a request as a single NDJSON line."""
|
|
return (json.dumps({"op": op, "params": params}) + "\n").encode("utf-8")
|
|
|
|
|
|
def decode_message(line: bytes) -> dict:
|
|
"""Decode a single NDJSON line (request or response) back into a dict."""
|
|
return json.loads(line.decode("utf-8"))
|