saphid--frame-control/tests/test_server.py

"""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)


@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()