From db5f85c9228788f1f6c4a7bfebd37a760ad869df Mon Sep 17 00:00:00 2001 From: Gerardo Betancourt Date: Tue, 27 Oct 2026 23:54:41 -0400 Subject: [PATCH] Fix private production pilot recovery, screen transfer and MCP dispatch --- RELEASE-NOTES.md | 14 + .../expanded/framebridge_expanded/broker.py | 24 +- source/expanded/framebridge_expanded/fleet.py | 29 +- source/expanded/framebridge_expanded/node.py | 69 ++- .../real_screen/screen_capture.py | 2 +- .../real_screen/screen_frontdoor.py | 131 +++++- .../real_screen/screen_rpc.py | 14 +- .../real_screen/screen_transport.py | 5 +- .../expanded/framebridge_expanded/runtime.py | 19 +- source/packaging/build_expanded.py | 8 +- source/portable/front_door_mcp.py | 59 ++- .../frozen_screen_receiver_f6b14c52.json.gz | Bin 0 -> 20290 bytes source/tests/test_frontdoor_dispatch.py | 316 +++++++++++++ source/tests/test_frontdoor_thread_state.py | 224 +++++++++ source/tests/test_node_recovery.py | 341 ++++++++++++++ source/tests/test_private_candidate_prerm.py | 91 ++++ source/tests/test_screen_registration.py | 24 +- source/tests/test_screen_transfer_budget.py | 431 ++++++++++++++++++ 18 files changed, 1735 insertions(+), 66 deletions(-) create mode 100644 source/tests/fixtures/frozen_screen_receiver_f6b14c52.json.gz create mode 100644 source/tests/test_frontdoor_dispatch.py create mode 100644 source/tests/test_frontdoor_thread_state.py create mode 100644 source/tests/test_node_recovery.py create mode 100644 source/tests/test_private_candidate_prerm.py create mode 100644 source/tests/test_screen_transfer_budget.py diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md index 6d6cd2d..abede3c 100644 --- a/RELEASE-NOTES.md +++ b/RELEASE-NOTES.md @@ -1,5 +1,19 @@ # FrameBridge release notes +## Unreleased — private production pilot fixes + +Private pilot validation remains in progress; these changes are not a public release. + +- Recover from classified transient broker failures with capped backoff and fresh signed nonces. Result acknowledgement retries retain the computed result and never reexecute a command. TLS, authentication and malformed-response failures remain terminal. +- Dispatch blocking MCP operations in bounded workers, preserve synchronous internal callbacks, and protect shared initialization. Concurrent generated local job IDs are distinct. +- Negotiate larger original-PNG chunks per capture while retaining legacy receivers. Apply a cooperative transfer budget and require fresh observation after uncertain input replies; performed claims require explicit matching backend confirmation. +- Bind local real-terminal capture transport to the current node identity. +- Keep privileged-helper downgrade revocation tied to the fixed 0.3.2 compatibility floor, allowing private candidates to return to 0.3.2 without revoking grants. + +Validation: 721 integrated source tests passed with zero failures, errors or skips. Frozen 0.3.2 source and signed Debian artifact remained unchanged. Focused tests cover real vendored MCP dispatch, retry/no-replay behavior, frozen screen-receiver compatibility and isolated maintainer-script semantics. + +Known limits: the cooperative capture budget can be exceeded by an already-running lower transport call. A broker restart can reject an unknown in-flight result; a lost poll reply can orphan a request. These cases do not reconstruct or replay commands. Live fleet acceptance remains separate from source-test success. + ## 0.3.1 — Community Release Installer: FrameBridge-0.3.1.deb diff --git a/source/expanded/framebridge_expanded/broker.py b/source/expanded/framebridge_expanded/broker.py index 0f9cb6f..6c35c25 100644 --- a/source/expanded/framebridge_expanded/broker.py +++ b/source/expanded/framebridge_expanded/broker.py @@ -28,6 +28,8 @@ class BrokerError(ValueError): """A fixed public failure code; never include HTTP bodies or credentials.""" +_SETTINGS_LOCK = threading.RLock() + _state = None _url = None _control_path = None @@ -80,18 +82,22 @@ def validate_endpoint(url): def configure(broker_url=None, state_dir=None, control_key_path=None): global _state, _url, _control_path - _state = state_path(state_dir) - config = _config(_state) - port = config.get("ports", {}).get("broker", 20878) - _url = validate_endpoint(broker_url or config.get("broker_control_url") or config.get("broker_url") - or f"http://127.0.0.1:{port}") - _control_path = Path(control_key_path or _state / "broker" / "control.key") + # Workers must see one fully validated profile, including explicit changes. + with _SETTINGS_LOCK: + state = state_path(state_dir) + config = _config(state) + port = config.get("ports", {}).get("broker", 20878) + url = validate_endpoint(broker_url or config.get("broker_control_url") or config.get("broker_url") + or f"http://127.0.0.1:{port}") + control_path = Path(control_key_path or state / "broker" / "control.key") + _state, _url, _control_path = state, url, control_path def _settings(): - if _state is None: - configure() - return _state, _url, _control_path + with _SETTINGS_LOCK: + if _state is None: + configure() + return _state, _url, _control_path def _private_read(path): diff --git a/source/expanded/framebridge_expanded/fleet.py b/source/expanded/framebridge_expanded/fleet.py index 5e0719b..393aa11 100644 --- a/source/expanded/framebridge_expanded/fleet.py +++ b/source/expanded/framebridge_expanded/fleet.py @@ -8,10 +8,13 @@ from pathlib import Path import re import socket import sys +import threading import time import uuid from typing import Any +_BOOTSTRAP_LOCK = threading.RLock() + JOB_RE = re.compile(r"^[A-Za-z0-9._-]{1,80}$") READ_CAPABILITIES = frozenset({ "system.status", "hardware.summary", "storage.inspect", @@ -56,18 +59,20 @@ def original_paths() -> tuple[Path, Path]: def bootstrap_original() -> None: """Extend the original package search path without copying or merging it.""" - broker_source, fleet_source = original_paths() - for path in (fleet_source, broker_source): - if str(path) in sys.path: - sys.path.remove(str(path)) - sys.path.insert(0, str(path)) - import framebridge_linux_chatgpt - package_file = Path(framebridge_linux_chatgpt.__file__).resolve() - _require(package_file.is_relative_to(broker_source.resolve()), "ORIGINAL_PACKAGE_SHADOWED") - for root in (broker_source, fleet_source): - path = str(root / "framebridge_linux_chatgpt") - if path not in framebridge_linux_chatgpt.__path__: - framebridge_linux_chatgpt.__path__.append(path) + # Registration now dispatches in workers; protect shared import paths. + with _BOOTSTRAP_LOCK: + broker_source, fleet_source = original_paths() + for path in (fleet_source, broker_source): + if str(path) in sys.path: + sys.path.remove(str(path)) + sys.path.insert(0, str(path)) + import framebridge_linux_chatgpt + package_file = Path(framebridge_linux_chatgpt.__file__).resolve() + _require(package_file.is_relative_to(broker_source.resolve()), "ORIGINAL_PACKAGE_SHADOWED") + for root in (broker_source, fleet_source): + path = str(root / "framebridge_linux_chatgpt") + if path not in framebridge_linux_chatgpt.__path__: + framebridge_linux_chatgpt.__path__.append(path) def state_directory() -> Path: diff --git a/source/expanded/framebridge_expanded/node.py b/source/expanded/framebridge_expanded/node.py index 473110a..fc98166 100644 --- a/source/expanded/framebridge_expanded/node.py +++ b/source/expanded/framebridge_expanded/node.py @@ -1,18 +1,27 @@ """Original FBPAIR identity and outbound protocol for an explicitly enrolled node.""" from __future__ import annotations +import errno import hashlib +import http.client import importlib import json +import logging import os import socket +import ssl import time import uuid +import urllib.error from pathlib import Path from . import broker +_LOG = logging.getLogger(__name__) +_RETRY_MAX_SECONDS = 30 + + class NodeError(ValueError): pass @@ -177,6 +186,54 @@ def dispatch_request(request, adapter_request=None, state_dir=None): return {"ok": False, "error": "NODE_DISPATCH_FAILED"} +def _retryable_broker_failure(exc): + # post_json intentionally hides transport details from its public error. Its + # exception context still distinguishes interruptions from TLS/JSON failures; + # never log that context or treat an unknown failure as transient. + code = str(exc) + if code in {"BROKER_HTTP_REJECTED_502", "BROKER_HTTP_REJECTED_503", "BROKER_HTTP_REJECTED_504"}: + return True + if code != "BROKER_TRANSPORT_FAILED": + return False + cause = exc.__context__ + if isinstance(cause, urllib.error.URLError): + cause = cause.reason + if isinstance(cause, ssl.SSLError): + return False + if isinstance(cause, socket.gaierror): + return cause.errno == socket.EAI_AGAIN + if isinstance(cause, (ConnectionError, TimeoutError, http.client.IncompleteRead)): + return True + return isinstance(cause, OSError) and cause.errno in { + errno.ECONNREFUSED, errno.ECONNRESET, errno.ECONNABORTED, errno.ETIMEDOUT, + errno.EPIPE, errno.ENETUNREACH, errno.EHOSTUNREACH, errno.ENETDOWN, + } + + +def _post_with_recovery(url, operation, payload, *, ctx, state, outbound, recover): + delay = 1 + retries = 0 + while True: + # A new signature/nonce is mandatory: the previous attempt may have been + # verified even when its response was lost. Do not replay its envelope. + envelope = outbound._signed_envelope(ctx, operation, payload) + try: + result = broker.post_json(url, envelope, state_dir=state) + except broker.BrokerError as exc: + if not recover or not _retryable_broker_failure(exc): + raise + retries += 1 + _LOG.warning("NODE_BROKER_RETRY operation=%s error=%s delay_seconds=%d", + operation, str(exc), delay) + # Signals and cancellation exceptions are deliberately not caught. + time.sleep(delay) + delay = min(delay * 2, _RETRY_MAX_SECONDS) + else: + if retries: + _LOG.warning("NODE_BROKER_RECOVERED operation=%s retries=%d", operation, retries) + return result + + def run(state_dir=None, once=False): state, config, ctx, identity = _context(state_dir) url = broker.validate_endpoint(config.get("broker_url", "")) @@ -187,16 +244,20 @@ def run(state_dir=None, once=False): raise NodeError("PAIRING_BINDING_MISMATCH") outbound = _original("outbound_agent") while True: - poll = broker.post_json(url + "/fbpair/v1/node/poll", - outbound._signed_envelope(ctx, "node.poll", {}), state_dir=state) + poll = _post_with_recovery(url + "/fbpair/v1/node/poll", "node.poll", {}, + ctx=ctx, state=state, outbound=outbound, recover=not once) if poll.get("ok") is not True: raise NodeError("NODE_POLL_REJECTED") req = poll.get("request") result = {"ok": True, "idle": True} if isinstance(req, dict): value = dispatch_request(req, state_dir=state) - posted = broker.post_json(url + "/fbpair/v1/node/result", outbound._signed_envelope( - ctx, "node.result", {"request_id": req.get("request_id"), "result": value}), state_dir=state) + # Original put_result only records this request's result; repeating + # that acknowledgement cannot execute the command. Keep the value + # here and never return to dispatch/poll while its delivery is pending. + posted = _post_with_recovery(url + "/fbpair/v1/node/result", "node.result", + {"request_id": req.get("request_id"), "result": value}, + ctx=ctx, state=state, outbound=outbound, recover=not once) if posted.get("ok") is not True: raise NodeError("NODE_RESULT_REJECTED") result = {"ok": True, "idle": False, "request_id": req.get("request_id")} diff --git a/source/expanded/framebridge_expanded/real_screen/screen_capture.py b/source/expanded/framebridge_expanded/real_screen/screen_capture.py index 3d1598e..c2c3ba9 100644 --- a/source/expanded/framebridge_expanded/real_screen/screen_capture.py +++ b/source/expanded/framebridge_expanded/real_screen/screen_capture.py @@ -32,7 +32,7 @@ import zlib MAX_PIXELS = 4096 * 2160 MAX_FRAMEBUFFER_BYTES = 128 * 1024 * 1024 MAX_IMAGE_BYTES = 32 * 1024 * 1024 -MAX_CHUNK_BYTES = 2400 +MAX_CHUNK_BYTES = 45000 CAPTURE_ID = re.compile(r"[0-9a-f]{32}\Z") SESSION_ID = re.compile(r"[A-Za-z0-9_-]{1,64}\Z") DISPLAY_ID = re.compile(r":[0-9]{1,5}(?:\.[0-9]{1,3})?\Z") diff --git a/source/expanded/framebridge_expanded/real_screen/screen_frontdoor.py b/source/expanded/framebridge_expanded/real_screen/screen_frontdoor.py index e0fddf9..55cd1fb 100644 --- a/source/expanded/framebridge_expanded/real_screen/screen_frontdoor.py +++ b/source/expanded/framebridge_expanded/real_screen/screen_frontdoor.py @@ -13,12 +13,17 @@ import math import re import shlex import struct +import time import uuid from typing import Callable CHUNK_BYTES = 2100 MAX_IMAGE_BYTES = 32 * 1024 * 1024 MAX_STDOUT_BYTES = 4096 +# Only the explicit typed screen transport preserves this larger stdout budget. +EXPANDED_STDOUT_BYTES = 65536 +EXPANDED_CHUNK_BYTES = 45000 +TRANSFER_TIMEOUT_SECONDS = 45 RPC_COMMAND = '/usr/bin/python3 -B -m framebridge_expanded.real_screen.screen_rpc ' HEX_ID = re.compile(r"[0-9a-f]{32}\Z") SHA = re.compile(r"[0-9a-f]{64}\Z") @@ -28,14 +33,25 @@ ERROR = re.compile(r"[A-Z0-9_]{1,100}\Z") READ = {"readOnlyHint": True, "destructiveHint": False, "idempotentHint": False, "openWorldHint": False} CHANGE = {**READ, "readOnlyHint": False} +# These explicit backend checks run before an input capture is consumed. Generic +# dispatch failures do not prove whether input was sent, even with valid framing. +INPUT_REJECTION_CODES = { + "REQUEST_INVALID", "OPERATION_INVALID", "UNKNOWN_REQUEST_FIELDS", + "SESSION_ID_INVALID", "KEY_INVALID", "TEXT_INVALID", "CAPTURE_ID_INVALID", + "CAPTURE_RECORD_INVALID", "CAPTURE_RECORD_MISMATCH", + "USE_GRAPHICAL_CUA_FOR_X11_INPUT", "VISIBLE_INPUT_UNAVAILABLE", + "CAPTURE_ALREADY_USED_OBSERVE_AGAIN", "SESSION_NOT_VISIBLE", +} class ScreenError(RuntimeError): """A bounded public error suitable for the MCP response.""" - def __init__(self, code, details=None): + def __init__(self, code, details=None, *, backend_rejection=False): super().__init__(code) self.details = details or {} + # Local proof, never accepted from a remote JSON field. + self.backend_rejection = backend_rejection def require(ok, code): @@ -62,7 +78,7 @@ class _AuthenticatedReply(dict): _IDENTITY_FIELDS = ("capture_id", "node", "sha256", "session_id", "bytes", "width", "height", "source", "screen_type", - "full_screen", "authentic_pixels") + "full_screen", "authentic_pixels", "max_chunk_bytes") def __init__(self, payload, owner, node, node_id): super().__init__(payload) @@ -80,23 +96,51 @@ class _AuthenticatedReply(dict): class ScreenClient: - """Transport only; execution policy and node resolution stay in the host.""" + """Transport only; execution policy and node resolution stay in the host. + + The monotonic transfer deadline bounds admission of RPCs and successful + publication. It is cooperative: the executor can spend its existing bounded + transport grace on the final in-flight RPC. Late replies never become a + successful image, and no further chunks are requested after the deadline. + """ def __init__(self, exec_request: Callable, image_factory: Callable, - resolve_node: Callable | None = None): + resolve_node: Callable | None = None, *, + stdout_limit_bytes: int = MAX_STDOUT_BYTES, + transfer_timeout_s: int = TRANSFER_TIMEOUT_SECONDS): + if (type(stdout_limit_bytes) is not int or + stdout_limit_bytes not in (MAX_STDOUT_BYTES, EXPANDED_STDOUT_BYTES)): + raise ValueError("unsupported screen stdout budget") + if type(transfer_timeout_s) is not int or not 1 <= transfer_timeout_s <= 120: + raise ValueError("screen transfer timeout must be 1..120 seconds") + self.stdout_limit_bytes = stdout_limit_bytes + self.chunk_bytes = (EXPANDED_CHUNK_BYTES + if stdout_limit_bytes == EXPANDED_STDOUT_BYTES else CHUNK_BYTES) + self.transfer_timeout_s = transfer_timeout_s self.exec_request = exec_request self.image_factory = image_factory self.resolve_node = resolve_node or (lambda node: node.upper()) self._reply_owner = object() - def remote(self, node, request): + def remote(self, node, request, *, deadline=None, submitted=None): + timeout_s = 30 + if deadline is not None: + remaining = deadline - time.monotonic() + # The authenticated EXEC protocol accepts whole seconds only. Rounding + # down prevents any individual request exceeding the remaining budget. + require(remaining >= 1, "SCREEN_TRANSFER_TIMEOUT") + timeout_s = min(timeout_s, int(remaining)) payload = base64.b64encode(encode(request)).decode("ascii") script = RPC_COMMAND + shlex.quote(payload) + "\n" digest = hashlib.sha256(script.encode("utf-8")).hexdigest() job_id = "cua-screen-" + uuid.uuid4().hex - outer = self.exec_request(script=script, authorized=True, timeout_s=30, + if submitted is not None: + submitted["attempted"] = True + outer = self.exec_request(script=script, authorized=True, timeout_s=timeout_s, job_id=job_id, sha256=digest, include_stdout=True, target=node) + if deadline is not None: + require(time.monotonic() < deadline, "SCREEN_TRANSFER_TIMEOUT") require(type(outer) is dict, "SCREEN_EXEC_FAILED") require(type(outer.get("exit_code")) in (int, str) and str(outer["exit_code"]) in ("0", "1"), "SCREEN_EXEC_FAILED") @@ -109,7 +153,7 @@ class ScreenClient: except (ValueError, AttributeError): raise ScreenError("SCREEN_EXEC_NODE_ID_INVALID") from None stdout = outer.get("stdout") - require(type(stdout) is str and len(stdout.encode("utf-8")) <= MAX_STDOUT_BYTES and + require(type(stdout) is str and len(stdout.encode("utf-8")) <= self.stdout_limit_bytes and outer.get("stdout_truncated") is not True, "SCREEN_STDOUT_INVALID") try: envelope = json.loads(stdout) @@ -129,12 +173,19 @@ class ScreenClient: if valid_reference(code.get("cause"), ERROR): details["cause"] = code["cause"] code = code.get("code") - raise ScreenError(code if valid_reference(code, ERROR) else "SCREEN_BACKEND_FAILED", details) + valid_code = valid_reference(code, ERROR) + raise ScreenError(code if valid_code else "SCREEN_BACKEND_FAILED", details, + backend_rejection=(valid_code and outer.get("ok") is False + and str(outer["exit_code"]) == "1" + and (code in INPUT_REJECTION_CODES + or details.get("performed") is False))) require(outer.get("ok") is True and str(outer["exit_code"]) == "0", "SCREEN_EXEC_FAILED") require(type(envelope.get("result")) is dict, "SCREEN_RESPONSE_INVALID") return _AuthenticatedReply(envelope["result"], self._reply_owner, node, node_id) - def image(self, node, manifest): + def image(self, node, manifest, *, deadline=None): + if deadline is None: + deadline = time.monotonic() + self.transfer_timeout_s capture_id = manifest.get("capture_id") total, digest = manifest.get("bytes"), manifest.get("sha256") require(valid_reference(capture_id, HEX_ID) and valid_reference(digest, SHA), @@ -151,11 +202,23 @@ class ScreenClient: "SCREEN_PIXEL_SOURCE_INVALID") require(manifest.get("full_screen") is True and manifest.get("authentic_pixels") is True, "SCREEN_PIXEL_PROVENANCE_INVALID") + # Capability belongs to this authenticated manifest, never a shared cache. + # A new central must retain the old request schema for a frozen receiver. + chunk_bytes = CHUNK_BYTES + if "max_chunk_bytes" in manifest: + capability = manifest["max_chunk_bytes"] + require(type(capability) is int and 1 <= capability <= EXPANDED_CHUNK_BYTES, + "SCREEN_TRANSFER_CAPABILITY_INVALID") + chunk_bytes = min(self.chunk_bytes, capability) parts = [] - count = math.ceil(total / CHUNK_BYTES) + count = math.ceil(total / chunk_bytes) for index in range(count): - part = self.remote(node, {"op": "read_chunk", "capture_id": capture_id, - "index": index, "expected_sha256": digest}) + request = {"op": "read_chunk", "capture_id": capture_id, + "index": index, "expected_sha256": digest} + # Omit the extension entirely on constrained legacy transports. + if chunk_bytes != CHUNK_BYTES: + request["chunk_bytes"] = chunk_bytes + part = self.remote(node, request, deadline=deadline) require(isinstance(part, _AuthenticatedReply) and part.bound_to(self._reply_owner, node, manifest.node_id), "SCREEN_CHUNK_NODE_MISMATCH") @@ -167,12 +230,12 @@ class ScreenClient: type(part.get("more")) is bool and part["more"] == (index + 1 < count), "SCREEN_CHUNK_MANIFEST_INVALID") text = part.get("data") - require(type(text) is str and len(text) <= 2800, "SCREEN_CHUNK_INVALID") + require(type(text) is str and len(text) <= 4 * math.ceil(chunk_bytes / 3), "SCREEN_CHUNK_INVALID") try: data = base64.b64decode(text, validate=True) except (ValueError, UnicodeError, binascii.Error): raise ScreenError("SCREEN_CHUNK_INVALID") from None - require(len(data) == min(CHUNK_BYTES, total - index * CHUNK_BYTES), + require(len(data) == min(chunk_bytes, total - index * chunk_bytes), "SCREEN_CHUNK_LENGTH_INVALID") parts.append(data) data = b"".join(parts) @@ -181,10 +244,15 @@ class ScreenClient: require(len(data) >= 33 and data[:8] == b"\x89PNG\r\n\x1a\n" and data[12:16] == b"IHDR" and struct.unpack(">II", data[16:24]) == (width, height), "SCREEN_PNG_INVALID") + require(time.monotonic() < deadline, "SCREEN_TRANSFER_TIMEOUT") return self.image_factory(data=data, format="png") def call(self, op, node, authorized, **kwargs): requested_node = node + action_confirmed = False + input_reply_received = False + submitted = {} + deadline = time.monotonic() + self.transfer_timeout_s try: require(authorized is True, "AUTHORIZED_TRUE_REQUIRED") require(valid_reference(node, NODE), "SCREEN_NODE_INVALID") @@ -202,11 +270,20 @@ class ScreenClient: if op == "text": require(type(kwargs.get("text")) is str and 0 < len(kwargs["text"].encode("utf-8")) <= 4096, "SCREEN_TEXT_INVALID") - manifest = self.remote(node, {"op": op, **kwargs}) + manifest = self.remote(node, {"op": op, **kwargs}, deadline=deadline, + submitted=submitted) + input_reply_received = op in ("key", "text") + action = manifest.get("action") + summary = action.get("result") if type(action) is dict else None + action_confirmed = (op in ("key", "text") and type(action) is dict and + action.get("kind") == op and + action.get("previous_capture_id") == kwargs["capture_id"] and + type(summary) is dict and summary.get("performed") is True) if op == "read_capture": require(manifest.get("capture_id") == kwargs["capture_id"], "SCREEN_CAPTURE_REFERENCE_MISMATCH") - image = self.image(node, manifest) + image = self.image(node, manifest, deadline=deadline) + require(time.monotonic() < deadline, "SCREEN_TRANSFER_TIMEOUT") metadata = {"ok": True, "node": node, "requested_node": requested_node, "node_id": manifest.node_id, "hostname": manifest["node"], "result": dict(manifest, transport_node=node, node_id=manifest.node_id)} @@ -217,9 +294,21 @@ class ScreenClient: metadata["result"]["authentic_pixels"] = True return [metadata, image] except ScreenError as exc: + if op in ("key", "text") and (input_reply_received or + (submitted.get("attempted") and not exc.backend_rejection)): + exc.details.setdefault("observe_again", True) + if action_confirmed: + exc.details.update(performed=True, capture_consumed=True) + exc.details.setdefault("capture_id", kwargs["capture_id"]) return {"ok": False, "node": requested_node, "error": str(exc), **exc.details} except Exception: - return {"ok": False, "node": requested_node, "error": "SCREEN_BACKEND_UNAVAILABLE"} + details = {} + if op in ("key", "text") and submitted.get("attempted"): + details.update(observe_again=True, capture_id=kwargs["capture_id"]) + if action_confirmed: + details.update(performed=True, capture_consumed=True) + return {"ok": False, "node": requested_node, + "error": "SCREEN_BACKEND_UNAVAILABLE", **details} def resolve_profile_node(alias): @@ -233,12 +322,16 @@ def resolve_profile_node(alias): def register_framebridge_screen_tools(mcp, exec_request, *, image_factory=None, - resolve_node=None): + resolve_node=None, + stdout_limit_bytes=MAX_STDOUT_BYTES, + transfer_timeout_s=TRANSFER_TIMEOUT_SECONDS): """Register real image tools. The host supplies its existing executor.""" if image_factory is None: from mcp.server.fastmcp import Image image_factory = Image - client = ScreenClient(exec_request, image_factory, resolve_node) + client = ScreenClient(exec_request, image_factory, resolve_node, + stdout_limit_bytes=stdout_limit_bytes, + transfer_timeout_s=transfer_timeout_s) @mcp.tool(annotations=dict(READ)) def framebridge_screen_observe(node: str, authorized: bool, diff --git a/source/expanded/framebridge_expanded/real_screen/screen_rpc.py b/source/expanded/framebridge_expanded/real_screen/screen_rpc.py index 7c1e596..80a211a 100644 --- a/source/expanded/framebridge_expanded/real_screen/screen_rpc.py +++ b/source/expanded/framebridge_expanded/real_screen/screen_rpc.py @@ -13,7 +13,7 @@ import stat import sys import time -from .screen_capture import CaptureError, capture_screen, read_chunk +from .screen_capture import CaptureError, capture_screen, read_chunk, MAX_CHUNK_BYTES from .screen_control import ControlError, act, bind_screen, snapshot_binding MAX_REQUEST = 16384 @@ -89,7 +89,10 @@ def write_record(path: Path, value, *, replace=False): def public_manifest(manifest): # Internal path is unnecessary to use or understand a captured screen. - return {k: v for k, v in manifest.items() if k not in {'path', '_action_consumed'}} + result = {k: v for k, v in manifest.items() if k not in {'path', '_action_consumed'}} + # An old receiver omits this capability and only accepts the original schema. + result['max_chunk_bytes'] = MAX_CHUNK_BYTES + return result def observe(directory: Path, requested_session_id=None): @@ -138,7 +141,7 @@ def validate_request(request): allowed = { 'observe': {'op', 'session_id'}, 'status': {'op'}, - 'read_chunk': {'op', 'capture_id', 'index', 'expected_sha256'}, + 'read_chunk': {'op', 'capture_id', 'index', 'expected_sha256', 'chunk_bytes'}, 'read_capture': {'op', 'capture_id'}, 'key': {'op', 'capture_id', 'session_id', 'key'}, 'text': {'op', 'capture_id', 'session_id', 'text'}, @@ -149,6 +152,9 @@ def validate_request(request): sid = request['session_id'] require(type(sid) is str and len(sid) <= 96 and re.fullmatch(r'[A-Za-z0-9_-]+', sid), 'SESSION_ID_INVALID') if op == 'read_chunk': + if 'chunk_bytes' in request: + require(type(request['chunk_bytes']) is int and + 1 <= request['chunk_bytes'] <= MAX_CHUNK_BYTES, 'CHUNK_BOUNDS_INVALID') require(type(request.get('index')) is int and request['index'] >= 0, 'CHUNK_INDEX_INVALID') require(type(request.get('expected_sha256')) is str and HEX_SHA.fullmatch(request['expected_sha256']), 'EXPECTED_SHA256_REQUIRED') if op == 'key': @@ -179,7 +185,7 @@ def dispatch(request, directory=None): if op == 'read_chunk': require(manifest['sha256'] == request['expected_sha256'], 'CAPTURE_SHA256_MISMATCH') chunk = read_chunk(directory, manifest['capture_id'], request['index'], - chunk_bytes=2100, expected_sha256=request['expected_sha256']) + chunk_bytes=request.get('chunk_bytes', 2100), expected_sha256=request['expected_sha256']) return {'ok': True, 'result': chunk} if manifest.get('screen_type') == 'x11': raise ScreenRpcError('USE_GRAPHICAL_CUA_FOR_X11_INPUT') diff --git a/source/expanded/framebridge_expanded/real_screen/screen_transport.py b/source/expanded/framebridge_expanded/real_screen/screen_transport.py index 4510c5b..289bf3b 100644 --- a/source/expanded/framebridge_expanded/real_screen/screen_transport.py +++ b/source/expanded/framebridge_expanded/real_screen/screen_transport.py @@ -17,7 +17,10 @@ import signal import subprocess import time -from .screen_frontdoor import RPC_COMMAND, MAX_STDOUT_BYTES +from .screen_frontdoor import RPC_COMMAND, EXPANDED_STDOUT_BYTES + +# This typed route bypasses the legacy 4096-byte administrative sanitizer. +MAX_STDOUT_BYTES = EXPANDED_STDOUT_BYTES from .screen_rpc import MAX_REQUEST, encode, validate_request, ScreenRpcError REQUEST_TYPE = "screen_rpc" diff --git a/source/expanded/framebridge_expanded/runtime.py b/source/expanded/framebridge_expanded/runtime.py index 9bb78ca..398cab4 100644 --- a/source/expanded/framebridge_expanded/runtime.py +++ b/source/expanded/framebridge_expanded/runtime.py @@ -16,6 +16,7 @@ import socket import stat import subprocess import sys +import threading import time from pathlib import Path from types import SimpleNamespace @@ -24,16 +25,20 @@ from .config import load_config, state_path, private_dir PREFIX = Path(os.environ.get("FRAMEBRIDGE_EXPANDED_PREFIX", Path(__file__).resolve().parents[2])) _STOP = False _LEGACY = None +_LEGACY_LOCK = threading.RLock() def _legacy(): global _LEGACY - if _LEGACY is None: - path = PREFIX / "bin/framebridge-original" - loader = importlib.machinery.SourceFileLoader("framebridge_original_launcher", str(path)) - spec = importlib.util.spec_from_loader(loader.name, loader) - _LEGACY = importlib.util.module_from_spec(spec) - loader.exec_module(_LEGACY) - return _LEGACY + # Publish only a fully initialized module; a failed load remains retryable. + with _LEGACY_LOCK: + if _LEGACY is None: + path = PREFIX / "bin/framebridge-original" + loader = importlib.machinery.SourceFileLoader("framebridge_original_launcher", str(path)) + spec = importlib.util.spec_from_loader(loader.name, loader) + module = importlib.util.module_from_spec(spec) + loader.exec_module(module) + _LEGACY = module + return _LEGACY def _args(state, config): return SimpleNamespace(state_dir=str(state), port=config["ports"]["mcp"], diff --git a/source/packaging/build_expanded.py b/source/packaging/build_expanded.py index 9a1ee01..670b052 100755 --- a/source/packaging/build_expanded.py +++ b/source/packaging/build_expanded.py @@ -18,6 +18,9 @@ from build_profiles import CANDIDATE_VERSION, PROFILES, SOURCE_VENDOR, VENDOR_RE BASE = Path(__file__).resolve().parents[1] ROOT = BASE.parent VERSION = CANDIDATE_VERSION +# Oldest package retaining the privileged helper lifecycle. Candidate/release +# version bumps must not move this downgrade revocation boundary. +PRIVILEGED_HELPER_FLOOR = "0.3.2" PUBLIC_FILENAME = "FrameBridge-0.3.1.deb" GPL_SHA256 = "3972dc9744f6499f0f9b2dbf76696f2ae7ad8af9b23dde66d6af86c9dfb36986" EPOCH = calendar.timegm((2026, 10, 27, 0, 0, 0)) @@ -140,7 +143,8 @@ def install_legal_material(docs): def maintainer_prerm(version): """Revoke helper grants before a downgrade removes the helper's code. - Equal-version reinstalls and forward upgrades retain managed credentials. + The argument is the fixed helper compatibility floor, not the package version. + Targets retaining the helper lifecycle preserve managed credentials. The fixed installed cleanup module validates the helper's root ownership, permissions and ancestors before running its bounded purge-all command. """ @@ -298,7 +302,7 @@ def main(argv=None): " Optional virtual displays are explicitly identified as virtual.\n" " No phone or Owner Device pairing. Services start after connection setup.\n" " Installing the package alone does not activate a connection.\n") - (control / "prerm").write_text(maintainer_prerm(VERSION)) + (control / "prerm").write_text(maintainer_prerm(PRIVILEGED_HELPER_FLOOR)) (control / "prerm").chmod(0o755) rows = [] for path in sorted(STAGE.rglob("*")): diff --git a/source/portable/front_door_mcp.py b/source/portable/front_door_mcp.py index 17671c4..29ca06f 100644 --- a/source/portable/front_door_mcp.py +++ b/source/portable/front_door_mcp.py @@ -1,5 +1,7 @@ #!/usr/bin/env python3 import base64 +from functools import partial, wraps +import inspect import hashlib import json import os @@ -8,6 +10,9 @@ import socket import time import urllib.error import urllib.request +import uuid + +import anyio from pathlib import Path from mcp import ClientSession @@ -55,6 +60,45 @@ mcp = FastMCP( mcp._mcp_server.version = PRODUCT_VERSION + +class _BlockingToolRegistrar: + """Offload only MCP dispatch while keeping synchronous callback bindings. + + Use the public SDK decorator; do not change its dispatcher or the functions + passed to ScreenClient/TTY as synchronous EXEC callbacks. AnyIO bounds worker + concurrency. Waiting cancellation does not replay or undo an operation: a + running worker is allowed to finish, including any side effects. + """ + + def __init__(self, server): + self.server = server + + def resource(self, *args, **kwargs): + return self.server.resource(*args, **kwargs) + + def tool(self, *args, **kwargs): + register = self.server.tool(*args, **kwargs) + + def decorate(function): + @wraps(function) + async def dispatch(*arguments, **keywords): + return await anyio.to_thread.run_sync( + partial(function, *arguments, **keywords), + abandon_on_cancel=False, + ) + + # Resolve future annotations in the defining module, including the + # screen presentation's Annotated return type, before registering. + dispatch.__signature__ = inspect.signature(function, eval_str=True) + register(dispatch) + return function + + return decorate + + +_blocking_tools = _BlockingToolRegistrar(mcp) + + def _registered_operations() -> list[str]: """Registered operations are not proof of their current execution readiness.""" return sorted(mcp._tool_manager._tools) @@ -395,7 +439,7 @@ def _framebridge_exec_geranas_impl( if not 1<=timeout_s<=600: return {"ok":False,"error":"INVALID_TIMEOUT"} if not job_id: - job_id="mcp-"+time.strftime("%Y%m%d-%H%M%S")+"-"+digest[:8] + job_id="mcp-"+time.strftime("%Y%m%d-%H%M%S")+"-"+digest[:8]+"-"+uuid.uuid4().hex[:12] if not JOB_RE.fullmatch(job_id): return {"ok":False,"error":"INVALID_JOB_ID"} req={ @@ -468,7 +512,7 @@ def framebridge_cloud_dsm(action: str, authorized: bool = False, params_json: st return _cloud_dsm_rpc({"action":action,"authorized":bool(authorized),"params":params}) -@mcp.tool() +@_blocking_tools.tool() def framebridge_exec( script: str, authorized: bool, @@ -641,7 +685,7 @@ def asus_cua_request(cua: dict, authorized: bool, timeout_s: int, job_id: str) - return {"ok":False,"error":"ASUS_CUA_TIMEOUT","request_id":rid} -@mcp.tool() +@_blocking_tools.tool() def framebridge_cua( action: str, authorized: bool, @@ -830,12 +874,15 @@ def _screen_exec_request( from framebridge_expanded.real_screen.screen_frontdoor import register_framebridge_screen_tools, resolve_profile_node -_real_screen_client = register_framebridge_screen_tools(mcp, _screen_exec_request, image_factory=Image, resolve_node=resolve_profile_node) +_real_screen_client = register_framebridge_screen_tools( + _blocking_tools, _screen_exec_request, image_factory=Image, + resolve_node=resolve_profile_node, stdout_limit_bytes=65536, transfer_timeout_s=45, +) from framebridge_expanded.real_screen.screen_presentation import register_screen_presentation -register_screen_presentation(mcp, _real_screen_client) +register_screen_presentation(_blocking_tools, _real_screen_client) from framebridge_expanded.tty_mcp import register_tools as register_tty_tools -register_tty_tools(mcp, framebridge_exec) +register_tty_tools(_blocking_tools, framebridge_exec, screen_client=_real_screen_client) if __name__ == "__main__": from framebridge_expanded.http_server import run_streamable_mcp diff --git a/source/tests/fixtures/frozen_screen_receiver_f6b14c52.json.gz b/source/tests/fixtures/frozen_screen_receiver_f6b14c52.json.gz new file mode 100644 index 0000000000000000000000000000000000000000..4e50e8e319d3cd870488b6255d29163f963ab8eb GIT binary patch literal 20290 zcmV(#K;*w4iwFP!00002|K)v)cG}36=&xwpYsQeq1pJEQbLe}CO%iT9wg=;M`q*A9 z0m9Y^NDLCO-5LM;-H&>!gmJoO=B{;fPA5pJ=dP;#s{Q!BF^Q*9wsH7e~x zdA?jk^yV(y-rL_e+}Ph6HXD1RurYZ%9B;qbZiKr#yL;im-of@{xI3OiyW{4;@F3jT zKL|(rlX2K=>}>CiCfm)uHwTl0gN@oomM#|~Xm#VKe|f#k7O#hK@;XYs1@p)JE=_i> zlj{xq-x2p7grj^JPJ@#;Sv~}x`$4)0zBHS`B2DvP9zR6WtQI8Ew`dV8qGTK`qH&N% zi&>mNr967b>(F4*9_8`3h$?1Lmc?n31KuGAnJ0m*U4m&&Vs>UvcyRb20=WVr;8jaCutt$IAENIe7jhLkLqa{ zX3_qxdK~4C^C)v3>&rZzieAQH9*(AA27^%DWaD_0*Yqb{G|J;yB+JR?T7ZB5ktUJq zXEI9isd~B#v%6_LRL}p+Qa$`xnBS>~RQ5H8w@^iPb#4}#-;dSrh3+#;VG(lu2#_)! z7R!utC-x%s}c{d9f(kL!)kxav;(Je(|8;nMU6;>0Pu2e= z>kr_Z4NlIEeue4^u=U;vtX2g+zkk;~`_SnRK3^R5I-S$*>H9M%*m&68#~I46?~7lz zaS5&y+Mt2J51qv#T~vC@1UDo-R1dF{0RD`l2^^?6iSxmrl10-=Ef}Ta=n!{?EQ5dW zQysrRH~5e7WjT))m1_``_whR^YJw5hbnuVu>Fa81K_Xbh}8G$yl_=7-#S;r(y6r zfpiA(*m$|br#Q*^X$or|~=_RNVk$`2#jZlNJ zwU7GU&z-^N{^0WTbEnsR-|ZaJ!O<%RfV#?vd;pG4G|_?$l+|yeymGy90U&@{L90d3 zAOa@8F})LCuQwin9&G5X#(&K6y=a!^x_pveZ`|1P#qo#9V@1~J!-l}}Fu3p68xPGU zPWgJ{o+;?{#*-dA&!}3n2XMXdq1QhD*ga~W3?x?Tb^iUb+cSVc;05I2fJ4B@0EQe6 z4IkFxdc&GuzyG`Vs#e7XhYgU={nLe8<7(&d#(~N-N-A=URJe&icz@RGeCVBBo*oY* z0`@F)29uI?uNtD^eDP@FgeY51b13)Tb3Wh@TwQN;0oHIH;Ivin@iGJEWaFl)-<;5q zrgz0Dn}4t{gYkx&CZLJmyxEM77(h2wF`4LL6wULX@@w=+xT_XmQ{w06aJsZ{a|y!p zg06e|xqZ?_xCO)~oD$bzu7aj==9whe)iu(yjv+0!D8fiD_$6o*w&=>?$U(sP6VX-y zu?gh&mm|3UN3jH+2XWkg;52Yebrx10Z|juM3alU{8psUTsy%AAmw#!Qm<;FeTl@Eu zHcUVws>{=h%k%TIUcX~*1b=HZW`!V`s~PlvKliut6G%F@clq%rzzvT`k71SJAO_KF zo<9cncff}sUoxLQB3d*^RB#A?#-?6W|UAr&;93<|Irk0d+2JB^iK*Z>TW*Fko$OEc`QvrnPWem(73> zauW~}<&HYL5Sz#tZJ$Tbd6>iUkE_nqyczQF1-Tj{4Z9M=3yJbH21Wyr z7On}LwP|`AC!>7ISOr&kOOhpiwY8O`Thr(^l4OTo%)?tg{dG91agL2<6%$lIC-Ef= zW)aUTE)X+3QKA6WuT&L*b>ZYaRGE903IK9BA~mjYRuAX%2oDQ1Qsfk%00Dky+idg* zkVb#d`Eu0h92;<-#L<)?%qWutDZ#+tZzi|Y5TQK$#kh{XB1*&`j9zKW3bcy9C1-=b zHA=KZV0Up5t}zOhnE`M-c|1gWv9r7MUM@Dify#yTHRuI2|22BV=6Tl6l2R`}Q#V_4 zYz_+FmRst~Xj(saY*-Qf%1XX1KYiJQl%sRWF58Bn>8yiaY4zb@rIO zChf#);1ck^#_LcroOM(R4G^hRs&e02^hY3z=ctS+Nw+`4L2bV~>39hY9oWhmTzP?4 zL=f00-Cm6d5bTSX&#I}$8oZAmDndtlK`Kj_o;o@o?50fm9=7kz^&X21Kc$<8@hecf3Chl?s%58qP{zzxDI4uXhDp3Y&Z=2z06irKE z2^2jX#&A;OC|iH}Af`r(Z_(IO16A48o3+%9AEFdw3WWjg<8gjxX#DOX6k&`Pz$T!K zGH?sZ_k{`%>12{c=9|aD8$=tZW>qhS^SSW^{KR39FXz+f3f>X_D{$m3-foqA8v?Ie z@(Iogh)ILtV;*IudUFBCd|_)hsU;|vF2?#3;KRClsaeWz;WWl;mBFn5YUTD&0oc_a zOX}GO{vO=c37kkk=_Ues8h+2<#1D8kfgkL>AZxy|%IdofLr%Tu{$sVYds_#vlNPWE zywmn>4XF7OjpbI)R%i)fv*uGlWry)Cu!;DK`W%3~0L-_yE4t-lxC0ou!`H8a19u?S zE(?BXsd@5f?9ImC#qR=y8VHcvy4Z&NU^l;Iw(hO`=&LL`+){lLo6_z4hI97lmgL-c zv-6*8%yp>jpf;%7DDD(xFFAex-nI%#~2mO(&1yDZLZ7G_^v zj*5G#1&n)1+P?)x0XU`GI-Obkfd|wQJ_U%HI22J6ejUi;&=Sqkvzm)8NZZMSD?yAo zJ2_M1_w7qH56Ng%a1?(Rjl7BICX|Vdf3Yad1W+r^`Lny{U=-)y_Sai!=r@w)6bn)DR z=m4v~+i)fWUI7|7>pFfD zBN&X*Ijom4CI|lXsg97-;w`^pjtk#z(_4VZ)teh#3ZJ(d{GzC&!>>ELMseVSaj~yr z@s9Q_oz!LnB#NCHa-+qaX~JOzqQ`3W5Nm9EiV&cu>{*jpb*sCPhqE*r%$ zQ@=$7^2Z1zp{hJ_1uno=Lfz_P%nc3@+te1y%mCQrdyG7B$xc&1muPF3@I1ErwQ zERo3$;n7L()Er1s6#OgvE*`wqe*~ed1<5J`S=CgP)Ch|(xkWZYP;k@w0c93koJRt- zSRiLb^nzI2hJpRkdo&dZ@2{x6LvHnFgVQvs8LEnLhJ8O$Z?fc$rP(q->u5o6a%hHoa20@wLUtzn4hkn3e_a1mWn2XH^$1~AysL_W`fcgtELnsUTe6W}5u=0T1 zM%7usnhBU{u>*_3|q%( zZtv|uY3d8sal49tyXXlhp9^V1K|cDf+LPBzbE%2p39@^nr_O;!xyuh~48ErwZ; zrD=lyVyW%DHvq@zAp|0cWRJHE^84Wezyi|_F8U zxGA93MpB`9KR`ClUNF$0DzE90z3x?8c}$b73EXZSI>rqRUKQL!K%fdIZj??_QqTZ*TAIm*4^)1PkaK#GS96e#DT? z&WfePmn)jg4dF_t=9zqyT@kdMGIGLbVyo{?jMd%N47r+YAdh^kfZ-)P@yscU!-W1q9>_w5eAO?fXy(M%2 zTuAL1B@)j;#D$)33VTKX35I^?TcXid5|_A#qEdcU9et40N#f3a4&GEv7PCNMYXS#Y z$BKtgM!C3GpsmoZT#uXJKU<D1baypYYYLvNHru>`2T8kWe4Xc{SG#R1KgSH7 z7r>Qf!nk4v3}ma3zG=X`qK7!k2kFrfEBU&J8bCDDBaKQL^I?506H9W@{#a~O0wCsK%Y;puGfTkq`jdV!zzv)=+Vda>4(no;P|rFJ^di9 zDwVEU+>C!A&#ZMX#|c4L*P5u@rmIp$8vbuWu~Gc!y*`CW&BBtFa4oDa-_m$InDa+c zWAM%2dOZ!}d=}1OIpUJ;L;?8cN7|i-WnsG!=lbeLR(NamUi13?{- z&|PwuP6~t468^CQr?s-pbXdB8{;dEfC6~5aP8Fb+`|>KCupD;G`;{(r^u>#F)9jog zQbP!-9ZeSbhUZiJ;@5)QF7$J)fGho5utge;I_?kFhgS@hNp=@ca$2hn$E~ z{}!m9#1%-a|ICmC!k5>2q6uq628bS@F*foX2)n`=(8bEcMnI-B5djd5o37Z7K4088 z?myv}g!p9!Oc%Gk31aIg#FJU}Hut1KuDW3iOnHu(`&soLlX4I2@2 zzsBKIq}i&8itiJuz#r;N#RUN_-fD7bR3!tz$vXB)0{*1aKE`XMQtMSALva)T(&Fll zgrcm?VW!d(O3Vp(S2!C0u9SRj>p?e{qAbF69A#hgbnZ4sphjK+j(rd=p=6TBBXVF> zK%XtMoZ1;ibR~(5@ghS1*}7|&NE=A1P5353Z4KP(Vyv$LCgO`SE@K^CwFpxbnvG^y zaMr$;dSjpVJ74;P{@K~!q}}__u>ebwTW08VBu|dlpe!4z9)#JQWVl@^5tZC*Vuvt6 zlR{|1BS|9o;$-Ovx=D!MxJsi7Rdsj5%&*WL)}c)WWrPvCV1K`~FZ&qT|Ar2wo8S&dSxnI5O#lGS8 zZL@Y|gnXIAIM`zva?zw+?>q}Zi7mVeD*-FXX0i{s{ZnAmtoH`imsm@Dago2aW|tSc z!uVfU#1}dC+S-=w63;0B;~AXP#e+SLIExcNQ^^R}O}QF`eQ_9GFABAQD?LEe7+SjL z_177;u?Db5t@u-LnepwV@*xyP=L!$Ol#A5i{tnI%+Gf|yw%Rr(6Sg+M;4kGK`L>EJm`B&904w2CXtx22{ zd;BTrMIaG>;}rrB#z2Y3v=p!gqcBMmij#1cB6cHN8F)LL(&vw7pE`B7y)V;vxQW{h z81!i{fwCB(0lS~URJ6|%n>Fx7^_YmJmLppDJ?u)#8 z2zei0w?wGxUsTcZ)9lfhd+?F2W1u9Y@A^a4vX7KHFF|RI`aTP=HK%VXeT zI4hM6f7!lKUNJ%$r1t?Ud87^rq;*$^WNM0NgOxc^m+6Vzv6kqF6gAM`G(2mCD-+@= z*2B8w_?4F;i_4;NUnf7o7S#;8Ev`hRkxh+c5J|l~LzhTZF%LCgDFFdMV~cQsb&S9* z^+}wJ2jUYidBH|hR`fV3o^;>!+C6t~7}_Tluw$jC*@GU`(ecTN?9oL8^)F}hC=p&X z^>A@ZRZvS~c`gj*H-B>rFQSZj9ao?(ba6CIvuI@>T=Aw*^EjG>%W2*NndUcuDxm*W z)&J(N;u$EZ5(>S2h;q44R`uhmB-HcySr^2RUgdcmhyPDmGpwIk`E}hcURAD22a64n z#c1_rC1^bGR0j5T719a*5Gbih=V`4uTGOeOKNfqtDW5@M#u9i!?OF9~TtG_cV<`aK z28x|uJSL+xL1z!XAU(Bn5bK6OGzyZ_N;2g~L_Q*wPRD~EVRwsIYaQ0Y(36W^uIwIA z#x-ft8duH6i0A3t8F7tiifeX}BoXNpu{R4dN|CD0CPvzr)V%tdV9c|rXZ|)Y%S|v| zKICWT?t!BFN+QGg>1`4Ls(lFd3nAJg#lyni?l*N0BR=y)x+&6MNb7in%?ZXK?Q@|C3a@SvCa zx=~&pG}#KB=Zi(?>UgE(&wyCrub~MSYerPp`chEsRL29+^z?8R(Rt*B_B1VDS{IC~ zu!(sKM^)uDBP@O4o4#W7eO0%KE72FhMQp1rwV7r32VqD~mV)+@k!uYZ@(Djl2j1n5 z7McUQ*H|4@@X75U%=N<8cx)wy!XkArn2PV5(|J*7am9_k2iN#LSZV=Y#R3cc{dz-c zqBjoBD*!zDgL@*@_=Pcny0bpyfjzCk5z*20P-}6Tazv8E#W z7LOwBKl~bAyq*jxn2bNhjCaSaPv8$eop4xR^@F+aYaH8UDBBQ%7-YO3TyoXnR^-MPZo{PHjCPm^w*{ znP@vC)TEn9P5bg1wnNfRmnOZ1XUoJ{;u_87_va|^Vn;Gov~{Ev7-)xEk=JN}6$s5c zRAU%Z4EeleMqS=0`a#Sw@S@;7AR;mDEX$+1a z90)bBaR6Qi6{Gggs2hjr7(UWTJc-Cd3NYd~yiUUV?L!W$r)%OX{Y00pY!MCb`IlF_ zEuwsLh>t_x5vWm+zp-ctLoK85eV1lAUI6obcfu#$Hqi8S z8o}2=NOM>Z<9M6WF-i407~x|O^JjAHIa0USss839S6~#sW#I6Mqh9!@e`d=?mSU`D zG;?Bc1;G+mCP@z(Is#6u+u zO3rD#*$47E5of?wwE)4#B)^088R1Lvt`a!MFhJEip@x{v>vnDwUwx1%;Q2-gdx#(F zbo$DI`q?yEUo8N zUnXStJLABRj%~40-C1`ra)5>v;|4VK-lVt|5N@xUE}%)4nu!->8G!Bx1rfbi1+H(Upye`yt~aWQ8>I0*F?GA;OL;e!z>so*oZo)!N~PG9*&yLmeN0`Iza-fmb@|3!DX!iLHd3a04D`W*#M z`Fl%4&E#K|CqePPTr1){J_L&)nI_{FnT%V)r8vxjiO=^P1uZh1tf(=Nv9BLscwpK8 zSPkDhVsYxq^Z>)>MD^kp|C7!bHeil+4e8ut1Y@%P$_Pg12{A~hF1b27uC81fkqVi-3Qe1u!b!675uIy)k>O&f zlmQJ{wrC7U;hT(-Qlv9-pwI^6M|2{&D7vmoWx_L>1shAupfvGg(=;4s6&s`JCRea{ zL34lSVAqX=Pi_96i?dV1#PnIS(^Y%;Q~R|0zH`xEBaX7MJ7#R2!nBI_k%_LJ znpw`^nrZ0>hZGl27BAt#sj6heSIowwVJ}z)7MW|V20(3!TNa_85gu&-%33WA3L%AR z8?(7c`5;RO(X(pK^5SC~YWX5db3YCOkax4FS#v*;S?DcE3fCK*FXx>jAOxiy|@579SuT(|9x)p&ViFo#gTXi*4F*-SYCRl} z2cf8gj#3Jr>YOzk_E?msf3SwS{@no{!+&+lj zHukm;cHbWC?wJ|&SkeA2rraiI2Ce!2ARS`lS575|PCaCJ&0)W4-dbWH}$@3_0 zUjGgasMGOsDox!vU>K&k<9iqo8bU@`pZoQon_%25s58e_a@56BKo;)rI-#=pU6=>= zm?}@C@!`}uRDPin@s zhS*?S?ML(Rt;}|}R)RY-shyR@u9&tCb&D@Z&!-(~FK81U@g96}-ZbEBUyU)`%`~&( zhn3_=TO+#~ei1F=jSc_e0h@z_Z^e;wv0ppCUyycO++20g0FF^}C&^WC_YTnF)57b1 z`yIdQheHuK{$uxjf1uvK;}5FwizCrS2Af7)pSm7)stuldm zn?yGW{_-)Mi7`>21BQbLp?^-s4x;z-b9FyW@AW87#TVJQF=u1SNB!Q(KtSneYh=7U zmM_P~%YVq1e;6&(m$v*88pS*ErP-UzIk7c0US7(VOSj=)<(IEU@t%CSFkb#9U*4-W zUcCYW@tz`A;LSl&#RBFtc>pFlDn}DzVgix6PE1EKrP^1y8l?N)p9`HIqwxA%eje(o z>M;Tow5y0Clr@nKwqN4uD=dW6L=kgU<}q93BXJQ*2pKG+)NdBaGSsW;iOp?KS@dT$ zlrb-w#(D4TsB>{KfW6)6d02DOTvq1ft3tTghPjiC0%IP+fAs~L=vT0n4U8?aeRw0& zZKydAd8OEoS)H`;NTczG=q!4W*{2Ex075cVF1BJjkv>1={-?_?YTjjOZoZ=cH)al$ z!O#AiA#gh&0y3YCpLp(9&9|Ny#Ns;8SuS(;^F`u-6EfzHr0%Oae%qLPDiA?RI2tLI z%JA+qi$zdUW?2?Yz?AGCM|Mn7wa2I%D+%Y^aejfL3({nrQ(erFi_NnaLHXkB=+{o4 z{e(QOEoqvQ(Y4(2F!a|;%0?o$y|JSU7e}&IjyIE70rM@uR@hSL$E+=lzlF7Hf;+de zAoN7c5D$jV6#!uoO+!LSd0JUO*QTW}A{A`pM-awpfQ*)l1-o%$2Rf?JTm-EXslHxe z_8wz=P`1j8cWxj%(Zy?$O>#3(3bG^owdjL|iRYw7_~i80N|tMtFlK(qpT$Bi zf=>w|5+jjwfWGEMLnhqFcGMqtto-fM*>T7EMhffg?NaB&wbr!<3|EO(I-0tp!&>GL zeiU4%XMOS0hrx`S{dtv*!|*zf-(lbI%qj(G*g$XLu?&}~<+iXG3MUJJ%yc>?Mm=XJ z@&H8}M#5|&6G{5u^&Q|&6i&h*1KDirvU|*Qfg6-Q$`%e=Ch>z|C1WP-G0EraB+6f7 z*#eVbjGX11f5zz)njp@`@ok(}ih(WN5DH#@k=4uw z9xFkyN9k-jLV>< zW*;tYDQFB#59dI}0fk#Km57(Kfm;kF@9)Fw>#GW1%U7?jZxC&B(^YSrPi~AwcldSH zyunf91}O=k=y%E1dB&KPKybO-1?@W7%T3kaNWs0Sh=r;v`uDG)UKZHaDno=ESWT{R zwHTgb<(d;6;Hp1_yNMM9A@nnl9N+DngvI-K8l9&3dmO!h$~o$E$i_#$;BlO!>DT4l z3nm*huo(YAa?bZkcGbAKlIWUJRW|7vQSJ@MRYq1@0^pP4Mi32gL)GAyU}yW~jIfyf zN%vE?KteF`gJ_;z2Qn*4+5i`#A~qRcRKKF6X@@e<{WqN6*Tdi&XK@383+G^XR#Bb| zPOSD-`k`ZrQy7`hOAxc(K1Ch-lWMlUoTGLJRg_k)#rkFcy^OWe@AN+L>j%{0ZEf%E zQ~X+;kFg}Zw39#bsvvGqvbqa0NiNL&W(^M>fBH!~rzHGmq@=8t#BC{09$D1yuDKtM z&OUv@GD`Z`g6oa{bXilABLK979lw~Vqe*}2^xMbne%ouDgn$+3 zQ*>sB7c&8oCE+}~OY;G2UA(5LR0|Qown!l@3kK;#CQ(LJw9JDcmEh8}9pj0R&17&F zj{$|-c`p#grdCIsHKItuYK^2cnRK4$QDjDQW!?8c`tk$>Yz|inhw-QH;`Fl>pt5O78Os;B(jDg(N*)Yg`E3Pq!jepXqgHgF(J_UPl)EwU+c-}B9*P8@Z zqn?ds=y~qGFhNBxeyE#iG|8QI)4Bb^1|_}HfLyOD0SLP#@zPloYJUtn`^ByQk*2d~ zJeW+wTf4P5ao3ipvCUi6q|5@?q)Z+%!lO8~HSu4?jPr(LUvWV_M9FNTkyh z#Ra{3&*o3Y$pTJlzgh|+6tpwIF7l_NtGye(Y;9O0HAD{4=lZ-o0GK%YvPn<@ zF@?~aigCWk=)}YSi#s%cQTebWjsuv^wUEWbLgD9%(?M|qSKFM4eU3i}5Y1;yy1Dee zE-l_PcS<2?kP|UDq9@G~p8I*C^0|A_eRt9!E^5&J^sf8i^6c`$>-~xfLeXgFAVl-v|s+0q=7R51Yew`7lrI=-TieJf$;b*pe41=uw}lL0b=@J2GS-6`lC zYL^bMQohLjU~4{J1j*WNPkSJztC+&PzZk6q?ZEQ^Q)bk z=UT?%_6X-!yEoOUYD&Ovj%mc!nlITqVwEdnzB@Qy;~BbmwmC69tGPL5Arno_DG}LP zGaQ&+AJou8Ri^}+U%femX4f4M69xhEU44`_^qD}25_`JDllNr%mHxyt+8>au=5^xz zr$jq%okNg0G9&?XKqn%leX=`p%2^g923+3k5RP>N!Y8o6_gqryP*l`$++^q=e6MRoFX=O0wjYUFzmV(=|+=3-Bh* zr8OGqXDJrh!Li@{)LG$~0dvRE|2^2+qx-}1t$@XTDdLHjji&SE+}eC* zh+6j;gq-igOmZbT1#%#;5125J7v*h(nqoj%itKOS7+!^26TB(>zQ0RKC&g|3kco+$ z@OxmHU*hCj>SQRK3SFPW)lP=_N-sm#5Q3Zx*E1>6-^fGomtbF}DG>ks&>f%!wEse< z!h-J&8f_1R$IZm_xA3^)m}vmH9X|{E@@bd(%>1|;;*VTCIg3zqOQicJ$x#6Rr5v9$ zgMbNA*|-v~(%Aj|YxF37DI7x2_D#qIMqZhsZ9q8XSUo_30-@N?UMJB#Uj5MiL&T)R zr=#QZl*8IXPvj(kt@iJtpm%;G?d>VW2RhURIe8GsNG8Ka)76ZoY1rkn5&qu7>7>K;-l>S3Cnl0xrx=5JSETUPOOT&K=+Fb`FP4gVKK_{T8PwhNc&>>Oc1bG}=CEm28 zU6gWgU^iFraN~!8o|!kq)Orf5vs15t;L2e|466C80_WovmYN+Y_!Z>raz&DkGsa&O z>t(7^f54=hB%|)>`DMRMJ}Q~Sf4`t#GLYv4{*r3{ErLI$7iSTIDFu3sMdKJ&c={-5 zhNKi2I^!E%RW32I4w`p@1QM|oX#J%-MZ_QqDGY#~p}e}WET{BmbuS_gnt8V@%M`AJSw&c<=ax}-?=3<<9to`WckYi^pkv?rVGnsHX3dp zgp<94y@Q>-Mz}W`zS)_KhI?^DTFv*yI!ytMQ2gCt(7g$|3GvauenKC~T)IG^lXh6gp{y`U2;i|44p?Pvd zU(1DcR_3aDH^mr-%5fDlw5rD?<%gj`!uCSMCq5hjYmUCRj3uCM<=gad@j@^Yy15ap2M z6brVu{t<5df$9{XBQkC!YdS^Xv4J_`K#xPWD?tmuY z=*#di<~x3g*~SqsH82YD6Fbc7V-H7u9Kmv9Xl%O-o_!FfeoGhAvGwH%cLSn)?EKX$ z90tb%um>T_x_-^&4$XXC+QAS7Kw$0W%OMD}0lD^r@EEfr_(Y_G!=nxddJe}YQf7Y7 zct#7~WlPRp+c~=60*!o+nR!qp1O3YVIadei*AlYI5D{I<#snQ;H_$_itQmn2gehs1 zWQbY_Q^=@?3bP{vM6!bE6~ia9NuyV(InC5up$5#s`B9>aj76tp6o+HR#{ zHWYhEd%I#}CDOE4WG$HrZYi$fk)b4gCA4j*EOir7#Imzg8>s=KaW9%rFzABmVuzv0Nh zs>*~p2qrNCYSIG=d&r(RAcR~jEY+6`Dg|x%n>gjj>CaQ3m~0YD09-_L#e<&!H$6)9 zgXrjanu*O0kf;G0aj3Y#nut?72`SQ-@xgPZ7MwklW~WO!q4N%%U267=VKR+Jr=&F! zndw75J7vwG5RYX%;!WQ)oDIiez$3-K>wpgNd=`!)Dz_g8GC?@w(28|3Shp>}$Ol49 zz&5yMBG4+vq7wheN}jM(#-J!F6trTp)IT%NU;mF~N&!C^u}~nkYm!Wo(#szac#WIY zL|^2Ju9jQ@{WBdh8p62Lp^(Se>D?b}UAT-83=7LHdT;SRwLk>n|3hjMr2o+%Ywa}F z@}S`HwfG-Qa*va2d83DwSejC@p$YGGA+)8YA(F(pub0oe5*$pG<4{{Z*`;fo3 zP^JNmN1qN-#8pZ0OeP01mSk}f6m0+-xUwqWRE5^M!Q`zeg5j+>E^|myQ&5$tmg8m^ zC1#`ffPpSbiH+hNuv?Lrf1qRfS;6(p@k;YIpUS&<5o{1mqeqE?H%fx3(t=2OB6wAI z9HgQ2#N#aXjEh4j1GzcXtjEJsTRg0n&(9Ppo*SGmr5|r_2R6c!t#qIAcZ-8Zm6q*q zXCM(7H`LM%Ji8~jxd8~Z3Gug%6_A-xi`!jqJ4oi(jxZ{ww0c2AoY0ppN3U~nes+4% zd1fgr5X(_2T%tp_D(;@agi6mU6~i=A-a!22<0$y*%f_#`;hTc%Fws+I2qGR|AwQoK zp}#*BtOqyNK33{vtC~%Vyq$-v|*UYssN=J>UW z2X*goRjA$)^EMA$3p>MzP%F-p)F@fyBwumyv0Yt-GD`y@7Z#T$XNBEv$WXB*nUF*m z{FahJuIj{Vi|p9}CETF`sU^pI_C&QXH2*#9s#Wk`C)WaOso4IBXS1x<0 z;Wv{+kL7o*Bv;tVeNIVA2N!3Ty`#=c#$##EZ2ajOHx$)sDgybl13T}XeeRsLPfG?z zfyWHTeKS3ZjhF*Ujec3%2cqBG=)upGqVe1B2isUV=U3#Gl0;#yOOvr8JwVvM9jiC`U5p6le_V zMa!q-Y?K!Otgb%Wa-vtN0(nvL#_%<3hdf`ub-Uh;m1D;O%qm&8fYpqFNos20IMKQI z#!mHgDps$$8*E<*=BiOC2&h2pf=iOI*vmU`$ISuPH79P4o4?VpD`r4;KB(($w3QT{ z286f!Ept{JM1B?A$)3>_hxsLDkv?cN3hP?hy*H|s2heAcjHlNa>!54t+IXSkb ztPbj>F|}66w(ewnf~}jQhGOqil7={`Sh+-aGj0VkX@QBV*IV9`&gqB#$5QALrI3!p zG!8>IDqxI?at@Kg0WJNRvD75g0djS!oMOPxpBf6S6%?q$>vumM_e7`&wk@bHlX*D$ z3d{dXw@bnm)!9GX-o5c8C0@;PlprY5(`;LuD49K+TGtygHK!I|(8gR|Qq3h* z(kf+)nEv(N3=bxO8lnBXyfLtEqKb!7u7YmZ%%{@6?0-D#0ZV^8==Ux?erL~d&^&6F zUuKow3T2~aO!R$6#nD&)#oeN5G=Wo63FAW7i%Uwn=0>jU20(PZ01?hq^>r?gX)Ya( zW2h|uh#}6?dBweRHp1!}b4ur~v9~tYXZDWSjUSohO7)LPC|3A{+C!B#f8Ou}7NxRA zKsBls;X)|}F^gIbDsw}0Y-fArP5Lvo=+0;voG%(KS9xZrA55&A~|l zyh02>~m!x9&P|3Vd7v)C2zf1ITdxCb6Ai-$WBC<%5s!O z*Ro9^3S!WcA26n#PM|00`{_YUD{3*R)VYt=tbHYG&M<)q>2t-K*F;*^YQfS1U=z5+ zYt<}j1^m?oQ*b4l5q1eTB{5%C1bcC<#g+YI$)+>Y4WUXHTCC#1#)vUOwfs~Vg)ian zpY6P$;U-^STg?=Wty?r z8J>yd<)E8d?b5)qi(ExdqCQ%k>y4AMqxK1>(bnTY+9FT95JImv$Z}2^*;{34m~Ee58Yvour#NjIy^MswQ`hulalCb;FXnhPjd>+%`!y1@d{D{m(tA^ZFT8H~bVXxrIieYrdT^Yw$7=W$ zzoqD0tBgyFm(DYVGYp{^TJUGa3osbUa-SG*3ER{1%eGT%0b&?h#|IuLWu$Cl$f$@$7m| z^Nrr}>>vAIxsf60nbr~~1E zzH-g1p;&HZi}}cMC>!m(ZEWv^jKdvBth-JSh6<2MHfJJEQ1wEd=Wu(P+j zKRlQmjK|~2+vfK6&i>mddK2z7w@0Q!*-!tXT#hgpyl^{`PFv49n)S|)4rSB;MyZ6q zcx~i@kvCoauW$pKjFNo%-;u`f(am2d8k3Gz%KBfri0QnYH75yZ%}CMhw!)62sujx@ zHAJfiBRpNL5EuCDw((gDx{*404~k%#M*KhIjChv4$>d7Fz0)E^9D<~hRvJ3%To$p z1?_i^`e(i0bx@#9>5(o(GchqEp9SiOHsAN!pE~b)-Qy1wd#ZhUjJc`%?S5yI42dgP ztA3Ya3}gVrW}T#&*MNZ+(QAWVZb@g%&1x0afgTWiunl!oDJ!Iv0Ql>mvT3#>?8G;N zp9}Wqm>G&^Q`KJ4=Hwjjwm0d{ba*LBj`XROVMYtSF%}!wj4rw!R5DmKMZS8Eo**}oS^{)uHWy60(psfP% z+$dXr5%5d^0{G5)$3^Jb<5AJO{@&7%s=1By&&~!X?cRsZX7Qd-WM`rXT&e}iss7A` zq3*oUpNOuoS-cK2OqGpm%VJ(KtxAoW&^3~87WYx{##7a- zFB3{hRy39`q^By_PGyZ(_$lEH?)nTtnHKZ$23cSE&Y442V2kTAOPK zMW!B3#H(o;N=SY+h(wZY{lQ)lRkJLPuvn7S(a)XEWE6Cn5Jv$g57_WsbJ@JUo@muB zd9`WW1Z>)xd|k%X`Ii>ND+&6zsV*co{V`@Y4S(NFLb98A{`=sxK{$GP8rt|DRc zRmExPsi?IV;QyPK7oEX}UiVZg;kCjKpR|yy0DRUqKp&@56cx2z5qjD`dCjoxTg20?u?cq{Wmdl!8Kou}zi! z32GzPKsuK?azmTd!mYE)C%|tthDOFo#hX)InVK#w%^KDR#%I&!Kr0@Mh!jRy*o21Z zhOi>gOuR_%XEE;9^bJc^cgj`ID7i($iDXk>=o3Rw7y(xnY3pdO8*p}vX^YKCFNIF{MIhPa}?DXaL`=&`gzV;VIsX;npa?3#6{=#!1( zY|rGmbpqHrN`M>cpPhGl?Y?1`D5O++AziSqbf_2?99) zw<4`#=gWELsE_(4j_WK9*owfQ9lp*0v7q1hOAQ5!C|k8)(=bIV_RGg(osOyGGd)uo zWp%bmMrWIb%~`N%nw0rgWb^v^q0!isvWL@hNdq$7=3L`l8T_)MF2WUYv1AWIq@_q& z$p4BgwT#~)gT1A3b)hR!X+ewhF^YW1lgu9dMn zRwE|KFTqffO6ht06nv)Oy-paaMINJ_g|ai?3E(FTCq?`pg&IV_%6R5=6FL$fnxpk| zfi=W(4$5g645`FHy7jvlBbcM16Y8qUln%b~GQG2YBVi)$FA73e%MH2bPfV^Puh!=L-u7tqwK(060N%HkebL9f<@D|Pa;whH2*+1z~Yf~`c#{?csPi|$uH6or26RH@Yv zG8VNm7xZywcEbhP%@k_dCuke>`v8W1Y;RVB%9bK5&$jb&YUP5a!DV6AIx)>n6(AZD zO0(Cu##Qlbw&akK=cgZ}PG}KL9|=h%WZov@fJ@-~S0SaJf=}V%D-DYgB^d;uHdAY> z5K|ti0JLO@#aR$SCqnM0rsJ>I_p2-mx76BH1#us44YXzZOu`U&2L06!0y|gt}Y{I#!DaI3t zn3U+%Teb&N+$OE=m8^>={QD-RsMuu3lTB$XHZWjxSc<(gi4&80vae0nV_RuVRpVl0 z1`APvaeuP}4}5`4YK)QJ9 z1Tm2j=j*__S~amfS$p7SeD0EUt&7!lQgDg!l(-`P^(F%944<4G{pz;|pu11BwyRHY zbaIF_@scWbz7$4`fpc4{1nu>6R~3OWUtpeS6%!JJGRT0Cn`HX5SuHZG4PVv=jkn8j zya$yVak~n5&Bm3YBy6JZPgRNRyf{}QKnwNRI!TwP;yVkeb0NUl3Jn$jCQy}dtl8x2 zEmSXzSiPWm0A-l>^t1+mn>K(=Z+bx#Yk?<)p%nr2CF9h0`a%oJL}=Zyl5r`TC8GXh zFd%VcFhG1f7)Y5!knG0{)~)jpBYOtu8rw6ha+S-AWt+RZGm3V058k}l-90#XI}AtR zr1^HQIc)6hOh&tJ_qHda=Hy^JdOO~K+Zb-|?2nq;(f$tbb1h0sD9gb&lLJ>tNbafI<$+O`sk@2lnqBMzIQAi7hk7$n&WtQ1BwS|68A^B6h zf7kEyJ_S*78z&e}8~tKL2-Zm)P08tdB>kw+9xqQv>6Gne*d76I0aacQ!O&-S(R9l3 z(EcY9N2Hn8G&q_P^2*8*NyN8+$@JrLh$-Qt>=_$kl`TY0$Ov3pD@28KDr2SxQOzQu ze28iV^`*R1vDIN}V z&IT0?{rmL>EPFvwIb{9QKxQIG0gx`BBDIwU}Bvn7SRcRywN_WJD-* zPl^T=Uz&OLV#-x*wPRL0=f!`qSu`1&0kL1e$+eBm*3Qi!YBTqPzck00MA7(9+CDq* za4ewb;UbGfs9F)XShi1n^KBBpH!loCfMpsmrY5|r80WyGnMPp{?zyN4<7~zTz}oP$ zdtP)?=fgqZp?0p26E%A~+-lw!L&nddj-Ok$IxR4-WvPkN#Ta=Aj!c~a0+zH7*sFP( z#SbpytOi3>nVny6#w*x$YHKCGuDbOy>ub(^P0sz9>lT(tB{flYm&HOfW}0zJS11|m z>DlQ~8~*B&Ppg@@d2zd(v0Z#2B)MvIWiWCYhKxwQDkQn=%7{ts%I3axE*-iuIS^Q1 zwqMSqtqOVMQVkDJjJ~Zpd>V1TK5XF)3cMDY@h+V|tWv?iLIiwQf!_=GNJWG zVnBa6i5msCHh`a24h?k|@|g0N#-k<1Ct?TIl``-HFD4 zrxFyihG~8m%;R~K{Y^MK-lNYqw!{wHP#hChPa;xCn{KB<$p@2llf8)~yzj)nvjQo& z-grSGIBZ4E@kV(?A!lkCPb0@ZjhJaPOc15a!x3#*{HY&DlW;lBFT@ME&>UYiV|HF{ z+(Swi7%mp!V};l=_yDrJ^WjNy#Nikdcw;&qgx4$ueiX@U1?i$bYCaz)H76~1)z3+( zhmb~&8U)Oi@$tCx`SkMSq$VQ#o1f0RHu75$mw;wdsLPq=2a|EOcDm1idG$dO-AiLq zml2aIP?v!+Dza0aLd{sC&S$(+klE-w`3hDx26I~37_FcggqV(nYVtP92eWWqDNy@p zCJs(Zuk}`8tqY-`bUnbZjc4L7T5^|7wj|Eg@|7nZB*nSSqggE&hzv(ODgIHxkc?_P zwV+YouKEI;$t!*71cfbztlG_S_BOwupXWnD7D)QOK-n8Z2DzIN(gTBAuAPnQd0iIkkO zTK}ufnv`PKvUKObnm2TfScuCiSZwLZ_+93hIiyat&7SJS*nLz<+IkkPNUDXTCyob7 zX~GkgWQado(xB~5;DjvMP$o()^F?FzB>@r$Ug@v#~!qh6D>E^%}C01@2Xs zpr(E)@T}DrP41;!r@dHfCSRjngK32AG+|b%mLJWK=Ann2f_fy9&^GAVr4Uwbll9U$ zRmiNjt*F55dTd$ZP%e(!`QK7>)XrxDK{^@iOjsd4eiveZju81e)JtJd=wF2>_&dAaX zMbB?sy?pULL5+#lQDq+!uhsrCxsMtLldA$+S;X^7wT?b?3(hfM@+JNNw0^u^ykL>D zG)f5do;7FErt;)Bo}OwOXnTlE*T&(-WPjM)9qny58*ih=cyDj`w)u9_eDgNkIT#&; ZyQ722!NLB) dict: + calls.append(1) + started.set() + release.wait(2) + finished.set() + return {"ok": True} + + async def dispatch(): + with anyio.CancelScope() as scope: + scopes.append(scope) + await mcp.call_tool("bounded_operation", {}) + + async with anyio.create_task_group() as group: + group.start_soon(dispatch) + await until_set(started) + scopes[0].cancel() + await anyio.sleep(.02) + self.assertFalse(finished.is_set()) + release.set() + self.assertTrue(finished.is_set()) + self.assertEqual(calls, [1]) + + +class AutomaticJobTests(unittest.TestCase): + def test_identical_concurrent_auto_jobs_are_distinct_and_explicit_ids_unchanged(self): + from concurrent.futures import ThreadPoolExecutor + tree = ast.parse(PORTABLE.read_text()) + function = next(node for node in tree.body if isinstance(node, ast.FunctionDef) + and node.name == "_framebridge_exec_geranas_impl") + namespace = dict(hashlib=hashlib, time=time, uuid=uuid, + JOB_RE=re.compile(r"^[A-Za-z0-9._-]{1,80}$"), + adapter_request=lambda payload: {"ok": True, "job_id": payload["job_id"]}, + sanitize_text=lambda text: text) + exec(compile(ast.Module(body=[function], type_ignores=[]), str(PORTABLE), "exec"), namespace) + execute = namespace["_framebridge_exec_geranas_impl"] + with mock.patch.object(time, "strftime", return_value="20261028-000000"): + with ThreadPoolExecutor(max_workers=8) as pool: + results = list(pool.map(lambda _: execute("same-command", True), range(16))) + self.assertEqual(len({result["job_id"] for result in results}), 16) + self.assertTrue(all(namespace["JOB_RE"].fullmatch(result["job_id"]) for result in results)) + first = execute("same-command", True, job_id="owner-idempotent-id") + second = execute("same-command", True, job_id="owner-idempotent-id") + self.assertEqual(first["job_id"], "owner-idempotent-id") + self.assertEqual(first["job_id"], second["job_id"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_frontdoor_thread_state.py b/source/tests/test_frontdoor_thread_state.py new file mode 100644 index 0000000..6d04927 --- /dev/null +++ b/source/tests/test_frontdoor_thread_state.py @@ -0,0 +1,224 @@ +"""Deterministic worker initialization races, using only in-memory fixtures.""" +import ast +from concurrent.futures import ThreadPoolExecutor +from pathlib import Path +import sys +import threading +from types import SimpleNamespace +import unittest +from unittest import mock + +WORK = Path(__file__).resolve().parents[1] +EXPANDED = WORK / "expanded/framebridge_expanded" + + +def definitions(module, names, namespace): + tree = ast.parse((EXPANDED / module).read_text()) + selected = [node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name in names] + exec(compile(ast.Module(body=selected, type_ignores=[]), str(EXPANDED / module), "exec"), namespace) + return namespace + + +class ObservedRLock: + """Signal when the competing worker reaches the lock, without timing guesses.""" + def __init__(self): + self.lock = threading.RLock() + self.waiting = threading.Event() + self.watch = None + + def __enter__(self): + if threading.get_ident() == self.watch: + self.waiting.set() + self.lock.acquire() + return self + + def __exit__(self, *exc): + self.lock.release() + + def watched(self, function): + self.watch = threading.get_ident() + return function() + + +class WorkerInitializationTests(unittest.TestCase): + def test_runtime_publishes_complete_module_once_to_competing_workers(self): + started, release = threading.Event(), threading.Event() + lock, module = ObservedRLock(), SimpleNamespace() + loads = [] + + def initialize(target): + loads.append(target) + started.set() + if not release.wait(2): + raise AssertionError("fixture load was not released") + target.settings = "complete" + + loader = SimpleNamespace(name="fixture", exec_module=initialize) + namespace = definitions("runtime.py", {"_legacy"}, dict( + _LEGACY=None, _LEGACY_LOCK=lock, PREFIX=Path("/fixture"), + importlib=SimpleNamespace( + machinery=SimpleNamespace(SourceFileLoader=lambda *args: loader), + util=SimpleNamespace(spec_from_loader=lambda *args: object(), + module_from_spec=lambda spec: module)))) + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(namespace["_legacy"]) + try: + self.assertTrue(started.wait(1)) + self.assertIsNone(namespace["_LEGACY"]) + second = pool.submit(lock.watched, namespace["_legacy"]) + self.assertTrue(lock.waiting.wait(1)) + self.assertFalse(second.done()) + finally: + release.set() + self.assertIs(first.result(1), module) + self.assertIs(second.result(1), module) + self.assertEqual(module.settings, "complete") + self.assertEqual(loads, [module]) + + def test_runtime_failed_initialization_remains_retryable(self): + attempts = [] + module = SimpleNamespace() + + def initialize(target): + attempts.append(target) + if len(attempts) == 1: + raise RuntimeError("fixture failure") + target.settings = "complete" + + loader = SimpleNamespace(name="fixture", exec_module=initialize) + namespace = definitions("runtime.py", {"_legacy"}, dict( + _LEGACY=None, _LEGACY_LOCK=threading.RLock(), PREFIX=Path("/fixture"), + importlib=SimpleNamespace( + machinery=SimpleNamespace(SourceFileLoader=lambda *args: loader), + util=SimpleNamespace(spec_from_loader=lambda *args: object(), + module_from_spec=lambda spec: module)))) + with self.assertRaisesRegex(RuntimeError, "fixture failure"): + namespace["_legacy"]() + self.assertIsNone(namespace["_LEGACY"]) + self.assertIs(namespace["_legacy"](), module) + self.assertEqual(len(attempts), 2) + self.assertEqual(module.settings, "complete") + + def test_fleet_bootstrap_serializes_shared_path_check_remove_and_append(self): + started, release = threading.Event(), threading.Event() + lock = ObservedRLock() + broker, fleet = Path("/fixture/broker"), Path("/fixture/fleet") + + class Paths(list): + def remove(self, value): + if not started.is_set(): + started.set() + if not release.wait(2): + raise AssertionError("fixture path update was not released") + super().remove(value) + + paths = Paths([str(broker), str(fleet)]) + package = SimpleNamespace(__file__=str(broker / "framebridge_linux_chatgpt/__init__.py"), + __path__=[]) + namespace = definitions("fleet.py", {"bootstrap_original"}, dict( + _BOOTSTRAP_LOCK=lock, original_paths=lambda: (broker, fleet), Path=Path, + sys=SimpleNamespace(path=paths), + _require=lambda valid, code: self.assertTrue(valid, code))) + with mock.patch.dict(sys.modules, {"framebridge_linux_chatgpt": package}): + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(namespace["bootstrap_original"]) + try: + self.assertTrue(started.wait(1)) + second = pool.submit(lock.watched, namespace["bootstrap_original"]) + self.assertTrue(lock.waiting.wait(1)) + self.assertFalse(second.done()) + finally: + release.set() + first.result(1) + second.result(1) + self.assertEqual(paths.count(str(broker)), 1) + self.assertEqual(paths.count(str(fleet)), 1) + self.assertCountEqual(package.__path__, [str(root / "framebridge_linux_chatgpt") + for root in (broker, fleet)]) + + + def broker_namespace(self, config): + def validate(url): + if url == "invalid-fixture": + raise ValueError("fixture endpoint rejected") + return url + + return definitions("broker.py", {"configure", "_settings"}, dict( + _state=None, _url=None, _control_path=None, + _SETTINGS_LOCK=ObservedRLock(), Path=Path, + state_path=lambda state=None: Path(state or "/fixture/default"), + _config=config, validate_endpoint=validate)) + + def test_broker_initial_settings_are_published_as_one_complete_profile(self): + started, release = threading.Event(), threading.Event() + loads = [] + + def config(state): + loads.append(state) + started.set() + if not release.wait(2): + raise AssertionError("fixture config was not released") + return {"ports": {"broker": 23456}} + + namespace = self.broker_namespace(config) + lock = namespace["_SETTINGS_LOCK"] + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(namespace["_settings"]) + try: + self.assertTrue(started.wait(1)) + self.assertEqual(tuple(namespace[key] for key in ("_state", "_url", "_control_path")), + (None, None, None)) + second = pool.submit(lock.watched, namespace["_settings"]) + self.assertTrue(lock.waiting.wait(1)) + self.assertFalse(second.done()) + finally: + release.set() + expected = (Path("/fixture/default"), "http://127.0.0.1:23456", + Path("/fixture/default/broker/control.key")) + self.assertEqual(first.result(1), expected) + self.assertEqual(second.result(1), expected) + self.assertEqual(loads, [Path("/fixture/default")]) + + def test_broker_explicit_profile_change_cannot_expose_mixed_endpoint_or_key_path(self): + started, release = threading.Event(), threading.Event() + + def config(state): + if state == Path("/fixture/next"): + started.set() + if not release.wait(2): + raise AssertionError("fixture config was not released") + return {"ports": {"broker": 23456}} + + namespace = self.broker_namespace(config) + previous = namespace["_settings"]() + lock = namespace["_SETTINGS_LOCK"] + with ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(namespace["configure"], "https://fixture.example:24443", + "/fixture/next", "/fixture/next/explicit-control.key") + try: + self.assertTrue(started.wait(1)) + self.assertEqual(tuple(namespace[key] for key in ("_state", "_url", "_control_path")), + previous) + second = pool.submit(lock.watched, namespace["_settings"]) + self.assertTrue(lock.waiting.wait(1)) + self.assertFalse(second.done()) + finally: + release.set() + first.result(1) + self.assertEqual(second.result(1), (Path("/fixture/next"), + "https://fixture.example:24443", Path("/fixture/next/explicit-control.key"))) + + def test_broker_failed_validation_publishes_nothing_and_can_retry(self): + namespace = self.broker_namespace(lambda state: {}) + with self.assertRaisesRegex(ValueError, "fixture endpoint rejected"): + namespace["configure"]("invalid-fixture", "/fixture/rejected") + self.assertEqual(tuple(namespace[key] for key in ("_state", "_url", "_control_path")), + (None, None, None)) + previous = namespace["_settings"]() + with self.assertRaisesRegex(ValueError, "fixture endpoint rejected"): + namespace["configure"]("invalid-fixture", "/fixture/rejected") + self.assertEqual(namespace["_settings"](), previous) + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_node_recovery.py b/source/tests/test_node_recovery.py new file mode 100644 index 0000000..fb96402 --- /dev/null +++ b/source/tests/test_node_recovery.py @@ -0,0 +1,341 @@ +"""Socket fault injection against the real FBPAIR broker and enrolled identities.""" +import errno +import hashlib +import selectors +import shlex +import signal +import socket +import ssl +import subprocess +import sys +import threading +import time +import unittest +import urllib.error +from unittest import mock + +from framebridge_expanded import broker, node +import test_fleet_broker_node as fixtures + + +class StopLoop(BaseException): + pass + + +class NodeRecoveryTests(unittest.TestCase): + setUp = fixtures.BrokerNodeTests.setUp + tearDown = fixtures.BrokerNodeTests.tearDown + make_node = fixtures.BrokerNodeTests.make_node + pair = fixtures.BrokerNodeTests.pair + + @staticmethod + def disconnect(handler): + handler.close_connection = True + handler.connection.shutdown(socket.SHUT_RDWR) + handler.connection.close() + + def test_transient_poll_disconnects_backoff_cap_and_inventory_recovery(self): + state, ident = self.pair("recover-node") + node.run(state, once=True) + store = self.server.expanded_store + store.node_last_seen[ident] = time.time() - 10 + self.assertFalse(broker.list_confirmed_nodes()[0]["ready"]) + identity_before = (state / "node/identity.json").read_bytes() + original = self.server.RequestHandlerClass.do_POST + real_post = broker.post_json + attempts, waits, envelopes = [], [], [] + recovered = [] + + def fault(handler): + if handler.path.endswith("/node/poll"): + attempts.append(1) + if len(attempts) <= 8: + self.disconnect(handler) + return + return original(handler) + + def post(url, envelope, **kwargs): + if url.endswith("/node/poll"): + envelopes.append(envelope) + result = real_post(url, envelope, **kwargs) + if url.endswith("/node/poll"): + recovered.append(True) + return result + + def pause(seconds): + if recovered: + raise StopLoop + self.assertFalse(store.node_runtime_status(ident)["ready"]) + waits.append(seconds) + + with mock.patch.object(self.server.RequestHandlerClass, "do_POST", fault), \ + mock.patch.object(broker, "post_json", side_effect=post), \ + mock.patch.object(node, "time", wraps=time) as clock, \ + self.assertLogs(node._LOG, level="WARNING") as logs: + clock.sleep.side_effect = pause + with self.assertRaises(StopLoop): + node.run(state) + self.assertEqual(waits, [1, 2, 4, 8, 16, 30, 30, 30]) + self.assertEqual(len(attempts), 9) + self.assertEqual(len({e["nonce"] for e in envelopes}), 9) + self.assertTrue(store.node_runtime_status(ident)["ready"]) + self.assertEqual((state / "node/identity.json").read_bytes(), identity_before) + self.assertTrue(any("NODE_BROKER_RECOVERED" in line for line in logs.output)) + self.assertTrue(all("signature" not in line and "payload" not in line for line in logs.output)) + + def test_result_disconnect_before_and_after_acceptance_never_reexecutes(self): + state, ident = self.pair("ack-node") + marker = self.root / "executed.txt" + real_post = broker.post_json + original_post = self.server.RequestHandlerClass.do_POST + original_send = self.server.RequestHandlerClass.send_json + for after_acceptance in (False, True): + with self.subTest(after_acceptance=after_acceptance): + marker.write_text("") + script = "printf x >> " + shlex.quote(str(marker)) + payload = {"authorized": True, "job_id": "recovery-command", "timeout_s": 10, + "script": script, "sha256": hashlib.sha256(script.encode()).hexdigest()} + queued, error = self.server.expanded_store.queue_admin(ident, "admin.exec", payload) + self.assertIsNone(error) + posted, dispatched, waits, dropped, delivered = [], [], [], [], [] + + def execute(value): + dispatched.append(value["job_id"]) + subprocess.run(["/bin/sh", "-c", value["script"]], check=True) + return {"ok": True, "exit_code": "0"} + + def fault_post(handler): + if not after_acceptance and handler.path.endswith("/node/result") and not dropped: + dropped.append(True) + self.disconnect(handler) + return + return original_post(handler) + + def fault_send(handler, code, value): + if after_acceptance and handler.path.endswith("/node/result") and not dropped: + dropped.append(True) + self.disconnect(handler) + return + return original_send(handler, code, value) + + def post(url, envelope, **kwargs): + if url.endswith("/node/result"): + posted.append(envelope) + result = real_post(url, envelope, **kwargs) + if url.endswith("/node/result"): + delivered.append(True) + return result + + def pause(seconds): + if delivered: + raise StopLoop + waits.append(seconds) + + with mock.patch.object(self.server.RequestHandlerClass, "do_POST", fault_post), \ + mock.patch.object(self.server.RequestHandlerClass, "send_json", fault_send), \ + mock.patch.object(broker, "post_json", side_effect=post), \ + mock.patch("framebridge_expanded.runtime.adapter_request", side_effect=execute), \ + mock.patch.object(node, "time", wraps=time) as clock, \ + self.assertLogs(node._LOG, level="WARNING"): + clock.sleep.side_effect = pause + with self.assertRaises(StopLoop): + node.run(state) + self.assertEqual(marker.read_text(), "x") + self.assertEqual(dispatched, ["recovery-command"]) + self.assertEqual(waits, [1]) + self.assertEqual(len(posted), 2) + self.assertNotEqual(posted[0]["nonce"], posted[1]["nonce"]) + self.assertEqual(posted[0]["payload"], posted[1]["payload"]) + saved = self.server.expanded_store.get_result(queued["request_id"]) + self.assertEqual(saved["state"], "complete") + self.assertTrue(saved["result"]["ok"]) + + def test_broker_restart_after_execution_is_terminal_without_replay(self): + state, ident = self.pair("orphan-node") + marker = self.root / "orphan-effect.txt" + script = "printf x >> " + shlex.quote(str(marker)) + payload = {"authorized": True, "job_id": "orphan-command", "timeout_s": 10, + "script": script, "sha256": hashlib.sha256(script.encode()).hexdigest()} + queued, error = self.server.expanded_store.queue_admin(ident, "admin.exec", payload) + self.assertIsNone(error) + real_post = broker.post_json + result_attempts = [] + + def execute(value): + subprocess.run(["/bin/sh", "-c", value["script"]], check=True) + return {"ok": True, "exit_code": "0"} + + def post(url, envelope, **kwargs): + if url.endswith("/node/result"): + result_attempts.append(1) + self.assertEqual(marker.read_text(), "x") + self.server.shutdown() + self.server.server_close() + self.thread.join(2) + self.server = broker.make_server(port=self.port, state_dir=self.central) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + self.assertIsNone(self.server.expanded_store.get_result(queued["request_id"])) + return real_post(url, envelope, **kwargs) + + with mock.patch.object(broker, "post_json", side_effect=post), \ + mock.patch("framebridge_expanded.runtime.adapter_request", side_effect=execute) as adapter, \ + mock.patch.object(node, "time", wraps=time) as clock: + with self.assertRaisesRegex(broker.BrokerError, "^BROKER_HTTP_REJECTED_400$"): + node.run(state) + adapter.assert_called_once() + clock.sleep.assert_not_called() + self.assertEqual(result_attempts, [1]) + self.assertEqual(marker.read_text(), "x") + # A later operator restart may poll, but must never reconstruct/replay + # the orphaned command from its lost acknowledgement. + self.assertTrue(node.run(state, once=True)["idle"]) + self.assertEqual(marker.read_text(), "x") + + def test_gateway_failures_recover_without_replaying_signed_envelope(self): + state, _ = self.pair("gateway-node") + original = self.server.RequestHandlerClass.do_POST + codes, waits, successes = [502, 503, 504], [], [] + + def gateway(handler): + if codes: + handler.send_json(codes.pop(0), {"ok": False}) + return + successes.append(True) + return original(handler) + + def pause(seconds): + if successes: + raise StopLoop + waits.append(seconds) + + with mock.patch.object(self.server.RequestHandlerClass, "do_POST", gateway), \ + mock.patch.object(node, "time", wraps=time) as clock, \ + self.assertLogs(node._LOG, level="WARNING"): + clock.sleep.side_effect = pause + with self.assertRaises(StopLoop): + node.run(state) + self.assertEqual(waits, [1, 2, 4]) + + def test_auth_and_invalid_json_fail_without_retry(self): + state, _ = self.pair("reject-node") + for code, body in [(403, b'{"ok":false}'), (500, b'{"ok":false}'), + (200, b'not-json'), (200, b'{"ok":false}')]: + with self.subTest(code=code, body=body): + attempts = [] + + def reject(handler): + attempts.append(1) + handler.send_response(code) + handler.send_header("Content-Length", str(len(body))) + handler.end_headers() + handler.wfile.write(body) + + with mock.patch.object(self.server.RequestHandlerClass, "do_POST", reject), \ + mock.patch.object(node, "time", wraps=time) as clock: + with self.assertRaises((broker.BrokerError, node.NodeError)): + node.run(state) + clock.sleep.assert_not_called() + self.assertEqual(len(attempts), 1) + + def test_tls_and_unknown_transport_causes_are_terminal(self): + state, _ = self.pair("tls-node") + for cause in [urllib.error.URLError(ssl.SSLCertVerificationError("untrusted test CA")), + urllib.error.URLError(socket.gaierror(socket.EAI_NONAME, "invalid host")), + OSError(errno.EACCES, "denied"), ValueError("invalid response")]: + with self.subTest(cause=type(cause).__name__), \ + mock.patch("urllib.request.OpenerDirector.open", side_effect=cause) as opened, \ + mock.patch.object(node, "time", wraps=time) as clock: + with self.assertRaises(broker.BrokerError): + node.run(state) + self.assertEqual(opened.call_count, 1) + clock.sleep.assert_not_called() + + def test_once_is_fail_fast_and_retry_wait_propagates_cancellation(self): + state, _ = self.pair("cancel-node") + attempts = [] + + def fault(handler): + attempts.append(1) + self.disconnect(handler) + + with mock.patch.object(self.server.RequestHandlerClass, "do_POST", fault), \ + mock.patch.object(node, "time", wraps=time) as clock: + with self.assertRaises(broker.BrokerError): + node.run(state, once=True) + clock.sleep.assert_not_called() + clock.sleep.side_effect = KeyboardInterrupt + with self.assertLogs(node._LOG, level="WARNING"), self.assertRaises(KeyboardInterrupt): + node.run(state) + self.assertEqual(len(attempts), 2) + clock.sleep.assert_called_once_with(1) + + def test_real_broker_restart_recovers_same_node_process_and_identity(self): + state, ident = self.pair("restart-node") + identity_before = (state / "node/identity.json").read_bytes() + code = "from framebridge_expanded.node import run; run(state_dir=" + repr(str(state)) + ")" + child = subprocess.Popen([sys.executable, "-u", "-c", code], + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) + + def wait_ready(): + deadline = time.monotonic() + 8 + while time.monotonic() < deadline: + self.assertIsNone(child.poll(), "node exited during recovery") + if self.server.expanded_store.node_runtime_status(ident)["ready"]: + return + threading.Event().wait(0.02) + self.fail("node heartbeat did not recover") + + try: + wait_ready() + self.server.shutdown() + self.server.server_close() + self.thread.join(2) + with selectors.DefaultSelector() as selector: + selector.register(child.stderr, selectors.EVENT_READ) + self.assertTrue(selector.select(10), "child did not report broker interruption") + self.assertIn("NODE_BROKER_RETRY", child.stderr.readline()) + self.assertIsNone(child.poll()) + self.server = broker.make_server(port=self.port, state_dir=self.central) + self.assertFalse(self.server.expanded_store.node_runtime_status(ident)["ready"]) + self.thread = threading.Thread(target=self.server.serve_forever, daemon=True) + self.thread.start() + wait_ready() + self.assertEqual((state / "node/identity.json").read_bytes(), identity_before) + self.assertEqual(broker.list_confirmed_nodes()[0]["node_id"], ident) + finally: + child.terminate() + try: + child.wait(timeout=3) + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=3) + child.stderr.close() + + def test_sigterm_stops_real_process_during_retry_wait(self): + state, _ = self.pair("stop-node") + self.server.shutdown() + self.server.server_close() + self.thread.join(2) + code = "from framebridge_expanded.node import run; run(state_dir=" + repr(str(state)) + ")" + child = subprocess.Popen([sys.executable, "-u", "-c", code], + stdout=subprocess.DEVNULL, stderr=subprocess.PIPE, text=True) + try: + with selectors.DefaultSelector() as selector: + selector.register(child.stderr, selectors.EVENT_READ) + self.assertTrue(selector.select(10), "child did not report retry") + self.assertIn("NODE_BROKER_RETRY", child.stderr.readline()) + started = time.monotonic() + child.terminate() + self.assertEqual(child.wait(timeout=3), -signal.SIGTERM) + self.assertLess(time.monotonic() - started, 3) + finally: + if child.poll() is None: + child.kill() + child.wait(timeout=3) + child.stderr.close() + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_private_candidate_prerm.py b/source/tests/test_private_candidate_prerm.py new file mode 100644 index 0000000..4b74291 --- /dev/null +++ b/source/tests/test_private_candidate_prerm.py @@ -0,0 +1,91 @@ +"""Private package versions retain the fixed privileged-helper rollback floor. + +Only a temporary maintainer script is run. Cleanup Python is replaced with a +recording stub; no installed helper, profile or package transaction is touched. +""" +import ast +import json +from pathlib import Path +import shlex +import subprocess +import sys +import tempfile +import unittest + +WORK = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(WORK / "packaging")) +import build_expanded + +PRIVATE_VERSION = "0.3.2+private.20261028.1" + + +class PrivateCandidatePrermTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + self.control = self.root / "DEBIAN" + self.control.mkdir() + self.log = self.root / "cleanup-calls.jsonl" + self.python_stub = self.root / "record-cleanup" + self.python_stub.write_text( + "#!/usr/bin/python3\nimport json,sys\nfrom pathlib import Path\n" + + "with Path(" + repr(str(self.log)) + ").open('a') as stream:\n" + + " stream.write(json.dumps(sys.argv[1:])+'\\n')\n") + self.python_stub.chmod(0o755) + + def candidate_script(self, package_version=PRIVATE_VERSION): + # Execute the real production write expression with an arbitrary package + # VERSION, rather than reconstructing its floor selection in the test. + tree = ast.parse(Path(build_expanded.__file__).read_text()) + main = next(node for node in tree.body + if isinstance(node, ast.FunctionDef) and node.name == "main") + writers = [node for node in ast.walk(main) + if isinstance(node, ast.Expr) and isinstance(node.value, ast.Call) + and isinstance(node.value.func, ast.Attribute) + and node.value.func.attr == "write_text" + and node.value.args and isinstance(node.value.args[0], ast.Call) + and isinstance(node.value.args[0].func, ast.Name) + and node.value.args[0].func.id == "maintainer_prerm"] + self.assertEqual(len(writers), 1) + namespace = dict(control=self.control, VERSION=package_version, + PRIVILEGED_HELPER_FLOOR=build_expanded.PRIVILEGED_HELPER_FLOOR, + maintainer_prerm=build_expanded.maintainer_prerm) + exec(compile(ast.Module(body=writers, type_ignores=[]), build_expanded.__file__, "exec"), namespace) + script = self.control / "prerm" + text = script.read_text() + self.assertIn('"$2" lt 0.3.2;', text) + self.assertNotIn(package_version, text) + # The real dpkg is used only for its read-only version validator and + # comparator. Every privileged cleanup command is a recording stub. + script.write_text(text.replace("/usr/bin/python3", shlex.quote(str(self.python_stub)))) + return script + + def upgrade_target(self, target, package_version=PRIVATE_VERSION): + script = self.candidate_script(package_version) + result = subprocess.run(["/bin/sh", str(script), "upgrade", target], + capture_output=True, text=True, timeout=5) + self.assertEqual(result.returncode, 0, result.stderr) + return [json.loads(line) for line in self.log.read_text().splitlines()] if self.log.exists() else [] + + def test_private_to_official_032_preserves_helper_grants(self): + self.assertEqual(self.upgrade_target("0.3.2"), []) + + def test_target_below_032_revokes_helper_grants(self): + calls = self.upgrade_target("0.3.1") + self.assertEqual(len(calls), 1) + self.assertEqual(calls[0][:3], ["-I", "-S", "-c"]) + self.assertIn('["purge_credential_helpers"]()', calls[0][3]) + + def test_same_private_version_preserves_helper_grants(self): + self.assertEqual(self.upgrade_target(PRIVATE_VERSION), []) + + def test_forward_upgrade_preserves_helper_grants(self): + self.assertEqual(self.upgrade_target("0.3.3"), []) + + def test_arbitrary_future_package_version_keeps_the_032_floor(self): + self.assertEqual(self.upgrade_target("0.3.2", package_version="9.9.9+private.1"), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_screen_registration.py b/source/tests/test_screen_registration.py index 89360a9..f36c4bb 100644 --- a/source/tests/test_screen_registration.py +++ b/source/tests/test_screen_registration.py @@ -3,6 +3,10 @@ The bounded executor is a fixture. This is not VM, GUI, TTY or input acceptance. """ import ast +import base64 +import json +import math +import shlex from pathlib import Path import unittest @@ -13,9 +17,26 @@ from test_screen_frontdoor import Executor, Image, MCP, NODE_ID PORTABLE = Path(__file__).resolve().parents[1] / "portable/front_door_mcp.py" +class TypedChunkExecutor(Executor): + """Model the explicitly negotiated typed route without changing legacy fixtures.""" + def __call__(self, **kwargs): + outer = super().__call__(**kwargs) + request = json.loads(base64.b64decode(shlex.split(kwargs["script"])[-1])) + if request["op"] == "read_chunk" and request.get("chunk_bytes"): + size, index = request["chunk_bytes"], request["index"] + envelope = json.loads(outer["stdout"]) + envelope["result"].update( + total_chunks=math.ceil(len(self.data) / size), + data=base64.b64encode(self.data[index * size:(index + 1) * size]).decode(), + more=(index + 1) * size < len(self.data), + ) + outer["stdout"] = json.dumps(envelope) + return outer + + class PortableScreenRegistrationTests(unittest.TestCase): def registration(self, target="LOCAL", corrupt=None): - fixture = Executor(hostname="fb-central-lab" if target == "LOCAL" else "fb-node-lab") + fixture = TypedChunkExecutor(hostname="fb-central-lab" if target == "LOCAL" else "fb-node-lab") calls, legacy_calls = [], [] def expanded(alias, script, authorized, timeout_s=120, job_id="", sha256="", include_stdout=False): @@ -51,6 +72,7 @@ class PortableScreenRegistrationTests(unittest.TestCase): framebridge_exec=legacy, register_framebridge_screen_tools=register_framebridge_screen_tools, resolve_profile_node=lambda node: node) + scope["_blocking_tools"] = scope["mcp"] # This fixture calls registrations synchronously. # Load the definition and registration from production source rather # than recreating the adapter inside this test. module = ast.Module(body=definitions + registrations, type_ignores=[]) diff --git a/source/tests/test_screen_transfer_budget.py b/source/tests/test_screen_transfer_budget.py new file mode 100644 index 0000000..c904c36 --- /dev/null +++ b/source/tests/test_screen_transfer_budget.py @@ -0,0 +1,431 @@ +"""Deterministic protocol fixtures only: never observe or control a desktop.""" +import base64 +import concurrent.futures +import contextlib +import hashlib +import gzip +import io +import json +import math +import os +from pathlib import Path +import shlex +import sys +import types +import tempfile +import threading +import unittest +from unittest.mock import patch + +from framebridge_expanded.real_screen import screen_rpc as rpc +from framebridge_expanded.real_screen import screen_transport as transport +from framebridge_expanded.real_screen.screen_frontdoor import ( + CHUNK_BYTES, EXPANDED_CHUNK_BYTES, EXPANDED_STDOUT_BYTES, + MAX_STDOUT_BYTES, RPC_COMMAND, ScreenClient, ScreenError, +) +from test_screen_frontdoor import CAPTURE_ID, NODE_ID, Image, png, Executor + + +class Clock: + now = 100.0 + + def __call__(self): + return self.now + + +class ProtocolFixture: + """Saved generated PNG fixture through actual producer, RPC and typed transport. + + The private records model a previously captured artifact for protocol tests; + they are generated test pixels and do not certify any real screen observation. + """ + def __init__(self, directory, width=384, height=256): + self.directory = directory + self.data = png(width, height) + self.manifest = { + 'capture_id': CAPTURE_ID, 'node': 'test-fixture-host', 'uid': os.getuid(), + 'source': 'linux-framebuffer', 'screen_type': 'tty', 'active_vt': 'tty1', + 'bytes': len(self.data), 'sha256': hashlib.sha256(self.data).hexdigest(), + 'width': width, 'height': height, 'synthetic': False, + 'pixel_source_verified': True, 'full_screen': True, 'authentic_pixels': True, + 'session_id': 'screen_fixture', 'requested_session_visible': False, + 'control_available': True, '_action_consumed': False, + 'binding': {'window_id': 'fixture-window', 'geometry': [0, 0, width, height]}, + } + path = directory / (CAPTURE_ID + '.png') + path.write_bytes(self.data) + path.chmod(0o600) + for suffix in ('.json', '.bound.json'): + rpc.write_record(directory / (CAPTURE_ID + suffix), self.manifest) + self.calls, self.reply_sizes = [], [] + self.rpc, self.transport = rpc, transport + self.variant = None + self.clock, self.latency = None, 0 + + def __call__(self, **kwargs): + request = json.loads(base64.b64decode(shlex.split(kwargs['script'])[-1])) + self.calls.append((request, kwargs['timeout_s'])) + if self.clock is not None: + self.clock.now += self.latency + payload = {key: kwargs[key] for key in ( + 'script', 'authorized', 'timeout_s', 'job_id', 'sha256', 'include_stdout')} + payload.update(version=1, mode='EXEC', request_type='screen_rpc') + + def runner(argument, timeout_s): + # No real observe/input functions are invoked by this fixture executor. + if request['op'] in ('key', 'text'): + manifest = self.rpc.public_manifest(self.manifest) + manifest['action'] = {'kind': request['op'], 'previous_capture_id': request['capture_id'], + 'result': {'performed': True}} + envelope = {'ok': True, 'result': manifest} + else: + envelope = self.rpc.dispatch(request, self.directory) + if request['op'] == 'read_chunk' and self.variant == 'corrupt': + encoded = envelope['result']['data'] + data = base64.b64decode(encoded) + envelope['result']['data'] = base64.b64encode(bytes([data[0] ^ 1]) + data[1:]).decode() + return 0, self.rpc.encode(envelope) + b'\n', b'' + + outer = self.transport.handle_request(payload, runner=runner) + outer.update(target=kwargs['target'], node_id=NODE_ID, + mcp_job_id=kwargs['job_id'], mcp_script_sha256=kwargs['sha256']) + self.reply_sizes.append(len(outer['stdout'].encode())) + if self.variant == 'truncated' and request['op'] == 'read_chunk': + outer['stdout_truncated'] = True + return outer + + +@contextlib.contextmanager +def frozen_receiver(testcase): + """Load the hashed base receiver fixture, including in exported source archives. + + The capsule contains only the five original real-screen modules from the + stated immutable revision. Its source hashes and outer digest are verified; + no Git checkout, installed receiver, or live desktop is used by this test. + """ + revision = 'f6b14c52109e0d55b919f179a38c8a4c8f8868f9' + path = Path(__file__).resolve().parent / 'fixtures/frozen_screen_receiver_f6b14c52.json.gz' + raw = path.read_bytes() + testcase.assertEqual(hashlib.sha256(raw).hexdigest(), '47670e1325c0f77fabeea9c69c09923c3fd0ed2463a8300a4940bb5e4bafffc5') + capsule = json.loads(gzip.decompress(raw)) + testcase.assertEqual(capsule['revision'], revision) + names = ('screen_frontdoor', 'screen_capture', 'screen_control', 'screen_rpc', 'screen_transport') + testcase.assertEqual(set(capsule['files']), set(names)) + package_name = '_framebridge_frozen_screen_fixture' + package = types.ModuleType(package_name) + package.__path__ = [] + with patch.dict(sys.modules, {package_name: package}): + modules = {} + for name in names: + entry = capsule['files'][name] + source = entry['source'] + testcase.assertEqual(hashlib.sha256(source.encode()).hexdigest(), entry['sha256']) + full_name = package_name + '.' + name + module = types.ModuleType(full_name) + module.__package__ = package_name + sys.modules[full_name] = module + exec(compile(source, revision + ':' + name + '.py', 'exec'), module.__dict__) + modules[name] = module + yield modules['screen_rpc'], modules['screen_transport'] + + +class ScreenTransferBudgetTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.directory = Path(self.temp.name) + self.fixture = ProtocolFixture(self.directory) + + def client(self, **kwargs): + return ScreenClient(self.fixture, Image, **kwargs) + + def read(self, client): + return client.call('read_capture', 'GERAMEG', True, capture_id=CAPTURE_ID) + + def test_expanded_complete_original_through_producer_rpc_and_transport(self): + result = self.read(self.client(stdout_limit_bytes=EXPANDED_STDOUT_BYTES)) + self.assertEqual(result[1].data, self.fixture.data) + self.assertEqual(hashlib.sha256(result[1].data).hexdigest(), self.fixture.manifest['sha256']) + self.assertEqual(len(self.fixture.calls), 1 + math.ceil(len(self.fixture.data) / EXPANDED_CHUNK_BYTES)) + self.assertTrue(all(r['chunk_bytes'] == EXPANDED_CHUNK_BYTES for r, _ in self.fixture.calls[1:])) + self.assertLessEqual(max(self.fixture.reply_sizes), EXPANDED_STDOUT_BYTES) + self.assertGreater(max(self.fixture.reply_sizes), MAX_STDOUT_BYTES) + # A full-sized serialized reply is measured, not estimated from base64 alone. + self.assertGreater(max(self.fixture.reply_sizes), 60000) + self.assertGreater(EXPANDED_STDOUT_BYTES - max(self.fixture.reply_sizes), 4096) + for key in ('capture_id', 'node', 'session_id', 'width', 'height', 'binding', 'requested_session_visible'): + self.assertEqual(result[0]['result'][key], self.fixture.manifest[key]) + self.assertEqual(result[0]['node_id'], NODE_ID) + legacy_calls = 1 + math.ceil(len(self.fixture.data) / CHUNK_BYTES) + self.assertGreater(legacy_calls, 10 * len(self.fixture.calls)) + + def test_legacy_omits_extension_and_stays_within_4096_bytes(self): + result = self.read(self.client()) + self.assertEqual(result[1].data, self.fixture.data) + self.assertEqual(len(self.fixture.calls), 1 + math.ceil(len(self.fixture.data) / CHUNK_BYTES)) + self.assertTrue(all('chunk_bytes' not in r for r, _ in self.fixture.calls)) + self.assertLessEqual(max(self.fixture.reply_sizes), MAX_STDOUT_BYTES) + + def test_new_central_with_actual_frozen_receiver_uses_legacy_schema(self): + with frozen_receiver(self) as (frozen_rpc, frozen_transport): + self.fixture.rpc, self.fixture.transport = frozen_rpc, frozen_transport + self.assertNotIn('max_chunk_bytes', frozen_rpc.public_manifest(self.fixture.manifest)) + # Only the capture itself is mocked: all frozen protocol and PNG reads + # execute the actual base code. Never access a live desktop in tests. + with patch.object(frozen_rpc, 'observe', return_value=self.fixture.manifest): + for op, arguments in [('observe', {}), ('read_capture', {'capture_id': CAPTURE_ID})]: + with self.subTest(op=op): + self.fixture.calls.clear() + result = self.client(stdout_limit_bytes=EXPANDED_STDOUT_BYTES).call( + op, 'GERAMEG', True, **arguments) + self.assertEqual(result[1].data, self.fixture.data) + self.assertEqual(result[0]['result']['sha256'], self.fixture.manifest['sha256']) + self.assertNotIn('max_chunk_bytes', result[0]['result']) + self.assertTrue(all('chunk_bytes' not in request for request, _ in self.fixture.calls)) + self.assertEqual(len(self.fixture.calls), 1 + math.ceil(len(self.fixture.data) / CHUNK_BYTES)) + self.assertLessEqual(max(self.fixture.reply_sizes), MAX_STDOUT_BYTES) + + def test_capability_is_exact_bounded_int_and_cannot_expand_transport_budget(self): + original = self.fixture.rpc.public_manifest + for capability in (True, 0, -1, EXPANDED_CHUNK_BYTES + 1, 1.5, '45000'): + with self.subTest(capability=capability), patch.object( + self.fixture.rpc, 'public_manifest', + side_effect=lambda value: dict(original(value), max_chunk_bytes=capability)): + self.fixture.calls.clear() + result = self.read(self.client(stdout_limit_bytes=EXPANDED_STDOUT_BYTES)) + self.assertEqual(result['error'], 'SCREEN_TRANSFER_CAPABILITY_INVALID') + self.assertEqual(len(self.fixture.calls), 1) + self.fixture.calls.clear() + result = self.read(self.client()) + self.assertEqual(result[1].data, self.fixture.data) + self.assertTrue(all('chunk_bytes' not in request for request, _ in self.fixture.calls)) + self.assertLessEqual(max(self.fixture.reply_sizes), MAX_STDOUT_BYTES) + + def test_ambiguous_submitted_input_failures_require_observe_without_claiming_performed(self): + variants = ('non_ok', 'truncated', 'malformed', 'binding') + for op, arguments in [('key', {'key': 'ENTER'}), ('text', {'text': 'fixture-only'})]: + for variant in variants: + with self.subTest(op=op, variant=variant): + calls = [] + fixture = Executor() + def execute(**kwargs): + calls.append(kwargs) + outer = fixture(**kwargs) + if variant == 'non_ok': + return {'ok': False, 'error': 'SCREEN_RPC_FAILED'} + if variant == 'truncated': + outer['stdout_truncated'] = True + elif variant == 'malformed': + outer['stdout'] = '{invalid fixture' + elif variant == 'binding': + outer['mcp_script_sha256'] = '0' * 64 + return outer + result = ScreenClient(execute, Image).call( + op, 'GERAMEG', True, capture_id=CAPTURE_ID, **arguments) + self.assertFalse(result['ok']) + self.assertTrue(result['observe_again']) + self.assertEqual(result['capture_id'], CAPTURE_ID) + self.assertNotIn('performed', result) + self.assertNotIn('capture_consumed', result) + self.assertEqual(len(calls), 1) + + def test_unconfirmed_input_image_integrity_failure_does_not_assert_performed(self): + for variant in ('dimensions', 'chunk_hash', 'image_hash'): + with self.subTest(variant=variant): + fixture = Executor(variant) + result = ScreenClient(fixture, Image).call( + 'key', 'GERAMEG', True, capture_id=CAPTURE_ID, key='ENTER') + self.assertFalse(result['ok']) + self.assertTrue(result['observe_again']) + self.assertNotIn('performed', result) + self.assertNotIn('capture_consumed', result) + self.assertEqual(sum(request['op'] == 'key' for request in fixture.requests), 1) + + def test_generic_rpc_exception_after_consumption_remains_uncertain(self): + attempts = [] + def execute(**kwargs): + request = shlex.split(kwargs['script'])[-1] + output = io.StringIO() + def interrupted(*args, **arguments): + attempts.append('fixture-input-attempt') + raise OSError('fixture interruption after consuming capture') + with patch.object(rpc, 'state_directory', return_value=self.directory), \ + patch.object(rpc, 'act', side_effect=interrupted), contextlib.redirect_stdout(output): + code = rpc.main([request]) + return {'ok': False, 'exit_code': str(code), 'target': kwargs['target'], 'node_id': NODE_ID, + 'mcp_job_id': kwargs['job_id'], 'mcp_script_sha256': kwargs['sha256'], + 'stdout': output.getvalue(), 'stdout_truncated': False} + result = ScreenClient(execute, Image).call( + 'key', 'GERAMEG', True, capture_id=CAPTURE_ID, key='ENTER') + self.assertEqual(result['error'], 'SCREEN_REQUEST_FAILED') + self.assertTrue(result['observe_again']) + self.assertNotIn('performed', result) + self.assertNotIn('capture_consumed', result) + self.assertEqual(attempts, ['fixture-input-attempt']) + self.assertTrue(rpc.load_capture(self.directory, CAPTURE_ID)['_action_consumed']) + + def test_chunk_rejection_after_unconfirmed_input_still_requires_observation(self): + fixture = Executor() + def execute(**kwargs): + result = fixture(**kwargs) + if fixture.requests[-1]['op'] == 'read_chunk': + result.update(ok=False, exit_code='1', stdout=json.dumps( + {'ok': False, 'error': {'code': 'UNKNOWN_REQUEST_FIELDS'}})) + return result + result = ScreenClient(execute, Image).call( + 'key', 'GERAMEG', True, capture_id=CAPTURE_ID, key='ENTER') + self.assertEqual(result['error'], 'UNKNOWN_REQUEST_FIELDS') + self.assertTrue(result['observe_again']) + self.assertNotIn('performed', result) + self.assertNotIn('capture_consumed', result) + self.assertEqual([request['op'] for request in fixture.requests], ['key', 'read_chunk']) + + def test_known_backend_rejection_keeps_original_meaning_without_uncertainty(self): + for op, arguments in [('key', {'key': 'ENTER'}), ('text', {'text': 'fixture-only'})]: + with self.subTest(op=op): + fixture = Executor('error_exit1') + result = ScreenClient(fixture, Image).call( + op, 'GERAMEG', True, capture_id=CAPTURE_ID, **arguments) + self.assertEqual(result['error'], 'SESSION_NOT_VISIBLE') + self.assertNotIn('observe_again', result) + self.assertNotIn('performed', result) + self.assertEqual(len(fixture.requests), 1) + + def test_truncation_and_corruption_never_return_partial_success(self): + for variant, error in [('truncated', 'SCREEN_STDOUT_INVALID'), ('corrupt', 'SCREEN_IMAGE_HASH_MISMATCH')]: + with self.subTest(variant=variant): + self.fixture.variant = variant + result = self.read(self.client(stdout_limit_bytes=EXPANDED_STDOUT_BYTES)) + self.assertIs(result['ok'], False) + self.assertEqual(result['error'], error) + self.assertNotIn('result', result) + + def test_monotonic_total_deadline_propagates_remaining_whole_seconds(self): + clock = Clock() + self.fixture.clock, self.fixture.latency = clock, 1.2 + with patch('framebridge_expanded.real_screen.screen_frontdoor.time.monotonic', clock): + result = self.read(self.client(stdout_limit_bytes=EXPANDED_STDOUT_BYTES, transfer_timeout_s=4)) + self.assertEqual(result['error'], 'SCREEN_TRANSFER_TIMEOUT') + self.assertEqual([t for _, t in self.fixture.calls], [4, 2, 1]) + self.assertLess(clock.now, 104) + self.assertNotIn('result', result) + + def test_late_executor_response_cannot_publish_stale_success(self): + clock = Clock() + self.fixture.clock, self.fixture.latency = clock, 6 + with patch('framebridge_expanded.real_screen.screen_frontdoor.time.monotonic', clock): + result = self.read(self.client(transfer_timeout_s=5)) + self.assertEqual(result['error'], 'SCREEN_TRANSFER_TIMEOUT') + self.assertEqual(len(self.fixture.calls), 1) + + def test_deadline_after_confirmed_input_preserves_no_replay_warning(self): + clock = Clock() + self.fixture.clock, self.fixture.latency = clock, 1.2 + with patch('framebridge_expanded.real_screen.screen_frontdoor.time.monotonic', clock): + result = self.client(transfer_timeout_s=3).call( + 'key', 'GERAMEG', True, capture_id=CAPTURE_ID, key='ENTER') + self.assertEqual(result['error'], 'SCREEN_TRANSFER_TIMEOUT') + self.assertTrue(result['performed']) + self.assertTrue(result['capture_consumed']) + self.assertTrue(result['observe_again']) + self.assertEqual(result['capture_id'], CAPTURE_ID) + + def test_executor_exception_during_input_requires_fresh_observation(self): + def failed(**kwargs): + raise TimeoutError('test fixture timeout') + result = ScreenClient(failed, Image).call( + 'key', 'GERAMEG', True, capture_id=CAPTURE_ID, key='ENTER') + self.assertFalse(result['ok']) + self.assertTrue(result['observe_again']) + self.assertNotIn('performed', result) + self.assertEqual(result['capture_id'], CAPTURE_ID) + + def test_chunk_extension_is_bounded_strict_and_legacy_default_remains(self): + request = {'op': 'read_chunk', 'capture_id': CAPTURE_ID, 'index': 0, + 'expected_sha256': self.fixture.manifest['sha256']} + self.assertEqual(rpc.dispatch(request, self.directory)['result']['chunk_bytes'], CHUNK_BYTES) + for value in (True, 0, -1, EXPANDED_CHUNK_BYTES + 1, 1.5, '45000'): + with self.subTest(value=value), self.assertRaisesRegex(rpc.ScreenRpcError, 'CHUNK_BOUNDS_INVALID'): + rpc.validate_request(dict(request, chunk_bytes=value)) + with self.assertRaisesRegex(rpc.ScreenRpcError, 'UNKNOWN_REQUEST_FIELDS'): + rpc.validate_request(dict(request, extra='invalid')) + + def test_client_stdout_budgets_reject_one_byte_over_exact_boundary(self): + for budget in (MAX_STDOUT_BYTES, EXPANDED_STDOUT_BYTES): + for over in (0, 1): + with self.subTest(budget=budget, over=over): + def execute(**kwargs): + envelope = {'ok': True, 'result': {'padding': ''}} + envelope['result']['padding'] = 'x' * (budget + over - len(rpc.encode(envelope))) + return {'ok': True, 'exit_code': '0', 'target': kwargs['target'], 'node_id': NODE_ID, + 'mcp_job_id': kwargs['job_id'], 'mcp_script_sha256': kwargs['sha256'], + 'stdout': rpc.encode(envelope).decode(), 'stdout_truncated': False} + client = ScreenClient(execute, Image, stdout_limit_bytes=budget) + if over: + with self.assertRaisesRegex(ScreenError, 'SCREEN_STDOUT_INVALID'): + client.remote('GERAMEG', {'op': 'status'}) + else: + self.assertIn('padding', client.remote('GERAMEG', {'op': 'status'})) + + def test_typed_transport_exact_serialized_boundary(self): + script = RPC_COMMAND + shlex.quote(base64.b64encode(rpc.encode({'op': 'status'})).decode()) + '\n' + payload = {'version': 1, 'mode': 'EXEC', 'authorized': True, 'script': script, + 'timeout_s': 30, 'job_id': 'test-screen-boundary', 'sha256': hashlib.sha256(script.encode()).hexdigest(), + 'include_stdout': True, 'request_type': 'screen_rpc'} + for over in (0, 1): + envelope = {'ok': True, 'result': {'padding': ''}} + envelope['result']['padding'] = 'x' * (EXPANDED_STDOUT_BYTES + over - len(rpc.encode(envelope)) - 1) + stdout = rpc.encode(envelope) + b'\n' + self.assertEqual(len(stdout), EXPANDED_STDOUT_BYTES + over) + result = transport.handle_request(payload, runner=lambda *args: (0, stdout, b'')) + self.assertEqual(result['ok'], not over) + if over: + self.assertIn('SCREEN_RPC_OUTPUT_LIMIT', result['stdout']) + + def test_main_serializes_competing_inputs_and_consumes_capture_once(self): + request = base64.b64encode(rpc.encode({'op': 'key', 'capture_id': CAPTURE_ID, 'key': 'ENTER'})).decode() + barrier = threading.Barrier(2) + def invoke(): + barrier.wait(timeout=2) + return rpc.main([request]) + with patch.object(rpc, 'state_directory', return_value=self.directory), \ + patch.object(rpc, 'act', return_value={'ok': True, 'performed': True}) as act, \ + patch.object(rpc, 'observe', return_value=dict(self.fixture.manifest)), \ + patch('builtins.print'), concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + outcomes = list(pool.map(lambda _: invoke(), range(2))) + self.assertEqual(sorted(outcomes), [0, 1]) + self.assertEqual(act.call_count, 1) + self.assertTrue(rpc.load_capture(self.directory, CAPTURE_ID)['_action_consumed']) + + def test_main_lock_covers_dispatch_in_competing_rpc_entries(self): + first_entered, second_entered, release = threading.Event(), threading.Event(), threading.Event() + count = 0 + mutex = threading.Lock() + def dispatch(*args): + nonlocal count + with mutex: + count += 1 + current = count + if current == 1: + first_entered.set() + self.assertTrue(release.wait(2)) + else: + second_entered.set() + return {'ok': True, 'result': {}} + request = base64.b64encode(rpc.encode({'op': 'status'})).decode() + with patch.object(rpc, 'state_directory', return_value=self.directory), \ + patch.object(rpc, 'dispatch', side_effect=dispatch), patch('builtins.print'), \ + concurrent.futures.ThreadPoolExecutor(max_workers=2) as pool: + first = pool.submit(rpc.main, [request]) + try: + self.assertTrue(first_entered.wait(2)) + second = pool.submit(rpc.main, [request]) + self.assertFalse(second_entered.wait(0.1)) + finally: + release.set() + self.assertEqual((first.result(timeout=2), second.result(timeout=2)), (0, 0)) + self.assertTrue(second_entered.is_set()) + + +if __name__ == '__main__': + unittest.main()