ysj7705andClaude Opus 5.5
7513fef6f8
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 <noreply@anthropic.com>
91 lines
3.0 KiB
Python
91 lines
3.0 KiB
Python
"""Steam Deck side effects: locating Mudfish, running its daemon, reading routes and the public IP."""
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import time
|
|
import urllib.request
|
|
from typing import Callable, Optional, Tuple
|
|
|
|
from .client import ssl_context
|
|
from .routes import delete_routes # noqa: F401 (re-exported for the controller)
|
|
|
|
MUDFISH_ROOT = "/opt/mudfish"
|
|
DAEMON_UNIT = "mudfish-decky"
|
|
DEFAULT_API = "http://127.0.0.1:8282"
|
|
|
|
|
|
def _version_key(name: str):
|
|
try:
|
|
return tuple(int(p) for p in name.split("."))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def find_mudfish(root: str = MUDFISH_ROOT) -> Optional[Tuple[str, str]]:
|
|
"""(daemon binary, var dir) of the highest installed Mudfish version."""
|
|
if not os.path.isdir(root):
|
|
return None
|
|
versions = []
|
|
for name in os.listdir(root):
|
|
key = _version_key(name)
|
|
binary = os.path.join(root, name, "bin", "mudrun-headless")
|
|
if key is not None and os.path.isfile(binary):
|
|
versions.append((key, name))
|
|
if not versions:
|
|
return None
|
|
name = max(versions)[1]
|
|
return os.path.join(root, name, "bin", "mudrun-headless"), os.path.join(root, name, "var")
|
|
|
|
|
|
def api_base(var_dir: str) -> str:
|
|
try:
|
|
with open(os.path.join(var_dir, ".ipport")) as f:
|
|
ipport = f.read().strip()
|
|
except OSError:
|
|
return DEFAULT_API
|
|
return f"http://{ipport}" if ipport else DEFAULT_API
|
|
|
|
|
|
def ensure_daemon(bin_path: str, is_up: Callable[[], bool], run=subprocess.run, sleep=time.sleep) -> None:
|
|
"""Start the launcher as a transient systemd unit so it outlives desktop/game mode switches."""
|
|
if is_up():
|
|
return
|
|
run(["systemd-run", f"--unit={DAEMON_UNIT}", "--collect", bin_path, "-B"], capture_output=True, timeout=10)
|
|
for _ in range(20):
|
|
sleep(0.5)
|
|
if is_up():
|
|
return
|
|
|
|
|
|
def sleep_offset(clock=time.clock_gettime, boot=getattr(time, "CLOCK_BOOTTIME", None),
|
|
mono=time.CLOCK_MONOTONIC) -> float:
|
|
"""Total time spent suspended since boot: BOOTTIME keeps counting during suspend, MONOTONIC doesn't."""
|
|
return clock(boot) - clock(mono)
|
|
|
|
|
|
def tap_up() -> bool:
|
|
return os.path.exists("/sys/class/net/tap0")
|
|
|
|
|
|
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:
|
|
data = json.load(resp)
|
|
return {"ip": data.get("ip"), "country": data.get("country")}
|
|
except Exception:
|
|
return None
|