#!/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 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 click_pixels(action, width, height, space): if space == "normalized_1000": x = integer(action.get("x"), 0, 1000) y = integer(action.get("y"), 0, 1000) return (x * (width - 1) + 500) // 1000, (y * (height - 1) + 500) // 1000 return integer(action.get("x"), 0, width - 1), integer(action.get("y"), 0, height - 1) 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") space = request.get("coordinateSpace", "pixels") if space not in ("pixels", "normalized_1000"): raise ValueError("Unknown coordinate space") 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": click_pixels(action, width, height, space) if action.get("button", "left") not in ("left", "middle", "right"): raise ValueError("Unknown mouse button") 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") space = request.get("coordinateSpace", "pixels") clicks = [] completed = 0 for action in actions: kind = action["type"] try: if kind == "click": button = {"left": "1", "middle": "2", "right": "3"}[action.get("button", "left")] x, y = click_pixels(action, width, height, space) run("xdotool", "mousemove", "--sync", str(x), str(y), "click", button) clicks.append({"supplied": [action["x"], action["y"]], "pixels": [x, y]}) 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"]) buffer = io.BytesIO() screenshot.save(buffer, format="PNG") print(json.dumps({ "observationId": str(uuid.uuid4()), "capturedAt": datetime.datetime.now(datetime.timezone.utc).isoformat(), "width": screenshot.width, "height": screenshot.height, "completedActions": completed, "coordinateSpace": space, "clicks": clicks, "image": base64.b64encode(buffer.getvalue()).decode(), })) if __name__ == "__main__": try: main() except Exception as error: print(str(error), file=sys.stderr) sys.exit(1)