From 7513fef6f8eff50357876363e2308a6b087541ec Mon Sep 17 00:00:00 2001 From: Sejong Yoo Date: Sat, 24 Oct 2026 21:42:12 +0900 Subject: [PATCH] feat: phone login via QR when the saved Mudfish login expires MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When autologin is rejected the panel offers '휴대폰으로 로그인 (QR)'. It opens a 5-minute LAN relay to the launcher UI guarded by a one-time token (first phone to open the QR URL is the only client allowed) and shows the URL as a QR code. The relay closes as soon as a new session appears, on cancel, on expiry, and on unload. Co-Authored-By: Claude Opus 5.5 --- main.py | 15 ++- package.json | 1 + pnpm-lock.yaml | 8 ++ py_modules/mudfish_decky/controller.py | 79 ++++++++++++- py_modules/mudfish_decky/relay.py | 151 +++++++++++++++++++++++++ py_modules/mudfish_decky/system.py | 10 ++ src/LoginModal.tsx | 88 ++++++++++++++ src/Panel.tsx | 9 ++ src/backend.ts | 4 + tests/test_controller.py | 107 +++++++++++++++++- tests/test_relay.py | 111 ++++++++++++++++++ tests/test_system.py | 10 ++ 12 files changed, 586 insertions(+), 7 deletions(-) create mode 100644 py_modules/mudfish_decky/relay.py create mode 100644 src/LoginModal.tsx create mode 100644 tests/test_relay.py diff --git a/main.py b/main.py index 184757a..9accb87 100644 --- a/main.py +++ b/main.py @@ -2,6 +2,7 @@ import asyncio import logging import os import sys +import urllib.parse sys.path.insert(0, os.path.join(os.path.dirname(__file__), "py_modules")) @@ -10,9 +11,11 @@ import decky # noqa: E402 from mudfish_decky import system # noqa: E402 from mudfish_decky.client import MudfishClient # noqa: E402 from mudfish_decky.controller import Controller # noqa: E402 +from mudfish_decky.relay import LoginRelay # noqa: E402 from mudfish_decky.store import Store # noqa: E402 RESUME_POLL_SECONDS = 2 +LOGIN_RELAY_SECONDS = 300 class Plugin: @@ -23,9 +26,19 @@ class Plugin: installed = system.find_mudfish() base = system.api_base(installed[1]) if installed else system.DEFAULT_API store = Store(os.path.join(decky.DECKY_PLUGIN_SETTINGS_DIR, "settings.json")) - self.controller = Controller(MudfishClient(base), system, store) + target = urllib.parse.urlsplit(base) + self.controller = Controller( + MudfishClient(base), system, store, + relay_factory=lambda: LoginRelay(target=(target.hostname, target.port or 80), ttl=LOGIN_RELAY_SECONDS), + ) return self.controller + async def start_login(self) -> dict: + return await asyncio.to_thread(self._get_controller().start_login) + + async def stop_login(self) -> dict: + return await asyncio.to_thread(self._get_controller().stop_login) + async def get_state(self) -> dict: return await asyncio.to_thread(self._get_controller().state) diff --git a/package.json b/package.json index 51aee70..1358484 100644 --- a/package.json +++ b/package.json @@ -20,6 +20,7 @@ }, "dependencies": { "@decky/api": "^1.1.3", + "qrcode-generator": "1.4.4", "react-icons": "^5.3.0", "tslib": "^2.7.0" }, diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 2b06ad6..4d503da 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -11,6 +11,9 @@ importers: '@decky/api': specifier: ^1.1.3 version: 1.1.3 + qrcode-generator: + specifier: 1.4.4 + version: 1.4.4 react-icons: specifier: ^5.3.0 version: 5.7.0(react@19.3.1) @@ -771,6 +774,9 @@ packages: resolution: {integrity: sha512-qcJu88Q2IWqJsDD529JKMdwGm/dvInW4HvQnRwiH9JtihJvzGOscDtHE3x1pBKeUOTysQ8kVmLnJ2kJu7yhcGA==} engines: {node: '>=12'} + qrcode-generator@1.4.4: + resolution: {integrity: sha512-HM7yY8O2ilqhmULxGMpcHSF1EhJJ9yBj8gvDEuZ6M+KGJ0YY2hKpnXvRD+hZPLrDVck3ExIGhmPtSdcjC+guuw==} + queue-microtask@1.2.3: resolution: {integrity: sha512-NuaNSa6flKT5JaSYQzJok04JzTL1CA6aGhv5rfLW3PgqA+M2ChpZQnAC8h8i4ZFkBS8X5RqkDBHA7r4hej3K9A==} @@ -1599,6 +1605,8 @@ snapshots: picomatch@4.0.7: {} + qrcode-generator@1.4.4: {} + queue-microtask@1.2.3: {} react-icons@5.7.0(react@19.3.1): diff --git a/py_modules/mudfish_decky/controller.py b/py_modules/mudfish_decky/controller.py index ce4be96..72ce4f6 100644 --- a/py_modules/mudfish_decky/controller.py +++ b/py_modules/mudfish_decky/controller.py @@ -21,6 +21,8 @@ MSG_MISSING = "데스크탑 모드에서 미꾸라지를 설치하세요 (docs.m MSG_STILL_UP = "터널이 아직 남아 있어요. 끄기를 다시 눌러 주세요" MSG_NO_RESPONSE = "미꾸라지가 응답하지 않아요. 터널이 남아 있을 수 있으니 끄기를 눌러 주세요" MSG_SLEEP = "절전에서 깨어나 VPN을 껐어요. 필요하면 다시 켜 주세요" +MSG_LOGGED_IN = "로그인됐어요. 이제 켤 수 있어요" +MSG_NO_LAN = "와이파이 주소를 찾지 못해 휴대폰 로그인을 열 수 없어요" CONNECT_TIMEOUT = 30 DISCONNECT_TIMEOUT = 10 @@ -32,15 +34,20 @@ log = logging.getLogger("mudfish_decky") class Controller: - def __init__(self, client, system, store, clock=time.time, sleep=time.sleep): + def __init__(self, client, system, store, clock=time.time, sleep=time.sleep, relay_factory=None): self.client = client self.system = system self.store = store self.clock = clock self.sleep = sleep + self.relay_factory = relay_factory self._lock = threading.Lock() self._error: Optional[str] = None self._notice: Optional[str] = None + self._need_login = False + self._failed_jwt: Optional[str] = None + self._relay = None + self._relay_url: Optional[str] = None self._static: Optional[list] = None self._ip_cache = (0.0, None) self._sleep_offset: Optional[float] = None @@ -60,6 +67,10 @@ class Controller: try: st = self._signed_in(installed) self._static_nodes(st) + except SigninError as e: + log.warning("prepare: saved login rejected: %s", e) + self._error = MSG_RELOGIN + self._mark_need_login() except Exception as e: log.warning("prepare: %s", e) @@ -77,6 +88,7 @@ class Controller: started = False try: st = self._signed_in(installed) + self._need_login = False if self._is_on(st): return self._state() self._refresh_rtts() @@ -116,6 +128,7 @@ class Controller: if self._is_on(st): self._reapply(st) except SigninError: + self._mark_need_login() return self._fail(MSG_RELOGIN) except Exception as e: log.exception("node change failed") @@ -137,9 +150,31 @@ class Controller: self._error = MSG_STILL_UP return True - def shutdown(self) -> None: - """Plugin unload (Deck shutdown/restart, Decky restart): never leave a tunnel behind.""" + def start_login(self) -> dict: + """Open the short-lived LAN relay to the launcher UI; the panel shows its URL as a QR code.""" with self._lock: + if self._relay is not None and self._relay.active: + return self._state() + ip = self.system.lan_ip() + if not ip or self.relay_factory is None: + self._notice = MSG_NO_LAN + return self._state() + relay = self.relay_factory() + port = relay.start() + self._relay = relay + self._relay_url = f"http://{ip}:{port}/_mfd/{relay.token}" + log.info("login relay open on %s:%d for %ss", ip, port, int(relay.expires_at - self.clock())) + return self._state() + + def stop_login(self) -> dict: + with self._lock: + self._close_relay() + return self._state() + + def shutdown(self) -> None: + """Plugin unload (Deck shutdown/restart, Decky restart): never leave a tunnel or relay behind.""" + with self._lock: + self._close_relay() if self._tunnel_present(): log.info("unload; turning VPN off") self._teardown() @@ -150,11 +185,42 @@ class Controller: self._error = message return self._state() + def _mark_need_login(self) -> None: + """Remember which session failed so state() can tell when a new login (e.g. from the phone) lands.""" + self._need_login = True + try: + st = self.client.status() + self._failed_jwt = st.get("jwt") if st.get("isSignedIn") else None + except Exception: + self._failed_jwt = None + def _abort(self, started: bool, message: str) -> dict: + if message == MSG_RELOGIN: + self._mark_need_login() if started and not self._teardown(): message = MSG_STILL_UP return self._fail(message) + def _track_login(self, st: Optional[dict]) -> None: + """Drop an expired relay; once a new session appears after a failed login, close the relay.""" + if self._relay is not None and not self._relay.active: + self._close_relay() + if not (self._need_login and st and st.get("isSignedIn") and st.get("jwt")): + return + if st["jwt"] == self._failed_jwt: + return + self._need_login = False + self._failed_jwt = None + if self._error == MSG_RELOGIN: + self._error = None + self._close_relay() + self._notice = MSG_LOGGED_IN + + def _close_relay(self) -> None: + if self._relay is not None: + self._relay.stop() + self._relay = self._relay_url = None + def _wait(self, cond, seconds) -> bool: for _ in range(seconds): if cond(): @@ -291,7 +357,8 @@ class Controller: def _state(self) -> dict: data = self.store.load() out = {"phase": "off", "message": None, "node_mode": data.get("node_mode") or "auto", "node": None, - "ip": None, "country": None, "connected_since": None, "tunnel": False} + "ip": None, "country": None, "connected_since": None, "tunnel": False, + "need_login": False, "login": None} if not self.system.find_mudfish(): out.update(phase="missing", message=MSG_MISSING) return out @@ -299,6 +366,10 @@ class Controller: st = self.client.status() except Exception: st = None + self._track_login(st) + out["need_login"] = self._need_login + if self._relay is not None: + out["login"] = {"url": self._relay_url, "expires_at": self._relay.expires_at} tap = self.system.tap_up() on = bool(st) and self._is_on(st) out["tunnel"] = tap or bool(st and st.get("core_running")) diff --git a/py_modules/mudfish_decky/relay.py b/py_modules/mudfish_decky/relay.py new file mode 100644 index 0000000..c07ee17 --- /dev/null +++ b/py_modules/mudfish_decky/relay.py @@ -0,0 +1,151 @@ +"""Short-lived LAN relay to the Mudfish launcher UI so a phone can log in by scanning a QR code. + +The launcher only listens on 127.0.0.1. While a login is pending we open a random LAN port that +forwards raw TCP to it, guarded by a one-time token: the first client to request /_mfd/ becomes +the only client allowed through; everyone else gets 403. The relay closes itself after `ttl` seconds. +""" +import logging +import secrets +import socket +import threading +import time +from typing import Optional + +TOKEN_PREFIX = "/_mfd/" +PEEK_TIMEOUT = 10 + +log = logging.getLogger("mudfish_decky") + +_REDIRECT = b"HTTP/1.1 302 Found\r\nLocation: /\r\nContent-Length: 0\r\nConnection: close\r\n\r\n" +_FORBIDDEN = b"HTTP/1.1 403 Forbidden\r\nContent-Length: 9\r\nConnection: close\r\n\r\nForbidden" + + +def parse_request_path(data: bytes) -> Optional[str]: + """Path from an HTTP request line, or None if the first line is incomplete or not HTTP.""" + line, sep, _ = data.partition(b"\r\n") + if not sep: + return None + parts = line.split(b" ") + if len(parts) != 3 or not parts[2].startswith(b"HTTP/"): + return None + try: + return parts[1].decode("ascii") + except UnicodeDecodeError: + return None + + +class Gate: + def __init__(self, token: str): + self.token = token + self.client: Optional[str] = None + self._lock = threading.Lock() + + def decide(self, ip: str, path: Optional[str]) -> str: + is_token = path is not None and path.split("?", 1)[0] == TOKEN_PREFIX + self.token + with self._lock: + if self.client is None and is_token: + self.client = ip + if ip != self.client: + return "deny" + return "authorize" if is_token else "pipe" + + +def _pipe(src, dst): + try: + while True: + data = src.recv(65536) + if not data: + break + dst.sendall(data) + except OSError: + pass + finally: + for s in (src, dst): + try: + s.shutdown(socket.SHUT_RDWR) + except OSError: + pass + + +class LoginRelay: + def __init__(self, target=("127.0.0.1", 8282), ttl: int = 300, clock=time.time): + self.target = target + self.ttl = ttl + self.clock = clock + self.token = secrets.token_urlsafe(16) + self.gate = Gate(self.token) + self.expires_at = 0.0 + self._srv: Optional[socket.socket] = None + self._stopped = threading.Event() + self._conns = set() + self._conns_lock = threading.Lock() + + @property + def active(self) -> bool: + return self._srv is not None and not self._stopped.is_set() and self.clock() < self.expires_at + + def start(self, host: str = "0.0.0.0") -> int: + self._srv = socket.create_server((host, 0)) + self._srv.settimeout(1) + self.expires_at = self.clock() + self.ttl + threading.Thread(target=self._serve, daemon=True).start() + return self._srv.getsockname()[1] + + def stop(self) -> None: + self._stopped.set() + if self._srv is not None: + # shutdown() first: close() alone does not stop a listener another thread is blocked on. + for op in (lambda: self._srv.shutdown(socket.SHUT_RDWR), self._srv.close): + try: + op() + except OSError: + pass + with self._conns_lock: + conns, self._conns = self._conns, set() + for c in conns: + try: + c.close() + except OSError: + pass + + def _track(self, sock): + with self._conns_lock: + self._conns.add(sock) + + def _serve(self): + while self.active: + try: + conn, addr = self._srv.accept() + except socket.timeout: + continue + except OSError: + break + self._track(conn) + threading.Thread(target=self._handle, args=(conn, addr[0]), daemon=True).start() + self.stop() + log.info("login relay closed") + + def _handle(self, conn, ip): + try: + conn.settimeout(PEEK_TIMEOUT) + path = parse_request_path(conn.recv(4096, socket.MSG_PEEK)) + decision = self.gate.decide(ip, path) + if decision in ("authorize", "deny"): + conn.recv(4096) # consume the request; closing with unread data would send RST + conn.sendall(_REDIRECT if decision == "authorize" else _FORBIDDEN) + conn.shutdown(socket.SHUT_WR) + conn.close() + if decision == "authorize": + log.info("login relay: phone %s authorized", ip) + return + conn.settimeout(None) + upstream = socket.create_connection(self.target, timeout=5) + upstream.settimeout(None) + self._track(upstream) + threading.Thread(target=_pipe, args=(conn, upstream), daemon=True).start() + _pipe(upstream, conn) + except OSError: + try: + conn.close() + except OSError: + pass diff --git a/py_modules/mudfish_decky/system.py b/py_modules/mudfish_decky/system.py index 18b021e..48211a4 100644 --- a/py_modules/mudfish_decky/system.py +++ b/py_modules/mudfish_decky/system.py @@ -71,6 +71,16 @@ def route_text(run=subprocess.run) -> str: return run(["ip", "-4", "route"], capture_output=True, text=True, timeout=5).stdout +def lan_ip(run=subprocess.run) -> Optional[str]: + """The Deck's source address for outbound traffic (its Wi-Fi/LAN IP when the VPN is off).""" + try: + out = run(["ip", "-4", "route", "get", "1.1.1.1"], capture_output=True, text=True, timeout=5).stdout + except Exception: + return None + parts = out.split() + return parts[parts.index("src") + 1] if "src" in parts[:-1] else None + + def public_ip() -> Optional[dict]: try: with urllib.request.urlopen("https://ipinfo.io/json", timeout=8, context=ssl_context()) as resp: diff --git a/src/LoginModal.tsx b/src/LoginModal.tsx new file mode 100644 index 0000000..cdf26ae --- /dev/null +++ b/src/LoginModal.tsx @@ -0,0 +1,88 @@ +import { DialogButton, ModalRoot } from "@decky/ui"; +import qrcode from "qrcode-generator"; +import { ReactElement, useEffect, useState } from "react"; +import { getState, startLogin, stopLogin, VpnState } from "./backend"; + +const POLL_MS = 2000; + +function QrSvg({ text, size }: { text: string; size: number }) { + const qr = qrcode(0, "M"); + qr.addData(text); + qr.make(); + const count = qr.getModuleCount(); + const quiet = 2; + const cells: ReactElement[] = []; + for (let r = 0; r < count; r++) { + for (let c = 0; c < count; c++) { + if (qr.isDark(r, c)) cells.push(); + } + } + const dim = count + quiet * 2; + return ( + + + {cells} + + ); +} + +function remaining(expiresAt: number): string { + const s = Math.max(0, Math.round(expiresAt - Date.now() / 1000)); + return `${Math.floor(s / 60)}:${String(s % 60).padStart(2, "0")}`; +} + +export function LoginModal({ closeModal }: { closeModal?: () => void }) { + const [state, setState] = useState(null); + const [, tick] = useState(0); + + useEffect(() => { + let alive = true; + startLogin().then((s) => alive && setState(s)); + const id = setInterval(async () => { + try { + const s = await getState(); + if (!alive) return; + setState(s); + if (!s.need_login) closeModal?.(); // logged in: backend already closed the relay + } catch (e) { + console.error("mudfish: getState failed", e); + } + tick((n) => n + 1); + }, POLL_MS); + return () => { + alive = false; + clearInterval(id); + }; + }, []); + + const cancel = () => { + stopLogin(); + closeModal?.(); + }; + + const login = state?.login; + return ( + +
+
+ {login ? :
} +
+
+
휴대폰으로 미꾸라지 로그인
+ {login ? ( + <> +
1. 스팀덱과 같은 와이파이에 연결된 휴대폰으로 QR을 찍으세요.
+
2. 열린 미꾸라지 화면에서 '로그인 기억하기'를 체크하고 로그인하세요.
+
3. 로그인이 확인되면 이 창은 자동으로 닫혀요.
+
{login.url}
+
{`남은 시간 ${remaining(login.expires_at)} · 처음 QR을 찍은 휴대폰만 접속할 수 있어요`}
+ + ) : ( +
{state?.message ?? "로그인 창을 여는 중…"}
+ )} + 닫기 +
+
+ + ); +} diff --git a/src/Panel.tsx b/src/Panel.tsx index 422f680..9a5e27b 100644 --- a/src/Panel.tsx +++ b/src/Panel.tsx @@ -4,8 +4,10 @@ import { Field, PanelSection, PanelSectionRow, + showModal, ToggleField, } from "@decky/ui"; +import { LoginModal } from "./LoginModal"; import { useEffect, useRef, useState } from "react"; import { connect, @@ -158,6 +160,13 @@ export function Panel() { )} + {state.need_login && ( + + showModal()}> + 휴대폰으로 로그인 (QR) + + + )} setAdvanced(!advanced)}> {advanced ? "고급 ▾" : "고급 ▸"} diff --git a/src/backend.ts b/src/backend.ts index 8b5dbba..4af6e9e 100644 --- a/src/backend.ts +++ b/src/backend.ts @@ -17,6 +17,8 @@ export interface VpnState { country: string | null; connected_since: number | null; tunnel: boolean; + need_login: boolean; + login: { url: string; expires_at: number } | null; } export const getState = callable<[], VpnState>("get_state"); @@ -24,3 +26,5 @@ export const connect = callable<[], VpnState>("connect"); export const disconnect = callable<[], VpnState>("disconnect"); export const setNode = callable<[choice: string], VpnState>("set_node"); export const listNodes = callable<[], NodeInfo[]>("list_nodes"); +export const startLogin = callable<[], VpnState>("start_login"); +export const stopLogin = callable<[], VpnState>("stop_login"); diff --git a/tests/test_controller.py b/tests/test_controller.py index 8dea160..4a9cf38 100644 --- a/tests/test_controller.py +++ b/tests/test_controller.py @@ -3,8 +3,8 @@ import tempfile import unittest from mudfish_decky.client import AuthError, SigninError -from mudfish_decky.controller import (MSG_MISSING, MSG_NO_RESPONSE, MSG_RELOGIN, MSG_RTT, MSG_SLEEP, MSG_STILL_UP, - MSG_TIMEOUT, Controller) +from mudfish_decky.controller import (MSG_LOGGED_IN, MSG_MISSING, MSG_NO_LAN, MSG_NO_RESPONSE, MSG_RELOGIN, MSG_RTT, + MSG_SLEEP, MSG_STILL_UP, MSG_TIMEOUT, Controller) from mudfish_decky.store import Store from tests.test_nodes import STATIC @@ -20,6 +20,10 @@ class FakeSystem: self.deleted = [] self.daemon_calls = 0 self.ip = {"ip": "13.73.234.162", "country": "JP"} + self.lan = "192.168.1.50" + + def lan_ip(self): + return self.lan def find_mudfish(self): return ("/opt/mudfish/6.6.0/bin/mudrun-headless", "/opt/mudfish/6.6.0/var") if self.installed else None @@ -41,6 +45,25 @@ class FakeSystem: return self.ip +class FakeRelay: + token = "tok" + expires_at = 1300.0 + + def __init__(self): + self.active = False + self.started = False + self.start_calls = 0 + + def start(self, host="0.0.0.0"): + self.started = True + self.active = True + self.start_calls += 1 + return 40123 + + def stop(self): + self.active = False + + class FakeClient: def __init__(self, system): self.system = system @@ -333,6 +356,86 @@ class ControllerTest(unittest.TestCase): self.assertEqual(st["message"], MSG_RTT) self.assertEqual(self.client.start_calls, 0) + # ---- phone (QR) login -------------------------------------------------------------------- + + def make_with_relay(self): + self.relay = FakeRelay() + return Controller(self.client, self.system, Store(self.path), clock=lambda: self.now, + sleep=lambda s: None, relay_factory=lambda: self.relay) + + def test_connect_signin_failure_flags_need_login(self): + self.client.signin_error = SigninError("401") + self.assertTrue(self.make().connect()["need_login"]) + + def test_prepare_signin_failure_flags_need_login(self): + self.client.signin_error = SigninError("401") + c = self.make() + c.prepare() + st = c.state() + self.assertTrue(st["need_login"]) + self.assertEqual(st["message"], MSG_RELOGIN) + + def test_prepare_network_failure_does_not_flag_need_login(self): + self.client.signin_error = OSError("name resolution") + c = self.make() + c.prepare() + self.assertFalse(c.state()["need_login"]) + + def test_need_login_false_by_default(self): + self.assertFalse(self.make().state()["need_login"]) + + def test_start_login_exposes_qr_url(self): + c = self.make_with_relay() + st = c.start_login() + self.assertEqual(st["login"], {"url": "http://192.168.1.50:40123/_mfd/tok", "expires_at": 1300.0}) + self.assertTrue(self.relay.started) + self.assertEqual(self.client.start_calls, 0) + + def test_start_login_reuses_active_relay(self): + c = self.make_with_relay() + c.start_login() + c.start_login() + self.assertEqual(self.relay.start_calls, 1) + + def test_start_login_without_lan_ip_shows_message(self): + self.system.lan = None + st = self.make_with_relay().start_login() + self.assertIsNone(st["login"]) + self.assertEqual(st["message"], MSG_NO_LAN) + self.assertFalse(self.relay.started) + + def test_state_closes_relay_after_signin(self): + self.client.signin_error = SigninError("401") + self.client.st["isSignedIn"] = False + c = self.make_with_relay() + c.connect() + c.start_login() + self.client.st["isSignedIn"] = True # user logged in from the phone + st = c.state() + self.assertFalse(self.relay.active) + self.assertIsNone(st["login"]) + self.assertFalse(st["need_login"]) + self.assertEqual(st["phase"], "off") + self.assertEqual(st["message"], MSG_LOGGED_IN) + + def test_stop_login_closes_relay(self): + c = self.make_with_relay() + c.start_login() + self.assertIsNone(c.stop_login()["login"]) + self.assertFalse(self.relay.active) + + def test_expired_relay_drops_from_state(self): + c = self.make_with_relay() + c.start_login() + self.relay.active = False + self.assertIsNone(c.state()["login"]) + + def test_shutdown_closes_relay(self): + c = self.make_with_relay() + c.start_login() + c.shutdown() + self.assertFalse(self.relay.active) + # ---- sleep / shutdown -------------------------------------------------------------------- def test_resume_jump_turns_tunnel_off_with_notice(self): diff --git a/tests/test_relay.py b/tests/test_relay.py new file mode 100644 index 0000000..ccfa8e5 --- /dev/null +++ b/tests/test_relay.py @@ -0,0 +1,111 @@ +import socket +import threading +import unittest + +from mudfish_decky.relay import Gate, LoginRelay, parse_request_path + + +def http_get(port, path): + with socket.create_connection(("127.0.0.1", port), timeout=5) as s: + s.sendall(f"GET {path} HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n".encode()) + chunks = [] + while True: + data = s.recv(4096) + if not data: + break + chunks.append(data) + return b"".join(chunks).decode() + + +class FakeTarget: + """Minimal HTTP server standing in for the Mudfish launcher.""" + + def __init__(self): + self.srv = socket.create_server(("127.0.0.1", 0)) + self.port = self.srv.getsockname()[1] + threading.Thread(target=self._serve, daemon=True).start() + + def _serve(self): + while True: + try: + conn, _ = self.srv.accept() + except OSError: + return + with conn: + conn.recv(4096) + conn.sendall(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello") + + def close(self): + self.srv.close() + + +class GateTest(unittest.TestCase): + def test_denies_until_token_seen(self): + g = Gate("tok") + self.assertEqual(g.decide("10.0.0.2", "/"), "deny") + self.assertEqual(g.decide("10.0.0.2", "/_mfd/wrong"), "deny") + + def test_first_token_holder_is_the_only_client(self): + g = Gate("tok") + self.assertEqual(g.decide("10.0.0.2", "/_mfd/tok"), "authorize") + self.assertEqual(g.decide("10.0.0.2", "/do/mudrun/status"), "pipe") + self.assertEqual(g.decide("10.0.0.3", "/_mfd/tok"), "deny") + self.assertEqual(g.decide("10.0.0.3", "/"), "deny") + + def test_query_string_on_token_path(self): + self.assertEqual(Gate("tok").decide("10.0.0.2", "/_mfd/tok?x=1"), "authorize") + + +class ParseTest(unittest.TestCase): + def test_parse_request_path(self): + self.assertEqual(parse_request_path(b"GET /_mfd/tok HTTP/1.1\r\nHost: x\r\n\r\n"), "/_mfd/tok") + + def test_parse_partial_or_garbage(self): + self.assertIsNone(parse_request_path(b"GET /incomplete")) + self.assertIsNone(parse_request_path(b"\x16\x03\x01binary")) + + +class LoginRelayTest(unittest.TestCase): + def setUp(self): + self.target = FakeTarget() + self.relay = LoginRelay(target=("127.0.0.1", self.target.port), ttl=60) + self.port = self.relay.start(host="127.0.0.1") + + def tearDown(self): + self.relay.stop() + self.target.close() + + def test_token_is_random_and_url_safe(self): + other = LoginRelay(target=("127.0.0.1", 1)) + self.assertNotEqual(self.relay.token, other.token) + self.assertGreaterEqual(len(self.relay.token), 16) + self.assertRegex(self.relay.token, r"^[A-Za-z0-9_-]+$") + + def test_forbidden_before_token(self): + self.assertIn("403", http_get(self.port, "/").splitlines()[0]) + + def test_token_redirects_then_proxies(self): + first = http_get(self.port, f"/_mfd/{self.relay.token}") + self.assertIn("302", first.splitlines()[0]) + self.assertIn("Location: /", first) + self.assertTrue(http_get(self.port, "/").endswith("hello")) + + def test_stop_closes_port(self): + self.relay.stop() + self.assertFalse(self.relay.active) + with self.assertRaises(OSError): + socket.create_connection(("127.0.0.1", self.port), timeout=2).close() + + def test_expires_after_ttl(self): + now = [1000.0] + relay = LoginRelay(target=("127.0.0.1", self.target.port), ttl=300, clock=lambda: now[0]) + relay.start(host="127.0.0.1") + self.assertEqual(relay.expires_at, 1300.0) + self.assertTrue(relay.active) + now[0] = 1301.0 + self.assertFalse(relay.active) + relay.stop() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_system.py b/tests/test_system.py index e395e25..803438e 100644 --- a/tests/test_system.py +++ b/tests/test_system.py @@ -35,6 +35,16 @@ class SystemTest(unittest.TestCase): with tempfile.TemporaryDirectory() as var: self.assertEqual(api_base(var), "http://127.0.0.1:8282") + def test_lan_ip_from_route_get(self): + from mudfish_decky.system import lan_ip + out = "1.1.1.1 via 192.168.1.1 dev wlan0 src 192.168.1.50 uid 0 \n cache \n" + self.assertEqual(lan_ip(run=lambda cmd, **kw: types.SimpleNamespace(stdout=out, returncode=0)), + "192.168.1.50") + + def test_lan_ip_none_without_route(self): + from mudfish_decky.system import lan_ip + self.assertIsNone(lan_ip(run=lambda cmd, **kw: types.SimpleNamespace(stdout="", returncode=2))) + def test_sleep_offset_is_boottime_minus_monotonic(self): from mudfish_decky.system import sleep_offset clocks = {"boot": 500.0, "mono": 420.0}