steam-frame/saphid--frame-control
1
0 Fork 0
mirror of https://github.com/saphid/frame-control.git synced 2026-10-06 07:00:37 +02:00
Code Issues Packages Projects Releases 6 Wiki Activity
Files
screenshot-copy
saphid--frame-control/tests/test_server.py
T
saphidandClaude Opus 5.5 83d74548cd Screenshots: Copy button, right-click menu, and new shots appear on their own 
Each screenshot card and the viewer get a Copy button that puts the image on
the clipboard (natively in the desktop app, as PNG in a browser). Right-click
a screenshot to open, copy, save, show it in Finder, or copy its path or name;
right-click the viewer to copy or save. The shelf re-lists the Frame's
screenshots every 8 s while the window is visible and connected, redraws only
when something changed, and keeps thumbnails it already has. Switching
headsets clears the list and ignores answers still on their way.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-29 15:08:36 +10:00

441 lines
23 KiB
Python
Raw Permalink Blame History

"""Frame Control server checks that need no headset.
Starts ui/server.py against an SSH alias that can't resolve, then exercises the
request guards and input validation, which all run before any SSH call.
Run: python3 -m unittest discover -s tests
"""
import sandbox # noqa: F401 (first: keeps tests off real data and services)
import http.client
import io
import json
import os
import socket
import struct
import subprocess
import sys
import tempfile
import time
import unittest
import zipfile
from pathlib import Path
from urllib.parse import quote
ROOT = Path(__file__).resolve().parent.parent
def free_port():
with socket.socket() as s:
s.bind(("127.0.0.1", 0))
return s.getsockname()[1]
class ServerGuards(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.port = free_port()
cls.ssh_dir = tempfile.mkdtemp(prefix="frame-control-ssh-") # an empty ~/.ssh: no headsets set up
env = {**os.environ, "FRAME_ALIAS": "frame-control-test.invalid", "PYTHONDONTWRITEBYTECODE": "1",
"FRAME_CONTROL_SSH_DIR": cls.ssh_dir}
cls.log = tempfile.TemporaryFile()
cls.proc = subprocess.Popen([sys.executable, str(ROOT / "ui" / "server.py"), "--port", str(cls.port)],
env=env, stdout=cls.log, stderr=subprocess.STDOUT)
for _ in range(100):
try:
if cls.request("GET", "/")[0] == 200:
return
except Exception:
pass
time.sleep(0.05)
cls.proc.kill()
cls.log.seek(0)
raise RuntimeError("server didn't start:\n" + cls.log.read().decode(errors="replace"))
@classmethod
def tearDownClass(cls):
cls.proc.terminate()
cls.proc.wait(timeout=10)
cls.log.close()
@classmethod
def request(cls, method, path, body=None, headers=None):
conn = http.client.HTTPConnection("127.0.0.1", cls.port, timeout=10)
data = body if isinstance(body, bytes) else json.dumps(body).encode() if body is not None else None
conn.request(method, path, body=data, headers=headers or {})
r = conn.getresponse()
payload = r.read()
conn.close()
return r.status, dict(r.getheaders()), payload
def post(self, path, body):
status, _, payload = self.request("POST", path, body, {"X-Frame-UI": "1", "Content-Type": "application/json"})
return status, json.loads(payload)
def test_page_served_with_identifying_and_anti_framing_headers(self):
status, headers, payload = self.request("GET", "/")
self.assertEqual(status, 200)
self.assertTrue(headers["Server"].startswith("FrameControl"))
self.assertEqual(headers["X-Frame-Options"], "DENY")
self.assertIn(b"<html", payload.lower())
def test_foreign_host_rejected(self):
# DNS rebinding: a hostile name pointed at 127.0.0.1.
for path in ("/", "/api/status"):
status, _, _ = self.request("GET", path, headers={"Host": f"evil.example:{self.port}", "X-Frame-UI": "1"})
self.assertEqual(status, 403, path)
def test_api_needs_custom_header(self):
# <img src> and plain form posts from other sites can't set it.
self.assertEqual(self.request("POST", "/api/comfort", {"action": "start"})[0], 403)
self.assertEqual(self.request("GET", "/api/status")[0], 403)
self.assertEqual(self.request("GET", "/api/screenshot?view=headset")[0], 403)
self.assertEqual(self.request("GET", "/api/shots")[0], 403)
self.assertEqual(self.request("GET", "/api/stream")[0], 403)
self.assertEqual(self.request("GET", "/api/shots/image?id=1/250820/20260925225208_1.jpg")[0], 403)
self.assertEqual(self.request("POST", "/api/launch", {"appid": "620"})[0], 403)
def test_captures_are_not_cacheable(self):
# Headset captures show everything on screen; nothing may cache them.
_, headers, _ = self.request("GET", "/api/screenshot", headers={"X-Frame-UI": "1"})
self.assertEqual(headers.get("Cache-Control"), "no-store")
self.assertIn("frame-ancestors 'none'", headers.get("Content-Security-Policy", ""))
def test_input_validation(self):
cases = [
("/api/comfort", {"action": "poweroff"}),
("/api/comfort", {"action": "start", "minutes": 0}),
("/api/launch", {"appid": "620; rm -rf ~"}),
("/api/launch", {"appid": ""}),
("/api/flatpak", {"id": "org.example.App;id", "action": "install"}),
("/api/flatpak", {"id": "org.example.App", "action": "explode"}),
("/api/volume", {"level": 1.5}),
("/api/clipboard", {"text": ""}),
("/api/open", {"what": "anything-else"}),
("/api/open", {"what": "shot", "id": "1/250820/../../.ssh/id_ed25519"}),
("/api/open", {"what": "shot"}),
("/api/shots/save", {"ids": []}),
("/api/shots/save", {"ids": "1/250820/20260925225208_1.jpg"}),
("/api/shots/save", {"ids": [1]}),
("/api/shots/save", {"ids": ["1/250820/../../.ssh/id_ed25519"]}),
("/api/shots/save", {"ids": ["1/250820/20260925225208_1.jpg; rm -rf ~"]}),
]
for path, body in cases:
status, payload = self.post(path, body)
self.assertEqual(status, 400, f"{path} {body} -> {payload}")
def test_showing_a_shot_needs_it_saved_here(self):
status, payload = self.post("/api/open", {"what": "shot", "id": "1/250820/19990101000000_1.jpg"})
self.assertEqual(status, 404, payload)
def test_screenshot_ids_checked_before_ssh(self):
for shot in ("../../etc/passwd", "1/250820/x.jpg", "1/2/20260925225208_1.jpg;id", "1/250820/20260925225208_1.gif"):
status, _, _ = self.request("GET", f"/api/shots/image?id={quote(shot)}", headers={"X-Frame-UI": "1"})
self.assertEqual(status, 400, shot)
def test_stream_settings_checked_before_ssh(self):
for query in ("h=480", "fps=24", "h=abc", "h=1080&fps=120"):
status, _, _ = self.request("GET", f"/api/stream?{query}", headers={"X-Frame-UI": "1"})
self.assertEqual(status, 400, query)
def test_bad_bodies(self):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=10)
conn.request("POST", "/api/launch", body=b"{not json", headers={"X-Frame-UI": "1"})
self.assertEqual(conn.getresponse().status, 400)
conn.close()
status, _ = self.post("/api/launch", ["not", "an", "object"])
self.assertEqual(status, 400)
def test_title_upload_is_inspected_then_discarded(self):
# A zip holding a Windows x86-64 program: inspected locally, no SSH until install.
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w") as z:
z.writestr("Tiny Game/Tiny Game.exe",
b"MZ" + b"\0" * 0x3A + struct.pack("<I", 0x40) + b"PE\0\0" + struct.pack("<HHIIIHH", 0x8664, 1, 0, 0, 0, 0xF0, 0x22))
status, _, payload = self.request("POST", "/api/upload", buf.getvalue(),
{"X-Frame-UI": "1", "X-Mode": "title", "X-Filename": quote("Tiny Game-win64.zip")})
r = json.loads(payload)
self.assertEqual(status, 200, r)
self.assertEqual((r["plan"]["id"], r["plan"]["target"], r["plan"]["runtime"]),
("Tiny_Game", "Tiny Game.exe", "proton-experimental"))
self.assertNotIn("root", r["plan"])
self.assertEqual(self.post("/api/titles", {"action": "discard", "token": r["token"]})[0], 200)
self.assertEqual(self.post("/api/titles", {"action": "install", "token": r["token"]})[0], 400)
def test_title_input_validation(self):
status, _, _ = self.request("POST", "/api/upload", b"not a zip",
{"X-Frame-UI": "1", "X-Mode": "title", "X-Filename": "x.zip"})
self.assertEqual(status, 400)
for body in ({"action": "inspect", "path": "relative/game.zip"},
{"action": "inspect", "path": "/nonexistent/frame-control/game.zip"},
{"action": "install", "token": "nope"},
{"action": "launch", "id": "x; rm -rf ~"},
{"action": "remove", "id": "../etc"},
{"action": "explode"}):
status, payload = self.post("/api/titles", body)
self.assertEqual(status, 400, f"{body} -> {payload}")
self.assertEqual(self.request("GET", "/api/titles/job?token=nope", headers={"X-Frame-UI": "1"})[0], 404)
self.assertEqual(self.request("POST", "/api/titles", {"action": "list"})[0], 403)
def test_web_install_needs_the_app_page(self):
# A website can only open frame-control:// links; it can't call these itself.
link = {"url": "https://cdn.example.com/game.apk"}
self.assertEqual(self.request("POST", "/api/webinstall/check", link)[0], 403)
self.assertEqual(self.request("POST", "/api/webinstall/start", {"id": "x"})[0], 403)
status, _, _ = self.request("POST", "/api/webinstall/check", link,
{"X-Frame-UI": "1", "Host": f"evil.example:{self.port}"})
self.assertEqual(status, 403)
def test_web_install_validation(self):
for body in ({}, {"url": 5}, {"url": "http://cdn.example.com/game.apk"}, {"url": "https://10.0.0.2/game.apk"},
{"url": "https://u:p@example.com/game.apk"}, {"url": "https://example.com/"},
{"url": "https://1.1.1.1/game.sh"}, {"manifest": "file:///etc/passwd"},
{"manifest": "https://example.com/m.json", "url": "https://example.com/g.apk"}):
status, payload = self.post("/api/webinstall/check", body)
self.assertEqual(status, 400, f"{body} -> {payload}")
# Only an id from /check starts an install, and only once.
self.assertEqual(self.post("/api/webinstall/start", {"id": "made-up"})[0], 400)
self.assertEqual(self.request("GET", "/api/webinstall/job?id=x", headers={"X-Frame-UI": "1"})[0], 404)
self.assertEqual(self.post("/api/webinstall/cancel", {"job": "x"})[0], 404)
def test_unreachable_frame_is_one_clear_offline_error(self):
status, _, payload = self.request("GET", "/api/status", headers={"X-Frame-UI": "1"})
body = json.loads(payload)
self.assertEqual(status, 503, body)
self.assertTrue(body["offline"])
self.assertIn("Can't find the Frame", body["error"])
self.assertIn("frame-control-test.invalid", body["detail"]) # ssh's own words stay available
def test_flatpak_install_runs_as_a_job(self):
status, started = self.post("/api/flatpak", {"id": "org.example.App", "action": "install"})
self.assertEqual(status, 200, started)
for _ in range(200):
status, _, payload = self.request("GET", f"/api/job?id={started['job']}", headers={"X-Frame-UI": "1"})
job = json.loads(payload)
if job["done"]:
break
time.sleep(0.05)
self.assertEqual(status, 200)
self.assertTrue(job["done"])
self.assertIn("Can't find the Frame", job["error"])
self.assertEqual(self.request("GET", "/api/job?id=nope", headers={"X-Frame-UI": "1"})[0], 404)
def test_android_install_checks_the_package_before_starting(self):
status, body = self.post("/api/android", {"action": "install", "package": "org.example.not.in.catalogue"})
self.assertNotEqual(status, 200, body)
self.assertNotIn("job", body)
def test_android_install_of_another_version_runs_as_a_job(self):
pkg = "org.example.frame_control.not_in_any_repo"
sys.path.insert(0, str(ROOT / "ui"))
import frame_apk_versions
# The job must fail on the cached lookup, before any download: nothing is cached for this package.
self.assertEqual(frame_apk_versions._versions(pkg, cached_only=True)[0], [])
status, started = self.post("/api/android", {"action": "install", "package": pkg,
"url": f"https://f-droid.org/repo/{pkg}_1.apk"})
self.assertEqual(status, 200, started)
for _ in range(200):
job = json.loads(self.request("GET", f"/api/job?id={started['job']}", headers={"X-Frame-UI": "1"})[2])
if job["done"]:
break
time.sleep(0.05)
self.assertTrue(job["done"])
self.assertIn("no longer available", job["error"])
def test_unknown_routes(self):
self.assertEqual(self.request("GET", "/nope")[0], 404)
self.assertEqual(self.post("/api/nope", {})[0], 404)
class OneServer(unittest.TestCase):
"""Two servers for one user would each connect and edit headsets on their own."""
def start(self, env):
proc = subprocess.Popen([sys.executable, str(ROOT / "ui" / "server.py"), "--port", "0"], env=env,
stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True)
self.addCleanup(lambda: (proc.terminate(), proc.wait(10), proc.stdout.close()))
return proc
def test_a_second_server_is_refused_until_the_first_exits(self):
data = tempfile.mkdtemp(prefix="frame-one-server-")
env = {**os.environ, "FRAME_CONTROL_DATA_DIR": data, "FRAME_ALIAS": "frame-control-test.invalid",
"FRAME_CONTROL_SERVER_WAIT": "1"}
first = self.start(env)
self.assertIn("Frame Control on", first.stdout.readline())
second = subprocess.run([sys.executable, str(ROOT / "ui" / "server.py"), "--port", "0"], env=env,
capture_output=True, text=True, timeout=60)
self.assertEqual(second.returncode, 1)
self.assertIn("already running", second.stderr)
first.terminate()
first.wait(10)
self.assertIn("Frame Control on", self.start(env).stdout.readline())
def test_a_private_server_runs_alongside_but_cant_change_headsets(self):
"""The MCP adapter starts its own server (FRAME_PRIVATE_SSH=1) while the app runs."""
data = tempfile.mkdtemp(prefix="frame-one-server-")
env = {**os.environ, "FRAME_CONTROL_DATA_DIR": data, "FRAME_ALIAS": "frame-control-test.invalid",
"FRAME_CONTROL_SERVER_WAIT": "1"}
self.assertIn("Frame Control on", self.start(env).stdout.readline())
private = self.start({**env, "FRAME_PRIVATE_SSH": "1"})
line = private.stdout.readline()
self.assertIn("Frame Control on", line)
port = int(line.split("http://127.0.0.1:")[1].split()[0])
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
conn.request("POST", "/api/devices", body=json.dumps({"action": "use", "id": "x"}),
headers={"Content-Type": "application/json", "X-Frame-UI": "1", "Host": f"127.0.0.1:{port}"})
r = conn.getresponse()
self.assertEqual(r.status, 403, r.read())
class ArtworkSettings(unittest.TestCase):
"""The settings panel's endpoints, with and without the page's X-Frame-UI key."""
def test_settings_need_and_accept_the_ui_key(self):
with tempfile.TemporaryDirectory() as home:
port = free_port()
env = {**os.environ, "FRAME_ALIAS": "frame-control-test.invalid", "PYTHONDONTWRITEBYTECODE": "1",
"HOME": home, "APPDATA": home, "XDG_DATA_HOME": home}
for name in ("STEAMGRIDDB_API_KEY", "FRAME_STEAMGRIDDB_API_KEY", "FRAME_UI_KEY"):
env.pop(name, None)
proc = subprocess.Popen([sys.executable, str(ROOT / "ui" / "server.py"), "--port", str(port)],
env=env, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
try:
def request(method, path, body=None, headers=None):
conn = http.client.HTTPConnection("127.0.0.1", port, timeout=10)
conn.request(method, path, body=json.dumps(body).encode() if body is not None else None,
headers=headers or {})
r = conn.getresponse()
payload = r.read()
conn.close()
return r.status, payload
for _ in range(100):
try:
if request("GET", "/")[0] == 200:
break
except OSError:
time.sleep(0.05)
key = {"X-Frame-UI": "1", "Content-Type": "application/json"}
self.assertEqual(request("GET", "/api/settings/artwork")[0], 403)
self.assertEqual(request("POST", "/api/settings/artwork", {"steamgriddb_api_key": "abc"})[0], 403)
status, payload = request("GET", "/api/settings/artwork", headers=key)
self.assertEqual((status, json.loads(payload)["steamgriddb_configured"]), (200, False))
status, payload = request("POST", "/api/settings/artwork", {"steamgriddb_api_key": "abc_1"}, key)
self.assertEqual((status, json.loads(payload)["steamgriddb_configured"]), (200, True))
self.assertNotIn(b"abc_1", payload)
status, payload = request("GET", "/api/settings/artwork", headers=key)
self.assertTrue(json.loads(payload)["steamgriddb_configured"])
self.assertEqual(request("POST", "/api/settings/artwork", {"steamgriddb_api_key": "a b"}, key)[0], 400)
finally:
proc.terminate()
proc.wait(timeout=10)
def test_panel_script_uses_the_keyed_api_helper(self):
script = (ROOT / "ui" / "artwork-settings.js").read_text(encoding="utf-8")
self.assertNotIn("fetch(", script)
self.assertIn("api('/api/settings/artwork'", script)
page = (ROOT / "ui" / "index.html").read_text(encoding="utf-8")
self.assertLess(page.index("async function api("), page.index('<script src="/artwork-settings.js">'))
@unittest.skipIf(os.name == "nt", "runs on the Frame (Linux); local-bin/ssh is a POSIX shell script")
class LocalMode(unittest.TestCase):
"""FRAME_LOCAL=1, as the iPhone app starts the server on the Frame: its own key
guards /api/, and ssh goes to ui/local-bin/ssh, which runs commands here."""
KEY = "0123456789abcdef0123456789abcdef"
@classmethod
def setUpClass(cls):
env = {**os.environ, "FRAME_LOCAL": "1", "FRAME_UI_KEY": cls.KEY, "FRAME_DEVICE": "iPhone",
"PYTHONDONTWRITEBYTECODE": "1"}
cls.log = tempfile.TemporaryFile()
cls.proc = subprocess.Popen([sys.executable, str(ROOT / "ui" / "server.py"), "--port", "0", "--exit-on-eof"],
env=env, stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=cls.log, text=True)
line = cls.proc.stdout.readline()
cls.port = int(line.split("127.0.0.1:")[1].split()[0]) # --port 0: the server prints the port it took
@classmethod
def tearDownClass(cls):
cls.proc.stdin.close() # --exit-on-eof: the phone disconnecting
cls.proc.wait(timeout=15)
cls.proc.stdout.close()
cls.log.close()
def request(self, method, path, body=None, key=KEY):
conn = http.client.HTTPConnection("127.0.0.1", self.port, timeout=20)
conn.request(method, path, body=json.dumps(body).encode() if body is not None else None,
headers={"X-Frame-UI": key, "Content-Type": "application/json"})
r = conn.getresponse()
data = json.loads(r.read() or b"{}")
conn.close()
return r.status, data
def test_needs_the_session_key(self):
self.assertEqual(self.request("GET", "/api/host", key="1")[0], 403)
self.assertEqual(self.request("GET", "/api/host", key="")[0], 403)
status, host = self.request("GET", "/api/host")
self.assertEqual(status, 200)
self.assertEqual(host, {"os": "SteamOS", "fileManager": None, "computer": "iPhone", "mobile": True})
def test_commands_run_locally(self):
# frame_titles lists ~/devkit-game here; with nothing there, the list is empty rather than an ssh error.
status, body = self.request("GET", "/api/titles")
self.assertEqual(status, 200, body)
self.assertIsInstance(body["titles"], list)
def test_open_is_for_the_app_and_power_needs_a_password(self):
self.assertEqual(self.request("POST", "/api/open", {"what": "terminal"})[0], 400)
status, body = self.request("POST", "/api/open", {"what": "reboot"})
self.assertEqual(status, 400)
self.assertIn("password", body["error"])
self.assertEqual(self.request("POST", "/api/open", {"what": "reboot", "password": "a\nb"})[0], 400)
class UnreachableMessages(unittest.TestCase):
"""Only ssh's own connection failures are reworded; other errors keep their text."""
@classmethod
def setUpClass(cls):
sys.path.insert(0, str(ROOT / "ui"))
import server
cls.server = server
def test_ssh_connection_failures(self):
cases = {
"ssh: Could not resolve hostname frame: nodename nor servname provided": "Can't find",
"ssh: connect to host frame.local port 22: Operation timed out": "isn't answering",
"ssh: connect to host 192.168.1.9 port 22: Host is down": "isn't answering",
"ssh: connect to host 192.168.1.9 port 22: No route to host": "isn't answering",
"ssh: connect to host 192.168.1.9 port 22: Connection refused": "refused",
"steamos@192.168.1.9: Permission denied (publickey,password).": "SSH key",
"Host key verification failed.": "identity changed",
"kex_exchange_identification: read: Connection reset by peer": "dropped",
"Timed out talking to frame": "too long",
}
for raw, words in cases.items():
body, status = self.server.error_body(raw)
self.assertEqual(status, 503, raw)
self.assertIn(words, body["error"], raw)
self.assertTrue(body["offline"])
def test_other_errors_pass_through(self):
for raw in ("bad Flatpak app ID", "error: No remote refs found for 'org.example.App'",
"cp: cannot open 'x': Permission denied", "timed out waiting for Steam"):
self.assertEqual(self.server.error_body(raw), ({"error": raw}, None), raw)
class StatusProbe(unittest.TestCase):
# frame_status.py only ever runs on the Frame (Linux); it needs os.statvfs.
@unittest.skipIf(os.name == "nt", "Frame-side script; POSIX only")
def test_runs_off_device_and_prints_one_json_object(self):
# The probe runs on the Frame; elsewhere every field must degrade to null/empty.
out = subprocess.run([sys.executable, str(ROOT / "ui" / "frame_status.py")],
capture_output=True, text=True, timeout=60)
self.assertEqual(out.returncode, 0, out.stderr)
data = json.loads(out.stdout)
for key in ("hostname", "battery", "disk", "services", "games", "flatpaks"):
self.assertIn(key, data)
if __name__ == "__main__":
unittest.main()
Reference in new issue View Git Blame Copy Permalink