Mohamed Ayoub Eleuch
8a2a1a2b8b
Count real UTF-8 bytes in bounded chunks and retain authenticated omission for malformed Unicode. Preserve existing wire/resource limits and completion semantics. Validation: producer/client regressions, fresh real Linux Cutter CLI/MCP output and Evidence recovery, and complete exact-head CI passed. Hosted unrelated CLI timeout and unchanged rerun results remain recorded in the PR. Fixes #3330.
600 lines
27 KiB
Python
600 lines
27 KiB
Python
"""Producer-level regression tests for Cutter's authenticated local protocol."""
|
|
|
|
import hashlib
|
|
import hmac
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import types
|
|
import unittest
|
|
import uuid
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
|
|
def _install_host_stubs():
|
|
cutter = types.ModuleType("cutter")
|
|
cutter.CutterPlugin = type("CutterPlugin", (), {})
|
|
sys.modules["cutter"] = cutter
|
|
qtcore = types.ModuleType("PySide6.QtCore")
|
|
qtcore.QObject = type("QObject", (), {})
|
|
qtcore.Signal = lambda *_args: types.SimpleNamespace(connect=lambda _handler: None)
|
|
qtcore.QCoreApplication = types.SimpleNamespace(applicationVersion=lambda: "Cutter test host")
|
|
pyside = types.ModuleType("PySide6")
|
|
pyside.QtCore = qtcore
|
|
sys.modules["PySide6"] = pyside
|
|
sys.modules["PySide6.QtCore"] = qtcore
|
|
|
|
|
|
_install_host_stubs()
|
|
MODULE_PATH = Path(__file__).parents[2] / "bridge" / "cutter" / "rea_bridge.py"
|
|
spec = importlib.util.spec_from_file_location("rea_cutter_bridge_test", MODULE_PATH)
|
|
bridge = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = bridge
|
|
spec.loader.exec_module(bridge)
|
|
|
|
|
|
def _frame(connection):
|
|
data = bytearray()
|
|
while b"\n" not in data:
|
|
data.extend(connection.recv(4096))
|
|
if len(data) > bridge.MAX_FRAME_BYTES:
|
|
raise AssertionError("oversized test frame")
|
|
line, remainder = bytes(data).split(b"\n", 1)
|
|
if remainder:
|
|
raise AssertionError("unexpected pipelined frame")
|
|
return json.loads(line)
|
|
|
|
|
|
def _send(connection, value):
|
|
connection.sendall(json.dumps(value, separators=(",", ":")).encode() + b"\n")
|
|
|
|
|
|
class Dispatcher:
|
|
generation = 4
|
|
current_file = "/tmp/sample.bin"
|
|
cutter_version = "test"
|
|
|
|
def __init__(self):
|
|
self.dispatch = self
|
|
self.commands = []
|
|
self.completed_commands = {}
|
|
self.lock = threading.Lock()
|
|
self.command_started = threading.Event()
|
|
self.finish_command = threading.Event()
|
|
self.status_queued = threading.Event()
|
|
|
|
def _read_current_file(self):
|
|
return self.current_file
|
|
|
|
def can_accept_command(self):
|
|
return len(self.completed_commands) < bridge.MAX_COMPLETION_RECEIPTS
|
|
|
|
def emit(self, request, reply):
|
|
def run():
|
|
kind = request.get("kind")
|
|
if kind == "command":
|
|
self.command_started.set()
|
|
elif kind == "status":
|
|
self.status_queued.set()
|
|
with self.lock:
|
|
if kind == "command":
|
|
self.finish_command.wait(2)
|
|
self.commands.append(kind)
|
|
if kind == "status":
|
|
reply.update({"ok": True, "session_id": request["session_id"], "document_generation": self.generation,
|
|
"current_file": self.current_file, "cutter_version": self.cutter_version,
|
|
"identity_status": "partial", "completed_commands": [
|
|
{"request_id": request_id, "execution_state": state}
|
|
for request_id, state in self.completed_commands.items()
|
|
]})
|
|
elif kind == "ack":
|
|
reply.update({"ok": True, "acknowledged": self.completed_commands.pop(request["request_id"], None) is not None})
|
|
else:
|
|
reply.update({"ok": True, "execution_state": "complete"})
|
|
if kind == "command" and isinstance(request.get("request_id"), str):
|
|
self.completed_commands[request["request_id"]] = "complete"
|
|
request["_event"].set()
|
|
threading.Thread(target=run, daemon=True).start()
|
|
|
|
|
|
class ServeDispatcher:
|
|
"""Small UI-dispatcher stand-in for real producer/client interoperability."""
|
|
|
|
generation = 4
|
|
current_file = "/tmp/cutter-interop-雪.bin"
|
|
cutter_version = "fixture"
|
|
|
|
def __init__(self):
|
|
self.dispatch = self
|
|
self.completed_commands = {}
|
|
self.fail_next_ack = False
|
|
self.malformed_next_ack = False
|
|
self.hide_receipts_after_ack_failure = False
|
|
|
|
def can_accept_command(self):
|
|
return len(self.completed_commands) < bridge.MAX_COMPLETION_RECEIPTS
|
|
|
|
def _read_current_file(self):
|
|
return self.current_file
|
|
|
|
def emit(self, request, reply):
|
|
if request.get("kind") == "status":
|
|
reply.update({
|
|
"ok": True,
|
|
"session_id": request["session_id"],
|
|
"document_generation": self.generation,
|
|
"current_file": self.current_file,
|
|
"cutter_version": self.cutter_version,
|
|
"identity_status": "partial",
|
|
"completed_commands": [
|
|
{"request_id": request_id, "execution_state": state}
|
|
for request_id, state in self.completed_commands.items()
|
|
] if not self.hide_receipts_after_ack_failure else [],
|
|
})
|
|
elif request.get("kind") == "ack":
|
|
if self.malformed_next_ack:
|
|
self.malformed_next_ack = False
|
|
reply.update({"ok": True, "session_id": request["session_id"]})
|
|
request["_event"].set()
|
|
return
|
|
if self.fail_next_ack:
|
|
self.fail_next_ack = False
|
|
self.hide_receipts_after_ack_failure = True
|
|
reply.update({"ok": False, "acknowledged": False})
|
|
request["_event"].set()
|
|
return
|
|
reply.update({
|
|
"ok": True,
|
|
"acknowledged": self.completed_commands.pop(request["request_id"], None) is not None,
|
|
})
|
|
self.hide_receipts_after_ack_failure = False
|
|
elif request.get("kind") == "command" and request.get("expected_generation") != self.generation:
|
|
reply.update({
|
|
"ok": False,
|
|
"error": "active-document-generation-changed",
|
|
"execution_state": "not_started",
|
|
})
|
|
self.completed_commands[request["request_id"]] = "not_started"
|
|
elif request.get("kind") == "command" and request.get("command") == "fixture-unknown":
|
|
reply.update({
|
|
"ok": False,
|
|
"error": "fixture-response-lost-after-completion",
|
|
"execution_state": "unknown",
|
|
})
|
|
self.completed_commands[request["request_id"]] = "complete"
|
|
elif request.get("kind") == "command" and request.get("command") == "fixture-fail-ack":
|
|
reply.update({
|
|
"ok": True,
|
|
"output": "ack will fail once",
|
|
"current_file": self.current_file,
|
|
"document_generation": self.generation,
|
|
"cutter_version": self.cutter_version,
|
|
"identity_status": "partial",
|
|
"execution_state": "complete",
|
|
})
|
|
self.completed_commands[request["request_id"]] = "complete"
|
|
self.fail_next_ack = True
|
|
elif request.get("kind") == "command" and request.get("command") == "fixture-malformed-ack":
|
|
reply.update({
|
|
"ok": True,
|
|
"output": "ack reply will be malformed once",
|
|
"current_file": self.current_file,
|
|
"document_generation": self.generation,
|
|
"cutter_version": self.cutter_version,
|
|
"identity_status": "partial",
|
|
"execution_state": "complete",
|
|
})
|
|
self.completed_commands[request["request_id"]] = "complete"
|
|
self.malformed_next_ack = True
|
|
else:
|
|
reply.update({
|
|
"ok": True,
|
|
"output": 'round-trip: 雪 ☃️ quote=" slash=\\\\ line=\\n tab=\\t',
|
|
"current_file": self.current_file,
|
|
"document_generation": self.generation,
|
|
"cutter_version": self.cutter_version,
|
|
"identity_status": "partial",
|
|
"execution_state": "complete",
|
|
})
|
|
if request.get("kind") == "command":
|
|
self.completed_commands[request["request_id"]] = "complete"
|
|
request["_event"].set()
|
|
|
|
|
|
def run_serve_mode():
|
|
server = bridge._BridgeServer(ServeDispatcher())
|
|
server.start()
|
|
try:
|
|
deadline = time.monotonic() + 5
|
|
while not server.descriptor.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.005)
|
|
if not server.descriptor.exists():
|
|
raise RuntimeError("bridge descriptor was not published")
|
|
print("ready", flush=True)
|
|
sys.stdin.buffer.readline()
|
|
finally:
|
|
server.stop()
|
|
server.join(timeout=3)
|
|
|
|
|
|
class ResponseEncodingTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.server = bridge._BridgeServer.__new__(bridge._BridgeServer)
|
|
self.server.token = "test-token"
|
|
self.server.session_id = str(uuid.uuid4())
|
|
|
|
def assert_authenticated_response(self, frame):
|
|
expected = hmac.new(
|
|
self.server.token.encode(),
|
|
f"rea-cutter-response-v2:{self.server.session_id}:{'a' * 64}:{'b' * 64}:{frame['response']}".encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
self.assertTrue(hmac.compare_digest(frame["proof"], expected))
|
|
return json.loads(frame["response"])
|
|
|
|
def test_large_mixed_unicode_output_fits_the_authenticated_frame(self):
|
|
for suffix in ("a", "é", "雪", "😀"):
|
|
with self.subTest(suffix=suffix):
|
|
output = "a" * (5 * 1024 * 1024 - 1) + suffix
|
|
value = {"ok": True, "output": output, "execution_state": "complete"}
|
|
encoded = self.server._encode_authenticated_response(value, "a" * 64, "b" * 64)
|
|
self.assertIsNotNone(encoded)
|
|
self.assertLessEqual(len(encoded), bridge.MAX_FRAME_BYTES)
|
|
frame = json.loads(encoded)
|
|
self.assertEqual(self.assert_authenticated_response(frame), value)
|
|
self.assertIsNone(self.server._encode_authenticated_response({
|
|
"ok": True, "output": "😀" * (5 * 1024 * 1024),
|
|
"execution_state": "complete",
|
|
}, "a" * 64, "b" * 64))
|
|
|
|
def test_unencodable_output_preserves_authenticated_completion(self):
|
|
for index, surrogate in ((0, "\ud800"), (5 * 1024 * 1024 - 1, "\udfff")):
|
|
with self.subTest(index=index):
|
|
output = "a" * index + surrogate + "a" * (5 * 1024 * 1024 - index - 1)
|
|
value = {"ok": True, "output": output, "execution_state": "complete"}
|
|
sender, receiver = socket.socketpair()
|
|
with sender, receiver:
|
|
receiver.settimeout(2)
|
|
self.server._send_authenticated(sender, value, "a" * 64, "b" * 64)
|
|
response = self.assert_authenticated_response(_frame(receiver))
|
|
self.assertTrue(response["ok"])
|
|
self.assertEqual(response["execution_state"], "complete")
|
|
self.assertIsNone(response["output"])
|
|
self.assertTrue(response["output_truncated"])
|
|
|
|
|
|
class DocumentIdentityTest(unittest.TestCase):
|
|
def test_document_config_drives_generation_and_command_admission(self):
|
|
first = "/tmp/Cutter 雪 #1?.elf"
|
|
second = "/tmp/another document.elf"
|
|
configuration = {"file.path": first}
|
|
core = types.SimpleNamespace(
|
|
getConfig=configuration.__getitem__,
|
|
)
|
|
commands = []
|
|
with patch.object(bridge.cutter, "core", lambda: core, create=True), patch.object(
|
|
bridge.cutter, "cmd", lambda command: commands.append(command) or "observed", create=True
|
|
):
|
|
dispatcher = bridge._RequestDispatcher()
|
|
self.assertEqual(dispatcher.current_file, first)
|
|
self.assertEqual(dispatcher.generation, 0)
|
|
|
|
def handle(**values):
|
|
request = {**values, "_event": threading.Event(), "_cancelled": threading.Event()}
|
|
reply = {}
|
|
dispatcher._handle(request, reply)
|
|
self.assertTrue(request["_event"].is_set())
|
|
return reply
|
|
|
|
same = handle(kind="status")
|
|
self.assertEqual(same["current_file"], first)
|
|
self.assertEqual(same["document_generation"], 0)
|
|
configuration["file.path"] = second
|
|
changed = handle(kind="status")
|
|
self.assertEqual(changed["current_file"], second)
|
|
self.assertEqual(changed["document_generation"], 1)
|
|
stale = handle(kind="command", request_id=str(uuid.uuid4()), expected_generation=0, command="iI")
|
|
self.assertEqual(stale["error"], "active-document-generation-changed")
|
|
self.assertEqual(stale["execution_state"], "not_started")
|
|
self.assertEqual(commands, [])
|
|
current = handle(kind="command", request_id=str(uuid.uuid4()), expected_generation=1, command="iI")
|
|
self.assertEqual(current["output"], "observed")
|
|
self.assertEqual(current["document_generation"], 1)
|
|
self.assertEqual(commands, ["iI"])
|
|
|
|
def test_absent_or_unreadable_document_config_remains_unknown(self):
|
|
class UnreadableCore:
|
|
def getConfig(self, _key):
|
|
raise RuntimeError("document config unavailable")
|
|
|
|
for core in [
|
|
types.SimpleNamespace(getConfig=lambda _key: ""),
|
|
types.SimpleNamespace(getConfig=lambda _key: None),
|
|
types.SimpleNamespace(getConfig=lambda _key: object()),
|
|
types.SimpleNamespace(),
|
|
UnreadableCore(),
|
|
]:
|
|
with self.subTest(core=type(core).__name__), patch.object(
|
|
bridge.cutter, "core", lambda: core, create=True
|
|
):
|
|
self.assertIsNone(bridge._RequestDispatcher._read_current_file())
|
|
|
|
|
|
class HostVersionTest(unittest.TestCase):
|
|
def test_host_version_reaches_status_and_command_without_backend_attribution(self):
|
|
backend_reads = []
|
|
core = types.SimpleNamespace(
|
|
getConfig=lambda _key: None,
|
|
getVersionInformation=lambda: backend_reads.append(True) or "Rizin 0.9.1",
|
|
)
|
|
application = sys.modules["PySide6.QtCore"].QCoreApplication
|
|
with patch.object(bridge.cutter, "core", lambda: core, create=True), patch.object(
|
|
application, "applicationVersion", return_value="2.5.0-HEAD-fixture"
|
|
), patch.object(bridge.cutter, "cmd", lambda _command: "observed", create=True):
|
|
dispatcher = bridge._RequestDispatcher()
|
|
self.assertEqual(dispatcher.cutter_version, "2.5.0-HEAD-fixture")
|
|
for values in [
|
|
{"kind": "status"},
|
|
{"kind": "command", "request_id": str(uuid.uuid4()), "expected_generation": 0, "command": "iI"},
|
|
]:
|
|
request = {**values, "_event": threading.Event(), "_cancelled": threading.Event()}
|
|
reply = {}
|
|
dispatcher._handle(request, reply)
|
|
self.assertTrue(request["_event"].is_set())
|
|
self.assertEqual(reply["cutter_version"], "2.5.0-HEAD-fixture")
|
|
self.assertEqual(backend_reads, [])
|
|
|
|
def test_unavailable_host_version_stays_unknown_without_backend_fallback(self):
|
|
core = types.SimpleNamespace(
|
|
getConfig=lambda _key: None,
|
|
getVersionInformation=lambda: "Rizin 0.9.1",
|
|
)
|
|
application = sys.modules["PySide6.QtCore"].QCoreApplication
|
|
with patch.object(bridge.cutter, "core", lambda: core, create=True):
|
|
for value in ["", None, object()]:
|
|
with self.subTest(value=type(value).__name__), patch.object(
|
|
application, "applicationVersion", return_value=value
|
|
):
|
|
self.assertIsNone(bridge._RequestDispatcher().cutter_version)
|
|
with patch.object(application, "applicationVersion", side_effect=RuntimeError("host unavailable")):
|
|
self.assertIsNone(bridge._RequestDispatcher().cutter_version)
|
|
|
|
|
|
class ProtocolProducerTest(unittest.TestCase):
|
|
def test_rejects_legacy_peers_before_ui_dispatch(self):
|
|
with socket.create_connection(("127.0.0.1", self.server.port), timeout=2) as connection:
|
|
connection.sendall((json.dumps({"kind": "rea-auth-v1", "nonce": "a" * 64}) + "\n").encode())
|
|
with connection.makefile("rb") as reader:
|
|
reply = json.loads(reader.readline())
|
|
self.assertEqual(reply, {"ok": False, "error": "authentication-failed"})
|
|
self.assertEqual(self.dispatcher.commands, [])
|
|
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
os.environ["REA_CUTTER_BRIDGE_DIR"] = self.temp.name
|
|
self.dispatcher = Dispatcher()
|
|
self.server = bridge._BridgeServer(self.dispatcher)
|
|
bridge.SOCKET_TIMEOUT_SECONDS = 0.1
|
|
self.server.start()
|
|
deadline = time.monotonic() + 2
|
|
while not self.server.descriptor.exists() and time.monotonic() < deadline:
|
|
time.sleep(0.005)
|
|
self.assertTrue(self.server.descriptor.exists())
|
|
|
|
def tearDown(self):
|
|
self.server.stop()
|
|
self.server.join(timeout=2)
|
|
os.environ.pop("REA_CUTTER_BRIDGE_DIR", None)
|
|
self.temp.cleanup()
|
|
|
|
def exchange(self, request):
|
|
token = self.server.token.encode()
|
|
client_nonce = os.urandom(32).hex()
|
|
with socket.create_connection(("127.0.0.1", self.server.port), timeout=1) as connection:
|
|
_send(connection, {"kind": "rea-auth-v2", "nonce": client_nonce})
|
|
challenge = _frame(connection)
|
|
self.assertEqual(challenge["session_id"], self.server.session_id)
|
|
server_nonce = challenge["nonce"]
|
|
expected_server = hmac.new(token, f"rea-cutter-server-v2:{self.server.session_id}:{client_nonce}:{server_nonce}".encode(), hashlib.sha256).hexdigest()
|
|
self.assertTrue(hmac.compare_digest(challenge["proof"], expected_server))
|
|
request_bytes = json.dumps(request, separators=(",", ":"))
|
|
proof = hmac.new(token, f"rea-cutter-client-v2:{self.server.session_id}:{client_nonce}:{server_nonce}:{request_bytes}".encode(), hashlib.sha256).hexdigest()
|
|
_send(connection, {"request": request_bytes, "proof": proof})
|
|
frame = _frame(connection)
|
|
if "response" not in frame:
|
|
return frame, client_nonce
|
|
response = frame["response"]
|
|
expected_response = hmac.new(token, f"rea-cutter-response-v2:{self.server.session_id}:{client_nonce}:{server_nonce}:{response}".encode(), hashlib.sha256).hexdigest()
|
|
self.assertTrue(hmac.compare_digest(frame["proof"], expected_response))
|
|
return json.loads(response), client_nonce
|
|
|
|
def test_request_session_uuid_is_checked_after_secret_proof(self):
|
|
spoofed = {"kind": "status", "session_id": "client-chosen-session"}
|
|
response, _ = self.exchange(spoofed)
|
|
self.assertEqual(response["error"], "authentication-failed")
|
|
self.assertEqual(self.dispatcher.commands, [])
|
|
|
|
def test_client_chosen_session_uuid_without_secret_proof_is_not_authenticated(self):
|
|
client_nonce = os.urandom(32).hex()
|
|
with socket.create_connection(("127.0.0.1", self.server.port), timeout=1) as connection:
|
|
_send(connection, {"kind": "rea-auth-v2", "nonce": client_nonce})
|
|
_frame(connection)
|
|
request_bytes = json.dumps({"kind": "status", "session_id": "client-chosen-session"}, separators=(",", ":"))
|
|
_send(connection, {"request": request_bytes, "proof": "0" * 64})
|
|
self.assertEqual(_frame(connection)["error"], "authentication-failed")
|
|
self.assertEqual(self.dispatcher.commands, [])
|
|
|
|
def test_cancelled_command_receipt_survives_later_command_until_acknowledged(self):
|
|
dispatcher = object.__new__(bridge._RequestDispatcher)
|
|
dispatcher.generation = 4
|
|
dispatcher.current_file = "/tmp/sample.bin"
|
|
dispatcher.cutter_version = "test"
|
|
dispatcher.completed_commands = {}
|
|
dispatcher._receipt_lock = threading.Lock()
|
|
dispatcher._refresh_document_identity = lambda: None
|
|
bridge.cutter.cmd = lambda command: f"output:{command}"
|
|
|
|
def handle(value):
|
|
request = {
|
|
**value,
|
|
"_event": threading.Event(),
|
|
"_cancelled": threading.Event(),
|
|
}
|
|
reply = {}
|
|
bridge._RequestDispatcher._handle(dispatcher, request, reply)
|
|
self.assertTrue(request["_event"].is_set())
|
|
return reply
|
|
|
|
request_a = "4b18ce4d-bf31-456e-ae31-e8c547b2f9db"
|
|
request_b = "8ca93d11-107d-41ed-8b40-31dd53cd26a6"
|
|
for request_id, command in ((request_a, "A"), (request_b, "B")):
|
|
result = handle({
|
|
"kind": "command",
|
|
"request_id": request_id,
|
|
"expected_generation": 4,
|
|
"command": command,
|
|
})
|
|
self.assertTrue(result["ok"])
|
|
|
|
status = handle({"kind": "status"})
|
|
self.assertEqual(
|
|
{receipt["request_id"] for receipt in status["completed_commands"]},
|
|
{request_a, request_b},
|
|
)
|
|
acknowledged = handle({"kind": "ack", "request_id": request_a})
|
|
self.assertTrue(acknowledged["acknowledged"])
|
|
after_ack = handle({"kind": "status"})
|
|
self.assertEqual(
|
|
[receipt["request_id"] for receipt in after_ack["completed_commands"]],
|
|
[request_b],
|
|
)
|
|
|
|
def test_stale_generation_rejections_are_terminal_and_acknowledged(self):
|
|
dispatcher = object.__new__(bridge._RequestDispatcher)
|
|
dispatcher.generation = 4
|
|
dispatcher.current_file = "/tmp/sample.bin"
|
|
dispatcher.cutter_version = "test"
|
|
dispatcher.completed_commands = {}
|
|
dispatcher._receipt_lock = threading.Lock()
|
|
dispatcher._refresh_document_identity = lambda: None
|
|
|
|
def handle(value):
|
|
request = {
|
|
**value,
|
|
"_event": threading.Event(),
|
|
"_cancelled": threading.Event(),
|
|
}
|
|
reply = {}
|
|
bridge._RequestDispatcher._handle(dispatcher, request, reply)
|
|
self.assertTrue(request["_event"].is_set())
|
|
return reply
|
|
|
|
invalid_id = str(uuid.UUID(int=bridge.MAX_COMPLETION_RECEIPTS + 10))
|
|
invalid = handle({
|
|
"kind": "command",
|
|
"request_id": invalid_id,
|
|
"expected_generation": 4,
|
|
"command": "",
|
|
})
|
|
self.assertEqual(invalid["execution_state"], "not_started")
|
|
self.assertTrue(handle({"kind": "ack", "request_id": invalid_id})["acknowledged"])
|
|
|
|
for index in range(bridge.MAX_COMPLETION_RECEIPTS + 1):
|
|
request_id = str(uuid.UUID(int=index + 1))
|
|
response = handle({
|
|
"kind": "command",
|
|
"request_id": request_id,
|
|
"expected_generation": 3,
|
|
"command": "iI",
|
|
})
|
|
self.assertEqual(response["execution_state"], "not_started")
|
|
self.assertEqual(response["error"], "active-document-generation-changed")
|
|
self.assertTrue(handle({"kind": "ack", "request_id": request_id})["acknowledged"])
|
|
|
|
self.assertEqual(dispatcher.completed_commands, {})
|
|
self.assertTrue(dispatcher.can_accept_command())
|
|
|
|
def test_poisoned_rejection_is_definitively_not_started_without_dispatch(self):
|
|
self.server.poisoned = True
|
|
response, _ = self.exchange({
|
|
"kind": "command",
|
|
"session_id": self.server.session_id,
|
|
"request_id": "4b18ce4d-bf31-456e-ae31-e8c547b2f9db",
|
|
})
|
|
self.assertEqual(response["error"], "session-poisoned-after-ui-timeout")
|
|
self.assertEqual(response["execution_state"], "not_started")
|
|
self.assertEqual(self.dispatcher.commands, [])
|
|
|
|
def test_receipt_capacity_rejects_before_ui_dispatch(self):
|
|
self.dispatcher.can_accept_command = lambda: False
|
|
response, _ = self.exchange({
|
|
"kind": "command",
|
|
"session_id": self.server.session_id,
|
|
"request_id": "4b18ce4d-bf31-456e-ae31-e8c547b2f9db",
|
|
})
|
|
self.assertEqual(response["error"], "completion-receipt-capacity-exceeded")
|
|
self.assertEqual(response["execution_state"], "not_started")
|
|
self.assertEqual(self.dispatcher.commands, [])
|
|
|
|
def test_status_probe_recovers_only_after_timed_out_ui_operation_finishes(self):
|
|
timed_out, _ = self.exchange({"kind": "command", "session_id": self.server.session_id})
|
|
self.assertEqual(timed_out["error"], "ui-thread-request-timeout")
|
|
self.assertTrue(self.server.poisoned)
|
|
self.assertTrue(self.dispatcher.command_started.wait(1))
|
|
result = {}
|
|
status_thread = threading.Thread(target=lambda: result.setdefault(
|
|
"response", self.exchange({"kind": "status", "session_id": self.server.session_id})[0]
|
|
))
|
|
status_thread.start()
|
|
self.assertTrue(self.dispatcher.status_queued.wait(1))
|
|
self.dispatcher.finish_command.set()
|
|
status_thread.join(timeout=2)
|
|
self.assertFalse(status_thread.is_alive())
|
|
response = result["response"]
|
|
self.assertTrue(response["ok"])
|
|
self.assertEqual(response["session_id"], self.server.session_id)
|
|
self.assertEqual(self.dispatcher.commands, ["command", "status"])
|
|
|
|
def test_escape_expansion_uses_a_signed_truthful_truncation_fallback(self):
|
|
sender, receiver = socket.socketpair()
|
|
try:
|
|
self.server._send_authenticated(sender, {
|
|
"ok": True,
|
|
"output": '"' * 6_000_000,
|
|
"execution_state": "complete",
|
|
"current_file": "/tmp/sample.bin",
|
|
"document_generation": 4,
|
|
"identity_status": "partial",
|
|
}, "a" * 64, "b" * 64)
|
|
frame_bytes = receiver.recv(bridge.MAX_FRAME_BYTES + 1)
|
|
self.assertLessEqual(len(frame_bytes), bridge.MAX_FRAME_BYTES)
|
|
frame = json.loads(frame_bytes)
|
|
response = frame["response"]
|
|
expected = hmac.new(
|
|
self.server.token.encode(),
|
|
f"rea-cutter-response-v2:{self.server.session_id}:{'a' * 64}:{'b' * 64}:{response}".encode(),
|
|
hashlib.sha256,
|
|
).hexdigest()
|
|
self.assertTrue(hmac.compare_digest(frame["proof"], expected))
|
|
value = json.loads(response)
|
|
self.assertEqual(value["ok"], True)
|
|
self.assertEqual(value["execution_state"], "complete")
|
|
self.assertTrue(value["output_truncated"])
|
|
self.assertIsNone(value["output"])
|
|
finally:
|
|
sender.close()
|
|
receiver.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
if sys.argv[1:] == ["--serve"]:
|
|
run_serve_mode()
|
|
else:
|
|
unittest.main()
|