#!/usr/bin/env python3 import base64 import datetime import json import os from pathlib import Path import subprocess import sys import uuid import fcntl import tempfile import accessibility STATE = Path.home() / ".local/state/desktop-harness" def run(*args, input=None): return subprocess.run(args, input=input, check=True, capture_output=True, timeout=10).stdout def integer(value, low, high): if type(value) is not int or not low <= value <= high: raise ValueError(f"Expected integer between {low} and {high}") return value def validate(request, width, height): if not isinstance(request, dict) or not isinstance(request.get("actions"), list): raise ValueError("Request must contain an actions array") actions = request["actions"] if len(actions) > 20: raise ValueError("At most 20 actions per request") total_wait = 0 for action in actions: if not isinstance(action, dict): raise ValueError("Actions must be objects") kind = action.get("type") if kind == "click": integer(action.get("x"), 0, width - 1) integer(action.get("y"), 0, height - 1) if action.get("button", "left") not in ("left", "middle", "right"): raise ValueError("Unknown mouse button") elif kind == "click_target": if not isinstance(action.get("target"), str) or not action["target"].isdigit(): raise ValueError("Target must be a numbered label string") if not isinstance(action.get("observationId"), str): raise ValueError("click_target requires observationId") if action is not actions[0] or sum(a.get("type") == "click_target" for a in actions) > 1: raise ValueError("click_target must be the first and only target click; capture again before another") elif kind == "scroll": if action.get("direction") not in ("up", "down"): raise ValueError("Unknown scroll direction") integer(action.get("steps"), 1, 30) elif kind == "keys": keys = action.get("keys") if not isinstance(keys, list) or not 1 <= len(keys) <= 8: raise ValueError("Supply 1–8 key names") if any(not isinstance(key, str) or not key or not all(c.isalnum() or c == "_" for c in key) for key in keys): raise ValueError("Invalid X11 key name") elif kind == "text": if not isinstance(action.get("text"), str) or len(action["text"]) > 20000 or "\0" in action["text"]: raise ValueError("Text must contain at most 20000 characters and no NUL") elif kind == "wait": total_wait += integer(action.get("milliseconds"), 0, 5000) else: raise ValueError(f"Unsupported action: {kind}") if total_wait > 5000: raise ValueError("Total requested wait exceeds 5 seconds") return actions def install_autostart(): directory = Path.home() / ".config/autostart" directory.mkdir(parents=True, exist_ok=True) # Desktop Exec quoting is not shell quoting; percent signs are field codes. helper = str(Path(__file__).resolve()).replace("%", "%%") for character in ("\\", '"', "`", "$"): helper = helper.replace(character, "\\" + character) (directory / "desktop-harness-session.desktop").write_text( "[Desktop Entry]\n" "Type=Application\n" "Name=Desktop harness session registration\n" f'Exec=python3 "{helper}" --register-session\n' "Terminal=false\n" "OnlyShowIn=XFCE;\n" ) def save_session(environment): # Readers must see either the previous complete environment or the new one. with tempfile.NamedTemporaryFile(mode="w", dir=STATE, delete=False) as file: temporary = Path(file.name) try: json.dump(environment, file) file.close() temporary.replace(STATE / "session.json") finally: temporary.unlink(missing_ok=True) def main(): STATE.mkdir(parents=True, exist_ok=True) if sys.argv[1:] == ["--install-autostart"]: install_autostart() print("Installed XFCE session registration autostart") return if sys.argv[1:] == ["--register-session"]: if not os.environ.get("DISPLAY") or os.environ.get("XDG_SESSION_TYPE") == "wayland": raise ValueError("Run registration from a terminal inside the XFCE/X11 desktop") environment = {key: os.environ[key] for key in ("DISPLAY", "XAUTHORITY", "DBUS_SESSION_BUS_ADDRESS") if key in os.environ} save_session(environment) print("Registered graphical session") return if not (STATE / "session.json").exists(): raise ValueError("No desktop session registered. Log into XFCE or run desktop.py --register-session in an XFCE terminal.") environment = json.loads((STATE / "session.json").read_text()) os.environ.update(environment) with (STATE / "desktop.lock").open("w") as lock: fcntl.flock(lock, fcntl.LOCK_EX | fcntl.LOCK_NB) from PIL import ImageGrab import io import time screenshot = ImageGrab.grab(xdisplay=os.environ["DISPLAY"]) width, height = screenshot.size raw = sys.stdin.read(256001) if len(raw) > 256000: raise ValueError("Request too large") request = json.loads(raw) actions = validate(request, width, height) expected = request.get("expectedSize") if expected is not None and expected != [width, height]: raise ValueError("Display geometry changed; capture again before acting") completed = 0 for action in actions: kind = action["type"] try: if kind == "click": button = {"left": "1", "middle": "2", "right": "3"}[action.get("button", "left")] run("xdotool", "mousemove", "--sync", str(action["x"]), str(action["y"]), "click", button) elif kind == "click_target": saved = json.loads((STATE / "targets.json").read_text()) current = accessibility.collect(width, height) x, y = accessibility.resolve_target(saved, current, action["observationId"], action["target"], width, height) run("xdotool", "mousemove", "--sync", str(x), str(y), "click", "1") elif kind == "scroll": run("xdotool", "click", "--repeat", str(action["steps"]), "--delay", "50", "4" if action["direction"] == "up" else "5") elif kind == "keys": run("xdotool", "key", "--clearmodifiers", "+".join(action["keys"])) elif kind == "text": # xclip owns the selection in the background. Replacing the clipboard is intentional. subprocess.run(["xclip", "-selection", "clipboard"], input=action["text"].encode(), check=True, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=5) run("xdotool", "key", "--clearmodifiers", "ctrl+v") elif kind == "wait": time.sleep(action["milliseconds"] / 1000) completed += 1 except Exception as error: raise RuntimeError(f"Action {completed} failed after {completed} completed actions; partial effects possible: {error}") from error screenshot = ImageGrab.grab(xdisplay=os.environ["DISPLAY"]) accessible = accessibility.collect(screenshot.width, screenshot.height) observation_id = str(uuid.uuid4()) (STATE / "targets.json").write_text(json.dumps({ "observationId": observation_id, "size": list(screenshot.size), "targets": accessible["targets"], })) raw_buffer = io.BytesIO() screenshot.save(raw_buffer, format="PNG") buffer = io.BytesIO() accessibility.annotate(screenshot, accessible["targets"]).save(buffer, format="PNG") print(json.dumps({ "observationId": observation_id, "targets": accessible["targets"], "accessibilityWarning": accessible["warning"], "rawImage": base64.b64encode(raw_buffer.getvalue()).decode(), "capturedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "width": screenshot.width, "height": screenshot.height, "completedActions": completed, "image": base64.b64encode(buffer.getvalue()).decode(), })) if __name__ == "__main__": try: main() except Exception as error: print(str(error), file=sys.stderr) sys.exit(1)