mirror of
https://github.com/Droid-Deck/DroidDeck.git
synced 2026-10-14 19:27:09 -04:00
340 lines
17 KiB
Python
340 lines
17 KiB
Python
"""The guest agent: its request files, exec, errors and the DevTools client, without a session."""
|
|
import base64
|
|
import hashlib
|
|
import importlib.machinery
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import socket
|
|
import struct
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
from unittest.mock import patch
|
|
import tracemalloc
|
|
|
|
SCRIPT = Path(__file__).resolve().parents[1] / "linuxfs/overlay/usr/local/bin/droiddeck-agent"
|
|
loader = importlib.machinery.SourceFileLoader("droiddeck_agent", str(SCRIPT))
|
|
spec = importlib.util.spec_from_loader(loader.name, loader)
|
|
agent = importlib.util.module_from_spec(spec)
|
|
loader.exec_module(agent)
|
|
|
|
|
|
class FakeDevTools(threading.Thread):
|
|
"""GET /json lists one target; its WebSocket answers Runtime.evaluate the way Chromium does."""
|
|
|
|
def __init__(self, answer):
|
|
super().__init__(daemon=True)
|
|
self.answer = answer
|
|
self.server = socket.socket()
|
|
self.server.bind(("127.0.0.1", 0))
|
|
self.server.listen(4)
|
|
self.port = self.server.getsockname()[1]
|
|
self.seen = []
|
|
|
|
def run(self):
|
|
while True:
|
|
try:
|
|
conn, _ = self.server.accept()
|
|
except OSError:
|
|
return
|
|
threading.Thread(target=self.serve, args=(conn,), daemon=True).start()
|
|
|
|
def serve(self, conn):
|
|
head = b""
|
|
while b"\r\n\r\n" not in head:
|
|
head += conn.recv(4096)
|
|
request = head.decode()
|
|
if request.startswith("GET /json"):
|
|
body = json.dumps([{"title": "SharedJSContext", "url": "https://steamloopback.host/", "type": "page",
|
|
"webSocketDebuggerUrl": "ws://127.0.0.1:%d/devtools/page/1" % self.port}]).encode()
|
|
conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: %d\r\n\r\n" % len(body) + body)
|
|
conn.close()
|
|
return
|
|
key = [l.split(":", 1)[1].strip() for l in request.split("\r\n") if l.lower().startswith("sec-websocket-key")][0]
|
|
accept = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode()).digest()).decode()
|
|
conn.sendall(("HTTP/1.1 101 Switching Protocols\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n"
|
|
"Sec-WebSocket-Accept: %s\r\n\r\n" % accept).encode())
|
|
b0, b1 = conn.recv(2)
|
|
n = b1 & 0x7F
|
|
if n == 126:
|
|
n = struct.unpack(">H", conn.recv(2))[0]
|
|
mask = conn.recv(4)
|
|
data = b""
|
|
while len(data) < n:
|
|
data += conn.recv(n - len(data))
|
|
message = json.loads(bytes(b ^ mask[i % 4] for i, b in enumerate(data)))
|
|
self.seen.append(message)
|
|
for reply in ({"method": "Runtime.consoleAPICalled", "params": {}}, dict(self.answer, id=message["id"])):
|
|
payload = json.dumps(reply).encode()
|
|
header = bytes([0x81, 126]) + struct.pack(">H", len(payload)) if len(payload) > 125 else bytes([0x81, len(payload)])
|
|
conn.sendall(header + payload)
|
|
time.sleep(0.2)
|
|
conn.close()
|
|
|
|
|
|
class AgentTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.temp = tempfile.TemporaryDirectory()
|
|
self.addCleanup(self.temp.cleanup)
|
|
self.dir = Path(self.temp.name)
|
|
(self.dir / "resp").mkdir()
|
|
|
|
def ask(self, request):
|
|
path = self.dir / "req.json"
|
|
path.write_text(json.dumps(request))
|
|
agent.answer(self.dir, path)
|
|
self.assertFalse(path.exists(), "the request file is consumed")
|
|
return json.loads((self.dir / "resp" / "req.json").read_text())
|
|
|
|
def vr_helper(self, body):
|
|
helper = self.dir / "droiddeck-steamvr"
|
|
helper.write_text("#!/bin/sh\n" + body)
|
|
old = os.environ.get("DROIDDECK_VR_HELPER")
|
|
os.environ["DROIDDECK_VR_HELPER"] = str(helper)
|
|
self.addCleanup(lambda: os.environ.pop("DROIDDECK_VR_HELPER") if old is None
|
|
else os.environ.__setitem__("DROIDDECK_VR_HELPER", old))
|
|
return helper
|
|
|
|
def test_vr_status_is_parsed(self):
|
|
self.vr_helper('echo "installed=1 running=0 installing=0 driver=1"\n')
|
|
answer = self.ask({"kind": "vr", "action": "status"})
|
|
self.assertTrue(answer["ok"])
|
|
self.assertEqual((True, False, False, True),
|
|
(answer["installed"], answer["running"], answer["installing"], answer["driver"]))
|
|
|
|
def test_vr_runs_only_its_own_actions(self):
|
|
log = self.dir / "args"
|
|
self.vr_helper('echo "$@" > %s\n' % log)
|
|
self.assertTrue(self.ask({"kind": "vr", "action": "start"})["ok"])
|
|
self.assertEqual("start", log.read_text().strip())
|
|
answer = self.ask({"kind": "vr", "action": "rm -rf /"})
|
|
self.assertFalse(answer["ok"])
|
|
self.assertEqual("INVALID_REQUEST", answer["error"]["code"])
|
|
|
|
def test_vr_without_a_helper_is_unavailable(self):
|
|
os.environ["DROIDDECK_VR_HELPER"] = str(self.dir / "missing")
|
|
self.addCleanup(os.environ.pop, "DROIDDECK_VR_HELPER")
|
|
answer = self.ask({"kind": "vr", "action": "status"})
|
|
self.assertEqual("UNAVAILABLE", answer["error"]["code"])
|
|
|
|
def test_exec_output_and_status(self):
|
|
answer = self.ask({"kind": "exec", "argv": ["sh", "-c", "echo out; echo err >&2; exit 3"], "env": {"X": "1"}})
|
|
self.assertTrue(answer["ok"])
|
|
self.assertEqual(3, answer["exitCode"])
|
|
self.assertEqual("out\n", answer["stdout"]["text"])
|
|
self.assertEqual("err\n", answer["stderr"]["text"])
|
|
self.assertFalse(answer["timedOut"])
|
|
|
|
def test_exec_environment_stdin_and_cap(self):
|
|
answer = self.ask({"kind": "exec", "argv": ["sh", "-c", "printf %s \"$X\"; cat"], "env": {"X": "a"}, "stdin": "b"})
|
|
self.assertEqual("ab", answer["stdout"]["text"])
|
|
big = self.ask({"kind": "exec", "argv": ["sh", "-c", "head -c 300000 /dev/zero"]})
|
|
self.assertTrue(big["stdout"]["truncated"])
|
|
self.assertEqual(agent.OUTPUT_CAP, len(big["stdout"]["text"]))
|
|
|
|
def test_exec_timeout_kills_the_process_group(self):
|
|
started = time.monotonic()
|
|
answer = self.ask({"kind": "exec", "argv": ["sh", "-c", "sleep 30 & sleep 30"], "timeout": 0.5})
|
|
self.assertTrue(answer["timedOut"])
|
|
self.assertLess(time.monotonic() - started, 10)
|
|
|
|
def test_exec_memory_is_bounded_while_draining_large_output(self):
|
|
tracemalloc.start()
|
|
try:
|
|
answer = agent.run_exec({"argv": [sys.executable, "-c", "import os; [os.write(1, b'x' * 65536) for _ in range(1024)]"]})
|
|
_, peak = tracemalloc.get_traced_memory()
|
|
finally:
|
|
tracemalloc.stop()
|
|
self.assertEqual(0, answer["exitCode"])
|
|
self.assertEqual(agent.OUTPUT_CAP, len(answer["stdout"]["text"]))
|
|
self.assertTrue(answer["stdout"]["truncated"])
|
|
self.assertLess(peak, 6 * 1024 * 1024)
|
|
|
|
def test_exec_drains_both_streams_while_writing_stdin(self):
|
|
program = "import os, sys; os.write(1, b'x' * 300000); print(len(sys.stdin.read()), file=sys.stderr)"
|
|
answer = self.ask({"kind": "exec", "argv": [sys.executable, "-c", program], "stdin": "y" * 100000})
|
|
self.assertFalse(answer["timedOut"])
|
|
self.assertTrue(answer["stdout"]["truncated"])
|
|
self.assertEqual("100000\n", answer["stderr"]["text"])
|
|
|
|
def test_flooding_both_streams_still_times_out(self):
|
|
program = "import os\nwhile True:\n os.write(1, b'o' * 65536)\n os.write(2, b'e' * 65536)"
|
|
answer = self.ask({"kind": "exec", "argv": [sys.executable, "-c", program], "timeout": 0.5})
|
|
self.assertTrue(answer["timedOut"])
|
|
for name in ("stdout", "stderr"):
|
|
self.assertEqual(agent.OUTPUT_CAP, len(answer[name]["text"]))
|
|
self.assertTrue(answer[name]["truncated"])
|
|
|
|
def test_timeout_applies_when_parent_exits_with_inherited_pipes(self):
|
|
started = time.monotonic()
|
|
answer = self.ask({"kind": "exec", "argv": ["sh", "-c", "sleep 30 & exit 0"], "timeout": 0.5})
|
|
self.assertTrue(answer["timedOut"])
|
|
self.assertLess(time.monotonic() - started, 5)
|
|
|
|
def test_invalid_timeout_and_input_are_refused_before_exec(self):
|
|
for value in (0, -1, 601, "NaN", "Infinity", "invalid"):
|
|
answer = self.ask({"kind": "exec", "argv": ["true"], "timeout": value})
|
|
self.assertEqual("INVALID_REQUEST", answer["error"]["code"])
|
|
for fields in ({"stdin": 1}, {"env": []}, {"env": {"BAD=KEY": "value"}}, {"argv": ["true\0"]}):
|
|
answer = self.ask({"kind": "exec", "argv": ["true"], **fields})
|
|
self.assertEqual("INVALID_REQUEST", answer["error"]["code"])
|
|
|
|
def test_errors_are_answers(self):
|
|
self.assertEqual("INVALID_REQUEST", self.ask({"kind": "exec", "argv": []})["error"]["code"])
|
|
self.assertEqual("EXEC_FAILED", self.ask({"kind": "exec", "argv": ["/no/such/program"]})["error"]["code"])
|
|
self.assertEqual("UNKNOWN_REQUEST", self.ask({"kind": "nope"})["error"]["code"])
|
|
self.assertEqual("INVALID_APP_ID", self.ask({"kind": "quit", "appId": "x"})["error"]["code"])
|
|
(self.dir / "req.json").write_text("{not json")
|
|
agent.answer(self.dir, self.dir / "req.json")
|
|
self.assertEqual("INVALID_REQUEST", json.loads((self.dir / "resp" / "req.json").read_text())["error"]["code"])
|
|
|
|
def test_request_shape_and_size_are_bounded(self):
|
|
self.assertEqual("INVALID_REQUEST", self.ask([])["error"]["code"])
|
|
self.assertEqual("INVALID_REQUEST", self.ask({"kind": "ping", "unused": "x" * agent.REQUEST_CAP})["error"]["code"])
|
|
|
|
def test_replies_are_owner_only_even_when_replacing_a_public_file(self):
|
|
path = self.dir / "private.json"
|
|
path.write_text("old")
|
|
path.chmod(0o644)
|
|
agent.write_json(path, {"ok": True})
|
|
self.assertEqual(0o600, path.stat().st_mode & 0o777)
|
|
|
|
def test_oversized_websocket_frames_are_refused_before_reading_payload(self):
|
|
ws = agent.WebSocket.__new__(agent.WebSocket)
|
|
ws.buffer = bytes([0x81, 127]) + struct.pack(">Q", agent.RESPONSE_CAP + 1)
|
|
with self.assertRaises(agent.AgentError) as caught:
|
|
ws.recv()
|
|
self.assertEqual("RESPONSE_TOO_LARGE", caught.exception.code)
|
|
|
|
def test_devtools_targets_cannot_connect_to_remote_hosts(self):
|
|
for url in ("ws://example.com/devtools", "ws://[email protected]/devtools", "wss://127.0.0.1/devtools"):
|
|
with patch.object(agent.socket, "create_connection") as connect:
|
|
with self.assertRaises(agent.AgentError) as caught:
|
|
agent.WebSocket(url, 1)
|
|
self.assertEqual("INVALID_TARGET", caught.exception.code)
|
|
connect.assert_not_called()
|
|
|
|
def test_cdp_without_a_port(self):
|
|
os.environ.pop("BL_CDP_PORT", None)
|
|
self.assertEqual("NO_DEVTOOLS", self.ask({"kind": "cdp", "expression": "1"})["error"]["code"])
|
|
|
|
def test_cdp_eval(self):
|
|
server = FakeDevTools({"result": {"result": {"type": "number", "value": 42}}})
|
|
server.start()
|
|
self.addCleanup(server.server.close)
|
|
(self.dir / "cdp-port").write_text("%d\n" % server.port)
|
|
answer = self.ask({"kind": "cdp", "expression": "6 * 7"})
|
|
self.assertTrue(answer["ok"], answer)
|
|
self.assertEqual(42, answer["value"])
|
|
self.assertEqual("SharedJSContext", answer["target"])
|
|
self.assertEqual("Runtime.evaluate", server.seen[0]["method"])
|
|
self.assertEqual("6 * 7", server.seen[0]["params"]["expression"])
|
|
targets = self.ask({"kind": "cdp", "op": "targets"})
|
|
self.assertEqual("SharedJSContext", targets["targets"][0]["title"])
|
|
missing = self.ask({"kind": "cdp", "expression": "1", "target": "Nope"})
|
|
self.assertEqual("NO_TARGET", missing["error"]["code"])
|
|
|
|
def test_steam_selection(self):
|
|
with patch.object(agent, "cdp_eval", return_value={"value": "3180310"}) as evaluate:
|
|
answer = self.ask({"kind": "steam-selection"})
|
|
self.assertEqual(3180310, answer["appId"])
|
|
self.assertIn("document.activeElement", evaluate.call_args.args[1])
|
|
self.assertEqual("Steam Big Picture Mode", evaluate.call_args.kwargs["target"])
|
|
|
|
def test_steam_selection_allows_no_highlight(self):
|
|
with patch.object(agent, "cdp_eval", return_value={"value": None}):
|
|
self.assertIsNone(self.ask({"kind": "steam-selection"})["appId"])
|
|
|
|
def test_steam_selection_rejects_invalid_app_id(self):
|
|
with patch.object(agent, "cdp_eval", return_value={"value": "4294967296"}):
|
|
answer = self.ask({"kind": "steam-selection"})
|
|
self.assertEqual("INVALID_SELECTION", answer["error"]["code"])
|
|
|
|
def test_shortcut_cgameid_waits_for_exact_steam_overview_mapping(self):
|
|
game_id = str((3551182975 << 32) | 0x02000000)
|
|
self.assertEqual(3551182975, agent.steam_shortcut_app_id(game_id))
|
|
self.assertIsNone(agent.steam_shortcut_app_id("413150"))
|
|
self.assertIsNone(agent.steam_shortcut_app_id(str((3551182975 << 32) | 0x03000000)))
|
|
clock = [0.0]
|
|
with patch.object(agent, "cdp_eval", side_effect=[{"value": None}, {"value": "3551182975"}]) as evaluate, \
|
|
patch.object(agent.time, "monotonic", side_effect=lambda: clock[0]), \
|
|
patch.object(agent.time, "sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay)):
|
|
result = agent.wait_for_steam_shortcut_overview(self.dir, game_id, timeout=1)
|
|
self.assertEqual({"supported": True, "ready": True, "appId": 3551182975}, result)
|
|
expression = evaluate.call_args_list[0].args[1]
|
|
self.assertIn("GetAppOverviewByGameID", expression)
|
|
self.assertIn(json.dumps(game_id), expression)
|
|
self.assertNotIn("RequestAppDetails", expression)
|
|
self.assertFalse(evaluate.call_args_list[0].kwargs["await_promise"])
|
|
|
|
def test_wrong_overview_app_id_keeps_shortcut_url_held_until_timeout(self):
|
|
game_id = str((3537741571 << 32) | 0x02000000)
|
|
clock = [0.0]
|
|
with patch.object(agent, "cdp_eval", return_value={"value": "646570"}) as evaluate, \
|
|
patch.object(agent.time, "monotonic", side_effect=lambda: clock[0]), \
|
|
patch.object(agent.time, "sleep", side_effect=lambda delay: clock.__setitem__(0, clock[0] + delay)):
|
|
with self.assertRaises(agent.AgentError) as raised:
|
|
agent.wait_for_steam_shortcut_overview(self.dir, game_id, timeout=0.5)
|
|
self.assertEqual("STEAM_OVERVIEW_TIMEOUT", raised.exception.code)
|
|
self.assertGreater(evaluate.call_count, 0)
|
|
self.assertTrue(all("RequestAppDetails" not in call.args[1] for call in evaluate.call_args_list))
|
|
|
|
def test_normal_steam_ids_and_unknown_cgameids_skip_overview_wait(self):
|
|
for game_id in ("413150", str((123456789 << 32) | 0x03000000), "18446744073709551616"):
|
|
with self.subTest(game_id=game_id), patch.object(agent, "cdp_eval") as evaluate:
|
|
result = agent.wait_for_steam_shortcut_overview(self.dir, game_id)
|
|
self.assertFalse(result["supported"])
|
|
evaluate.assert_not_called()
|
|
|
|
def test_cdp_exception(self):
|
|
server = FakeDevTools({"result": {"result": {"type": "object"}, "exceptionDetails": {
|
|
"text": "Uncaught", "exception": {"description": "ReferenceError: SteamClient is not defined"}}}})
|
|
server.start()
|
|
self.addCleanup(server.server.close)
|
|
(self.dir / "cdp-port").write_text(str(server.port))
|
|
answer = self.ask({"kind": "cdp", "expression": "SteamClient"})
|
|
self.assertEqual("JS_EXCEPTION", answer["error"]["code"])
|
|
self.assertIn("ReferenceError", answer["error"]["message"])
|
|
|
|
def test_serve_answers_request_files(self):
|
|
server = subprocess.Popen([sys.executable, str(SCRIPT), "--no-focus"],
|
|
env=dict(os.environ, BL_LAUNCH_DIR=str(self.dir)),
|
|
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
|
|
def stop_server():
|
|
server.terminate()
|
|
server.wait(timeout=5)
|
|
self.addCleanup(stop_server)
|
|
agent_dir = self.dir / "agent"
|
|
for _ in range(100):
|
|
if (agent_dir / "hello.json").exists():
|
|
break
|
|
time.sleep(0.02)
|
|
hello = json.loads((agent_dir / "hello.json").read_text())
|
|
self.assertEqual(server.pid, hello["pid"])
|
|
self.assertFalse(hello["focus"])
|
|
(agent_dir / "req" / "a.json.tmp").write_text(json.dumps({"kind": "ping"}))
|
|
os.replace(agent_dir / "req" / "a.json.tmp", agent_dir / "req" / "a.json")
|
|
for _ in range(200):
|
|
if (agent_dir / "resp" / "a.json").exists():
|
|
break
|
|
time.sleep(0.02)
|
|
self.assertEqual(agent.VERSION, json.loads((agent_dir / "resp" / "a.json").read_text())["version"])
|
|
|
|
def test_game_pids_reads_the_steam_environment(self):
|
|
import subprocess
|
|
if not Path("/proc").is_dir():
|
|
self.skipTest("needs /proc")
|
|
child = subprocess.Popen(["sleep", "30"], env=dict(os.environ, SteamAppId="424242"))
|
|
self.addCleanup(lambda: (child.kill(), child.wait()))
|
|
time.sleep(0.2)
|
|
self.assertIn(child.pid, agent.game_pids(424242))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|