diff --git a/source/expanded/framebridge_expanded/native_host.py b/source/expanded/framebridge_expanded/native_host.py index 1a33ebe..74d5795 100644 --- a/source/expanded/framebridge_expanded/native_host.py +++ b/source/expanded/framebridge_expanded/native_host.py @@ -5,7 +5,9 @@ import json import os from pathlib import Path import sys +import subprocess from . import display_context as display +from . import x11_text META = ("input_may_have_been_sent", "do_not_replay", "performed", "display_context") @@ -17,12 +19,15 @@ def _input_primitive(argv): "getwindowgeometry", "getwindowname", "getwindowpid", "search", "version", "help"} -def install_cua_guard(cua, *, env=None, checker=None): +def install_cua_guard(cua, *, env=None, checker=None, text_sender=None): env = os.environ if env is None else env checker = display.check if checker is None else checker original_run = cua._run original_handle = cua.handle_cua_message - state = {"active": False, "input_started": False, "failed_code": ""} + original_context = getattr(cua, "_action_context", None) + text_sender = x11_text.bounded_type_text if text_sender is None else text_sender + state = {"active": False, "input_started": False, "failed_code": "", "action_context": None, + "type_keys": False} def verify(): try: @@ -31,11 +36,60 @@ def install_cua_guard(cua, *, env=None, checker=None): state["failed_code"] = str(exc) raise cua.CuaTransportError(str(exc), str(exc)) from None + if original_context is not None: + def guarded_context(message): + context = original_context(message) + if (not isinstance(message, dict) or str(message.get("action", "")).strip().upper() != "TYPE" + or str(message.get("type_mode", "KEYS")).upper().strip() != "KEYS"): + return context + # Bind the original authorization/window/geometry gate, including + # when the caller omitted an explicit expected window identifier. + try: + bound = (int(context["before_id"]), + (int(context["width"]), int(context["height"]))) + if bound[0] <= 0 or min(bound[1]) <= 0: + raise ValueError() + state["action_context"] = bound + except (KeyError, TypeError, ValueError): + state["failed_code"] = "TYPE_PRIMITIVE_UNSUPPORTED" + raise cua.CuaTransportError("TYPE_PRIMITIVE_UNSUPPORTED", "TYPE_PRIMITIVE_UNSUPPORTED") from None + return context + cua._action_context = guarded_context + + def before_input(): + verify() + state["input_started"] = True + + def text_primitive(argv, timeout): + args = list(argv) + context = state.get("action_context") + if (len(args) != 7 or args[:6] != ["xdotool", "type", "--clearmodifiers", "--delay", "10", "--"] + or type(args[6]) is not str or context is None or not state.get("active")): + state["failed_code"] = "TYPE_PRIMITIVE_UNSUPPORTED" + raise cua.CuaTransportError("TYPE_PRIMITIVE_UNSUPPORTED", "TYPE_PRIMITIVE_UNSUPPORTED") + try: + text_sender(args[6], expected_window=context[0], expected_geometry=context[1], + before_input=before_input, timeout=timeout) + except x11_text.TextError as error: + state["failed_code"] = str(error) + raise cua.CuaTransportError(str(error), str(error)) from None + except cua.CuaTransportError: + raise + except Exception: + state["failed_code"] = "TYPE_NATIVE_FAILURE" + raise cua.CuaTransportError("TYPE_NATIVE_FAILURE", "TYPE_NATIVE_FAILURE") from None + return subprocess.CompletedProcess(args, 0, stdout="", stderr="") + def guard_run(run, track_input): def guarded(argv, timeout=15): # The captured pointer helper calls its base runner after an internal # query. Guard that final runner too, immediately before subprocess. verify() + args = list(argv) + if (track_input and len(args) >= 2 and Path(str(args[0])).name == "xdotool" + and (str(args[1]).lower() == "type" or + (state["type_keys"] and any(str(arg).lower() == "type" for arg in args[1:])))): + return text_primitive(args, timeout) if track_input and _input_primitive(argv): state["input_started"] = True return run(argv, timeout=timeout) @@ -47,7 +101,10 @@ def install_cua_guard(cua, *, env=None, checker=None): guarded_run = guard_run(original_run, primitive_run is None) def guarded_handle(message): - state.update(active=True, input_started=False, failed_code="") + state.update(active=True, input_started=False, failed_code="", action_context=None) + state["type_keys"] = (isinstance(message, dict) + and str(message.get("action", "")).strip().upper() == "TYPE" + and str(message.get("type_mode", "KEYS")).upper().strip() == "KEYS") verify() result = original_handle(message) verify() diff --git a/source/expanded/framebridge_expanded/x11_text.py b/source/expanded/framebridge_expanded/x11_text.py new file mode 100644 index 0000000..2ca2763 --- /dev/null +++ b/source/expanded/framebridge_expanded/x11_text.py @@ -0,0 +1,624 @@ +"""X11 text using only mappings in the validated current keyboard group. + +The 2016 libxdo character search is keycode-first across all groups. Its +sender then locks/restores groups around XTEST events. Here libxdo supplies +read-only mapping data; only XTestFakeKeyEvent emits input. No clipboard, +keyboard binding, group-lock or keyboard-map setter is used. +""" +from __future__ import annotations + +import ctypes as C +from dataclasses import dataclass +import hashlib +import os +from pathlib import Path +import json +import selectors +import signal +import subprocess +import sys +import time +from typing import Callable + + +class TextError(RuntimeError): + """Fixed public error code, without text, clipboard or credential contents.""" + + +@dataclass(frozen=True) +class Mapping: + code: int + group: int + symbol: int + modifiers: int + character: int + needs_binding: int = 0 + + +@dataclass(frozen=True) +class Snapshot: + group: int + locked_group: int + base_group: int + latched_group: int + mods: int + base_mods: int + latched_mods: int + locked_mods: int + window: int + width: int + height: int + keymap: str + held: tuple[int, ...] + + +@dataclass(frozen=True) +class Stroke: + code: int + modifiers: tuple[int, ...] + mask: int + character: int + + +def require(value, code): + if not value: + raise TextError(code) + + +def plan_text(text: str, mappings: tuple[Mapping, ...], snapshot: Snapshot, + modifier_keys: dict[int, int], lookup: Callable[[int, int, int], int]) -> tuple[Stroke, ...]: + """Plan the entire string before clearing modifiers or emitting any event.""" + require(type(text) is str and 0 < len(text) <= 4096 and "\0" not in text, + "TYPE_TEXT_INVALID") + # Group zero is the native context validated by this revision. Do not guess + # XTEST device mappings for another group or consume one-shot accessibility + # state. Unsupported contexts fail without changing that state. + require(snapshot.group == snapshot.locked_group == 0 and snapshot.base_group == 0 + and snapshot.latched_group == 0, "TYPE_KEYBOARD_GROUP_UNSUPPORTED") + require(snapshot.latched_mods == 0, "TYPE_LATCHED_MODIFIERS_UNSUPPORTED") + # Symbol lookup alone cannot prove that locked Ctrl/Alt/Super will not + # invoke an application shortcut. Only conventional Caps/Num lock bits + # participate in this text backend; held modifiers are refused natively. + require(not snapshot.locked_mods & ~(2 | 16), "TYPE_LOCKED_MODIFIERS_UNSUPPORTED") + allowed = 0 + for mask in modifier_keys: + allowed |= mask + choices: dict[int, Stroke] = {} + candidates = sorted(mappings, key=lambda item: (item.modifiers.bit_count(), item.code)) + for item in candidates: + if item.group != snapshot.group or item.needs_binding or item.modifiers & ~allowed: + continue + require(8 <= item.code <= 255 and 0 <= item.modifiers <= 255, + "TYPE_KEYMAP_INVALID") + character = lookup(item.code, snapshot.locked_mods | item.modifiers, snapshot.group) + if character and character not in choices: + modifiers = tuple(modifier_keys[mask] for mask in sorted(modifier_keys) + if item.modifiers & mask) + choices[character] = Stroke(item.code, modifiers, item.modifiers, character) + require(all(ord(character) in choices for character in text), + "TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP") + return tuple(choices[ord(character)] for character in text) + + +def _same_context(current: Snapshot, initial: Snapshot, *, cleared: bool): + require(current.window == initial.window, "TYPE_ACTIVE_WINDOW_CHANGED") + require((current.width, current.height) == (initial.width, initial.height), + "TYPE_GEOMETRY_CHANGED") + require((current.group, current.locked_group, current.base_group, current.latched_group) + == (initial.group, initial.locked_group, 0, 0), "TYPE_KEYBOARD_GROUP_CHANGED") + require(current.latched_mods == 0 and current.locked_mods == initial.locked_mods, + "TYPE_KEYBOARD_STATE_CHANGED") + require(current.keymap == initial.keymap, "TYPE_KEYMAP_CHANGED") + if cleared: + require(current.mods == current.locked_mods, "TYPE_MODIFIERS_NOT_CLEARED") + + +def type_text(text: str, *, expected_window: int | None = None, + expected_geometry: tuple[int, int] | None = None, + before_input: Callable[[], None], timeout: float = 20, + backend=None, clock=time.monotonic, sleep=time.sleep): + """Emit a prevalidated plan and conservatively restore all attempted keys. + + before_input is the existing native-host display/authority guard. It is + called immediately before every stroke and before the first state change. + Caller ownership of uncertainty remains unchanged: after this callback has + run, a failure must not invite replay or assert that nothing was performed. + """ + owned = backend is None + try: + backend = NativeBackend() if owned else backend + except (AttributeError, OSError) as error: + raise TextError("TYPE_NATIVE_LIBRARY_UNAVAILABLE") from error + initial = None + released_held: list[int] = [] + down: list[int] = [] + attempted = False + failure = None + restore_failed = False + try: + require(type(timeout) in (int, float) and 0 < timeout <= 120, "TYPE_TIMEOUT_INVALID") + deadline = clock() + timeout + initial = backend.snapshot() + require(initial.window > 0, "TYPE_ACTIVE_WINDOW_UNAVAILABLE") + require(expected_window is None or initial.window == expected_window, + "TYPE_ACTIVE_WINDOW_CHANGED") + require(expected_geometry is None or (initial.width, initial.height) == expected_geometry, + "TYPE_GEOMETRY_CHANGED") + strokes = plan_text(text, backend.mappings, initial, backend.modifier_keys, backend.lookup) + if hasattr(backend, "validate_strokes"): + backend.validate_strokes(strokes, initial.locked_mods) + require(clock() < deadline and len(strokes) * 0.01 < timeout, + "TYPE_BUDGET_TOO_SMALL") + # This validation is still before ANY release/press. In particular an + # unmapped final character cannot cause a typed prefix or modifier clear. + _same_context(backend.snapshot(), initial, cleared=False) + if initial.held: + before_input() + attempted = True + _same_context(backend.snapshot(), initial, cleared=False) + for code in initial.held: + released_held.append(code) # A failed call may still have acted. + backend.event(code, False) + backend.sync() + _same_context(backend.snapshot(), initial, cleared=True) + for stroke in strokes: + require(clock() < deadline, "TYPE_DEADLINE_EXCEEDED") + _same_context(backend.snapshot(), initial, cleared=True) + require(backend.lookup(stroke.code, initial.locked_mods | stroke.mask, + initial.group) == stroke.character, "TYPE_KEYMAP_CHANGED") + before_input() + attempted = True + # Permission can wait on the parent display guard. Revalidate + # after that wait, at the actual event boundary as well. + _same_context(backend.snapshot(), initial, cleared=True) + require(backend.lookup(stroke.code, initial.locked_mods | stroke.mask, + initial.group) == stroke.character, "TYPE_KEYMAP_CHANGED") + for code in stroke.modifiers: + down.append(code) + backend.event(code, True) + down.append(stroke.code) + backend.event(stroke.code, True) + backend.event(stroke.code, False) + down.remove(stroke.code) + for code in reversed(stroke.modifiers): + backend.event(code, False) + down.remove(code) + backend.sync() + sleep(0.01) + _same_context(backend.snapshot(), initial, cleared=True) + except BaseException as error: + failure = error + finally: + # Releases and restoration are cleanup, not text replay. Attempt all of + # them even if a prior event, guard or restore call failed. + for code in reversed(down): + try: + backend.event(code, False) + except BaseException: + restore_failed = True + for code in released_held: + try: + backend.event(code, True) + except BaseException: + restore_failed = True + if attempted: + try: + backend.sync() + final = backend.snapshot() + require(initial is not None and final.held == initial.held + and final.group == initial.group and final.locked_mods == initial.locked_mods, + "TYPE_MODIFIER_RESTORE_UNCERTAIN") + except BaseException: + restore_failed = True + if owned: + try: + backend.close() + except BaseException: + if attempted: + restore_failed = True + elif failure is None: + failure = TextError("TYPE_NATIVE_CLOSE_FAILED") + if restore_failed: + raise TextError("TYPE_MODIFIER_RESTORE_UNCERTAIN") from failure + if failure is not None: + raise failure + + +def bounded_type_text(text: str, *, expected_window: int, + expected_geometry: tuple[int, int], before_input, + timeout: float = 20): + """Bound native Xlib work by a child process and gate every input batch. + + The child never receives permission to emit until whole-text preflight has + passed. An ACK is conservative evidence of possible input. A stalled X + server can prevent cleanup; any such termination therefore remains uncertain + after an ACK and must never invite replay. + """ + require(type(timeout) in (int, float) and 0 < timeout <= 120, "TYPE_TIMEOUT_INVALID") + require(type(text) is str and 0 < len(text) <= 4096 and "\0" not in text, + "TYPE_TEXT_INVALID") + deadline = time.monotonic() + timeout + selector = selectors.DefaultSelector() + try: + process = subprocess.Popen([sys.executable, "-I", "-S", "-B", + str(Path(__file__).resolve()), "--worker"], + stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.DEVNULL, bufsize=0) + except BaseException: + selector.close() + raise + buffer = b"" + complete = False + try: + request = json.dumps({"text": text, "window": expected_window, + "geometry": expected_geometry, "timeout": timeout}, + ensure_ascii=True).encode("ascii") + b"\n" + # The request is bounded below PIPE_BUF by ASCII length for normal + # strings, but Unicode escaping can be larger: use a writable selector + # rather than allowing pipe backpressure to block the native host. + selector.register(process.stdin, selectors.EVENT_WRITE) + offset = 0 + while offset < len(request): + remaining = deadline - time.monotonic() + require(remaining > 0 and selector.select(remaining), "TYPE_WORKER_TIMEOUT") + offset += os.write(process.stdin.fileno(), request[offset:offset + 4096]) + selector.unregister(process.stdin) + selector.register(process.stdout, selectors.EVENT_READ) + while True: + remaining = deadline - time.monotonic() + require(remaining > 0 and selector.select(remaining), "TYPE_WORKER_TIMEOUT") + chunk = os.read(process.stdout.fileno(), 4096) + require(bool(chunk), "TYPE_WORKER_EXIT_UNCERTAIN") + buffer += chunk + require(len(buffer) <= 4096, "TYPE_WORKER_PROTOCOL_INVALID") + while b"\n" in buffer: + line, buffer = buffer.split(b"\n", 1) + if line == b"INPUT": + before_input() + require(os.write(process.stdin.fileno(), b"GO\n") == 3, + "TYPE_WORKER_PROTOCOL_INVALID") + elif line == b"OK": + require(not buffer, "TYPE_WORKER_PROTOCOL_INVALID") + complete = True + return + elif line.startswith(b"ERROR TYPE_"): + code = line[6:].decode("ascii", errors="strict") + require(len(code) <= 96 and all(c.isupper() or c == "_" for c in code), + "TYPE_WORKER_PROTOCOL_INVALID") + raise TextError(code) + else: + raise TextError("TYPE_WORKER_PROTOCOL_INVALID") + except (OSError, ValueError, UnicodeError) as error: + raise TextError("TYPE_WORKER_IO_UNCERTAIN") from error + finally: + selector.close() + if process.poll() is None: + if not complete: + process.terminate() # child handler runs restoration finally + try: + process.wait(timeout=0.5) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=0.5) + process.stdin.close() + process.stdout.close() + + +def _worker(): + # SIGTERM must unwind normal Python paths, including a child waiting for + # permission. If Xlib is stuck, the parent bounds cleanup and kills it. + def stopped(_number, _frame): + raise TextError("TYPE_WORKER_CANCELLED") + signal.signal(signal.SIGTERM, stopped) + def permission(): + sys.stdout.write("INPUT\n"); sys.stdout.flush() + require(sys.stdin.buffer.readline(16) == b"GO\n", "TYPE_WORKER_CANCELLED") + try: + raw = sys.stdin.buffer.readline(65537) + require(len(raw) <= 65536 and raw.endswith(b"\n"), "TYPE_WORKER_REQUEST_INVALID") + request = json.loads(raw) + type_text(request["text"], expected_window=request["window"], + expected_geometry=tuple(request["geometry"]), + before_input=permission, timeout=request["timeout"]) + print("OK", flush=True) + except BaseException as error: + code = str(error) if isinstance(error, TextError) else "TYPE_NATIVE_FAILURE" + print("ERROR " + code, flush=True) + + +# Public 2016 libxdo ABI. Debian preserves extra fields AFTER this prefix; +# deliberately never describe, write or depend on the rest of xdo_t. +class CharCode(C.Structure): + _fields_ = [("key", C.c_uint32), ("code", C.c_ubyte), ("symbol", C.c_ulong), + ("group", C.c_int), ("modmask", C.c_int), ("needs_binding", C.c_int)] + + +class XdoPrefix(C.Structure): + _fields_ = [("xdpy", C.c_void_p), ("display_name", C.c_void_p), + ("charcodes", C.POINTER(CharCode)), ("charcodes_len", C.c_int)] + + +class XkbState(C.Structure): + # Authoritative XKBstr.h, not the obsolete HTML manual's byte fields. + _fields_ = [("group", C.c_ubyte), ("locked_group", C.c_ubyte), + ("base_group", C.c_ushort), ("latched_group", C.c_ushort), + ("mods", C.c_ubyte), ("base_mods", C.c_ubyte), + ("latched_mods", C.c_ubyte), ("locked_mods", C.c_ubyte), + ("compat_state", C.c_ubyte), ("grab_mods", C.c_ubyte), + ("compat_grab_mods", C.c_ubyte), ("lookup_mods", C.c_ubyte), + ("compat_lookup_mods", C.c_ubyte), ("ptr_buttons", C.c_ushort)] + + +class ModifierMap(C.Structure): + _fields_ = [("max_keypermod", C.c_int), ("modifiermap", C.POINTER(C.c_ubyte))] + + +class SymMap(C.Structure): + _fields_ = [("kt_index", C.c_ubyte * 4), ("group_info", C.c_ubyte), + ("width", C.c_ubyte), ("offset", C.c_ushort)] + + +class ClientMapPrefix(C.Structure): + _fields_ = [("size_types", C.c_ubyte), ("num_types", C.c_ubyte), ("types", C.c_void_p), + ("size_syms", C.c_ushort), ("num_syms", C.c_ushort), ("syms", C.c_void_p), + ("key_sym_map", C.POINTER(SymMap))] + + +class ServerMapPrefix(C.Structure): + _fields_ = [("num_acts", C.c_ushort), ("size_acts", C.c_ushort), + ("acts", C.POINTER(C.c_ubyte * 8)), ("behaviors", C.c_void_p), + ("key_acts", C.POINTER(C.c_ushort))] + + +class KeyboardPrefix(C.Structure): + _fields_ = [("display", C.c_void_p), ("flags", C.c_ushort), ("device", C.c_ushort), + ("min_key_code", C.c_ubyte), ("max_key_code", C.c_ubyte), ("ctrls", C.c_void_p), + ("server", C.POINTER(ServerMapPrefix)), ("map", C.POINTER(ClientMapPrefix))] + + +def validate_actions(actions, strokes, modifier_keys, locked_mods=0): + """No group/latch/lock/redirect/pointer actions on any generated key.""" + modifier_masks = {code: mask for mask, code in modifier_keys.items()} + character_codes = {stroke.code for stroke in strokes} + codes = character_codes | {code for stroke in strokes for code in stroke.modifiers} + for code in codes: + require(code in actions, "TYPE_KEY_ACTIONS_UNAVAILABLE") + for action in actions[code]: + require(len(action) == 8, "TYPE_KEY_ACTIONS_INVALID") + if action[0] == 0: # NoAction + require(code in character_codes, "TYPE_KEY_ACTION_UNSUPPORTED") + continue + require(code not in character_codes and code in modifier_masks, + "TYPE_KEY_ACTION_UNSUPPORTED") + mask = modifier_masks[code] + # XKB ClearLocks only unlocks action modifiers (X.Org table16.2). + # Permit it only when those bits have no locked state to alter. + # No latch/lock/group/virtual modifier side effects are allowed. + require(action[0] == 1 and action[1] in (0, 1, 4, 5) and action[2] == mask + and action[3] == mask and not any(action[4:]), + "TYPE_KEY_ACTION_UNSUPPORTED") + require(not (action[1] & 1 and locked_mods & mask), + "TYPE_KEY_ACTION_WOULD_CLEAR_LOCK") + + +def trusted_library(name: str): + path = Path("/usr/lib/x86_64-linux-gnu") / name + try: + resolved = path.resolve(strict=True) + for part in [resolved, *resolved.parents]: + info = part.lstat() + require(info.st_uid == 0 and not info.st_mode & 0o022, + "TYPE_NATIVE_LIBRARY_UNTRUSTED") + require(resolved.is_file(), "TYPE_NATIVE_LIBRARY_UNTRUSTED") + return C.CDLL(str(resolved)) + except (OSError, ValueError) as error: + raise TextError("TYPE_NATIVE_LIBRARY_UNAVAILABLE") from error + + +class NativeBackend: + """Read-only XKB/libxdo inspection plus direct XTEST key events only.""" + + def __init__(self): + self.context = None + self.xdo = trusted_library("libxdo.so.3") + self.x11 = trusted_library("libX11.so.6") + self.xtst = trusted_library("libXtst.so.6") + require(C.sizeof(C.c_void_p) == 8 and C.sizeof(C.c_ulong) == 8 + and C.sizeof(CharCode) == 32 and C.sizeof(XkbState) == 18 + and C.sizeof(SymMap) == 8 and KeyboardPrefix.server.offset == 24 + and KeyboardPrefix.map.offset == 32 and ServerMapPrefix.key_acts.offset == 24 + and ClientMapPrefix.key_sym_map.offset == 32, + "TYPE_NATIVE_ABI_UNSUPPORTED") + self._bind() + require(self.xdo.xdo_version() == b"3.20160805.1", "TYPE_NATIVE_ABI_UNSUPPORTED") + self.context = self.xdo.xdo_new(None) + require(bool(self.context), "TYPE_DISPLAY_UNAVAILABLE") + try: + prefix = XdoPrefix.from_address(self.context) + require(prefix.xdpy and prefix.charcodes and 0 < prefix.charcodes_len <= 65536, + "TYPE_NATIVE_KEYMAP_INVALID") + self.display = prefix.xdpy + self.mappings = tuple(Mapping(item.code, item.group, item.symbol, item.modmask, + item.key, item.needs_binding) + for item in prefix.charcodes[:prefix.charcodes_len]) + require(all(8 <= item.code <= 255 and 0 <= item.group <= 3 + and 0 <= item.modifiers <= 255 and item.needs_binding in (0, 1) + for item in self.mappings), "TYPE_NATIVE_KEYMAP_INVALID") + self.characters = {item.symbol: item.character for item in self.mappings} + require(self.xtst.XTestQueryExtension(self.display, C.byref(C.c_int()), + C.byref(C.c_int()), C.byref(C.c_int()), + C.byref(C.c_int())) != 0, "TYPE_XTEST_UNAVAILABLE") + self.modifier_keys, self.modifier_codes, self.locking_codes, self.group_codes = self._modifiers() + except BaseException: + self.close() + raise + + def _bind(self): + def bind(lib, name, result, *args): + value = getattr(lib, name); value.restype = result; value.argtypes = list(args) + bind(self.xdo, "xdo_version", C.c_char_p) + bind(self.xdo, "xdo_new", C.c_void_p, C.c_char_p) + bind(self.xdo, "xdo_free", None, C.c_void_p) + bind(self.xdo, "xdo_get_active_window", C.c_int, C.c_void_p, C.POINTER(C.c_ulong)) + bind(self.x11, "XkbGetState", C.c_int, C.c_void_p, C.c_uint, C.POINTER(XkbState)) + bind(self.x11, "XkbLookupKeySym", C.c_int, C.c_void_p, C.c_ubyte, C.c_uint, + C.POINTER(C.c_uint), C.POINTER(C.c_ulong)) + bind(self.x11, "XkbKeycodeToKeysym", C.c_ulong, C.c_void_p, C.c_ubyte, C.c_int, C.c_int) + bind(self.x11, "XkbGetMap", C.POINTER(KeyboardPrefix), C.c_void_p, C.c_uint, C.c_uint) + bind(self.x11, "XkbFreeKeyboard", None, C.POINTER(KeyboardPrefix), C.c_uint, C.c_int) + bind(self.x11, "XQueryKeymap", C.c_int, C.c_void_p, C.c_void_p) + bind(self.x11, "XGetModifierMapping", C.POINTER(ModifierMap), C.c_void_p) + bind(self.x11, "XFreeModifiermap", C.c_int, C.POINTER(ModifierMap)) + bind(self.x11, "XDisplayKeycodes", C.c_int, C.c_void_p, C.POINTER(C.c_int), C.POINTER(C.c_int)) + bind(self.x11, "XGetKeyboardMapping", C.POINTER(C.c_ulong), C.c_void_p, C.c_ubyte, + C.c_int, C.POINTER(C.c_int)) + bind(self.x11, "XFree", C.c_int, C.c_void_p) + bind(self.x11, "XDefaultScreen", C.c_int, C.c_void_p) + bind(self.x11, "XDisplayWidth", C.c_int, C.c_void_p, C.c_int) + bind(self.x11, "XDisplayHeight", C.c_int, C.c_void_p, C.c_int) + bind(self.x11, "XSync", C.c_int, C.c_void_p, C.c_int) + bind(self.xtst, "XTestQueryExtension", C.c_int, C.c_void_p, + C.POINTER(C.c_int), C.POINTER(C.c_int), C.POINTER(C.c_int), C.POINTER(C.c_int)) + bind(self.xtst, "XTestFakeKeyEvent", C.c_int, C.c_void_p, C.c_uint, C.c_int, C.c_ulong) + + def _modifier_data(self): + pointer = self.x11.XGetModifierMapping(self.display) + require(bool(pointer), "TYPE_MODIFIER_MAP_UNAVAILABLE") + try: + width = pointer.contents.max_keypermod + require(0 < width <= 32, "TYPE_MODIFIER_MAP_INVALID") + values = tuple(pointer.contents.modifiermap[:8 * width]) + require(all(code == 0 or 8 <= code <= 255 for code in values), + "TYPE_MODIFIER_MAP_INVALID") + return width, values + finally: + self.x11.XFreeModifiermap(pointer) + + def _modifiers(self): + width, values = self._modifier_data() + safe, codes, locking, groups = {}, set(), set(), set() + for index in range(8): + for code in values[index * width:(index + 1) * width]: + if not code: + continue + codes.add(code) + symbol = self.x11.XkbKeycodeToKeysym(self.display, code, 0, 0) + if index == 0 and symbol in (0xFFE1, 0xFFE2): + safe.setdefault(1, code) + elif symbol == 0xFE03: # ISO_Level3_Shift, not a group switch. + safe.setdefault(1 << index, code) + if symbol in (0xFFE5, 0xFFE6, 0xFF7F, 0xFF14): + locking.add(code) + if symbol == 0xFF7E or 0xFE06 <= symbol <= 0xFE0F: + groups.add(code) + code_masks = {code: sum(1 << index for index in range(8) + if code in values[index * width:(index + 1) * width]) + for code in codes} + safe = {mask: code for mask, code in safe.items() if code_masks[code] == mask} + return safe, codes, locking, groups + + def _map_token(self): + low, high, width = C.c_int(), C.c_int(), C.c_int() + self.x11.XDisplayKeycodes(self.display, C.byref(low), C.byref(high)) + require(8 <= low.value <= high.value <= 255, "TYPE_NATIVE_KEYMAP_INVALID") + pointer = self.x11.XGetKeyboardMapping(self.display, low.value, + high.value - low.value + 1, C.byref(width)) + require(bool(pointer), "TYPE_NATIVE_KEYMAP_INVALID") + try: + require(0 < width.value <= 64, "TYPE_NATIVE_KEYMAP_INVALID") + data = C.string_at(pointer, (high.value - low.value + 1) * width.value * C.sizeof(C.c_ulong)) + finally: + self.x11.XFree(pointer) + action_data = b"".join(bytes((code, len(actions))) + self.behaviors[code] + b"".join(actions) + for code, actions in self._actions().items()) + return hashlib.sha256(data + bytes(self._modifier_data()[1]) + action_data).hexdigest() + + def _actions(self): + # KeySyms | KeyActions | KeyBehaviors, using the authoritative public + # XKBstr.h LP64 prefixes only. All indices are bounded before slicing. + pointer = self.x11.XkbGetMap(self.display, 2 | 16 | 32, 256) + require(bool(pointer), "TYPE_KEY_ACTIONS_UNAVAILABLE") + try: + keyboard = pointer.contents + require(8 <= keyboard.min_key_code <= keyboard.max_key_code <= 255 + and keyboard.server and keyboard.map, "TYPE_KEY_ACTIONS_INVALID") + server, client = keyboard.server.contents, keyboard.map.contents + require(client.key_sym_map and 0 <= server.num_acts <= server.size_acts <= 65535 + and (server.num_acts == 0 or server.acts), "TYPE_KEY_ACTIONS_INVALID") + result, behaviors = {}, {} + for code in range(keyboard.min_key_code, keyboard.max_key_code + 1): + entry = client.key_sym_map[code] + groups = entry.group_info & 0x0F + require(groups <= 4 and entry.width <= 32, "TYPE_KEY_ACTIONS_INVALID") + count = groups * entry.width + offset = server.key_acts[code] if server.key_acts else 0 + require(not offset or (0 < count and offset + count <= server.num_acts), + "TYPE_KEY_ACTIONS_INVALID") + result[code] = (tuple(bytes(server.acts[index]) for index in range(offset, offset + count)) + if offset else (bytes(8),)) + behaviors[code] = (C.string_at(server.behaviors + 2 * code, 2) + if server.behaviors else bytes(2)) + self.behaviors = behaviors + return result + finally: + self.x11.XkbFreeKeyboard(pointer, 0, True) + + def validate_strokes(self, strokes, locked_mods=0): + actions = self._actions() + codes = {stroke.code for stroke in strokes} | {code for stroke in strokes for code in stroke.modifiers} + require(all(self.behaviors.get(code) == bytes(2) for code in codes), + "TYPE_KEY_BEHAVIOR_UNSUPPORTED") + validate_actions(actions, strokes, self.modifier_keys, locked_mods) + + def snapshot(self): + state, window = XkbState(), C.c_ulong() + require(self.x11.XkbGetState(self.display, 256, C.byref(state)) == 0, + "TYPE_KEYBOARD_STATE_UNAVAILABLE") + require(self.xdo.xdo_get_active_window(self.context, C.byref(window)) == 0, + "TYPE_ACTIVE_WINDOW_UNAVAILABLE") + keys = C.create_string_buffer(32) + self.x11.XQueryKeymap(self.display, keys) + held = tuple(sorted(code for code in self.modifier_codes + if keys.raw[code // 8] & (1 << (code % 8)))) + # XKB actions (e.g. grp:alt_shift_toggle) can be attached even to + # ordinary-looking modifier keysyms. No modifier release is safe to + # infer from keysym alone. This revision therefore requires none held. + require(not held, "TYPE_HELD_MODIFIERS_UNSUPPORTED") + require(not state.base_mods or held, "TYPE_MODIFIER_STATE_UNSUPPORTED") + screen = self.x11.XDefaultScreen(self.display) + return Snapshot(state.group, state.locked_group, state.base_group, state.latched_group, + state.mods, state.base_mods, state.latched_mods, state.locked_mods, + window.value, self.x11.XDisplayWidth(self.display, screen), + self.x11.XDisplayHeight(self.display, screen), self._map_token(), held) + + def lookup(self, code, modifiers, group): + used, symbol = C.c_uint(), C.c_ulong() + if not self.x11.XkbLookupKeySym(self.display, code, modifiers | (group << 13), + C.byref(used), C.byref(symbol)): + return 0 + value = symbol.value + if value in self.characters: + return self.characters[value] + if 32 <= value <= 126 or 160 <= value <= 255: + return value + if 0x01000000 <= value <= 0x0110FFFF: + return value & 0xFFFFFF + return 0 + + def event(self, code, pressed): + require(8 <= code <= 255, "TYPE_KEYCODE_INVALID") + require(self.xtst.XTestFakeKeyEvent(self.display, code, bool(pressed), 0) != 0, + "TYPE_XTEST_EVENT_UNCERTAIN") + + def sync(self): + self.x11.XSync(self.display, 0) + + def close(self): + if self.context: + self.xdo.xdo_free(self.context) + self.context = None + + +if __name__ == "__main__": + if sys.argv[1:] == ["--worker"]: + _worker() diff --git a/source/tests/test_native_dispatch.py b/source/tests/test_native_dispatch.py index bc15690..97ea1d7 100644 --- a/source/tests/test_native_dispatch.py +++ b/source/tests/test_native_dispatch.py @@ -63,7 +63,7 @@ class NativeDispatchTests(unittest.TestCase): def test_postflight_failure_after_input_retains_uncertain_outcome(self): calls = [] - cua = self.fake(calls, lambda engine: engine._run(["xdotool", "type", "x"])) + cua = self.fake(calls, lambda engine: engine._run(["xdotool", "key", "x"])) state = nh.install_cua_guard(cua, env={}, checker=self.checker(3)) with self.assertRaises(CuaError): cua.handle_cua_message({}) diff --git a/source/tests/test_native_host_x11.py b/source/tests/test_native_host_x11.py new file mode 100644 index 0000000..d6f57c5 --- /dev/null +++ b/source/tests/test_native_host_x11.py @@ -0,0 +1,338 @@ +"""Native-host TYPE interception tests with an entirely fake CUA/display. + +No captured code, browser, display, library loader, or native input is used. +""" +import importlib.util +import ast +import ctypes as C +from pathlib import Path +import sys +import subprocess +from types import ModuleType, SimpleNamespace +import unittest +from unittest.mock import Mock + + +PACKAGE = "_framebridge_native_host_x11_test" +ROOT = Path(__file__).parent +if not (ROOT / "x11_text.py").is_file(): + ROOT = ROOT.parent / "expanded/framebridge_expanded" +package = ModuleType(PACKAGE) +package.__path__ = [str(ROOT)] +display = ModuleType(PACKAGE + ".display_context") + + +class DisplayContextError(RuntimeError): + pass + + +display.DisplayContextError = DisplayContextError +display.check = lambda env: None +display.describe = lambda env: {"identity": "fixture-owner"} +sys.modules[PACKAGE] = package +sys.modules[display.__name__] = display +text_spec = importlib.util.spec_from_file_location(PACKAGE + ".x11_text", ROOT / "x11_text.py") +text_module = importlib.util.module_from_spec(text_spec) +sys.modules[text_spec.name] = text_module +text_spec.loader.exec_module(text_module) +spec = importlib.util.spec_from_file_location(PACKAGE + ".native_host", ROOT / "native_host.py") +native_host = importlib.util.module_from_spec(spec) +sys.modules[spec.name] = native_host +spec.loader.exec_module(native_host) + + +class CuaTransportError(RuntimeError): + def __init__(self, code, detail=None): + super().__init__(code) + self.code = code + self.detail = detail + + +class FakeCua: + CuaTransportError = CuaTransportError + + def __init__(self, *, with_base=False): + self.calls = [] + self.context_calls = [] + self.context = {"before_id": 101, "width": 1920, "height": 1080} + self.argv = None + self.mutate_context_after_capture = False + self.token = object() + if with_base: + self._run_v3_base = self._primitive + + def _primitive(self, argv, timeout=15): + self.calls.append((list(argv), timeout)) + return self.token + + def _run(self, argv, timeout=15): + if hasattr(self, "_run_v3_base"): + return self._run_v3_base(argv, timeout=timeout) + return self._primitive(argv, timeout=timeout) + + def _action_context(self, message, *args, **kwargs): + self.context_calls.append((message, args, kwargs)) + return self.context + + def handle_cua_message(self, message): + self._action_context(message) + if self.mutate_context_after_capture: + self.context.update(before_id=999, width=800, height=600) + argv = self.argv + if argv is None: + if message.get("action") == "TYPE": + argv = ["xdotool", "type", "--clearmodifiers", "--delay", "10", "--", message["text"]] + else: + argv = ["xdotool", "getactivewindow"] + return self._run(argv, timeout=7) + + +class NativeHostX11Tests(unittest.TestCase): + def install(self, cua=None, sender=None, checker=None): + cua = cua or FakeCua() + sender = sender or Mock(return_value=None) + checker = checker or Mock() + state = native_host.install_cua_guard(cua, env={"DISPLAY": ":fixture"}, + checker=checker, text_sender=sender) + return cua, sender, checker, state + + def test_exact_type_intercept_carries_original_context_and_timeout(self): + cua, sender, _, state = self.install() + result = cua.handle_cua_message({"action": "TYPE", "text": "-S /run/test"}) + self.assertIsInstance(result, subprocess.CompletedProcess) + self.assertEqual(result.returncode, 0) + self.assertEqual(result.stdout, "") + self.assertEqual(result.stderr, "") + self.assertEqual(result.args, ["xdotool", "type", "--clearmodifiers", "--delay", "10", "--", "-S /run/test"]) + self.assertEqual(cua.calls, []) + self.assertEqual(len(cua.context_calls), 1) + args, kwargs = sender.call_args + self.assertEqual(args, ("-S /run/test",)) + self.assertEqual(kwargs["expected_window"], 101) + self.assertEqual(kwargs["expected_geometry"], (1920, 1080)) + self.assertEqual(kwargs["timeout"], 7) + self.assertTrue(callable(kwargs["before_input"])) + self.assertFalse(state["input_started"]) + + def test_captured_context_is_a_snapshot_not_a_mutable_alias(self): + cua = FakeCua() + cua.mutate_context_after_capture = True + cua, sender, _, _ = self.install(cua) + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + self.assertEqual(sender.call_args.kwargs["expected_window"], 101) + self.assertEqual(sender.call_args.kwargs["expected_geometry"], (1920, 1080)) + + def test_intercepts_when_captured_runner_has_a_base_runner(self): + cua, sender, _, _ = self.install(FakeCua(with_base=True)) + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + sender.assert_called_once() + self.assertEqual(cua.calls, []) + + def test_unexpected_type_arguments_fail_closed(self): + forms = [ + ["xdotool", "type", "--delay", "10", "--", "/"], + ["xdotool", "type", "--clearmodifiers", "--delay", "0", "--", "/"], + ["xdotool", "type", "--clearmodifiers", "--delay", "10", "--", "/", "extra"], + ["xdotool", "type", "--window", "999", "--", "/"], + ["xdotool", "TYPE", "--clearmodifiers", "--delay", "10", "--", "/"], + ] + for argv in forms: + with self.subTest(argv=argv): + cua = FakeCua() + cua.argv = argv + cua, sender, _, state = self.install(cua) + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + self.assertEqual(cua.calls, []) + sender.assert_not_called() + self.assertFalse(state["input_started"]) + + def test_missing_original_context_rejects_before_sender_or_runner(self): + cua = FakeCua() + cua.context = {} + cua, sender, _, state = self.install(cua) + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + sender.assert_not_called() + self.assertEqual(cua.calls, []) + self.assertFalse(state["input_started"]) + + def test_non_type_query_passes_through_and_stays_read_only(self): + cua, sender, _, state = self.install() + result = cua.handle_cua_message({"action": "QUERY"}) + self.assertIs(result, cua.token) + self.assertEqual(cua.calls, [(["xdotool", "getactivewindow"], 7)]) + sender.assert_not_called() + self.assertFalse(state["input_started"]) + + def test_non_type_query_with_invalid_action_context_still_passes_through(self): + cua = FakeCua() + cua.context = {"unrelated_query_context": True} + cua, sender, _, state = self.install(cua) + self.assertIs(cua.handle_cua_message({"action": "QUERY"}), cua.token) + self.assertEqual(cua.calls, [(["xdotool", "getactivewindow"], 7)]) + sender.assert_not_called() + self.assertFalse(state["input_started"]) + + def test_search_literal_type_is_a_read_only_query(self): + cua = FakeCua() + cua.context = {"query_context": True} + cua.argv = ["xdotool", "search", "--name", "type"] + cua, sender, _, state = self.install(cua) + self.assertIs(cua.handle_cua_message({"action": "QUERY"}), cua.token) + self.assertEqual(cua.calls, [(["xdotool", "search", "--name", "type"], 7)]) + sender.assert_not_called() + self.assertFalse(state["input_started"]) + + def test_clipboard_context_passes_through_without_clipboard_access(self): + cua = FakeCua() + cua.context = {"clipboard_context_is_not_type_keys_context": True} + cua.argv = ["xdotool", "key", "--clearmodifiers", "ctrl+v"] + cua, sender, _, state = self.install(cua) + result = cua.handle_cua_message({"action": "TYPE", "type_mode": "CLIPBOARD_EXACT", "text": "fixture"}) + self.assertIs(result, cua.token) + self.assertEqual(cua.calls, [(["xdotool", "key", "--clearmodifiers", "ctrl+v"], 7)]) + sender.assert_not_called() + self.assertTrue(state["input_started"]) + + def test_actual_frozen_type_consumes_completed_process_contract(self): + candidates = [Path(__file__).with_name("frozen_type_fixture.py")] + for parent in [ROOT, *ROOT.parents]: + candidates.append(parent / "sources/original-asus/framebridge-native/host/framebridge_cua_linux.py") + source = next((path for path in candidates if path.is_file()), None) + if source is None: + self.skipTest("captured original _type source is not available in this checkout") + tree = ast.parse(source.read_text()) + definitions = [node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == "_type"] + self.assertTrue(definitions, str(source)) + cua = FakeCua() + namespace = {"CuaTransportError": CuaTransportError, + "_run": lambda argv, timeout: cua._run(argv, timeout=timeout), + "_action_context": lambda message: cua._action_context(message), + "_evidence_lines": lambda context: []} + exec(compile(ast.Module(body=[definitions[-1]], type_ignores=[]), str(source), "exec"), namespace) + cua.handle_cua_message = namespace["_type"] + cua, sender, _, state = self.install(cua) + result = cua.handle_cua_message({"action": "TYPE", "text": "-S /run/fixture"}) + self.assertTrue(result["ok"]) + self.assertEqual(result["action"], "TYPE") + self.assertIn("RESULTADO_FINAL=CUA_TYPE_PASS", result["_frame_output"]) + self.assertEqual(sender.call_args.kwargs["timeout"], 20) + self.assertEqual(cua.calls, []) + + def test_non_type_input_keeps_existing_input_tracking(self): + cua = FakeCua() + cua.argv = ["xdotool", "key", "Return"] + cua, sender, _, state = self.install(cua) + self.assertIs(cua.handle_cua_message({"action": "KEY"}), cua.token) + self.assertEqual(cua.calls, [(["xdotool", "key", "Return"], 7)]) + sender.assert_not_called() + self.assertTrue(state["input_started"]) + + def test_sender_preflight_failure_reports_no_input(self): + # Use the same error class the native sender exports, without loading + # or instantiating its native backend. + sender = Mock(side_effect=text_module.TextError("TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP")) + cua, _, _, state = self.install(sender=sender) + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "\u2603"}) + self.assertFalse(state["input_started"]) + self.assertEqual(cua.calls, []) + response = native_host.decorate_response({"executed": False}, state, {}) + self.assertFalse(response["input_may_have_been_sent"]) + self.assertFalse(response["do_not_replay"]) + self.assertFalse(response["performed"]) + + def test_input_started_only_when_sender_calls_guard(self): + observed = [] + holder = {} + + def sender(text, **kwargs): + observed.append(holder["state"]["input_started"]) + kwargs["before_input"]() + observed.append(holder["state"]["input_started"]) + + cua, _, _, state = self.install(sender=sender) + holder["state"] = state + cua.handle_cua_message({"action": "TYPE", "text": "a"}) + self.assertEqual(observed, [False, True]) + self.assertEqual(cua.calls, []) + + def test_authority_failure_inside_sender_guard_sends_no_input(self): + gate = {"sender_active": False} + + def checker(env): + if gate["sender_active"]: + raise DisplayContextError("IDENTITY_CHANGED") + + def sender(text, **kwargs): + gate["sender_active"] = True + kwargs["before_input"]() + cua, _, _, state = self.install(sender=sender, checker=checker) + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + self.assertFalse(state["input_started"]) + self.assertEqual(cua.calls, []) + + def test_post_input_failure_has_do_not_replay_and_no_performed_false(self): + # The injected sender uses the module instance imported by native_host. + def sender(text, **kwargs): + kwargs["before_input"]() + raise text_module.TextError("TYPE_KEYMAP_CHANGED") + + cua, _, _, state = self.install(sender=sender) + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + response = native_host.decorate_response({"executed": False}, state, {}) + self.assertTrue(response["input_may_have_been_sent"]) + self.assertTrue(response["do_not_replay"]) + self.assertNotIn("performed", response) + self.assertEqual(cua.calls, []) + + def test_successful_callback_marks_performed_in_existing_response(self): + sender = Mock(side_effect=lambda text, **kwargs: kwargs["before_input"]()) + cua, _, _, state = self.install(sender=sender) + cua.handle_cua_message({"action": "TYPE", "text": "/"}) + response = native_host.decorate_response({"executed": True}, state, {}) + self.assertTrue(response["performed"]) + self.assertTrue(response["input_may_have_been_sent"]) + self.assertFalse(response["do_not_replay"]) + + def test_new_message_does_not_reuse_previous_action_context(self): + cua, sender, _, state = self.install() + cua.handle_cua_message({"action": "TYPE", "text": "first"}) + native_host.decorate_response({"executed": True}, state, {}) + cua.context = {} + with self.assertRaises(CuaTransportError): + cua.handle_cua_message({"action": "TYPE", "text": "second"}) + self.assertEqual(sender.call_count, 1) + self.assertEqual(cua.calls, []) + + def test_native_snapshot_rejects_held_shift_before_any_input(self): + backend = object.__new__(text_module.NativeBackend) + backend.display = 1 + backend.context = 1 + backend.modifier_codes = {50} + + def keymap(display_id, buffer): + data = bytearray(32) + data[50 // 8] = 1 << (50 % 8) + C.memmove(buffer, bytes(data), 32) + return 1 + + def active_window(context, result): + result._obj.value = 101 + return 0 + + backend.x11 = SimpleNamespace(XkbGetState=Mock(return_value=0), + XQueryKeymap=keymap, XDefaultScreen=Mock()) + backend.xdo = SimpleNamespace(xdo_get_active_window=active_window) + backend.xtst = SimpleNamespace(XTestFakeKeyEvent=Mock()) + with self.assertRaisesRegex(text_module.TextError, "^TYPE_HELD_MODIFIERS_UNSUPPORTED$"): + backend.snapshot() + backend.xtst.XTestFakeKeyEvent.assert_not_called() + backend.x11.XDefaultScreen.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_x11_actions.py b/source/tests/test_x11_actions.py new file mode 100644 index 0000000..27acd38 --- /dev/null +++ b/source/tests/test_x11_actions.py @@ -0,0 +1,237 @@ +"""XKB action/behavior validation with bounded in-memory public ABI tables.""" +import ctypes as C +from pathlib import Path +import sys +from types import SimpleNamespace +import unittest +from unittest.mock import Mock + +try: + from framebridge_expanded import x11_text as subject +except ModuleNotFoundError as error: + if error.name != "framebridge_expanded": + raise + import x11_text as subject +from test_x11_text import KeyboardFixture + + +NO_ACTION = bytes(8) +SHIFT_ACTION = bytes([1, 0, 1, 1, 0, 0, 0, 0]) +SLASH = subject.Stroke(61, (), 0, ord("/")) +UPPERCASE_A = subject.Stroke(38, (50,), 1, ord("A")) + + +class ActionTableFixture: + def __init__(self): + self.symmap = (subject.SymMap * 256)() + self.keyacts = (C.c_ushort * 256)() + self.actions = ((C.c_ubyte * 8) * 8)() + self.behaviors = (C.c_ubyte * 512)() + self.symmap[50].group_info = 1 + self.symmap[50].width = 1 + self.keyacts[50] = 1 + self.actions[1][:] = SHIFT_ACTION + self.symmap[61].group_info = 1 + self.symmap[61].width = 1 + self.keyacts[61] = 2 + self.actions[2][:] = NO_ACTION + self.server = subject.ServerMapPrefix(3, 8, self.actions, + C.addressof(self.behaviors), self.keyacts) + self.client = subject.ClientMapPrefix() + self.client.key_sym_map = self.symmap + self.keyboard = subject.KeyboardPrefix() + self.keyboard.min_key_code = 8 + self.keyboard.max_key_code = 61 + self.keyboard.server = C.pointer(self.server) + self.keyboard.map = C.pointer(self.client) + self.pointer = C.pointer(self.keyboard) + self.backend = object.__new__(subject.NativeBackend) + self.backend.display = 1 + self.backend.modifier_keys = {1: 50} + self.backend.x11 = SimpleNamespace(XkbGetMap=Mock(return_value=self.pointer), + XkbFreeKeyboard=Mock()) + + def assert_freed(self, case): + self.backend.x11.XkbFreeKeyboard.assert_called_once_with(self.pointer, 0, True) + + def validate_text_without_input(self, text): + keyboard = KeyboardFixture() + keyboard.validate_strokes = self.backend.validate_strokes + subject.type_text(text, backend=keyboard, before_input=lambda: None, + clock=lambda: 0, sleep=lambda _: None) + return keyboard + + +class ActionValidationTests(unittest.TestCase): + def test_us_slash_and_shift_are_safe_and_read_tables_are_freed(self): + fixture = ActionTableFixture() + fixture.backend.validate_strokes((SLASH, UPPERCASE_A)) + fixture.backend.x11.XkbGetMap.assert_called_once_with(1, 2 | 16 | 32, 256) + fixture.assert_freed(self) + + def test_modifier_map_flag_is_safe_for_exact_shift_mask(self): + subject.validate_actions({38: (NO_ACTION,), 50: (bytes([1, 4, 1, 1, 0, 0, 0, 0]),)}, + (UPPERCASE_A,), {1: 50}) + + def test_group_latch_lock_redirect_and_pointer_actions_reject_before_events(self): + # XKB action types 2=LATCH_MODS, 3=LOCK_MODS, 4=SET_GROUP, + # 5=LATCH_GROUP, 6=LOCK_GROUP, 7=MOVE_PTR, 17=REDIRECT_KEY. + for action_type in (2, 3, 4, 5, 6, 7, 17): + with self.subTest(action_type=action_type): + fixture = ActionTableFixture() + fixture.actions[2][0] = action_type + keyboard = KeyboardFixture() + keyboard.validate_strokes = fixture.backend.validate_strokes + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTION_UNSUPPORTED$"): + subject.type_text("/", backend=keyboard, before_input=lambda: None, + clock=lambda: 0, sleep=lambda _: None) + self.assertEqual(keyboard.events, []) + fixture.assert_freed(self) + + def test_character_key_may_not_set_modifiers(self): + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTION_UNSUPPORTED$"): + subject.validate_actions({61: (SHIFT_ACTION,)}, (SLASH,), {1: 50}) + + def test_modifier_action_requires_exact_mask_and_no_extra_effects(self): + actions = [bytes([1, 0, 4, 4, 0, 0, 0, 0]), + bytes([1, 2, 1, 1, 0, 0, 0, 0]), + bytes([1, 0, 1, 1, 1, 0, 0, 0]), + bytes([1, 0, 1, 1, 0, 1, 0, 0]), NO_ACTION] + for action in actions: + with self.subTest(action=action): + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTION_UNSUPPORTED$"): + subject.validate_actions({38: (NO_ACTION,), 50: (action,)}, + (UPPERCASE_A,), {1: 50}) + + def test_clearlocks_flags_one_and_five_preserve_unrelated_numlock(self): + for flags in (1, 5): + with self.subTest(flags=flags): + fixture = ActionTableFixture() + fixture.actions[1][1] = flags + keyboard = KeyboardFixture(locked=16) + keyboard.validate_strokes = fixture.backend.validate_strokes + subject.type_text("A/", backend=keyboard, before_input=lambda: None, + clock=lambda: 0, sleep=lambda _: None) + self.assertEqual("".join(keyboard.output), "A/") + self.assertEqual(keyboard.initial.locked_mods, 16) + self.assertNotIn(77, [code for code, _ in keyboard.events]) + + def test_clearlocks_cannot_clear_an_existing_lock_on_action_modifier(self): + for flags in (1, 5): + with self.subTest(flags=flags): + fixture = ActionTableFixture() + fixture.actions[1][1] = flags + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTION_WOULD_CLEAR_LOCK$"): + fixture.backend.validate_strokes((UPPERCASE_A,), locked_mods=1) + fixture.assert_freed(self) + + def test_action_missing_or_wrong_width_fails_closed(self): + for actions, error in [({}, "TYPE_KEY_ACTIONS_UNAVAILABLE"), + ({61: (bytes(7),)}, "TYPE_KEY_ACTIONS_INVALID")]: + with self.subTest(actions=actions): + with self.assertRaisesRegex(subject.TextError, "^" + error + "$"): + subject.validate_actions(actions, (SLASH,), {1: 50}) + + def test_all_groups_and_levels_are_checked_not_just_first_action(self): + fixture = ActionTableFixture() + fixture.symmap[61].group_info = 2 + fixture.symmap[61].width = 2 + fixture.server.num_acts = 6 + fixture.actions[5][0] = 6 # Unsafe fourth entry, reached by bounded read. + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTION_UNSUPPORTED$"): + fixture.backend.validate_strokes((SLASH,)) + fixture.assert_freed(self) + + def test_planned_character_or_modifier_nondefault_behavior_is_rejected(self): + for code in (38, 50, 61): + for behavior_byte in (0, 1): + with self.subTest(code=code, behavior_byte=behavior_byte): + fixture = ActionTableFixture() + fixture.behaviors[2 * code + behavior_byte] = 1 + keyboard = KeyboardFixture() + keyboard.validate_strokes = fixture.backend.validate_strokes + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_BEHAVIOR_UNSUPPORTED$"): + subject.type_text("A/", backend=keyboard, before_input=lambda: None, + clock=lambda: 0, sleep=lambda _: None) + self.assertEqual(keyboard.events, []) + fixture.assert_freed(self) + + def test_unplanned_behavior_does_not_reject_safe_plan(self): + fixture = ActionTableFixture() + fixture.behaviors[2 * 40] = 1 + fixture.backend.validate_strokes((SLASH,)) + fixture.assert_freed(self) + + def test_zero_action_offsets_and_missing_optional_tables_default_to_no_action(self): + fixture = ActionTableFixture() + fixture.server.key_acts = C.POINTER(C.c_ushort)() + fixture.server.acts = C.POINTER(C.c_ubyte * 8)() + fixture.server.num_acts = 0 + fixture.server.behaviors = None + actions = fixture.backend._actions() + self.assertEqual(actions[61], (NO_ACTION,)) + self.assertEqual(fixture.backend.behaviors[61], bytes(2)) + subject.validate_actions(actions, (SLASH,), {1: 50}) + fixture.assert_freed(self) + + def test_bad_action_offset_or_span_is_rejected_and_freed(self): + cases = [(3, 1, 1), (2, 2, 1), (1, 0, 1), (65535, 1, 1)] + for offset, width, groups in cases: + with self.subTest(offset=offset, width=width, groups=groups): + fixture = ActionTableFixture() + fixture.keyacts[61] = offset + fixture.symmap[61].width = width + fixture.symmap[61].group_info = groups + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTIONS_INVALID$"): + fixture.backend._actions() + fixture.assert_freed(self) + + def test_invalid_dimensions_are_rejected_before_table_read_and_freed(self): + for attribute, value in [("group_info", 5), ("width", 33)]: + with self.subTest(attribute=attribute): + fixture = ActionTableFixture() + setattr(fixture.symmap[61], attribute, value) + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTIONS_INVALID$"): + fixture.backend._actions() + fixture.assert_freed(self) + + def test_invalid_keyboard_range_or_required_pointer_is_rejected_and_freed(self): + for fault in ("range", "server", "client", "symmap", "actions", "count"): + with self.subTest(fault=fault): + fixture = ActionTableFixture() + if fault == "range": + fixture.keyboard.min_key_code = 7 + elif fault == "server": + fixture.keyboard.server = C.POINTER(subject.ServerMapPrefix)() + elif fault == "client": + fixture.keyboard.map = C.POINTER(subject.ClientMapPrefix)() + elif fault == "symmap": + fixture.client.key_sym_map = C.POINTER(subject.SymMap)() + elif fault == "actions": + fixture.server.acts = C.POINTER(C.c_ubyte * 8)() + else: + fixture.server.num_acts = 9 + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTIONS_INVALID$"): + fixture.backend._actions() + fixture.assert_freed(self) + + def test_null_keyboard_pointer_reports_unavailable_without_freeing_null(self): + fixture = ActionTableFixture() + fixture.backend.x11.XkbGetMap.return_value = C.POINTER(subject.KeyboardPrefix)() + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEY_ACTIONS_UNAVAILABLE$"): + fixture.backend._actions() + fixture.backend.x11.XkbFreeKeyboard.assert_not_called() + + def test_public_xkb_prefix_abi_offsets(self): + self.assertEqual(C.sizeof(subject.SymMap), 8) + self.assertEqual(subject.SymMap.offset.offset, 6) + self.assertEqual(subject.ClientMapPrefix.key_sym_map.offset, 32) + self.assertEqual(subject.ServerMapPrefix.acts.offset, 8) + self.assertEqual(subject.ServerMapPrefix.behaviors.offset, 16) + self.assertEqual(subject.ServerMapPrefix.key_acts.offset, 24) + self.assertEqual(subject.KeyboardPrefix.server.offset, 24) + self.assertEqual(subject.KeyboardPrefix.map.offset, 32) + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_x11_text.py b/source/tests/test_x11_text.py new file mode 100644 index 0000000..4db184d --- /dev/null +++ b/source/tests/test_x11_text.py @@ -0,0 +1,360 @@ +"""Behavioral X11 text regressions; no display, libraries, or input devices. + +The fixture is an independent US physical-key model. It deliberately places +the earlier LATAM slash before the US slash, matching the ASUS failure class. +""" +import ast +import ctypes as C +from dataclasses import replace +import inspect +from pathlib import Path +import sys +from types import SimpleNamespace +import textwrap +import unittest +from unittest.mock import Mock, patch + +try: + from framebridge_expanded import x11_text as subject +except ModuleNotFoundError as error: + if error.name != "framebridge_expanded": + raise + import x11_text as subject + + +SHIFT, CAPS, NUM = 1, 2, 16 + + +class KeyboardFixture: + def __init__(self, *, held=(), locked=0): + # Physical X11 evdev codes and the US unshifted/shifted characters. + rows = [ + (10, "1", "!"), (11, "2", "@"), (12, "3", "#"), + (13, "4", "$"), (14, "5", "%"), (15, "6", "^"), + (16, "7", "&"), (17, "8", "*"), (18, "9", "("), + (19, "0", ")"), (20, "-", "_"), (21, "=", "+"), + (34, "[", "{"), (35, "]", "}"), (47, ";", ":"), + (48, "'", '"'), (49, "`", "~"), (51, "\\", "|"), + (59, ",", "<"), (60, ".", ">"), (61, "/", "?"), + (65, " ", " "), + ] + for start, letters in [(24, "qwertyuiop"), (38, "asdfghjkl"), (52, "zxcvbnm")]: + rows.extend((start + index, letter, letter.upper()) + for index, letter in enumerate(letters)) + self.physical = {code: (lower, upper) for code, lower, upper in rows} + self.modifier_keys = {SHIFT: 50} + mappings = [subject.Mapping(16, 1, ord("/"), SHIFT, ord("/"))] + for code, lower, upper in rows: + mappings.append(subject.Mapping(code, 0, ord(lower), 0, ord(lower))) + if lower != upper: + mappings.append(subject.Mapping(code, 0, ord(upper), SHIFT, ord(upper))) + self.mappings = tuple(mappings) + self.pressed = set(held) + self.initial = subject.Snapshot(0, 0, 0, 0, locked, 0, 0, locked, + 101, 1920, 1080, "fixture-map-v1", tuple(sorted(held))) + self.events = [] + self.output = [] + self.snapshot_calls = 0 + self.sync_calls = 0 + self.sleeps = 0 + self.on_snapshot = None + self.on_sleep = None + self.fail_events = {} + self.closed = False + + def lookup(self, code, mask, group): + if group == 1 and code == 16: + return ord("/") if mask & SHIFT else ord("7") + if group != 0 or code not in self.physical: + return 0 + lower, upper = self.physical[code] + shifted = bool(mask & SHIFT) + if lower.isalpha(): + shifted ^= bool(mask & CAPS) + return ord(upper if shifted else lower) + + def snapshot(self): + self.snapshot_calls += 1 + mods = self.initial.locked_mods + if self.pressed & {50, 62}: + mods |= SHIFT + if 37 in self.pressed: + mods |= 4 + value = replace(self.initial, mods=mods, + base_mods=mods & ~self.initial.locked_mods, + held=tuple(sorted(self.pressed & {37, 50, 62}))) + return self.on_snapshot(value, self.snapshot_calls) if self.on_snapshot else value + + def event(self, code, pressed): + self.events.append((code, pressed)) + if pressed: + self.pressed.add(code) + else: + self.pressed.discard(code) + if pressed and code in self.physical: + mask = self.initial.locked_mods | (SHIFT if self.pressed & {50, 62} else 0) + value = self.lookup(code, mask, self.initial.group) + self.output.append(chr(value) if value else "\ufffd") + error = self.fail_events.get(len(self.events)) + if error: + raise error + + def sync(self): + self.sync_calls += 1 + + def sleep(self, duration): + self.sleeps += 1 + if self.on_sleep: + self.on_sleep(self.sleeps) + + def close(self): + self.closed = True + + +class TextBehaviorTests(unittest.TestCase): + def run_text(self, text, keyboard=None, guard=None, **kwargs): + keyboard = keyboard or KeyboardFixture() + subject.type_text(text, backend=keyboard, before_input=guard or (lambda: None), + clock=lambda: 0, sleep=keyboard.sleep, **kwargs) + return keyboard + + def assert_error(self, code, operation): + with self.assertRaises(subject.TextError) as caught: + operation() + self.assertEqual(str(caught.exception), code) + + def test_slash_uses_current_us_key_even_with_earlier_latam_entry(self): + keyboard = self.run_text("/") + self.assertEqual(keyboard.events, [(61, True), (61, False)]) + self.assertEqual("".join(keyboard.output), "/") + + def test_original_shell_looking_text_is_only_exact_data(self): + text = "exec tmux -S /run/user/1000/fbe-tty-probe/sudo-consoles/sudo/s-c451.sock attach-session -t fb-real-f86" + keyboard = self.run_text(text) + self.assertEqual("".join(keyboard.output), text) + self.assertEqual(sum(event == (61, True) for event in keyboard.events), text.count("/")) + + def test_unsupported_last_character_does_not_even_release_held_modifiers(self): + keyboard = KeyboardFixture(held=(37, 50)) + self.assert_error("TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP", + lambda: self.run_text("already representable/\U0001f642", keyboard)) + self.assertEqual(keyboard.events, []) + self.assertEqual(keyboard.pressed, {37, 50}) + + def test_nonzero_group_or_group_latch_is_rejected_before_events(self): + cases = [dict(group=1, locked_group=1), dict(group=2, locked_group=2), + dict(base_group=1), dict(latched_group=1), dict(locked_group=1)] + for changed in cases: + with self.subTest(changed=changed): + keyboard = KeyboardFixture(held=(50,)) + keyboard.initial = replace(keyboard.initial, **changed) + self.assert_error("TYPE_KEYBOARD_GROUP_UNSUPPORTED", lambda: self.run_text("a", keyboard)) + self.assertEqual(keyboard.events, []) + + def test_latched_modifier_rejected_without_consuming_it(self): + keyboard = KeyboardFixture() + keyboard.initial = replace(keyboard.initial, latched_mods=SHIFT) + self.assert_error("TYPE_LATCHED_MODIFIERS_UNSUPPORTED", lambda: self.run_text("A", keyboard)) + self.assertEqual(keyboard.events, []) + + def test_locked_control_alt_or_super_rejected_before_any_event(self): + for mask in (4, 8, 64): + with self.subTest(mask=mask): + keyboard = KeyboardFixture(locked=mask) + self.assert_error("TYPE_LOCKED_MODIFIERS_UNSUPPORTED", + lambda: self.run_text("a/", keyboard)) + self.assertEqual(keyboard.events, []) + self.assertEqual(keyboard.initial.locked_mods, mask) + + def test_mid_text_focus_change_stops_after_prefix_without_replay(self): + keyboard = KeyboardFixture() + keyboard.on_sleep = lambda count: setattr(keyboard, "initial", replace(keyboard.initial, window=999)) + self.assert_error("TYPE_ACTIVE_WINDOW_CHANGED", lambda: self.run_text("abc", keyboard)) + self.assertEqual(keyboard.events, [(38, True), (38, False)]) + self.assertEqual("".join(keyboard.output), "a") + + def test_mid_text_group_change_stops_and_marks_restoration_uncertain(self): + keyboard = KeyboardFixture() + keyboard.on_sleep = lambda count: setattr(keyboard, "initial", replace(keyboard.initial, group=1)) + self.assert_error("TYPE_MODIFIER_RESTORE_UNCERTAIN", lambda: self.run_text("ab", keyboard)) + self.assertEqual(keyboard.events, [(38, True), (38, False)]) + + def test_mid_text_keymap_change_stops_after_prefix(self): + keyboard = KeyboardFixture() + keyboard.on_sleep = lambda count: setattr(keyboard, "initial", replace(keyboard.initial, keymap="changed")) + self.assert_error("TYPE_KEYMAP_CHANGED", lambda: self.run_text("ab", keyboard)) + self.assertEqual("".join(keyboard.output), "a") + self.assertEqual(keyboard.events, [(38, True), (38, False)]) + + def test_mid_text_geometry_change_stops_after_prefix(self): + keyboard = KeyboardFixture() + keyboard.on_sleep = lambda count: setattr(keyboard, "initial", replace(keyboard.initial, width=800)) + self.assert_error("TYPE_GEOMETRY_CHANGED", lambda: self.run_text("ab", keyboard)) + self.assertEqual("".join(keyboard.output), "a") + + def test_guard_failure_before_first_mutation_has_zero_events(self): + keyboard = KeyboardFixture(held=(37,)) + guard = Mock(side_effect=subject.TextError("AUTHORITY_EXPIRED")) + self.assert_error("AUTHORITY_EXPIRED", lambda: self.run_text("a", keyboard, guard)) + self.assertEqual(keyboard.events, []) + guard.assert_called_once_with() + + def test_guard_failure_mid_text_retains_prefix_without_replay(self): + keyboard = KeyboardFixture() + guard = Mock(side_effect=[None, subject.TextError("AUTHORITY_EXPIRED")]) + self.assert_error("AUTHORITY_EXPIRED", lambda: self.run_text("ab", keyboard, guard)) + self.assertEqual(keyboard.events, [(38, True), (38, False)]) + self.assertEqual("".join(keyboard.output), "a") + + def test_focus_change_during_permission_wait_emits_zero_events(self): + keyboard = KeyboardFixture() + + def guard(): + keyboard.initial = replace(keyboard.initial, window=999) + + permission = Mock(side_effect=guard) + self.assert_error("TYPE_ACTIVE_WINDOW_CHANGED", lambda: self.run_text("/", keyboard, permission)) + permission.assert_called_once_with() + self.assertEqual(keyboard.events, []) + + def test_keymap_change_during_permission_wait_emits_zero_events(self): + keyboard = KeyboardFixture() + + def guard(): + keyboard.initial = replace(keyboard.initial, keymap="changed-during-ack") + + permission = Mock(side_effect=guard) + self.assert_error("TYPE_KEYMAP_CHANGED", lambda: self.run_text("/", keyboard, permission)) + permission.assert_called_once_with() + self.assertEqual(keyboard.events, []) + + def test_failed_keydown_is_released_and_held_modifiers_restored(self): + keyboard = KeyboardFixture(held=(37,)) + keyboard.fail_events[3] = subject.TextError("EVENT_FAILED") + self.assert_error("EVENT_FAILED", lambda: self.run_text("A", keyboard)) + self.assertEqual(keyboard.events, [(37, False), (50, True), (38, True), + (38, False), (50, False), (37, True)]) + self.assertEqual(keyboard.pressed, {37}) + + def test_cleanup_continues_after_one_cleanup_event_fails(self): + keyboard = KeyboardFixture(held=(37, 62)) + keyboard.fail_events[4] = subject.TextError("KEYDOWN_FAILED") + keyboard.fail_events[5] = subject.TextError("RELEASE_FAILED") + self.assert_error("TYPE_MODIFIER_RESTORE_UNCERTAIN", lambda: self.run_text("A", keyboard)) + self.assertEqual(keyboard.events[-4:], [(38, False), (50, False), (37, True), (62, True)]) + self.assertEqual(keyboard.pressed, {37, 62}) + + def test_partial_modifier_clear_is_restored_after_event_failure(self): + keyboard = KeyboardFixture(held=(37, 50)) + keyboard.fail_events[1] = subject.TextError("CLEAR_FAILED") + self.assert_error("CLEAR_FAILED", lambda: self.run_text("a", keyboard)) + self.assertEqual(keyboard.events, [(37, False), (37, True)]) + self.assertEqual(keyboard.pressed, {37, 50}) + + def test_caps_lock_uses_lookup_result_without_toggling_lock(self): + keyboard = self.run_text("aA/", KeyboardFixture(locked=CAPS)) + self.assertEqual("".join(keyboard.output), "aA/") + self.assertEqual(keyboard.events, [(50, True), (38, True), (38, False), + (50, False), (38, True), (38, False), + (61, True), (61, False)]) + self.assertEqual(keyboard.initial.locked_mods, CAPS) + + def test_num_lock_is_preserved_and_does_not_add_numlock_events(self): + keyboard = self.run_text("1/A", KeyboardFixture(locked=NUM)) + self.assertEqual("".join(keyboard.output), "1/A") + self.assertNotIn(77, [code for code, _ in keyboard.events]) + self.assertEqual(keyboard.initial.locked_mods, NUM) + + def test_held_modifiers_clear_once_and_restore_after_success(self): + keyboard = self.run_text("a", KeyboardFixture(held=(37, 62))) + self.assertEqual(keyboard.events, [(37, False), (62, False), (38, True), + (38, False), (37, True), (62, True)]) + self.assertEqual(keyboard.pressed, {37, 62}) + + def test_too_small_budget_rejects_before_events(self): + keyboard = KeyboardFixture(held=(37,)) + self.assert_error("TYPE_BUDGET_TOO_SMALL", lambda: self.run_text("abc", keyboard, timeout=.02)) + self.assertEqual(keyboard.events, []) + + def test_expected_window_or_geometry_mismatch_has_zero_events(self): + for kwargs, code in [({"expected_window": 202}, "TYPE_ACTIVE_WINDOW_CHANGED"), + ({"expected_geometry": (800, 600)}, "TYPE_GEOMETRY_CHANGED")]: + with self.subTest(kwargs=kwargs): + keyboard = KeyboardFixture() + self.assert_error(code, lambda: self.run_text("a", keyboard, **kwargs)) + self.assertEqual(keyboard.events, []) + + def test_layout_binding_required_entry_never_becomes_typeable(self): + keyboard = KeyboardFixture() + keyboard.mappings += (subject.Mapping(120, 0, 0x01002603, 0, 0x2603, 1),) + keyboard.physical[120] = ("\u2603", "\u2603") + self.assert_error("TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP", + lambda: self.run_text("\u2603", keyboard)) + self.assertEqual(keyboard.events, []) + + +class NativeBoundaryTests(unittest.TestCase): + def test_primary_header_abi_sizes_and_offsets(self): + self.assertEqual(C.sizeof(subject.CharCode), 32) + self.assertEqual({name: getattr(subject.CharCode, name).offset for name, _ in subject.CharCode._fields_}, + {"key": 0, "code": 4, "symbol": 8, "group": 16, + "modmask": 20, "needs_binding": 24}) + self.assertEqual(C.sizeof(subject.XkbState), 18) + self.assertEqual({name: getattr(subject.XkbState, name).offset for name, _ in subject.XkbState._fields_}, + {"group": 0, "locked_group": 1, "base_group": 2, + "latched_group": 4, "mods": 6, "base_mods": 7, + "latched_mods": 8, "locked_mods": 9, "compat_state": 10, + "grab_mods": 11, "compat_grab_mods": 12, "lookup_mods": 13, + "compat_lookup_mods": 14, "ptr_buttons": 16}) + self.assertEqual(subject.XdoPrefix.charcodes.offset, 16) + self.assertEqual(subject.XdoPrefix.charcodes_len.offset, 24) + self.assertEqual([name for name, _ in subject.XdoPrefix._fields_], + ["xdpy", "display_name", "charcodes", "charcodes_len"]) + + def test_only_native_input_binding_is_xtest_key_event(self): + tree = ast.parse(textwrap.dedent(inspect.getsource(subject.NativeBackend._bind))) + bindings = {call.args[1].value for call in ast.walk(tree) + if isinstance(call, ast.Call) and isinstance(call.func, ast.Name) + and call.func.id == "bind"} + self.assertEqual(bindings & {"xdo_enter_text_window", "xdo_send_keysequence_window", + "XkbLockGroup", "XkbLatchGroup", "XChangeKeyboardMapping", + "XkbSetMap", "XSetSelectionOwner", "XSendEvent"}, set()) + self.assertIn("XTestFakeKeyEvent", bindings) + self.assertEqual({name for name in bindings if name.startswith("xdo_")}, + {"xdo_version", "xdo_new", "xdo_free", "xdo_get_active_window"}) + + def test_unknown_libxdo_abi_rejected_without_context_or_display(self): + fake = SimpleNamespace(xdo_version=Mock(return_value=b"4.20260303.1"), xdo_new=Mock()) + with patch.object(subject, "trusted_library", return_value=fake), patch.object(subject.NativeBackend, "_bind"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_NATIVE_ABI_UNSUPPORTED$"): + subject.NativeBackend() + fake.xdo_new.assert_not_called() + + def test_untrusted_library_ancestor_is_rejected_before_load(self): + for uid, mode in [(1000, 0o755), (0, 0o777), (0, 0o775)]: + with self.subTest(uid=uid, mode=mode): + leaf = Mock() + leaf.lstat.return_value = SimpleNamespace(st_uid=0, st_mode=0o755) + ancestor = Mock() + ancestor.lstat.return_value = SimpleNamespace(st_uid=uid, st_mode=mode) + leaf.parents = [ancestor] + fake_path = Mock() + fake_path.__truediv__ = Mock(return_value=fake_path) + fake_path.resolve.return_value = leaf + with patch.object(subject, "Path", return_value=fake_path), patch.object(subject.C, "CDLL") as loader: + with self.assertRaisesRegex(subject.TextError, "^TYPE_NATIVE_LIBRARY_UNTRUSTED$"): + subject.trusted_library("libX11.so.6") + loader.assert_not_called() + + def test_missing_library_is_rejected_without_loader(self): + fake_path = Mock() + fake_path.__truediv__ = Mock(return_value=fake_path) + fake_path.resolve.side_effect = FileNotFoundError("fixture missing") + with patch.object(subject, "Path", return_value=fake_path), patch.object(subject.C, "CDLL") as loader: + with self.assertRaisesRegex(subject.TextError, "^TYPE_NATIVE_LIBRARY_UNAVAILABLE$"): + subject.trusted_library("libX11.so.6") + loader.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/source/tests/test_x11_worker.py b/source/tests/test_x11_worker.py new file mode 100644 index 0000000..79fa88c --- /dev/null +++ b/source/tests/test_x11_worker.py @@ -0,0 +1,237 @@ +"""Exercise bounded worker pipes with real Python-only child processes. + +The worker command is replaced at the Popen boundary. No native backend or +display is instantiated; request/permission/deadline behavior uses real pipes. +""" +from contextlib import contextmanager +import json +import os +from pathlib import Path +import subprocess +import sys +import time +import unittest +from unittest.mock import Mock, patch + +try: + from framebridge_expanded import x11_text as subject +except ModuleNotFoundError as error: + if error.name != "framebridge_expanded": + raise + import x11_text as subject + + +STUB = r''' +import json, signal, sys, time +mode = sys.argv[1] +if mode == "hang_unread": + time.sleep(10) + raise SystemExit(2) +request = json.loads(sys.stdin.buffer.readline()) +if request["window"] != 101 or request["geometry"] != [1920,1080]: + print("ERROR TYPE_FIXTURE_CONTEXT_INVALID", flush=True) + raise SystemExit(2) +if mode == "preflight": + print("ERROR TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP", flush=True) +elif mode == "exit": + raise SystemExit(2) +elif mode == "bad": + print("NOT A VALID MESSAGE", flush=True) +elif mode == "oversize": + print("x" * 5000, flush=True) +elif mode == "hang_before": + time.sleep(10) +else: + for index in range(2 if mode == "twice" else 1): + print("INPUT", flush=True) + if sys.stdin.buffer.readline() != b"GO\n": + raise SystemExit(3) + if mode == "hang_after": + time.sleep(10) + elif mode == "ignore_terminate": + signal.signal(signal.SIGTERM, signal.SIG_IGN) + time.sleep(10) + elif mode == "error_after": + print("ERROR TYPE_KEYMAP_CHANGED", flush=True) + else: + print("OK", flush=True) +''' + + +class BoundedWorkerTests(unittest.TestCase): + def test_selector_failure_does_not_spawn_worker(self): + with patch.object(subject.selectors, "DefaultSelector", side_effect=OSError("fixture")), \ + patch.object(subject.subprocess, "Popen") as spawn: + with self.assertRaises(OSError): + subject.bounded_type_text("/", expected_window=101, expected_geometry=(1920, 1080), + before_input=lambda: None) + spawn.assert_not_called() + + def test_spawn_failure_closes_selector(self): + selector = Mock() + with patch.object(subject.selectors, "DefaultSelector", return_value=selector), \ + patch.object(subject.subprocess, "Popen", side_effect=OSError("fixture")): + with self.assertRaises(OSError): + subject.bounded_type_text("/", expected_window=101, expected_geometry=(1920, 1080), + before_input=lambda: None) + selector.close.assert_called_once_with() + + @contextmanager + def child(self, mode): + real_popen = subprocess.Popen + children = [] + + def spawn(argv, **kwargs): + self.assertEqual(argv, [sys.executable, "-I", "-S", "-B", str(Path(subject.__file__).resolve()), "--worker"]) + child = real_popen([sys.executable, "-u", "-c", STUB, mode], **kwargs) + children.append(child) + return child + + try: + with patch.object(subject.subprocess, "Popen", side_effect=spawn): + yield children + finally: + for process in children: + if process.poll() is None: + process.kill() + process.wait(timeout=2) + self.assertIsNotNone(process.returncode) + self.assertTrue(process.stdin.closed) + self.assertTrue(process.stdout.closed) + + def invoke(self, guard, *, text="-S /run/fixture", timeout=.5): + return subject.bounded_type_text(text, expected_window=101, + expected_geometry=(1920, 1080), + before_input=guard, timeout=timeout) + + def test_preflight_failure_does_not_request_any_permission(self): + guard = Mock() + with self.child("preflight"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_TEXT_UNREPRESENTABLE_IN_CURRENT_GROUP$"): + self.invoke(guard) + guard.assert_not_called() + + def test_one_input_batch_requires_one_guard_and_ack(self): + guard = Mock() + with self.child("success"): + self.assertIsNone(self.invoke(guard)) + guard.assert_called_once_with() + + def test_each_input_batch_requires_a_fresh_guard(self): + guard = Mock() + with self.child("twice"): + self.invoke(guard) + self.assertEqual(guard.call_count, 2) + + def test_timeout_before_permission_has_zero_guard_calls(self): + guard = Mock() + started = time.monotonic() + with self.child("hang_before"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_TIMEOUT$"): + self.invoke(guard, timeout=.15) + self.assertLess(time.monotonic() - started, 2) + guard.assert_not_called() + + def test_timeout_after_permission_retains_possible_input_evidence(self): + guard = Mock() + started = time.monotonic() + with self.child("hang_after"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_TIMEOUT$"): + self.invoke(guard, timeout=.15) + self.assertLess(time.monotonic() - started, 2) + guard.assert_called_once_with() + + def test_guard_failure_terminates_child_waiting_for_ack(self): + guard = Mock(side_effect=RuntimeError("AUTHORITY_EXPIRED")) + started = time.monotonic() + with self.child("success"): + with self.assertRaisesRegex(RuntimeError, "^AUTHORITY_EXPIRED$"): + self.invoke(guard) + self.assertLess(time.monotonic() - started, 2) + guard.assert_called_once_with() + + def test_stalled_child_that_ignores_termination_is_killed(self): + guard = Mock() + started = time.monotonic() + with self.child("ignore_terminate") as children: + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_TIMEOUT$"): + self.invoke(guard, timeout=.15) + self.assertLess(time.monotonic() - started, 2) + guard.assert_called_once_with() + self.assertLess(children[0].returncode, 0) + + def test_partial_input_error_preserves_permission_evidence(self): + guard = Mock() + with self.child("error_after"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_KEYMAP_CHANGED$"): + self.invoke(guard) + guard.assert_called_once_with() + + def test_unknown_protocol_line_fails_without_permission(self): + guard = Mock() + with self.child("bad"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_PROTOCOL_INVALID$"): + self.invoke(guard) + guard.assert_not_called() + + def test_oversized_worker_line_fails_without_permission(self): + guard = Mock() + with self.child("oversize"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_PROTOCOL_INVALID$"): + self.invoke(guard) + guard.assert_not_called() + + def test_worker_exit_is_not_mistaken_for_success(self): + guard = Mock() + with self.child("exit"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_EXIT_UNCERTAIN$"): + self.invoke(guard) + guard.assert_not_called() + + def test_large_unicode_request_cannot_block_on_unread_pipe(self): + guard = Mock() + started = time.monotonic() + with self.child("hang_unread"): + with self.assertRaisesRegex(subject.TextError, "^TYPE_WORKER_TIMEOUT$"): + self.invoke(guard, text="\U0001f642" * 4096, timeout=.15) + self.assertLess(time.monotonic() - started, 2) + guard.assert_not_called() + + def test_main_dispatch_runs_only_after_backend_definition(self): + # Replace only the NativeBackend AST in memory. This executes the real + # module/worker order with a constructor that cannot call native code. + wrapper = r''' +import ast, pathlib, sys +source = pathlib.Path(sys.argv[1]).read_text() +tree = ast.parse(source) +replacement = ast.parse("class NativeBackend:\n def __init__(self):\n raise TextError('TYPE_FIXTURE_NATIVE_BOUNDARY')\n").body[0] +for index, node in enumerate(tree.body): + if isinstance(node, ast.ClassDef) and node.name == 'NativeBackend': + tree.body[index] = replacement +ast.fix_missing_locations(tree) +sys.argv = [sys.argv[1], '--worker'] +namespace = {'__name__': '__main__', '__file__': sys.argv[0]} +exec(compile(tree, sys.argv[0], 'exec'), namespace) +''' + request = json.dumps({"text": "a", "window": 101, + "geometry": [1920, 1080], "timeout": 1}) + "\n" + result = subprocess.run([sys.executable, "-c", wrapper, str(Path(subject.__file__))], + input=request, text=True, capture_output=True, timeout=2) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout, "ERROR TYPE_FIXTURE_NATIVE_BOUNDARY\n") + + def test_real_isolated_worker_bootstrap_reaches_safe_display_preflight(self): + guard = Mock() + # This display number cannot refer to the user's real session. Native + # preflight can read libraries but cannot emit a GUI event. + with patch.dict(os.environ, {"DISPLAY": ":99999"}): + with self.assertRaises(subject.TextError) as caught: + self.invoke(guard, timeout=1) + self.assertIn(str(caught.exception), {"TYPE_DISPLAY_UNAVAILABLE", + "TYPE_NATIVE_LIBRARY_UNAVAILABLE", + "TYPE_NATIVE_ABI_UNSUPPORTED"}) + guard.assert_not_called() + + +if __name__ == "__main__": + unittest.main()