mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 06:00:33 +02:00
saphidandClaude Opus 5.5
0f770dc88a
The page was one 6,800px scroll with nine nav links (hidden below 1150px). It is now four tabs, Home, Games, Android and Tools, switched with 1-4; old section links still land on the right tab. When the Frame can't be reached, the server turns ssh's connection errors into one plain message (503, offline: true), the page shows a single banner with Retry and Set Up Connection, retries every 8 s, and reloads every panel when the Frame answers. Panels say "Waiting for the Frame" instead of raw ssh text. Flatpak and Android catalogue installs run as background jobs the page polls, so a slow install no longer holds a request for up to 15 minutes or reports a false failure; the bottom bar counts running installs. Files can be dropped anywhere in the window, as the README already said. Recent reports show the newest five, with Show all. Android display explains an empty or failed read. A topped-up headset on a charger reads as not charging rather than "still draining, using 0.0 W". Fixes a race where the catalogue and reports loads wrote the compat-db mirror's .tmp file at once and one failed with a 500. Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
268 lines
13 KiB
Python
268 lines
13 KiB
Python
"""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_unknown_routes(self):
|
|
self.assertEqual(self.request("GET", "/nope")[0], 404)
|
|
self.assertEqual(self.post("/api/nope", {})[0], 404)
|
|
|
|
|
|
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()
|