steam-frame/saphid--frame-control
1
0 Fork 0
mirror of https://github.com/saphid/frame-control.git synced 2026-10-06 00:00:21 +02:00
Code Issues Packages Projects Releases 6 Wiki Activity
Files
f81c98a87b68d618e5cdd4b0a6759917b91b66d9
saphid--frame-control/tests/test_server.py
T
saphidandClaude Opus 5.5 f81c98a87b Fix CI: title removal order, Windows fixtures and POSIX-only tests 
- Title removal ran the Steam shortcut tidy-up before steamos-delete, which
  finds the Proton prefix through that shortcut, so compatdata was left behind
  (e2e caught it). steamos-delete runs first again; art/collection tidy-up after.
- Test fixtures are byte-exact: never convert line endings (a text-looking
  fixture APK got CRLF on Windows and failed its SHA-256).
- Read index.html/artwork-settings.js as UTF-8 in tests; app-data backup and
  OBB shell tests run only on POSIX (they exercise the Frame-side scripts).
- e2e expects the icon under artwork/ now.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-09-28 23:59:44 +10:00

389 lines
20 KiB
Python
Raw 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 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()
env = {**os.environ, "FRAME_ALIAS": "frame-control-test.invalid", "PYTHONDONTWRITEBYTECODE": "1"}
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("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/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/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_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 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