1"""Guest half of the one-linux desktop tools.
2
3mcp_linux.py pipes this source plus a `main(request)` call into `python3 -` over
4SSH, so the guest needs no installed agent and always runs the current code.
5"""
6
7import base64
8import json
9import os
10import signal
11import struct
12import subprocess
13import tempfile
14import time
15
16UI_LINE_LIMIT = 1500
17
18
19def screen():
20 png = subprocess.run(["maim", "--hidecursor"], capture_output=True, check=True).stdout
21 width, height = struct.unpack(">II", png[16:24])
22 return {"png_b64": base64.b64encode(png).decode("ascii"), "w": width, "h": height}
23
24
25def output(*argv):
26 process = subprocess.run(argv, capture_output=True, text=True)
27 return process.stdout.strip() if process.returncode == 0 else None
28
29
30def active_window():
31 window = output("xdotool", "getactivewindow")
32 if not window:
33 return None
34 # Debian's xdotool predates getwindowclassname; WM_CLASS reads `= "inst", "Class"`.
35 wm_class = output("xprop", "-notype", "-id", window, "WM_CLASS") or ""
36 pid = output("xdotool", "getwindowpid", window)
37 return {"title": output("xdotool", "getwindowname", window) or "",
38 "class": wm_class.rsplit('"', 2)[-2] if wm_class.count('"') >= 2 else "",
39 "pid": int(pid) if pid else None}
40
41
42def run_script(script, timeout_ms, shot_delay_ms):
43 with tempfile.TemporaryFile() as out, tempfile.TemporaryFile() as err:
44 # Files, not pipes: an app the script starts in the background inherits
45 # these descriptors and would otherwise hold the SSH channel open.
46 process = subprocess.Popen(["bash", "-c", script], stdin=subprocess.DEVNULL,
47 stdout=out, stderr=err, start_new_session=True)
48 try:
49 code = process.wait(timeout_ms / 1000)
50 error = None
51 time.sleep(shot_delay_ms / 1000)
52 shot = screen()
53 except subprocess.TimeoutExpired:
54 shot = screen() # before the kill, while whatever blocked is on screen
55 os.killpg(process.pid, signal.SIGKILL)
56 process.wait()
57 code = None
58 error = "script timed out after %d ms, process group killed" % timeout_ms
59 out.seek(0)
60 err.seek(0)
61 return dict(shot, exit=code, error=error, win=active_window(),
62 stdout=out.read().decode("utf-8", "replace"),
63 stderr=err.read().decode("utf-8", "replace"))
64
65
66def ui_tree():
67 import gi
68 gi.require_version("Atspi", "2.0")
69 from gi.repository import Atspi, GLib
70
71 win = active_window()
72 desktop = Atspi.get_desktop(0)
73 apps = [desktop.get_child_at_index(i) for i in range(desktop.get_child_count())]
74 chosen = [app for app in apps if win and app.get_process_id() == win["pid"]] or apps
75 lines = []
76
77 def walk(node, depth):
78 if len(lines) >= UI_LINE_LIMIT:
79 return
80 states = node.get_state_set()
81 if depth and not states.contains(Atspi.StateType.SHOWING):
82 return
83 line = "%s%s %r" % (" " * depth, node.get_role_name(), node.get_name())
84 if depth:
85 try:
86 rect = node.get_extents(Atspi.CoordType.SCREEN)
87 line += " %d,%d,%d,%d" % (rect.x, rect.y, rect.width, rect.height)
88 except GLib.Error:
89 pass
90 text = node.get_text_iface()
91 if text:
92 value = Atspi.Text.get_text(text, 0, 200)
93 if value and value != node.get_name():
94 line += " = %r" % value
95 if states.contains(Atspi.StateType.FOCUSED):
96 line += " [focused]"
97 lines.append(line)
98 for i in range(node.get_child_count()):
99 walk(node.get_child_at_index(i), depth + 1)
100
101 for app in chosen:
102 walk(app, 0)
103 if len(lines) >= UI_LINE_LIMIT:
104 lines.append("... truncated at %d lines" % UI_LINE_LIMIT)
105 return {"win": win, "controls": "\n".join(lines)}
106
107
108def main(request):
109 request = json.loads(request)
110 verb = request["verb"]
111 if verb == "exec":
112 reply = run_script(request["script"], request["timeout_ms"],
113 request["shot_delay_ms"])
114 elif verb == "shot":
115 reply = dict(screen(), win=active_window())
116 elif verb == "ui":
117 reply = ui_tree()
118 else:
119 raise ValueError("unknown verb %r" % verb)
120 print(json.dumps(reply))