Initial commit
This commit is contained in:
+163
@@ -0,0 +1,163 @@
|
||||
#!/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 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 == "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 == "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,
|
||||
"image": base64.b64encode(buffer.getvalue()).decode(),
|
||||
}))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
try:
|
||||
main()
|
||||
except Exception as error:
|
||||
print(str(error), file=sys.stderr)
|
||||
sys.exit(1)
|
||||
Reference in New Issue
Block a user