mirror of
https://github.com/saphid/frame-control.git
synced 2026-10-06 05:02:50 +02:00
saphidandClaude Opus 5.5
636a4a47b7
Combines the three feature branches on bundle-deps. Conflicts in server.py,
index.html, preload.js, README and test_server.py keep both sides. The
web-install downloader now uses urllib's default HTTPS context, so the
bundled CA list from 770f26c applies to it on Windows too.
Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
208 lines
10 KiB
Python
208 lines
10 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_unknown_routes(self):
|
|
self.assertEqual(self.request("GET", "/nope")[0], 404)
|
|
self.assertEqual(self.post("/api/nope", {})[0], 404)
|
|
|
|
|
|
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()
|