gerardo/framebridge
1
0 Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Fix guarded X11 text mapping without group switching

Browse Source
This commit is contained in:
Gerardo Betancourt committed 2026-10-28 09:07:05 -04:00
1 parent 7a27f7fd55
commit b283999f5a
7 files changed
+1857 -4
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

source/expanded/framebridge_expanded/native_host.py
+60 -3
View File
@@ -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()
source/expanded/framebridge_expanded/x11_text.py
+624
View File
@@ -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()