mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-10-14 19:27:09 -04:00
Juan MartíandKurt Himebauch
435beb7f19
* feat(profiles): add persistent per-game compatibility profiles * fix(profiles): show Windows component overrides Include per-game Windows component selections in the resolved profile summary so the session overlay marks those games as custom and shows the selected components. * fix(session): detect Proton games across runtime paths * fix(profiles): select the active Gamescope game * fix(profiles): improve per-game reset feedback * fix(profiles): align session components with game overrides * fix(session): prefer selected Steam game during refresh * fix(session): resume profile following when reopening games * chore: keep local dev testing outside the repository * test(session): remove local dev package path * fix(session): preserve profiles when refresh fails * fix(session): reject empty game profile reset scope * perf(session): cache drawer inventory by refresh revision * fix(profiles): align game presets and profile reset behavior * fix(profiles): preserve originals and normalize shortcut ids * fix(profiles): keep editor scope stable after rebase * fix(tests): host Compose tests in both build variants * fix(profiles): recover interrupted component selection saves --------- Co-authored-by: Kurt Himebauch <[email protected]>
721 lines
28 KiB
Python
Executable File
721 lines
28 KiB
Python
Executable File
#!/usr/bin/python3
|
|
"""The guest side of the agent bridge (docs/agent-control.md).
|
|
|
|
Started by droiddeck-session inside the session, so whatever it runs has the session's own
|
|
environment: DISPLAY, the games' Xwayland, XDG_RUNTIME_DIR, and the preloads every guest process
|
|
gets - which is what lets it reach the Steam client's DevTools port, open to guest processes only.
|
|
|
|
The app talks to it through files under $BL_LAUNCH_DIR/agent (the session root, the same path on
|
|
both sides):
|
|
|
|
req/<id>.json a request the app wrote (tmp + rename)
|
|
resp/<id>.json the answer, written the same way
|
|
hello.json this daemon is up: pid, version, what it can do
|
|
focus.json gamescope's focus and every server's top-level windows, rewritten on change
|
|
cdp-port the client's DevTools port, written by droiddeck-session
|
|
|
|
Requests: ping, exec, focus, cdp, quit. Each answer is {"ok": true, ...} or
|
|
{"ok": false, "error": {"code", "message"}}.
|
|
"""
|
|
|
|
import base64
|
|
import json
|
|
import math
|
|
import os
|
|
import selectors
|
|
import signal
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
import urllib.request
|
|
import urllib.parse
|
|
from pathlib import Path
|
|
|
|
VERSION = 1
|
|
OUTPUT_CAP = 256 * 1024
|
|
REQUEST_CAP = 256 * 1024
|
|
RESPONSE_CAP = 1024 * 1024
|
|
FOCUS_PROPS = (
|
|
"GAMESCOPE_FOCUSED_APP", "GAMESCOPE_FOCUSED_WINDOW", "GAMESCOPE_FOCUSABLE_APPS",
|
|
"GAMESCOPE_FOCUSABLE_WINDOWS", "GAMESCOPECTRL_BASELAYER_APPID",
|
|
)
|
|
WINDOW_PROPS = (
|
|
"STEAM_GAME", "WM_STATE", "_NET_WM_PID", "WM_CLASS", "_NET_WM_WINDOW_OPACITY", "STEAM_OVERLAY",
|
|
"GAMESCOPE_EXTERNAL_OVERLAY", "STEAM_INPUT_FOCUS", "_NET_WM_STATE",
|
|
)
|
|
MAP_STATE = ("unmapped", "unviewable", "viewable")
|
|
WM_STATES = {0: "withdrawn", 1: "normal", 3: "iconic"}
|
|
|
|
|
|
def log(message):
|
|
print("== agent: " + message, flush=True)
|
|
|
|
|
|
def write_json(path, value):
|
|
tmp = path.with_name(path.name + ".tmp")
|
|
with os.fdopen(os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_TRUNC | os.O_NOFOLLOW, 0o600), "w") as stream:
|
|
os.fchmod(stream.fileno(), 0o600)
|
|
json.dump(value, stream, separators=(",", ":"))
|
|
os.replace(tmp, path)
|
|
|
|
|
|
class AgentError(Exception):
|
|
def __init__(self, code, message):
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
# --- X11: just enough of the protocol for root properties, the window tree and events ---------
|
|
|
|
class X:
|
|
"""One connection to an Xwayland server. The rootfs has xprop but no python X library."""
|
|
|
|
def __init__(self, display):
|
|
num = display.lstrip(":").split(".")[0]
|
|
last = None
|
|
for address in ("\0/tmp/.X11-unix/X" + num, "/tmp/.X11-unix/X" + num):
|
|
try:
|
|
self.s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
|
|
self.s.settimeout(3)
|
|
self.s.connect(address)
|
|
break
|
|
except OSError as e:
|
|
last = e
|
|
self.s.close()
|
|
else:
|
|
raise last
|
|
self.s.sendall(struct.pack("<BxHHHHxx", 0x6C, 11, 0, 0, 0))
|
|
head = self._read(8)
|
|
if head[0] != 1:
|
|
raise OSError("X connection refused on " + display)
|
|
setup = self._read(struct.unpack_from("<H", head, 6)[0] * 4)
|
|
vendor = struct.unpack_from("<H", setup, 16)[0]
|
|
formats = setup[21]
|
|
self.root = struct.unpack_from("<I", setup, 32 + (vendor + 3) // 4 * 4 + formats * 8)[0]
|
|
self.events = []
|
|
self._atoms = {}
|
|
self._names = {}
|
|
|
|
def close(self):
|
|
try:
|
|
self.s.close()
|
|
except OSError:
|
|
pass
|
|
|
|
def _read(self, n):
|
|
buf = b""
|
|
while len(buf) < n:
|
|
chunk = self.s.recv(n - len(buf))
|
|
if not chunk:
|
|
raise OSError("X connection closed")
|
|
buf += chunk
|
|
return buf
|
|
|
|
def request(self, data):
|
|
"""Send one request and return its reply, or None for an X error (a window that went)."""
|
|
self.s.sendall(data)
|
|
while True:
|
|
head = self._read(32)
|
|
if head[0] == 0:
|
|
return None
|
|
if head[0] == 1:
|
|
return head + self._read(struct.unpack_from("<I", head, 4)[0] * 4)
|
|
self.events.append(head)
|
|
|
|
def next_event(self):
|
|
if self.events:
|
|
return self.events.pop(0)
|
|
while True:
|
|
head = self._read(32)
|
|
if head[0] > 1:
|
|
return head
|
|
|
|
def atom(self, name):
|
|
if name not in self._atoms:
|
|
n = name.encode()
|
|
pad = (4 - len(n) % 4) % 4
|
|
r = self.request(struct.pack("<BBHHxx", 16, 0, 2 + (len(n) + pad) // 4, len(n)) + n + b"\0" * pad)
|
|
self._atoms[name] = struct.unpack_from("<I", r, 8)[0] if r else 0
|
|
return self._atoms[name]
|
|
|
|
def atom_name(self, atom):
|
|
if atom not in self._names:
|
|
r = self.request(struct.pack("<BxHI", 17, 2, atom))
|
|
self._names[atom] = r[32:32 + struct.unpack_from("<H", r, 8)[0]].decode(errors="replace") if r else str(atom)
|
|
return self._names[atom]
|
|
|
|
def select(self, window, mask):
|
|
self.s.sendall(struct.pack("<BxHIII", 2, 4, window, 0x800, mask))
|
|
|
|
def children(self, window):
|
|
r = self.request(struct.pack("<BxHI", 15, 2, window))
|
|
if not r:
|
|
return []
|
|
n = struct.unpack_from("<H", r, 16)[0]
|
|
return list(struct.unpack_from("<%dI" % n, r, 32))
|
|
|
|
def attributes(self, window):
|
|
r = self.request(struct.pack("<BxHI", 3, 2, window))
|
|
return (r[26], r[27], struct.unpack_from("<H", r, 12)[0]) if r else None
|
|
|
|
def geometry(self, window):
|
|
r = self.request(struct.pack("<BxHI", 14, 2, window))
|
|
return struct.unpack_from("<hhHH", r, 12) if r else None
|
|
|
|
def prop(self, window, name, length=1024):
|
|
atom = self.atom(name)
|
|
if not atom:
|
|
return None
|
|
r = self.request(struct.pack("<BBHIIIII", 20, 0, 6, window, atom, 0, 0, length))
|
|
if not r or r[1] == 0:
|
|
return None
|
|
fmt, n = r[1], struct.unpack_from("<I", r, 16)[0]
|
|
data = r[32:32 + n * fmt // 8]
|
|
if fmt == 32:
|
|
return list(struct.unpack("<%dI" % n, data))
|
|
return data
|
|
|
|
|
|
def windows(x):
|
|
"""Top-level windows that are mapped or carry an app id: what gamescope picks focus from."""
|
|
out = []
|
|
for win in x.children(x.root):
|
|
attrs = x.attributes(win)
|
|
if not attrs:
|
|
continue
|
|
props = {name: x.prop(win, name) for name in WINDOW_PROPS}
|
|
if attrs[0] != 2 and not props["STEAM_GAME"]:
|
|
continue
|
|
geo = x.geometry(win) or (0, 0, 0, 0)
|
|
cls = props["WM_CLASS"]
|
|
state = props["WM_STATE"]
|
|
entry = {
|
|
"window": win,
|
|
"map": MAP_STATE[attrs[0]] if attrs[0] < 3 else attrs[0],
|
|
"overrideRedirect": bool(attrs[1]),
|
|
"inputOnly": attrs[2] == 2,
|
|
"geometry": [geo[0], geo[1], geo[2], geo[3]],
|
|
"appId": props["STEAM_GAME"][0] if props["STEAM_GAME"] else None,
|
|
"wmState": WM_STATES.get(state[0], state[0]) if state else None,
|
|
"pid": props["_NET_WM_PID"][0] if props["_NET_WM_PID"] else None,
|
|
"class": cls.split(b"\0")[-2].decode(errors="replace") if cls and b"\0" in cls else None,
|
|
}
|
|
if props["_NET_WM_WINDOW_OPACITY"]:
|
|
entry["opacity"] = props["_NET_WM_WINDOW_OPACITY"][0]
|
|
for key, name in (("overlay", "STEAM_OVERLAY"), ("externalOverlay", "GAMESCOPE_EXTERNAL_OVERLAY"),
|
|
("inputFocus", "STEAM_INPUT_FOCUS")):
|
|
if props[name]:
|
|
entry[key] = props[name][0]
|
|
if props["_NET_WM_STATE"]:
|
|
entry["netState"] = [x.atom_name(a).replace("_NET_WM_STATE_", "") for a in props["_NET_WM_STATE"]]
|
|
out.append(entry)
|
|
return out
|
|
|
|
|
|
def displays():
|
|
seen = []
|
|
for d in (os.environ.get("DISPLAY"), os.environ.get("STEAM_GAME_DISPLAY_0")):
|
|
if d and d not in seen:
|
|
seen.append(d)
|
|
return seen
|
|
|
|
|
|
def focus_snapshot():
|
|
"""gamescope's focus as Steam sees it (root properties on the client's server) and every
|
|
server's windows. GAMESCOPE_FOCUSABLE_WINDOWS is the list the client orders the base layer
|
|
from: a game missing there stays behind the client however healthy its window looks."""
|
|
shown = displays()
|
|
if not shown:
|
|
raise AgentError("NO_DISPLAY", "This session has no X display")
|
|
snap = {"t": int(time.time() * 1000), "displays": {}}
|
|
for i, d in enumerate(shown):
|
|
x = X(d)
|
|
try:
|
|
if i == 0:
|
|
values = {name: x.prop(x.root, name) for name in FOCUS_PROPS}
|
|
trip = values["GAMESCOPE_FOCUSABLE_WINDOWS"] or []
|
|
snap["focusedApp"] = (values["GAMESCOPE_FOCUSED_APP"] or [None])[0]
|
|
snap["focusedWindow"] = (values["GAMESCOPE_FOCUSED_WINDOW"] or [None])[0]
|
|
snap["focusableApps"] = values["GAMESCOPE_FOCUSABLE_APPS"] or []
|
|
snap["focusableWindows"] = [
|
|
{"window": trip[k], "appId": trip[k + 1], "pid": trip[k + 2]} for k in range(0, len(trip) - 2, 3)
|
|
]
|
|
snap["baselayerAppIds"] = values["GAMESCOPECTRL_BASELAYER_APPID"] or []
|
|
snap["displays"][d] = windows(x)
|
|
finally:
|
|
x.close()
|
|
return snap
|
|
|
|
|
|
class FocusWatcher(threading.Thread):
|
|
"""Rewrites focus.json when gamescope's focus properties change, and every few seconds when
|
|
the windows do (a window can change without any root property moving)."""
|
|
|
|
def __init__(self, path):
|
|
super().__init__(daemon=True, name="focus")
|
|
self.path = path
|
|
self.changed = threading.Event()
|
|
self.last = None
|
|
self.seq = 0
|
|
|
|
def run(self):
|
|
threading.Thread(target=self.listen, daemon=True, name="focus-events").start()
|
|
while True:
|
|
self.changed.wait(timeout=5)
|
|
self.changed.clear()
|
|
time.sleep(0.2) # let a burst of property changes settle
|
|
try:
|
|
snap = focus_snapshot()
|
|
except (OSError, AgentError) as e:
|
|
log("focus snapshot failed: %s" % e)
|
|
time.sleep(2)
|
|
continue
|
|
body = {k: v for k, v in snap.items() if k != "t"}
|
|
if body == self.last:
|
|
continue
|
|
self.last = body
|
|
self.seq += 1
|
|
snap["seq"] = self.seq
|
|
write_json(self.path, snap)
|
|
|
|
def listen(self):
|
|
while True:
|
|
try:
|
|
x = X(os.environ["DISPLAY"])
|
|
wanted = {x.atom(name) for name in FOCUS_PROPS}
|
|
x.select(x.root, 0x400000) # PropertyChangeMask
|
|
# Waits for events as long as there are none: the connect timeout left in place
|
|
# here dropped and remade the connection, and took a snapshot, every few seconds.
|
|
x.s.settimeout(None)
|
|
self.changed.set()
|
|
while True:
|
|
event = x.next_event()
|
|
if event[0] & 0x7F == 28 and struct.unpack_from("<I", event, 8)[0] in wanted:
|
|
self.changed.set()
|
|
except OSError as e:
|
|
log("focus events: %s; reconnecting" % e)
|
|
time.sleep(2)
|
|
|
|
|
|
# --- Steam client DevTools (CDP) -----------------------------------------------------------------
|
|
|
|
class WebSocket:
|
|
"""A minimal RFC 6455 client: text frames out, text frames in. Enough for CDP."""
|
|
|
|
def __init__(self, url, timeout):
|
|
target = urllib.parse.urlsplit(url)
|
|
if (target.scheme != "ws" or target.hostname not in ("127.0.0.1", "::1", "localhost")
|
|
or target.username is not None or target.password is not None or target.fragment):
|
|
raise AgentError("INVALID_TARGET", "DevTools targets must use a loopback WebSocket")
|
|
hostport = target.netloc
|
|
path = target.path or "/"
|
|
if target.query:
|
|
path += "?" + target.query
|
|
host = "127.0.0.1" if target.hostname == "localhost" else target.hostname
|
|
self.s = socket.create_connection((host, target.port or 80), timeout=timeout)
|
|
try:
|
|
self._handshake(path, hostport)
|
|
except Exception:
|
|
self.s.close()
|
|
raise
|
|
|
|
def _handshake(self, path, hostport):
|
|
key = base64.b64encode(os.urandom(16)).decode()
|
|
self.s.sendall((
|
|
"GET %s HTTP/1.1\r\nHost: %s\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
"Sec-WebSocket-Key: %s\r\nSec-WebSocket-Version: 13\r\n\r\n" % (path, hostport, key)
|
|
).encode())
|
|
head = b""
|
|
while b"\r\n\r\n" not in head:
|
|
chunk = self.s.recv(4096)
|
|
if not chunk:
|
|
raise OSError("DevTools closed the connection during the handshake")
|
|
head += chunk
|
|
if len(head) > 16 * 1024:
|
|
raise AgentError("RESPONSE_TOO_LARGE", "The DevTools handshake exceeds 16 KiB")
|
|
status = head.split(b"\r\n", 1)[0]
|
|
if b" 101 " not in status:
|
|
raise OSError("DevTools refused the WebSocket: " + status.decode(errors="replace"))
|
|
self.buffer = head.split(b"\r\n\r\n", 1)[1]
|
|
|
|
def _read(self, n):
|
|
while len(self.buffer) < n:
|
|
chunk = self.s.recv(65536)
|
|
if not chunk:
|
|
raise OSError("DevTools closed the connection")
|
|
self.buffer += chunk
|
|
data, self.buffer = self.buffer[:n], self.buffer[n:]
|
|
return data
|
|
|
|
def send(self, text):
|
|
payload = text.encode()
|
|
header = bytearray([0x81])
|
|
n = len(payload)
|
|
if n < 126:
|
|
header.append(0x80 | n)
|
|
elif n < 65536:
|
|
header.append(0x80 | 126)
|
|
header += struct.pack(">H", n)
|
|
else:
|
|
header.append(0x80 | 127)
|
|
header += struct.pack(">Q", n)
|
|
mask = os.urandom(4)
|
|
header += mask
|
|
self.s.sendall(bytes(header) + bytes(b ^ mask[i % 4] for i, b in enumerate(payload)))
|
|
|
|
def recv(self):
|
|
message = b""
|
|
while True:
|
|
b0, b1 = self._read(2)
|
|
n = b1 & 0x7F
|
|
if n == 126:
|
|
n = struct.unpack(">H", self._read(2))[0]
|
|
elif n == 127:
|
|
n = struct.unpack(">Q", self._read(8))[0]
|
|
if n > RESPONSE_CAP or len(message) + n > RESPONSE_CAP:
|
|
raise AgentError("RESPONSE_TOO_LARGE", "The DevTools response exceeds 1 MiB")
|
|
mask = self._read(4) if b1 & 0x80 else None
|
|
data = self._read(n)
|
|
if mask:
|
|
data = bytes(b ^ mask[i % 4] for i, b in enumerate(data))
|
|
op = b0 & 0x0F
|
|
if op == 8:
|
|
raise OSError("DevTools closed the WebSocket")
|
|
if op == 9: # ping
|
|
continue
|
|
if op in (0, 1, 2):
|
|
message += data
|
|
if b0 & 0x80:
|
|
return message.decode(errors="replace")
|
|
|
|
def close(self):
|
|
try:
|
|
self.s.close()
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def cdp_port(agent_dir):
|
|
try:
|
|
port = (agent_dir / "cdp-port").read_text().strip()
|
|
except OSError:
|
|
port = os.environ.get("BL_CDP_PORT", "")
|
|
if not port.isdigit():
|
|
raise AgentError("NO_DEVTOOLS", "The Steam client's DevTools port is not known (no Steam client in this session yet)")
|
|
return int(port)
|
|
|
|
|
|
def cdp_targets(agent_dir, timeout=5):
|
|
url = "http://127.0.0.1:%d/json" % cdp_port(agent_dir)
|
|
try:
|
|
with urllib.request.urlopen(url, timeout=timeout) as r:
|
|
data = r.read(RESPONSE_CAP + 1)
|
|
if len(data) > RESPONSE_CAP:
|
|
raise AgentError("RESPONSE_TOO_LARGE", "The DevTools response exceeds 1 MiB")
|
|
return json.loads(data.decode())
|
|
except OSError as e:
|
|
raise AgentError("DEVTOOLS_UNREACHABLE", "Could not reach the client's DevTools at %s: %s" % (url, e))
|
|
|
|
|
|
def cdp_eval(agent_dir, expression, target=None, await_promise=True, timeout=15):
|
|
targets = [t for t in cdp_targets(agent_dir) if t.get("webSocketDebuggerUrl")]
|
|
want = target or "SharedJSContext"
|
|
match = [t for t in targets if want in (t.get("title") or "") or want in (t.get("url") or "")]
|
|
if not match:
|
|
names = ", ".join(sorted({t.get("title") or t.get("url") or "?" for t in targets}))
|
|
raise AgentError("NO_TARGET", "No DevTools target matches '%s' (have: %s)" % (want, names))
|
|
ws = WebSocket(match[0]["webSocketDebuggerUrl"], timeout)
|
|
try:
|
|
ws.send(json.dumps({"id": 1, "method": "Runtime.evaluate", "params": {
|
|
"expression": expression, "returnByValue": True, "awaitPromise": await_promise,
|
|
"userGesture": True,
|
|
}}))
|
|
deadline = time.monotonic() + timeout
|
|
while time.monotonic() < deadline:
|
|
message = json.loads(ws.recv())
|
|
if message.get("id") != 1:
|
|
continue
|
|
if "error" in message:
|
|
raise AgentError("CDP_ERROR", json.dumps(message["error"]))
|
|
result = message.get("result", {})
|
|
if result.get("exceptionDetails"):
|
|
details = result["exceptionDetails"]
|
|
text = (details.get("exception") or {}).get("description") or details.get("text") or "exception"
|
|
raise AgentError("JS_EXCEPTION", text)
|
|
value = result.get("result", {})
|
|
return {"target": match[0].get("title"), "type": value.get("type"),
|
|
"value": value.get("value"), "description": value.get("description")}
|
|
raise AgentError("TIMEOUT", "No answer from DevTools in %ss" % timeout)
|
|
finally:
|
|
ws.close()
|
|
|
|
|
|
def steam_selection(agent_dir, timeout=5):
|
|
result = cdp_eval(
|
|
agent_dir,
|
|
"""(() => {
|
|
let node = document.activeElement;
|
|
while (node) {
|
|
for (const name of ["data-id", "data-appid"]) {
|
|
const value = node.getAttribute?.(name);
|
|
if (/^[1-9][0-9]{0,9}$/.test(value || "")) {
|
|
const appId = Number(value);
|
|
if (Number.isSafeInteger(appId) && appId <= 4294967295) return value;
|
|
}
|
|
}
|
|
node = node.parentElement;
|
|
}
|
|
return null;
|
|
})()""",
|
|
target="Steam Big Picture Mode",
|
|
timeout=timeout,
|
|
)
|
|
value = result.get("value")
|
|
if value is None:
|
|
return {"appId": None}
|
|
if not isinstance(value, str) or not value.isdigit() or not 0 < int(value) <= 4294967295:
|
|
raise AgentError("INVALID_SELECTION", "Steam returned an invalid selected app id")
|
|
return {"appId": int(value)}
|
|
|
|
|
|
# --- Game processes ------------------------------------------------------------------------------
|
|
|
|
def game_pids(app_id):
|
|
"""Our processes started for this Steam app (Steam sets SteamAppId / SteamGameId in their
|
|
environment, and Proton's wine processes inherit it)."""
|
|
wanted = [("SteamAppId=%s" % app_id).encode(), ("SteamGameId=%s" % app_id).encode()]
|
|
me = os.getpid()
|
|
pids = []
|
|
for entry in os.listdir("/proc"):
|
|
if not entry.isdigit() or int(entry) == me:
|
|
continue
|
|
try:
|
|
with open("/proc/%s/environ" % entry, "rb") as f:
|
|
env = f.read().split(b"\0")
|
|
except OSError:
|
|
continue
|
|
if any(w in env for w in wanted):
|
|
pids.append(int(entry))
|
|
return pids
|
|
|
|
|
|
def quit_game(agent_dir, app_id, timeout):
|
|
if not str(app_id).isdigit():
|
|
raise AgentError("INVALID_APP_ID", "appId must be a decimal Steam app id")
|
|
steps = []
|
|
if game_pids(app_id):
|
|
try:
|
|
cdp_eval(agent_dir, "SteamClient.Apps.TerminateApp(%s, false)" % json.dumps(str(app_id)), timeout=5)
|
|
steps.append("client")
|
|
except AgentError as e:
|
|
steps.append("client unavailable: %s" % e)
|
|
deadline = time.monotonic() + timeout
|
|
while game_pids(app_id) and time.monotonic() < deadline:
|
|
time.sleep(0.5)
|
|
left = game_pids(app_id)
|
|
if left:
|
|
steps.append("SIGTERM %s" % left)
|
|
for pid in left:
|
|
try:
|
|
os.kill(pid, signal.SIGTERM)
|
|
except OSError:
|
|
pass
|
|
end = time.monotonic() + 5
|
|
while game_pids(app_id) and time.monotonic() < end:
|
|
time.sleep(0.5)
|
|
left = game_pids(app_id)
|
|
if left:
|
|
steps.append("SIGKILL %s" % left)
|
|
for pid in left:
|
|
try:
|
|
os.kill(pid, signal.SIGKILL)
|
|
except OSError:
|
|
pass
|
|
time.sleep(1)
|
|
return {"appId": int(app_id), "steps": steps, "remaining": game_pids(app_id)}
|
|
|
|
|
|
# --- Requests ------------------------------------------------------------------------------------
|
|
|
|
def run_exec(req):
|
|
argv = req.get("argv")
|
|
if not isinstance(argv, list) or not argv or not all(isinstance(a, str) and "\0" not in a for a in argv):
|
|
raise AgentError("INVALID_REQUEST", "exec needs argv, a non-empty list of strings")
|
|
timeout = request_timeout(req, 30)
|
|
env = dict(os.environ)
|
|
extra_env = req.get("env") if req.get("env") is not None else {}
|
|
if not isinstance(extra_env, dict) or any("\0" in str(k) or "=" in str(k) or "\0" in str(v) for k, v in extra_env.items()):
|
|
raise AgentError("INVALID_REQUEST", "env must be an object with valid environment entries")
|
|
env.update({str(k): str(v) for k, v in extra_env.items()})
|
|
stdin = req.get("stdin")
|
|
if stdin is not None and not isinstance(stdin, str):
|
|
raise AgentError("INVALID_REQUEST", "stdin must be a string")
|
|
started = time.monotonic()
|
|
try:
|
|
proc = subprocess.Popen(
|
|
argv, cwd=req.get("cwd") or None, env=env, start_new_session=True,
|
|
stdin=subprocess.PIPE if stdin is not None else subprocess.DEVNULL,
|
|
stdout=subprocess.PIPE, stderr=subprocess.PIPE,
|
|
)
|
|
except OSError as e:
|
|
raise AgentError("EXEC_FAILED", "%s: %s" % (argv[0], e))
|
|
captured = {"stdout": bytearray(), "stderr": bytearray()}
|
|
truncated = {"stdout": False, "stderr": False}
|
|
pending = memoryview(stdin.encode()) if stdin is not None else memoryview(b"")
|
|
deadline = started + timeout
|
|
timed_out = False
|
|
try:
|
|
with selectors.DefaultSelector() as selector:
|
|
for stream, name in ((proc.stdout, "stdout"), (proc.stderr, "stderr")):
|
|
os.set_blocking(stream.fileno(), False)
|
|
selector.register(stream, selectors.EVENT_READ, name)
|
|
if proc.stdin is not None:
|
|
os.set_blocking(proc.stdin.fileno(), False)
|
|
if pending:
|
|
selector.register(proc.stdin, selectors.EVENT_WRITE, "stdin")
|
|
else:
|
|
proc.stdin.close()
|
|
while selector.get_map() or proc.poll() is None:
|
|
remaining = deadline - time.monotonic()
|
|
if remaining <= 0:
|
|
timed_out = True
|
|
break
|
|
for key, _ in selector.select(min(0.1, remaining)):
|
|
stream, name = key.fileobj, key.data
|
|
if name == "stdin":
|
|
try:
|
|
pending = pending[os.write(stream.fileno(), pending[:65536]):]
|
|
except BrokenPipeError:
|
|
pending = pending[len(pending):]
|
|
if not pending:
|
|
selector.unregister(stream)
|
|
stream.close()
|
|
else:
|
|
data = os.read(stream.fileno(), 65536)
|
|
if not data:
|
|
selector.unregister(stream)
|
|
stream.close()
|
|
continue
|
|
available = OUTPUT_CAP - len(captured[name])
|
|
captured[name].extend(data[:available])
|
|
truncated[name] |= len(data) > available
|
|
if timed_out:
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
proc.wait(timeout=5)
|
|
finally:
|
|
if proc.poll() is None:
|
|
try:
|
|
os.killpg(proc.pid, signal.SIGKILL)
|
|
except ProcessLookupError:
|
|
pass
|
|
proc.wait(timeout=5)
|
|
for stream in (proc.stdin, proc.stdout, proc.stderr):
|
|
if stream is not None:
|
|
stream.close()
|
|
|
|
def text(name):
|
|
return {"text": captured[name].decode(errors="replace"), "truncated": truncated[name]}
|
|
|
|
return {"exitCode": proc.returncode, "timedOut": timed_out,
|
|
"seconds": round(time.monotonic() - started, 3), "stdout": text("stdout"), "stderr": text("stderr")}
|
|
|
|
|
|
def request_timeout(req, default):
|
|
try:
|
|
timeout = float(req.get("timeout", default))
|
|
except (TypeError, ValueError):
|
|
raise AgentError("INVALID_REQUEST", "timeout must be a positive number no greater than 600 seconds")
|
|
if not math.isfinite(timeout) or not 0 < timeout <= 600:
|
|
raise AgentError("INVALID_REQUEST", "timeout must be a positive number no greater than 600 seconds")
|
|
return timeout
|
|
|
|
|
|
def handle(agent_dir, req):
|
|
kind = req.get("kind")
|
|
if kind == "ping":
|
|
return {"pid": os.getpid(), "version": VERSION, "displays": displays()}
|
|
if kind == "exec":
|
|
return run_exec(req)
|
|
if kind == "focus":
|
|
return focus_snapshot()
|
|
if kind == "cdp":
|
|
if req.get("op") == "targets":
|
|
return {"targets": [{k: t.get(k) for k in ("title", "url", "type")} for t in cdp_targets(agent_dir)]}
|
|
expression = req.get("expression")
|
|
if not isinstance(expression, str) or not expression:
|
|
raise AgentError("INVALID_REQUEST", "cdp eval needs an expression")
|
|
return cdp_eval(agent_dir, expression, req.get("target"), bool(req.get("awaitPromise", True)),
|
|
request_timeout(req, 15))
|
|
if kind == "steam-selection":
|
|
return steam_selection(agent_dir, request_timeout(req, 5))
|
|
if kind == "quit":
|
|
return quit_game(agent_dir, req.get("appId"), request_timeout(req, 15))
|
|
raise AgentError("UNKNOWN_REQUEST", "Unknown agent request '%s'" % kind)
|
|
|
|
|
|
def answer(agent_dir, path):
|
|
rid = path.stem
|
|
try:
|
|
with path.open("rb") as stream:
|
|
data = stream.read(REQUEST_CAP + 1)
|
|
if len(data) > REQUEST_CAP:
|
|
raise ValueError("The request exceeds 256 KiB")
|
|
req = json.loads(data)
|
|
if not isinstance(req, dict):
|
|
raise ValueError("The request must be a JSON object")
|
|
except (OSError, ValueError) as e:
|
|
req = None
|
|
result = {"ok": False, "error": {"code": "INVALID_REQUEST", "message": str(e)}}
|
|
finally:
|
|
try:
|
|
path.unlink()
|
|
except OSError:
|
|
pass
|
|
if req is not None:
|
|
try:
|
|
result = dict(handle(agent_dir, req), ok=True)
|
|
except AgentError as e:
|
|
result = {"ok": False, "error": {"code": e.code, "message": str(e)}}
|
|
except Exception as e: # an answer of some kind always goes back
|
|
result = {"ok": False, "error": {"code": "AGENT_FAILED", "message": "%s: %s" % (type(e).__name__, e)}}
|
|
write_json(agent_dir / "resp" / (rid + ".json"), result)
|
|
|
|
|
|
def serve(agent_dir, focus=True, poll=0.1):
|
|
(agent_dir / "req").mkdir(parents=True, exist_ok=True)
|
|
(agent_dir / "resp").mkdir(parents=True, exist_ok=True)
|
|
can_focus = focus and bool(os.environ.get("DISPLAY"))
|
|
if can_focus:
|
|
FocusWatcher(agent_dir / "focus.json").start()
|
|
write_json(agent_dir / "hello.json", {"pid": os.getpid(), "version": VERSION, "started": int(time.time() * 1000),
|
|
"displays": displays(), "focus": can_focus})
|
|
log("serving %s (focus %s)" % (agent_dir, "on" if can_focus else "off"))
|
|
taken = set()
|
|
while True:
|
|
for path in sorted((agent_dir / "req").glob("*.json")):
|
|
if path.name in taken:
|
|
continue
|
|
taken.add(path.name)
|
|
threading.Thread(target=lambda p=path: (answer(agent_dir, p), taken.discard(p.name)),
|
|
daemon=True, name="req").start()
|
|
time.sleep(poll)
|
|
|
|
|
|
def main(argv):
|
|
base = os.environ.get("BL_LAUNCH_DIR")
|
|
if not base:
|
|
print("BL_LAUNCH_DIR is not set", file=sys.stderr)
|
|
return 2
|
|
serve(Path(base) / "agent", focus="--no-focus" not in argv)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main(sys.argv[1:]))
|