feat: build CouchOS Melee appliance
Some checks failed
ci / verify (push) Has been cancelled

This commit is contained in:
jackmhny
2026-08-03 02:52:28 +02:00
commit 04ef2a0d74
42 changed files with 3671 additions and 0 deletions

26
.github/workflows/ci.yml vendored Normal file
View File

@@ -0,0 +1,26 @@
name: ci
on:
push:
pull_request:
jobs:
verify:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/setup-python@v5
with:
python-version: '3.11'
- name: Unit tests
run: PYTHONPATH=src python -m unittest
- name: Compile checks
run: python -m py_compile src/couchd/*.py
- name: Shell syntax
run: |
bash -n setup.sh
for file in scripts/*.sh; do
bash -n "$file"
done
- name: Diff check
run: git diff --check

15
.gitignore vendored Normal file
View File

@@ -0,0 +1,15 @@
__pycache__/
*.py[cod]
.pytest_cache/
.coverage
htmlcov/
.mypy_cache/
.ruff_cache/
dist/
build/
*.egg-info/
.venv/
.env
config/couchd.token
logs/
artifacts/

127
README.md Normal file
View File

@@ -0,0 +1,127 @@
# CouchOS
CouchOS targets the `couchos` laptop: a Debian 13 living-room Melee appliance for the `couch` user. This repository is being built remotely from a separate workstation. `couchd` is intended to bind only to Tailnet IP `100.64.0.15`, and the dashboard is intended to be served same-origin at `couch.mhny.dev`.
As of Monday, August 3, 2026:
- The main root-side blocker on the target laptop is still interactive `sudo` access. The repo contains the setup and verification logic, but it has not claimed to apply privileged changes from this remote implementation workstation.
- The TV currently advertises only `1920x1080@60.00` on `HDMI-1`. There is no verified 120 Hz mode to automate today.
- The Xbox controller is known at `C8:3F:26:12:69:7B`. It is presently paired and connected, but it is not yet trusted under `jackmhny`.
- An HDMI PulseAudio sink is currently absent on the target laptop, so HDMI sink selection remains deferred until PulseAudio exposes one.
- Dashboard DNS already exists externally. This repo covers the dashboard files and Caddy snippet, not DNS creation.
## Layout
- `src/couchd/`: stdlib-only Python control daemon, launcher, discovery, and status probes
- `setup.sh`: idempotent root/couch setup with `--dry-run` and PASS/FAIL `--check`
- `scripts/check.sh`: local/CI verification entrypoint from a clean checkout
- `scripts/preflight.sh`: deployment evidence checks plus reboot/soak evidence capture
- `scripts/install_slippi.sh`: official launcher installer with checksum fail-closed behavior
- `scripts/backup_restore.sh`: scoped backup/restore without launcher account/session data, ROMs, replays, or tokens
- `systemd/systemd-user/`: user units installed under `/etc/systemd/user` and started from the `couch` graphical session
- `dashboard/`: same-origin static UI
- `config/`: non-secret config templates plus LightDM/logind/Caddy config
## Automated Vs Manual
Automated by this repo:
- Stable install under `/home/couch/couchos`
- LightDM autologin config for `couch` into XFCE
- logind lid-ignore on AC power
- `couchd` and display/audio user units bound to the `couch` graphical session via XFCE autostart entries
- Runtime config generation with the real `couch` UID and runtime dir
- First-apply bearer token generation with mode `0600`, preserving any existing token
- XFCE compositing off, blanking/suspend off, lock/screensaver off, notification daemon/update-notifier autostart suppressed
- Tailnet-only daemon binding, constant-time bearer auth, strict route allowlist, safe process adoption/stop logic
- Dashboard static assets and Caddy reverse-proxy snippet with server-side bearer injection
Still manual or operator-observed:
- Installing the privileged changes on the actual laptop
- Setting a real ISO path under `/home/couch`
- Confirming actual launcher fallback entries discovered on the machine
- First-run Slippi UI steps and controller profile mapping
- Confirming autologin/session behavior, five reboot controller behavior, and 30-minute gameplay/FPS observations
## `couchd`
Exposed routes:
- `GET /status`
- `POST /launch/melee`
- `POST /open/slippi`
- `POST /stop`
Implemented constraints:
- Tailscale IPv4 bind only; loopback and `0.0.0.0` are rejected
- Constant-time bearer token comparison
- Exact route allowlist with safe query-string stripping
- Nonempty action request bodies rejected
- Request body size capped
- Expected config/discovery/launch failures converted to JSON `409`; unexpected action failures return JSON `503`
- State persisted atomically and corrupt/stale state tolerated
- Existing couch-owned Slippi/Dolphin processes under allowed roots can be adopted to prevent duplicates
- Stop logic verifies PID identity and allowed roots before sending `TERM`/`KILL`
- Display status reports the active starred refresh, for example `HDMI-1 1920x1080@60.00`
- Launcher and Dolphin version probes degrade safely on missing commands or timeouts
## Setup
Root/system phase:
```bash
sudo ./setup.sh --phase root --dry-run
sudo ./setup.sh --phase root
```
Couch/user phase:
```bash
sudo ./setup.sh --phase couch --dry-run
sudo ./setup.sh --phase couch
```
Notes:
- `setup.sh` does not install Caddy. The deployed site can stay behind an existing Caddy `basicauth` block or another external public auth layer such as Cloudflare Access.
- The generated runtime config keeps `iso_path`, `launcher_shortcut`, and `xbox_bluetooth_address` null unless the operator supplies real values.
- Launcher fallback is not invented. If direct Dolphin launch is unavailable, configure only a discovered `gtk-launch` desktop entry that the target machine actually exposes.
- On first apply, `setup.sh` generates a cryptographically random bearer token at `/home/couch/.config/couchd/token` with mode `0600` and leaves an existing token untouched.
- To synchronize that token with Heavy Caddy without printing or committing it, read it locally on the laptop as root and copy the same value into the Caddy environment source used by the deployed site, then reload Caddy after updating that secret.
## Dashboard And Caddy
Use [config/caddy/couch.mhny.dev.caddy](/home/jackmhny/couchos/config/caddy/couch.mhny.dev.caddy) as the site snippet.
What this snippet assumes:
- An existing Caddy `basicauth` block, Cloudflare Access, or equivalent public auth already exists outside this repo
- The static dashboard is deployed at `/var/www/couchos`
- Caddy injects `Authorization: Bearer {env.COUCHD_BEARER_TOKEN}` server-side before proxying `/api/*` to `100.64.0.15:8765`
## Verification
Project checks:
```bash
./scripts/check.sh
```
Deployment checks:
```bash
./setup.sh --check
./scripts/preflight.sh quick
./scripts/preflight.sh five-reboots
./scripts/preflight.sh soak-30m-start
./scripts/preflight.sh soak-30m-finish
```
`five-reboots` and soak mode now record real boot IDs, controller snapshots, and honest operator-observation placeholders instead of just incrementing counters.
## Recovery
- `./scripts/backup_restore.sh backup [archive.tar.gz]`
- `./scripts/backup_restore.sh restore <archive.tar.gz>`
Backups are mode `0600` and intentionally exclude:
- launcher account/session data
- replays
- ROMs
- bearer tokens
## External References
- Slippi downloads: https://slippi.gg/downloads
- Slippi launcher publish metadata: https://github.com/project-slippi/slippi-launcher/blob/main/electron-builder.json

View File

@@ -0,0 +1,21 @@
couch.mhny.dev {
root * /var/www/couchos
handle_path /api/* {
reverse_proxy 100.64.0.15:8765 {
header_up Authorization "Bearer {env.COUCHD_BEARER_TOKEN}"
}
}
handle {
file_server
}
header {
Content-Security-Policy "default-src 'self'; img-src 'self' data:; style-src 'self' 'unsafe-inline'; script-src 'self'; connect-src 'self'; base-uri 'none'; frame-ancestors 'none'"
Permissions-Policy "camera=(), geolocation=(), microphone=()"
Referrer-Policy "no-referrer"
X-Content-Type-Options "nosniff"
X-Frame-Options "DENY"
}
}

View File

@@ -0,0 +1,12 @@
{
"bind_address": "100.64.0.15",
"port": 8765,
"token_file": "/home/couch/.config/couchd/token",
"couch_home": "/home/couch",
"iso_path": null,
"state_file": "/home/couch/.local/state/couchd/state.json",
"display": ":0",
"xdg_runtime_dir": null,
"xbox_bluetooth_address": null,
"launcher_shortcut": null
}

View File

@@ -0,0 +1,4 @@
[Seat:*]
autologin-user=couch
autologin-session=xfce
user-session=xfce

View File

@@ -0,0 +1,4 @@
[Login]
HandleLidSwitchExternalPower=ignore
HandleLidSwitchDocked=ignore
IdleAction=ignore

36
dashboard/app.js Normal file
View File

@@ -0,0 +1,36 @@
async function refreshStatus() {
const output = document.getElementById("output");
try {
const response = await fetch("/api/status", { credentials: "same-origin" });
const payload = await response.json();
document.getElementById("online").textContent = response.ok ? "online" : "offline";
document.getElementById("game-state").textContent = payload.state ?? "unknown";
document.getElementById("controller").textContent =
payload.controller?.connected === true ? "connected" :
payload.controller?.connected === false ? "disconnected" : "unknown";
document.getElementById("display").textContent = payload.display?.mode ?? "unknown";
output.textContent = JSON.stringify(payload, null, 2);
} catch (error) {
document.getElementById("online").textContent = "offline";
output.textContent = String(error);
}
}
async function postAction(path) {
const output = document.getElementById("output");
const response = await fetch(path, { method: "POST", credentials: "same-origin" });
const payload = await response.json();
output.textContent = JSON.stringify(payload, null, 2);
await refreshStatus();
}
document.querySelectorAll("[data-action]").forEach((button) => {
button.addEventListener("click", () => {
postAction(button.dataset.action).catch((error) => {
document.getElementById("output").textContent = String(error);
});
});
});
refreshStatus().catch(() => {});
setInterval(refreshStatus, 5000);

43
dashboard/index.html Normal file
View File

@@ -0,0 +1,43 @@
<!doctype html>
<html lang="en">
<head>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width, initial-scale=1">
<title>CouchOS</title>
<link rel="stylesheet" href="/style.css">
</head>
<body>
<main class="shell">
<section class="panel">
<p class="eyebrow">couch.mhny.dev</p>
<h1>CouchOS</h1>
<p class="lede">Living-room Melee controls over the Tailnet-backed API.</p>
<div class="status-grid">
<div>
<span class="label">Online</span>
<strong id="online">unknown</strong>
</div>
<div>
<span class="label">Game</span>
<strong id="game-state">unknown</strong>
</div>
<div>
<span class="label">Controller</span>
<strong id="controller">unknown</strong>
</div>
<div>
<span class="label">Display</span>
<strong id="display">unknown</strong>
</div>
</div>
<div class="actions">
<button data-action="/api/launch/melee">Play Melee</button>
<button data-action="/api/open/slippi">Open Slippi</button>
<button data-action="/api/stop" class="danger">Stop Game</button>
</div>
<pre id="output" aria-live="polite"></pre>
</section>
</main>
<script src="/app.js" defer></script>
</body>
</html>

119
dashboard/style.css Normal file
View File

@@ -0,0 +1,119 @@
:root {
--bg: #f4efe6;
--panel: rgba(255, 252, 246, 0.88);
--ink: #13212e;
--muted: #5b6670;
--accent: #a6412b;
--accent-dark: #7b2f1e;
--line: rgba(19, 33, 46, 0.12);
--shadow: 0 24px 80px rgba(19, 33, 46, 0.16);
}
* {
box-sizing: border-box;
}
body {
margin: 0;
min-height: 100vh;
font-family: "IBM Plex Sans", "Segoe UI", sans-serif;
color: var(--ink);
background:
radial-gradient(circle at top left, rgba(166, 65, 43, 0.18), transparent 28%),
linear-gradient(160deg, #efe5d2 0%, #f8f5ef 45%, #dfe9ea 100%);
}
.shell {
min-height: 100vh;
display: grid;
place-items: center;
padding: 20px;
}
.panel {
width: min(100%, 440px);
padding: 24px;
border: 1px solid var(--line);
border-radius: 24px;
background: var(--panel);
box-shadow: var(--shadow);
backdrop-filter: blur(10px);
}
.eyebrow,
.label {
text-transform: uppercase;
letter-spacing: 0.12em;
font-size: 0.72rem;
color: var(--muted);
}
h1 {
margin: 6px 0 8px;
font-family: "Space Grotesk", "IBM Plex Sans", sans-serif;
font-size: clamp(2rem, 8vw, 3.2rem);
}
.lede {
margin: 0 0 20px;
color: var(--muted);
}
.status-grid {
display: grid;
grid-template-columns: repeat(2, minmax(0, 1fr));
gap: 12px;
}
.status-grid div,
.actions button,
#output {
border-radius: 18px;
border: 1px solid var(--line);
background: rgba(255, 255, 255, 0.56);
}
.status-grid div {
padding: 14px;
}
.status-grid strong {
display: block;
margin-top: 6px;
font-size: 1rem;
}
.actions {
display: grid;
gap: 10px;
margin-top: 18px;
}
.actions button {
appearance: none;
padding: 14px 16px;
font: inherit;
font-weight: 700;
color: var(--ink);
cursor: pointer;
transition: transform 120ms ease, background 120ms ease;
}
.actions button:hover,
.actions button:focus-visible {
transform: translateY(-1px);
background: rgba(255, 255, 255, 0.9);
}
.actions .danger {
color: #fff;
background: linear-gradient(135deg, var(--accent), var(--accent-dark));
}
#output {
min-height: 90px;
margin: 18px 0 0;
padding: 14px;
white-space: pre-wrap;
color: var(--muted);
}

18
pyproject.toml Normal file
View File

@@ -0,0 +1,18 @@
[build-system]
requires = ["setuptools>=68"]
build-backend = "setuptools.build_meta"
[project]
name = "couchos"
version = "0.1.0"
description = "CouchOS living-room Melee appliance tooling"
requires-python = ">=3.11"
[tool.setuptools]
package-dir = {"" = "src"}
[tool.setuptools.packages.find]
where = ["src"]
[tool.pytest.ini_options]
testpaths = ["tests"]

135
scripts/backup_restore.sh Executable file
View File

@@ -0,0 +1,135 @@
#!/usr/bin/env bash
set -Eeuo pipefail
MODE="${1:-}"
ARCHIVE_PATH="${2:-}"
COUCH_HOME="${COUCH_HOME:-/home/couch}"
STAMP="$(date +%Y%m%d-%H%M%S)"
usage() {
cat <<'EOF'
Usage:
./scripts/backup_restore.sh backup [archive.tar.gz]
./scripts/backup_restore.sh restore <archive.tar.gz>
EOF
}
if [[ -z "${MODE}" ]]; then
usage >&2
exit 1
fi
build_backup_manifest() {
python3 - <<'PY' "${COUCH_HOME}"
import os
import sys
from pathlib import Path
couch_home = Path(sys.argv[1])
roots = [
Path(".config/couchd/runtime.json"),
Path(".config/dolphin-emu"),
Path(".config/Slippi"),
Path(".local/share/dolphin-emu"),
Path("Slippi"),
]
excluded_dirs = tuple(
item.lower()
for item in (
".config/Slippi Launcher",
".local/share/accounts",
".local/share/sessions",
"Slippi/Replay",
"Slippi/Replays",
)
)
excluded_suffixes = (".slp", ".iso", ".gcm", ".wbfs", ".nkit.iso")
for root in roots:
source = couch_home / root
if not source.exists():
continue
if source.is_file():
print(str(root), end="\0")
continue
for current_root, dir_names, file_names in os.walk(source):
rel_root = Path(current_root).relative_to(couch_home)
dir_names[:] = [
name
for name in dir_names
if str(rel_root / name).lower() not in excluded_dirs
]
for file_name in file_names:
rel_path = rel_root / file_name
rel_lower = str(rel_path).lower()
if any(rel_lower == item or rel_lower.startswith(f"{item}/") for item in excluded_dirs):
continue
if rel_lower.endswith(excluded_suffixes):
continue
print(str(rel_path), end="\0")
PY
}
validate_restore_archive() {
python3 - <<'PY' "${ARCHIVE_PATH}" "${COUCH_HOME}"
import sys
import tarfile
from pathlib import Path
archive_path = Path(sys.argv[1])
couch_home = Path(sys.argv[2]).resolve(strict=False)
with tarfile.open(archive_path, "r:*") as archive:
for member in archive.getmembers():
member_path = Path(member.name)
if member_path.is_absolute():
raise SystemExit(f"unsafe archive entry: {member.name}")
if any(part == ".." for part in member_path.parts):
raise SystemExit(f"unsafe archive entry: {member.name}")
if member.isdev():
raise SystemExit(f"unsafe archive entry: {member.name}")
target_path = (couch_home / member_path).resolve(strict=False)
if couch_home != target_path and couch_home not in target_path.parents:
raise SystemExit(f"unsafe archive entry: {member.name}")
if member.issym() or member.islnk():
link_target = Path(member.linkname)
if link_target.is_absolute():
raise SystemExit(f"unsafe archive entry: {member.name}")
resolved_link = (target_path.parent / link_target).resolve(strict=False)
if couch_home != resolved_link and couch_home not in resolved_link.parents:
raise SystemExit(f"unsafe archive entry: {member.name}")
PY
}
case "${MODE}" in
backup)
ARCHIVE_PATH="${ARCHIVE_PATH:-${COUCH_HOME}/Slippi/couchos-backup-${STAMP}.tar.gz}"
umask 077
mapfile -d '' manifest < <(build_backup_manifest)
if [[ "${#manifest[@]}" -eq 0 ]]; then
printf 'no backup sources found under %s\n' "${COUCH_HOME}" >&2
exit 1
fi
tar -C "${COUCH_HOME}" -czf "${ARCHIVE_PATH}" \
--exclude='.config/couchd/token' \
"${manifest[@]}"
chmod 0600 "${ARCHIVE_PATH}"
printf 'backup written to %s\n' "${ARCHIVE_PATH}"
;;
restore)
if [[ -z "${ARCHIVE_PATH}" || ! -f "${ARCHIVE_PATH}" ]]; then
printf 'restore archive is required\n' >&2
exit 1
fi
if ! validation_error="$(validate_restore_archive 2>&1)"; then
printf '%s\n' "${validation_error}" >&2
exit 1
fi
tar -C "${COUCH_HOME}" -xzf "${ARCHIVE_PATH}"
printf 'restored from %s\n' "${ARCHIVE_PATH}"
;;
*)
usage >&2
exit 1
;;
esac

62
scripts/check.sh Executable file
View File

@@ -0,0 +1,62 @@
#!/usr/bin/env bash
set -Eeuo pipefail
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
export PYTHONPATH="${REPO_ROOT}/src${PYTHONPATH:+:${PYTHONPATH}}"
run_step() {
local label="$1"
shift
printf '[check] %s\n' "${label}"
"$@"
}
secret_scan() {
python3 - "${REPO_ROOT}" <<'PY'
import re
import sys
from pathlib import Path
root = Path(sys.argv[1])
rom_suffixes = (".iso", ".gcm", ".wbfs", ".nkit.iso")
secret_pattern = re.compile(
r"BEGIN [A-Z ]+PRIVATE KEY|Authorization:\s*Bearer\s+[A-Za-z0-9._~-]{20,}"
)
findings: list[str] = []
for path in root.rglob("*"):
if not path.is_file() or any(part in {".git", "__pycache__", "artifacts"} for part in path.parts):
continue
relative = path.relative_to(root)
if path.name.lower().endswith(rom_suffixes):
findings.append(f"ROM file: {relative}")
continue
if relative in {Path("templates/couchd.token.example"), Path("scripts/check.sh")}:
continue
try:
text = path.read_text(encoding="utf-8")
except (UnicodeDecodeError, OSError):
continue
if secret_pattern.search(text):
findings.append(f"possible secret: {relative}")
if findings:
print("\n".join(findings))
raise SystemExit(1)
PY
}
run_step "unit tests" python3 -m unittest discover -v
run_step "setup shell tests" bash "${REPO_ROOT}/tests/test_setup_modes.sh"
run_step "shell syntax" bash -n "${REPO_ROOT}/setup.sh" "${REPO_ROOT}"/scripts/*.sh
run_step "python compile" python3 -m py_compile "${REPO_ROOT}"/src/couchd/*.py "${REPO_ROOT}"/tests/*.py
if command -v node >/dev/null 2>&1; then
run_step "dashboard JS syntax" node --check "${REPO_ROOT}/dashboard/app.js"
else
printf '[check] dashboard JS syntax skipped: node not installed\n'
fi
if git -C "${REPO_ROOT}" rev-parse --is-inside-work-tree >/dev/null 2>&1; then
run_step "git diff check" git -C "${REPO_ROOT}" diff --check
else
printf '[check] git diff check skipped: deployed tree has no .git directory\n'
fi
run_step "secret and ROM scan" secret_scan

View File

@@ -0,0 +1,85 @@
#!/usr/bin/env bash
set -Eeuo pipefail
log() {
printf '[display-audio] %s\n' "$*"
}
warn() {
printf '[display-audio] WARN: %s\n' "$*" >&2
logger -t couch-display-audio -- "$*" 2>/dev/null || true
}
pick_hdmi_refresh() {
awk '
$1 == "HDMI-1" && $2 == "connected" { in_hdmi=1; next }
in_hdmi && $0 !~ /^[[:space:]]/ { exit }
in_hdmi && $1 == "1920x1080" {
for (i = 2; i <= NF; i++) {
gsub(/[+*]/, "", $i)
if ($i ~ /^[0-9]+(\.[0-9]+)?$/) {
value = $i + 0
if (value > best) {
best = value
best_text = $i
}
}
}
}
END {
if (best_text != "") {
print best_text
}
}
'
}
choose_hdmi_profile() {
pactl list cards | awk '
/^Card #[0-9]+/ { card="" }
/^\tName: / { sub(/^\tName: /, "", $0); card=$0 }
/^\t\toutput:hdmi/ && /available: yes/ {
profile=$1
sub(/:$/, "", profile)
print card "\t" profile
exit
}
'
}
choose_hdmi_sink() {
pactl list short sinks | awk '$2 ~ /\.hdmi/ { print $2; exit }'
}
xrandr_output="$(xrandr --query)"
if ! grep -q '^HDMI-1 connected' <<<"${xrandr_output}"; then
warn "HDMI-1 is not connected; deferring display/audio switch"
exit 0
fi
refresh="$(printf '%s\n' "${xrandr_output}" | pick_hdmi_refresh)"
if [[ -z "${refresh}" ]]; then
warn "HDMI-1 is connected but no 1920x1080 refresh was discovered"
exit 0
fi
log "setting HDMI-1 primary at 1920x1080@${refresh} and disabling eDP-1"
xrandr --output HDMI-1 --primary --mode 1920x1080 --rate "${refresh}" --output eDP-1 --off
profile_line="$(choose_hdmi_profile || true)"
if [[ -z "${profile_line}" ]]; then
warn "HDMI audio sink/profile not available yet; display configured and audio switch deferred"
exit 0
fi
card_name="${profile_line%%$'\t'*}"
profile_name="${profile_line#*$'\t'}"
log "selecting HDMI audio profile ${profile_name} on ${card_name}"
pactl set-card-profile "${card_name}" "${profile_name}"
pactl list short sinks >/dev/null 2>&1 || true
sink_name="$(choose_hdmi_sink || true)"
if [[ -z "${sink_name}" ]]; then
warn "HDMI sink still unavailable after profile switch; audio switch deferred"
exit 0
fi
pactl set-default-sink "${sink_name}"

8
scripts/discover_controller.sh Executable file
View File

@@ -0,0 +1,8 @@
#!/usr/bin/env bash
set -Eeuo pipefail
ADDRESS="${1:-C8:3F:26:12:69:7B}"
bluetoothctl info "${ADDRESS}" || true
printf '\nKnown-safe template: templates/controllers/xbox-standard-controller.profile.ini\n'
printf 'Do not fill in SDL GUIDs or Dolphin device IDs until you confirm them on this machine.\n'

77
scripts/install_slippi.sh Executable file
View File

@@ -0,0 +1,77 @@
#!/usr/bin/env bash
set -Eeuo pipefail
API_URL="${SLIPPI_RELEASE_API_URL:-https://api.github.com/repos/project-slippi/slippi-launcher/releases/latest}"
INSTALL_DIR="${SLIPPI_INSTALL_DIR:-/home/couch/Applications}"
TMP_DIR="$(mktemp -d)"
trap 'rm -rf "${TMP_DIR}"' EXIT
log() {
printf '[install_slippi] %s\n' "$*"
}
require_cmd() {
command -v "$1" >/dev/null 2>&1 || {
printf 'missing required command: %s\n' "$1" >&2
exit 1
}
}
require_cmd curl
require_cmd jq
require_cmd install
mkdir -p "${INSTALL_DIR}"
validate_asset_url() {
local url="$1"
local expected_name_regex="$2"
if [[ ! "${url}" =~ ^https://github\.com/project-slippi/slippi-launcher/releases/download/ ]]; then
printf 'unexpected asset host or path: %s\n' "${url}" >&2
exit 1
fi
local name
name="$(basename "${url}")"
if [[ ! "${name}" =~ ${expected_name_regex} ]]; then
printf 'unexpected asset name: %s\n' "${name}" >&2
exit 1
fi
}
release_json="${TMP_DIR}/release.json"
curl -fsSL "${API_URL}" -o "${release_json}"
appimage_url="$(jq -r '[.assets[] | select(.name | test("^Slippi-Launcher-.*x86_64\\.AppImage$")) | .browser_download_url] | first // empty' "${release_json}")"
checksum_url="$(jq -r '[.assets[] | select(.name | test("(sha256|checksums?)(\\.txt)?$"; "i")) | .browser_download_url] | first // empty' "${release_json}")"
if [[ -z "${appimage_url}" ]]; then
printf 'could not find official Slippi Launcher AppImage asset\n' >&2
exit 1
fi
validate_asset_url "${appimage_url}" '^Slippi-Launcher-.*x86_64\.AppImage$'
appimage_name="$(basename "${appimage_url}")"
appimage_path="${TMP_DIR}/${appimage_name}"
curl -fsSL "${appimage_url}" -o "${appimage_path}"
if [[ -n "${checksum_url}" ]]; then
validate_asset_url "${checksum_url}" '^(sha256|checksums?)(\.[A-Za-z0-9]+)?$'
checksum_path="${TMP_DIR}/checksums.txt"
curl -fsSL "${checksum_url}" -o "${checksum_path}"
if grep -F "${appimage_name}" "${checksum_path}" >/dev/null 2>&1; then
(
cd "${TMP_DIR}"
grep -F "${appimage_name}" "${checksum_path}" | sha256sum --check --status
)
log "checksum verified for ${appimage_name}"
else
printf 'checksum asset present but does not include %s\n' "${appimage_name}" >&2
exit 1
fi
else
log "no upstream checksum asset published for this release; upstream verification unavailable"
fi
install -m 0755 "${appimage_path}" "${INSTALL_DIR}/${appimage_name}"
ln -sfn "${INSTALL_DIR}/${appimage_name}" "${INSTALL_DIR}/SlippiLauncher.AppImage"
log "installed ${INSTALL_DIR}/${appimage_name}"

9
scripts/launch_melee.sh Executable file
View File

@@ -0,0 +1,9 @@
#!/usr/bin/env bash
set -Eeuo pipefail
CONFIG_PATH="${COUCHOS_CONFIG:-/home/couch/.config/couchd/runtime.json}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
export PYTHONPATH="${REPO_ROOT}/src${PYTHONPATH:+:${PYTHONPATH}}"
exec python3 -m couchd --config "${CONFIG_PATH}" launch-melee

9
scripts/open_slippi.sh Executable file
View File

@@ -0,0 +1,9 @@
#!/usr/bin/env bash
set -Eeuo pipefail
CONFIG_PATH="${COUCHOS_CONFIG:-/home/couch/.config/couchd/runtime.json}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
export PYTHONPATH="${REPO_ROOT}/src${PYTHONPATH:+:${PYTHONPATH}}"
exec python3 -m couchd --config "${CONFIG_PATH}" open-slippi

346
scripts/preflight.sh Executable file
View File

@@ -0,0 +1,346 @@
#!/usr/bin/env bash
set -Eeuo pipefail
MODE="${1:-quick}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
export PYTHONPATH="${REPO_ROOT}/src${PYTHONPATH:+:${PYTHONPATH}}"
ARTIFACT_DIR="${ARTIFACT_DIR:-${REPO_ROOT}/artifacts}"
mkdir -p "${ARTIFACT_DIR}"
LOG_FILE="${ARTIFACT_DIR}/preflight-${MODE}.log"
STATE_FILE="${ARTIFACT_DIR}/five-reboots.jsonl"
SOAK_FILE="${ARTIFACT_DIR}/soak-30m.json"
FAILURES=0
COUCH_USER="couch"
RUNTIME_CONFIG="/home/couch/.config/couchd/runtime.json"
record() {
local status="$1"
shift
printf '%s %s\n' "${status}" "$*" | tee -a "${LOG_FILE}"
if [[ "${status}" == "FAIL" ]]; then
FAILURES=$((FAILURES + 1))
fi
}
run_as_couch() {
if [[ "${EUID}" -eq 0 ]]; then
runuser -u "${COUCH_USER}" -- "$@"
elif [[ "$(id -un)" == "${COUCH_USER}" ]]; then
"$@"
else
return 1
fi
}
runtime_field() {
local field="$1"
python3 - <<'PY' "${RUNTIME_CONFIG}" "${field}"
import json
import sys
from pathlib import Path
path = Path(sys.argv[1])
field = sys.argv[2]
data = json.loads(path.read_text(encoding="utf-8"))
value = data.get(field)
if value is None:
raise SystemExit(1)
print(value)
PY
}
controller_snapshot() {
local address
address="$(runtime_field xbox_bluetooth_address 2>/dev/null || true)"
if [[ -z "${address}" ]]; then
printf '{"address": null, "paired": false, "trusted": false, "connected": false}'
return 0
fi
local output
output="$(bluetoothctl info "${address}" 2>/dev/null || true)"
python3 - <<'PY' "${address}" "${output}"
import json
import sys
address = sys.argv[1]
text = sys.argv[2]
print(json.dumps({
"address": address,
"paired": "Paired: yes" in text,
"trusted": "Trusted: yes" in text,
"connected": "Connected: yes" in text,
}))
PY
}
status_json() {
python3 -m couchd --config "${RUNTIME_CONFIG}" status
}
finish() {
if [[ "${FAILURES}" -gt 0 ]]; then
exit 1
fi
}
quick_mode() {
: >"${LOG_FILE}"
if id "${COUCH_USER}" >/dev/null 2>&1; then
record PASS "couch user exists"
else
record FAIL "couch user missing"
fi
if [[ "$(stat -c '%a' /home/jackmhny 2>/dev/null || true)" == "700" ]]; then
record PASS "/home/jackmhny mode is 0700"
else
record FAIL "/home/jackmhny mode is not 0700"
fi
if run_as_couch bash -lc 'test ! -r /home/jackmhny && test ! -x /home/jackmhny'; then
record PASS "couch cannot read or traverse /home/jackmhny"
else
record FAIL "couch can read or traverse /home/jackmhny or couch access test could not run"
fi
if id -nG "${COUCH_USER}" 2>/dev/null | tr ' ' '\n' | grep -Eq '^(sudo|adm)$'; then
record FAIL "couch is still in sudo/adm"
else
record PASS "couch is not in sudo/adm"
fi
grep -Fq 'autologin-user=couch' /etc/lightdm/lightdm.conf.d/50-couchos.conf 2>/dev/null && \
grep -Fq 'autologin-session=xfce' /etc/lightdm/lightdm.conf.d/50-couchos.conf 2>/dev/null \
&& record PASS "LightDM autologin is set to couch/XFCE" \
|| record FAIL "LightDM autologin is not set to couch/XFCE"
grep -Fq 'HandleLidSwitchExternalPower=ignore' /etc/systemd/logind.conf.d/50-couchos.conf 2>/dev/null \
&& record PASS "logind ignores lid close on AC" \
|| record FAIL "logind lid policy on AC is not configured"
grep -Fq 'use_compositing" type="bool" value="false' /home/couch/.config/xfce4/xfconf/xfce-perchannel-xml/xfwm4.xml 2>/dev/null \
&& record PASS "XFCE compositing disabled" \
|| record FAIL "XFCE compositing not disabled"
grep -Fq 'do-not-disturb" type="bool" value="true' /home/couch/.config/xfce4/xfconf/xfce-perchannel-xml/xfce4-notifyd.xml 2>/dev/null \
&& record PASS "notifications DND enabled" \
|| record FAIL "notifications DND not enabled"
grep -Fq 'dpms-enabled" type="bool" value="false' /home/couch/.config/xfce4/xfconf/xfce-perchannel-xml/xfce4-power-manager.xml 2>/dev/null \
&& record PASS "power blanking/suspend disabled" \
|| record FAIL "power blanking/suspend not disabled"
if [[ -f "${RUNTIME_CONFIG}" ]]; then
record PASS "runtime config present"
else
record FAIL "runtime config missing"
fi
if [[ -f /home/couch/.config/couchd/token && "$(stat -c '%a' /home/couch/.config/couchd/token)" == "600" ]]; then
if [[ "$(tr -d '\r\n' </home/couch/.config/couchd/token)" == "replace-with-long-random-token" ]]; then
record FAIL "token still uses the template placeholder"
else
record PASS "token mode is 0600 and not placeholder"
fi
else
record FAIL "token missing or not 0600"
fi
local bind_address
bind_address="$(runtime_field bind_address 2>/dev/null || true)"
if [[ "${bind_address}" == "100.64.0.15" ]]; then
record PASS "runtime bind address is 100.64.0.15"
else
record FAIL "runtime bind address is not 100.64.0.15"
fi
if ss -ltn sport = :8765 2>/dev/null | awk '{print $4}' | grep -Fxq '100.64.0.15:8765'; then
record PASS "couchd listens only on 100.64.0.15:8765"
else
record FAIL "couchd listener is not exactly 100.64.0.15:8765"
fi
local controller
controller="$(controller_snapshot)"
if python3 - <<'PY' "${controller}"
import json
import sys
snapshot = json.loads(sys.argv[1])
if snapshot["address"] and snapshot["paired"] and snapshot["trusted"] and snapshot["connected"]:
raise SystemExit(0)
raise SystemExit(1)
PY
then
record PASS "controller is paired/trusted/connected"
else
record FAIL "controller is not paired/trusted/connected"
fi
local status_payload
if status_payload="$(status_json 2>/dev/null)"; then
record PASS "couchd status command succeeded"
if python3 - <<'PY' "${status_payload}"
import json
import sys
payload = json.loads(sys.argv[1])
mode = payload.get("display", {}).get("mode")
name = payload.get("display", {}).get("name")
if name == "HDMI-1" and mode == "1920x1080@60.00":
raise SystemExit(0)
raise SystemExit(1)
PY
then
record PASS "HDMI-1 active mode is 1920x1080@60.00"
else
record FAIL "HDMI-1 active mode is not 1920x1080@60.00"
fi
else
record FAIL "couchd status command failed"
fi
if pactl info 2>/dev/null | grep -Eiq '^Default Sink: .*hdmi'; then
record PASS "default audio sink is HDMI"
else
record FAIL "default audio sink is not HDMI"
fi
local couch_uid
couch_uid="$(id -u "${COUCH_USER}" 2>/dev/null || true)"
if [[ -n "${couch_uid}" ]] && XDG_RUNTIME_DIR="/run/user/${couch_uid}" run_as_couch systemctl --user is-active couchd.service >/dev/null 2>&1; then
record PASS "couchd user service is active"
else
record FAIL "couchd user service is not active"
fi
if [[ -f /home/couch/.config/autostart/couchd.desktop && -f /home/couch/.config/autostart/couch-display-audio.desktop ]]; then
record PASS "XFCE autostart entries exist for couch session units"
else
record FAIL "XFCE autostart entries for couch session units are missing"
fi
finish
}
five_reboots_mode() {
: >"${LOG_FILE}"
local boot_id
boot_id="$(cat /proc/sys/kernel/random/boot_id)"
local controller
controller="$(controller_snapshot)"
local status_payload
status_payload="$(status_json 2>/dev/null || printf '{}')"
python3 - <<'PY' "${STATE_FILE}" "${boot_id}" "${controller}" "${status_payload}"
import json
import sys
from pathlib import Path
path = Path(sys.argv[1])
boot_id = sys.argv[2]
controller = json.loads(sys.argv[3])
status_payload = json.loads(sys.argv[4])
existing = []
if path.exists():
existing = [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]
if not any(item.get("boot_id") == boot_id for item in existing):
existing.append({
"boot_id": boot_id,
"timestamp": __import__("datetime").datetime.now().isoformat(timespec="seconds"),
"controller": controller,
"display_mode": status_payload.get("display", {}).get("mode"),
"operator_login_observed": "SKIP",
"operator_fps_observed": "SKIP",
})
path.write_text("\n".join(json.dumps(item, sort_keys=True) for item in existing) + ("\n" if existing else ""), encoding="utf-8")
print(len(existing))
PY
local count
count="$(python3 - <<'PY' "${STATE_FILE}"
import json
import sys
from pathlib import Path
path = Path(sys.argv[1])
print(len([json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()]))
PY
)"
record PASS "recorded reboot observation for boot ${boot_id} (${count}/5 unique boots)"
record SKIP "operator_login_observed remains SKIP until human session confirmation is recorded"
record SKIP "operator_fps_observed remains SKIP until human 60 FPS confirmation is recorded"
finish
}
soak_30m_start() {
: >"${LOG_FILE}"
python3 - <<'PY' "${SOAK_FILE}" "$(cat /proc/sys/kernel/random/boot_id)" "$(controller_snapshot)"
import json
import sys
from datetime import datetime
from pathlib import Path
Path(sys.argv[1]).write_text(json.dumps({
"start_boot_id": sys.argv[2],
"started_at": datetime.now().isoformat(timespec="seconds"),
"controller_start": json.loads(sys.argv[3]),
"operator_fps_observed": "SKIP",
}, sort_keys=True) + "\n", encoding="utf-8")
PY
record PASS "30-minute soak started"
record SKIP "operator_fps_observed remains SKIP until a human records it after the session"
finish
}
soak_30m_finish() {
: >"${LOG_FILE}"
if [[ ! -f "${SOAK_FILE}" ]]; then
record FAIL "soak-30m-start was not run"
finish
fi
python3 - <<'PY' "${SOAK_FILE}" "$(cat /proc/sys/kernel/random/boot_id)" "$(controller_snapshot)" "${ARTIFACT_DIR}/soak-30m-finish.json"
import json
import sys
from datetime import datetime
from pathlib import Path
start = json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))
finished_at = datetime.now()
started_at = datetime.fromisoformat(start["started_at"])
elapsed = int((finished_at - started_at).total_seconds())
payload = {
**start,
"finish_boot_id": sys.argv[2],
"finished_at": finished_at.isoformat(timespec="seconds"),
"elapsed_seconds": elapsed,
"controller_finish": json.loads(sys.argv[3]),
}
Path(sys.argv[4]).write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
print(elapsed)
PY
local elapsed
elapsed="$(python3 - <<'PY' "${ARTIFACT_DIR}/soak-30m-finish.json"
import json
import sys
from pathlib import Path
print(json.loads(Path(sys.argv[1]).read_text(encoding="utf-8"))["elapsed_seconds"])
PY
)"
if [[ "${elapsed}" -ge 1800 ]]; then
record PASS "30-minute soak elapsed ${elapsed} seconds"
else
record FAIL "30-minute soak only ran ${elapsed} seconds"
fi
record SKIP "operator_fps_observed remains SKIP until a human records the observed FPS stability"
finish
}
case "${MODE}" in
quick) quick_mode ;;
five-reboots) five_reboots_mode ;;
soak-30m-start) soak_30m_start ;;
soak-30m-finish) soak_30m_finish ;;
*)
printf 'unknown mode: %s\n' "${MODE}" >&2
exit 1
;;
esac

View File

@@ -0,0 +1,22 @@
#!/usr/bin/env bash
set -Eeuo pipefail
COUCH_HOME="${COUCH_HOME:-/home/couch}"
REPLAY_DIR="${COUCH_HOME}/Slippi"
CHECKLIST="${REPLAY_DIR}/FIRST_RUN_CHECKLIST.md"
install -d -m 0700 "${REPLAY_DIR}"
cat <<EOF
Created/validated ${REPLAY_DIR}.
Manual first-run checklist:
1. Launch Slippi Launcher and install/update Slippi Dolphin.
2. Point Dolphin at your own NTSC 1.02 Melee image.
3. In Dolphin graphics, set Vulkan, fullscreen, 4:3, internal resolution 2x, and V-Sync off.
4. Confirm replays save under ${REPLAY_DIR}.
5. Create and save the named Xbox controller profile after device discovery.
Reference template:
${CHECKLIST}
EOF

9
scripts/stop_game.sh Executable file
View File

@@ -0,0 +1,9 @@
#!/usr/bin/env bash
set -Eeuo pipefail
CONFIG_PATH="${COUCHOS_CONFIG:-/home/couch/.config/couchd/runtime.json}"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
REPO_ROOT="$(cd "${SCRIPT_DIR}/.." && pwd)"
export PYTHONPATH="${REPO_ROOT}/src${PYTHONPATH:+:${PYTHONPATH}}"
exec python3 -m couchd --config "${CONFIG_PATH}" stop

464
setup.sh Executable file
View File

@@ -0,0 +1,464 @@
#!/usr/bin/env bash
set -Eeuo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
MODE="apply"
PHASE="all"
CHECK_FAILURES=0
COUCH_USER="couch"
COUCH_HOME="/home/${COUCH_USER}"
INSTALL_ROOT="${COUCH_HOME}/couchos"
RUNTIME_CONFIG_PATH="${COUCH_HOME}/.config/couchd/runtime.json"
TOKEN_PATH="${COUCH_HOME}/.config/couchd/token"
COUCH_UID=""
MISSING_UID_PLACEHOLDER="<missing-couch-uid>"
TOKEN_PLACEHOLDER="$(<"${ROOT_DIR}/templates/couchd.token.example")"
usage() {
cat <<'EOF'
Usage: ./setup.sh [--phase root|couch|all] [--check] [--dry-run]
Safe, idempotent CouchOS setup entrypoint.
- `--check` reports PASS/FAIL and exits nonzero when any validation fails.
- `--dry-run` prints intended changes without applying them.
- `--phase root` covers system setup and stable install location.
- `--phase couch` covers runtime config and per-user desktop/session state.
EOF
}
log() {
printf '[setup] %s\n' "$*"
}
record_check() {
local status="$1"
shift
printf '[check] %s %s\n' "${status}" "$*"
if [[ "${status}" == "FAIL" ]]; then
CHECK_FAILURES=$((CHECK_FAILURES + 1))
fi
}
run_cmd() {
if [[ "${MODE}" == "check" ]]; then
return 0
fi
if [[ "${MODE}" == "dry-run" ]]; then
printf '[dry-run] %s\n' "$*"
return 0
fi
"$@"
}
need_root() {
if [[ "${EUID}" -ne 0 && "${MODE}" != "check" ]]; then
printf 'root privileges are required for this phase\n' >&2
exit 1
fi
}
resolve_couch_uid() {
COUCH_UID="$(id -u "${COUCH_USER}")"
}
effective_couch_uid() {
if [[ -n "${COUCH_UID}" ]]; then
printf '%s\n' "${COUCH_UID}"
else
printf '%s\n' "${MISSING_UID_PLACEHOLDER}"
fi
}
check_path_exists() {
local path="$1"
if [[ -e "${path}" ]]; then
record_check PASS "${path} exists"
else
record_check FAIL "${path} missing"
fi
}
check_command_success() {
local description="$1"
shift
if "$@" >/dev/null 2>&1; then
record_check PASS "${description}"
else
record_check FAIL "${description}"
fi
}
install_file() {
local source="$1"
local target="$2"
local mode="$3"
local owner="$4"
local group="$5"
log "install ${target}"
if [[ "${MODE}" == "check" ]]; then
if [[ -f "${target}" ]]; then
record_check PASS "${target} present"
else
record_check FAIL "${target} missing"
fi
return 0
fi
run_cmd install -D -m "${mode}" -o "${owner}" -g "${group}" "${source}" "${target}"
}
install_text_file() {
local target="$1"
local mode="$2"
local owner="$3"
local group="$4"
local content="$5"
if [[ "${MODE}" == "check" ]]; then
check_path_exists "${target}"
return 0
fi
run_cmd install -d -m 0755 -o "${owner}" -g "${group}" "$(dirname "${target}")"
local tmp
tmp="$(mktemp)"
printf '%s' "${content}" >"${tmp}"
run_cmd install -m "${mode}" -o "${owner}" -g "${group}" "${tmp}" "${target}"
rm -f "${tmp}"
}
ensure_couch_user() {
if id -u "${COUCH_USER}" >/dev/null 2>&1; then
log "user ${COUCH_USER} already present"
resolve_couch_uid
else
if [[ "${MODE}" == "check" ]]; then
record_check FAIL "couch user missing"
COUCH_UID=""
return 0
fi
run_cmd useradd --create-home --home-dir "${COUCH_HOME}" --shell /bin/bash --user-group "${COUCH_USER}"
if id -u "${COUCH_USER}" >/dev/null 2>&1; then
resolve_couch_uid
else
COUCH_UID=""
fi
fi
}
remove_group_membership() {
local group_name="$1"
if ! getent group "${group_name}" >/dev/null 2>&1; then
return 0
fi
if id -nG "${COUCH_USER}" | tr ' ' '\n' | grep -Fxq "${group_name}"; then
if [[ "${MODE}" == "check" ]]; then
record_check FAIL "${COUCH_USER} is still a member of ${group_name}"
else
run_cmd gpasswd -d "${COUCH_USER}" "${group_name}"
fi
elif [[ "${MODE}" == "check" ]]; then
record_check PASS "${COUCH_USER} is not a member of ${group_name}"
fi
}
ensure_repo_install() {
if [[ "${MODE}" == "check" ]]; then
check_path_exists "${INSTALL_ROOT}/src/couchd/app.py"
return 0
fi
run_cmd install -d -m 0700 -o "${COUCH_USER}" -g "${COUCH_USER}" "${INSTALL_ROOT}"
run_cmd bash -lc "cd '${ROOT_DIR}' && tar --exclude='.git' --exclude='__pycache__' --exclude='.pytest_cache' --exclude='artifacts' -cf - . | tar -C '${INSTALL_ROOT}' -xf -"
run_cmd chown -R "${COUCH_USER}:${COUCH_USER}" "${INSTALL_ROOT}"
run_cmd chmod 0700 "${COUCH_HOME}"
}
ensure_runtime_config() {
local controller_address="${COUCHOS_XBOX_BLUETOOTH_ADDRESS:-}"
local bind_address="${COUCHOS_BIND_ADDRESS:-100.64.0.15}"
local existing_runtime="{}"
local runtime_uid
runtime_uid="$(effective_couch_uid)"
if [[ -f "${RUNTIME_CONFIG_PATH}" ]]; then
existing_runtime="$(cat "${RUNTIME_CONFIG_PATH}")"
fi
if [[ "${MODE}" == "check" ]]; then
check_path_exists "${RUNTIME_CONFIG_PATH}"
if [[ -f "${TOKEN_PATH}" ]]; then
local token_mode
token_mode="$(stat -c '%a' "${TOKEN_PATH}" 2>/dev/null || true)"
if [[ "${token_mode}" == "600" ]]; then
record_check PASS "${TOKEN_PATH} mode is 0600"
else
record_check FAIL "${TOKEN_PATH} mode is not 0600"
fi
if [[ "$(tr -d '\r\n' <"${TOKEN_PATH}")" == "${TOKEN_PLACEHOLDER}" ]]; then
record_check FAIL "${TOKEN_PATH} still uses the template placeholder token"
else
record_check PASS "${TOKEN_PATH} does not use the template placeholder token"
fi
else
record_check FAIL "${TOKEN_PATH} missing"
fi
return 0
fi
run_cmd install -d -m 0700 -o "${COUCH_USER}" -g "${COUCH_USER}" "${COUCH_HOME}/.config/couchd"
if [[ ! -f "${TOKEN_PATH}" ]]; then
if [[ "${MODE}" == "dry-run" ]]; then
printf '[dry-run] generate random bearer token at %s (mode 0600)\n' "${TOKEN_PATH}"
else
local token_tmp
token_tmp="$(mktemp)"
python3 - <<'PY' >"${token_tmp}"
import secrets
print(secrets.token_urlsafe(32))
PY
install -m 0600 -o "${COUCH_USER}" -g "${COUCH_USER}" "${token_tmp}" "${TOKEN_PATH}"
rm -f "${token_tmp}"
fi
fi
if [[ "${MODE}" == "dry-run" ]]; then
printf '[dry-run] generate %s with xdg_runtime_dir=/run/user/%s\n' "${RUNTIME_CONFIG_PATH}" "${runtime_uid}"
return 0
fi
EXISTING_RUNTIME_JSON="${existing_runtime}" \
RUNTIME_TARGET="${RUNTIME_CONFIG_PATH}" \
CONTROLLER_ADDRESS="${controller_address}" \
BIND_ADDRESS="${bind_address}" \
COUCH_UID="${runtime_uid}" \
python3 <<'PY'
import json
import os
from pathlib import Path
existing = json.loads(os.environ["EXISTING_RUNTIME_JSON"])
target = Path(os.environ["RUNTIME_TARGET"])
couch_home = Path("/home/couch")
payload = {
"bind_address": os.environ["BIND_ADDRESS"],
"port": int(existing.get("port", 8765)),
"token_file": str(couch_home / ".config" / "couchd" / "token"),
"couch_home": str(couch_home),
"iso_path": existing.get("iso_path"),
"state_file": str(couch_home / ".local" / "state" / "couchd" / "state.json"),
"display": existing.get("display", ":0"),
"xdg_runtime_dir": f"/run/user/{os.environ['COUCH_UID']}",
"xbox_bluetooth_address": os.environ["CONTROLLER_ADDRESS"] or existing.get("xbox_bluetooth_address"),
"launcher_shortcut": existing.get("launcher_shortcut"),
}
target.write_text(json.dumps(payload, indent=2, sort_keys=True) + "\n", encoding="utf-8")
PY
run_cmd chown "${COUCH_USER}:${COUCH_USER}" "${RUNTIME_CONFIG_PATH}"
run_cmd chmod 0640 "${RUNTIME_CONFIG_PATH}"
}
configure_user_session_files() {
local xprofile_content='xset s off
xset -dpms
xset s noblank
'
local xfwm4_content='<?xml version="1.0" encoding="UTF-8"?>
<channel name="xfwm4" version="1.0">
<property name="general" type="empty">
<property name="use_compositing" type="bool" value="false"/>
</property>
</channel>
'
local power_content='<?xml version="1.0" encoding="UTF-8"?>
<channel name="xfce4-power-manager" version="1.0">
<property name="xfce4-power-manager" type="empty">
<property name="blank-on-ac" type="int" value="0"/>
<property name="dpms-enabled" type="bool" value="false"/>
<property name="inactivity-on-ac" type="int" value="0"/>
<property name="logind-handle-lid-switch" type="bool" value="false"/>
</property>
</channel>
'
local notifyd_content='<?xml version="1.0" encoding="UTF-8"?>
<channel name="xfce4-notifyd" version="1.0">
<property name="do-not-disturb" type="bool" value="true"/>
</channel>
'
local screensaver_content='<?xml version="1.0" encoding="UTF-8"?>
<channel name="xfce4-screensaver" version="1.0">
<property name="lock" type="bool" value="false"/>
<property name="saver" type="empty">
<property name="enabled" type="bool" value="false"/>
<property name="mode" type="int" value="0"/>
</property>
</channel>
'
local autostart_override='[Desktop Entry]
Type=Application
Hidden=true
X-GNOME-Autostart-enabled=false
'
local couchd_autostart='[Desktop Entry]
Type=Application
Version=1.0
Name=CouchOS Control Daemon
OnlyShowIn=XFCE;
Exec=sh -lc "systemctl --user import-environment DISPLAY XAUTHORITY XDG_RUNTIME_DIR DBUS_SESSION_BUS_ADDRESS; systemctl --user daemon-reload; systemctl --user start couchd.service"
X-GNOME-Autostart-enabled=true
'
local display_audio_autostart='[Desktop Entry]
Type=Application
Version=1.0
Name=CouchOS HDMI Display And Audio
OnlyShowIn=XFCE;
Exec=sh -lc "systemctl --user import-environment DISPLAY XAUTHORITY XDG_RUNTIME_DIR DBUS_SESSION_BUS_ADDRESS; systemctl --user daemon-reload; systemctl --user start couch-display-audio.service"
X-GNOME-Autostart-enabled=true
'
install_text_file "${COUCH_HOME}/.xprofile" 0644 "${COUCH_USER}" "${COUCH_USER}" "${xprofile_content}"
install_text_file "${COUCH_HOME}/.config/xfce4/xfconf/xfce-perchannel-xml/xfwm4.xml" 0644 "${COUCH_USER}" "${COUCH_USER}" "${xfwm4_content}"
install_text_file "${COUCH_HOME}/.config/xfce4/xfconf/xfce-perchannel-xml/xfce4-power-manager.xml" 0644 "${COUCH_USER}" "${COUCH_USER}" "${power_content}"
install_text_file "${COUCH_HOME}/.config/xfce4/xfconf/xfce-perchannel-xml/xfce4-notifyd.xml" 0644 "${COUCH_USER}" "${COUCH_USER}" "${notifyd_content}"
install_text_file "${COUCH_HOME}/.config/xfce4/xfconf/xfce-perchannel-xml/xfce4-screensaver.xml" 0644 "${COUCH_USER}" "${COUCH_USER}" "${screensaver_content}"
install_text_file "${COUCH_HOME}/.config/autostart/xfce4-notifyd.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${autostart_override}"
install_text_file "${COUCH_HOME}/.config/autostart/update-notifier.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${autostart_override}"
install_text_file "${COUCH_HOME}/.config/autostart/light-locker.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${autostart_override}"
install_text_file "${COUCH_HOME}/.config/autostart/xscreensaver.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${autostart_override}"
install_text_file "${COUCH_HOME}/.config/autostart/couchd.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${couchd_autostart}"
install_text_file "${COUCH_HOME}/.config/autostart/couch-display-audio.desktop" 0644 "${COUCH_USER}" "${COUCH_USER}" "${display_audio_autostart}"
}
root_phase() {
need_root
log "root phase"
ensure_couch_user
run_cmd apt-get update
run_cmd apt-get install -y \
lightdm xfce4 xfce4-goodies unattended-upgrades x11-xserver-utils pulseaudio pavucontrol \
bluez blueman rfkill mesa-utils vulkan-tools jq curl
install_file "${ROOT_DIR}/config/lightdm/50-couchos.conf" "/etc/lightdm/lightdm.conf.d/50-couchos.conf" 0644 root root
install_file "${ROOT_DIR}/config/logind/50-couchos.conf" "/etc/systemd/logind.conf.d/50-couchos.conf" 0644 root root
install_file "${ROOT_DIR}/systemd/systemd-user/couchd.service" "/etc/systemd/user/couchd.service" 0644 root root
install_file "${ROOT_DIR}/systemd/systemd-user/couch-display-audio.service" "/etc/systemd/user/couch-display-audio.service" 0644 root root
install_text_file "/etc/apt/apt.conf.d/52couchos-unattended" 0644 root root $'APT::Periodic::Update-Package-Lists "1";\nAPT::Periodic::Unattended-Upgrade "1";\n'
remove_group_membership sudo
remove_group_membership adm
ensure_repo_install
if [[ "${MODE}" == "check" ]]; then
check_command_success "couch home mode is 0700" test "$(stat -c '%a' "${COUCH_HOME}" 2>/dev/null)" = "700"
else
run_cmd loginctl enable-linger "${COUCH_USER}"
run_cmd systemctl daemon-reload
fi
}
couch_phase() {
need_root
log "couch phase"
ensure_couch_user
run_cmd install -d -m 0700 -o "${COUCH_USER}" -g "${COUCH_USER}" \
"${COUCH_HOME}/Applications" "${COUCH_HOME}/Slippi" "${COUCH_HOME}/bin" "${COUCH_HOME}/.local/state/couchd"
ensure_runtime_config
configure_user_session_files
if [[ "${MODE}" != "check" ]]; then
run_cmd ln -sfn "${INSTALL_ROOT}/scripts/launch_melee.sh" "${COUCH_HOME}/bin/launch_melee.sh"
run_cmd ln -sfn "${INSTALL_ROOT}/scripts/open_slippi.sh" "${COUCH_HOME}/bin/open_slippi.sh"
run_cmd ln -sfn "${INSTALL_ROOT}/scripts/stop_game.sh" "${COUCH_HOME}/bin/stop_game.sh"
run_cmd ln -sfn "${INSTALL_ROOT}/scripts/slippi_first_run_helper.sh" "${COUCH_HOME}/bin/slippi_first_run_helper.sh"
run_cmd chown -h "${COUCH_USER}:${COUCH_USER}" "${COUCH_HOME}/bin/"*.sh
fi
}
check_runtime_sanity() {
if [[ ! -f "${RUNTIME_CONFIG_PATH}" ]]; then
record_check FAIL "${RUNTIME_CONFIG_PATH} missing"
return 0
fi
if [[ -z "${COUCH_UID}" ]]; then
record_check FAIL "runtime config could not be verified because couch uid is unresolved"
return 0
fi
if RUNTIME_TARGET="${RUNTIME_CONFIG_PATH}" COUCH_UID="${COUCH_UID}" python3 <<'PY'
import json
import os
from pathlib import Path
target = Path(os.environ["RUNTIME_TARGET"])
data = json.loads(target.read_text(encoding="utf-8"))
failures = []
if data.get("xdg_runtime_dir") != f"/run/user/{os.environ['COUCH_UID']}":
failures.append("runtime xdg_runtime_dir does not match couch uid")
if data.get("launcher_shortcut") == ["/usr/bin/xdg-open", "slippi://play"]:
failures.append("runtime launcher_shortcut still uses unverified slippi://play")
iso_path = data.get("iso_path")
if iso_path is not None and not str(iso_path).startswith("/home/couch/"):
failures.append("runtime iso_path is outside /home/couch")
if failures:
for failure in failures:
print(failure)
raise SystemExit(1)
PY
then
record_check PASS "runtime config placeholders resolved safely"
else
record_check FAIL "runtime config placeholders resolved safely"
fi
}
main() {
while [[ $# -gt 0 ]]; do
case "$1" in
--phase)
PHASE="$2"
shift 2
;;
--check)
MODE="check"
shift
;;
--dry-run)
MODE="dry-run"
shift
;;
--help|-h)
usage
exit 0
;;
*)
printf 'unknown argument: %s\n' "$1" >&2
usage >&2
exit 1
;;
esac
done
case "${PHASE}" in
root) root_phase ;;
couch) couch_phase ;;
all)
root_phase
couch_phase
;;
*)
printf 'unknown phase: %s\n' "${PHASE}" >&2
exit 1
;;
esac
if [[ "${MODE}" == "check" ]]; then
if id -u "${COUCH_USER}" >/dev/null 2>&1; then
resolve_couch_uid
fi
check_runtime_sanity
if [[ "${CHECK_FAILURES}" -gt 0 ]]; then
exit 1
fi
fi
}
if [[ "${BASH_SOURCE[0]}" == "$0" ]]; then
main "$@"
fi

1
src/couchd/__init__.py Normal file
View File

@@ -0,0 +1 @@
"""CouchOS daemon package."""

4
src/couchd/__main__.py Normal file
View File

@@ -0,0 +1,4 @@
from couchd.cli import entrypoint
raise SystemExit(entrypoint())

214
src/couchd/app.py Normal file
View File

@@ -0,0 +1,214 @@
"""HTTP control plane for CouchOS."""
from __future__ import annotations
import ipaddress
import hmac
import json
from dataclasses import dataclass, field
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from typing import Callable, Protocol
from urllib.parse import urlsplit
from couchd.discovery import DiscoveryError
TAILSCALE_CGNAT = ipaddress.ip_network("100.64.0.0/10")
MAX_CONTENT_LENGTH = 1024
class ConfigError(ValueError):
"""Raised when couchd configuration is invalid."""
@dataclass(frozen=True)
class Config:
bind_address: str
token: str
port: int = 8765
@dataclass(frozen=True)
class Response:
status_code: int
body: dict[str, object]
headers: dict[str, str] = field(
default_factory=lambda: {"Content-Type": "application/json; charset=utf-8"}
)
class Launcher(Protocol):
def launch_melee(self) -> dict[str, object]:
...
def open_slippi(self) -> dict[str, object]:
...
def stop(self) -> dict[str, object]:
...
class App:
def __init__(
self,
config: Config,
status_provider: Callable[[], dict[str, object]] | None = None,
launcher: Launcher | None = None,
) -> None:
self.config = config
self._status_provider = status_provider or self._default_status_provider
self._launcher = launcher
validate_bind_address(config.bind_address)
if not config.token.strip():
raise ConfigError("token must not be empty")
def handle_request(
self,
method: str,
path: str,
headers: dict[str, str] | None = None,
body: bytes | None = None,
) -> Response:
body = body or b""
headers = headers or {}
if not self._authorized(headers.get("Authorization")):
return Response(401, {"error": "missing or invalid bearer token"})
try:
content_length = int(headers.get("Content-Length", str(len(body))))
except ValueError:
return Response(400, {"error": "invalid content length"})
if content_length > MAX_CONTENT_LENGTH:
return Response(413, {"error": "request body too large"})
normalized_path = urlsplit(path).path or "/"
routes: dict[tuple[str, str], Callable[[], Response]] = {
("GET", "/status"): lambda: Response(200, self._status_provider()),
("POST", "/launch/melee"): self._launch_melee,
("POST", "/open/slippi"): self._open_slippi,
("POST", "/stop"): self._stop,
}
if method == "POST" and normalized_path in {"/launch/melee", "/open/slippi", "/stop"} and body:
return Response(400, {"error": "request body is not allowed"})
if (method, normalized_path) in routes:
try:
return routes[(method, normalized_path)]()
except (ConfigError, DiscoveryError, FileNotFoundError, PermissionError, ValueError) as exc:
return Response(409, {"error": str(exc)})
except Exception:
return Response(503, {"error": "launcher action failed"})
known_paths = {"/status", "/launch/melee", "/open/slippi", "/stop"}
if normalized_path in known_paths:
return Response(405, {"error": "method not allowed"})
return Response(404, {"error": "not found"})
def _authorized(self, authorization: str | None) -> bool:
if not authorization or not authorization.startswith("Bearer "):
return False
return hmac.compare_digest(authorization[7:], self.config.token)
def _launch_melee(self) -> Response:
if self._launcher is None:
return Response(503, {"error": "launcher unavailable"})
return Response(202, self._launcher.launch_melee())
def _open_slippi(self) -> Response:
if self._launcher is None:
return Response(503, {"error": "launcher unavailable"})
return Response(202, self._launcher.open_slippi())
def _stop(self) -> Response:
if self._launcher is None:
return Response(503, {"error": "launcher unavailable"})
return Response(202, self._launcher.stop())
@staticmethod
def _default_status_provider() -> dict[str, object]:
return {
"state": "unknown",
"pid": None,
"display": {"name": None, "mode": None},
"slippi_version": None,
"controller": {"connected": None, "name": None},
}
class CouchRequestHandler(BaseHTTPRequestHandler):
"""Bound request handler for a specific App instance."""
app: App
server_version = "couchd/0.1"
protocol_version = "HTTP/1.1"
def do_GET(self) -> None: # noqa: N802
self._handle()
def do_POST(self) -> None: # noqa: N802
self._handle()
def log_message(self, format: str, *args: object) -> None:
del format, args
def _handle(self) -> None:
try:
length = int(self.headers.get("Content-Length", "0"))
except ValueError:
self._write_response(Response(400, {"error": "invalid content length"}))
return
if length > MAX_CONTENT_LENGTH:
self._write_response(Response(413, {"error": "request body too large"}))
return
body = self.rfile.read(length) if length > 0 else b""
response = self.app.handle_request(
self.command,
self.path,
headers={key: value for key, value in self.headers.items()},
body=body,
)
self._write_response(response)
def _write_response(self, response: Response) -> None:
payload = json.dumps(response.body).encode("utf-8")
self.send_response(response.status_code)
for key, value in response.headers.items():
self.send_header(key, value)
self.send_header("Content-Length", str(len(payload)))
self.end_headers()
self.wfile.write(payload)
def make_http_server(app: App) -> ThreadingHTTPServer:
handler = type("BoundCouchRequestHandler", (CouchRequestHandler,), {"app": app})
return ThreadingHTTPServer((app.config.bind_address, app.config.port), handler)
def validate_bind_address(address: str) -> str:
ip = ipaddress.ip_address(address)
if ip.is_loopback or ip.is_unspecified:
raise ConfigError("loopback and unspecified addresses are not allowed")
if ip.version != 4 or ip not in TAILSCALE_CGNAT:
raise ConfigError("bind address must be a Tailscale IPv4 address")
return address
def load_token(token_path: str) -> str:
with open(token_path, "r", encoding="utf-8") as handle:
token = handle.read().strip()
if not token:
raise ConfigError(f"token file {token_path} is empty")
return token
def main(app: App) -> int:
server = make_http_server(app)
try:
server.serve_forever()
except KeyboardInterrupt:
return HTTPStatus.OK
finally:
server.server_close()
return HTTPStatus.OK

144
src/couchd/cli.py Normal file
View File

@@ -0,0 +1,144 @@
"""CLI entrypoints for couchd."""
from __future__ import annotations
import argparse
import json
import sys
from dataclasses import dataclass
from pathlib import Path
from couchd.app import App, Config, load_token, main as run_server
from couchd.discovery import DiscoveryConfig, SlippiDiscovery
from couchd.launcher import (
LaunchConfig,
LaunchService,
ProcessRegistry,
default_process_identity,
default_process_probe,
default_is_pid_running,
default_kill,
default_spawn,
default_terminate,
)
from couchd.status import build_default_status_collector
@dataclass(frozen=True)
class RuntimeConfig:
bind_address: str
port: int
token_file: Path
couch_home: Path
iso_path: Path | None
state_file: Path
display: str
launcher_shortcut: list[str] | None
xbox_bluetooth_address: str | None
xdg_runtime_dir: str | None
def parse_args(argv: list[str]) -> argparse.Namespace:
parser = argparse.ArgumentParser(prog="couchd")
parser.add_argument("--config", required=True, help="Path to runtime JSON config")
subparsers = parser.add_subparsers(dest="command", required=True)
subparsers.add_parser("serve")
subparsers.add_parser("status")
subparsers.add_parser("launch-melee")
subparsers.add_parser("open-slippi")
subparsers.add_parser("stop")
return parser.parse_args(argv)
def load_runtime_config(path: Path) -> RuntimeConfig:
data = json.loads(path.read_text(encoding="utf-8"))
return RuntimeConfig(
bind_address=data["bind_address"],
port=int(data.get("port", 8765)),
token_file=Path(data["token_file"]),
couch_home=Path(data["couch_home"]),
iso_path=Path(data["iso_path"]) if data.get("iso_path") else None,
state_file=Path(data["state_file"]),
display=data.get("display", ":0"),
launcher_shortcut=data.get("launcher_shortcut"),
xbox_bluetooth_address=data.get("xbox_bluetooth_address"),
xdg_runtime_dir=data.get("xdg_runtime_dir"),
)
def build_services(runtime: RuntimeConfig) -> tuple[App, LaunchService]:
token = load_token(str(runtime.token_file))
discovery = SlippiDiscovery(
DiscoveryConfig(
couch_home=runtime.couch_home,
applications_dir=runtime.couch_home / "Applications",
launcher_fallback=runtime.launcher_shortcut,
)
)
couch_uid = runtime.couch_home.stat().st_uid
allowed_roots = discovery.allowed_roots
registry = ProcessRegistry(
is_pid_running=default_is_pid_running,
spawn=default_spawn,
terminate=default_terminate,
kill=default_kill,
state_file=runtime.state_file,
process_probe=lambda: default_process_probe(couch_uid, allowed_roots),
process_identity=lambda pid: default_process_identity(pid, couch_uid, allowed_roots),
)
launcher = LaunchService(
discovery=discovery,
registry=registry,
config=LaunchConfig(
couch_home=runtime.couch_home,
iso_path=runtime.iso_path,
display=runtime.display,
xdg_runtime_dir=runtime.xdg_runtime_dir,
launcher_shortcut=runtime.launcher_shortcut,
),
)
collector = build_default_status_collector(
discovery=discovery,
xbox_bluetooth_address=runtime.xbox_bluetooth_address,
)
def status_provider() -> dict[str, object]:
current = launcher.current_state()
return collector.collect(state=str(current["state"]), pid=current["pid"])
app = App(
Config(bind_address=runtime.bind_address, token=token, port=runtime.port),
status_provider=status_provider,
launcher=launcher,
)
return app, launcher
def command_status(app: App) -> int:
response = app.handle_request("GET", "/status", headers={"Authorization": f"Bearer {app.config.token}"})
print(json.dumps(response.body, indent=2, sort_keys=True))
return 0
def command_action(app: App, path: str) -> int:
response = app.handle_request("POST", path, headers={"Authorization": f"Bearer {app.config.token}"})
print(json.dumps(response.body, indent=2, sort_keys=True))
return 0 if response.status_code < 400 else 1
def entrypoint(argv: list[str] | None = None) -> int:
args = parse_args(argv or sys.argv[1:])
runtime = load_runtime_config(Path(args.config))
app, _launcher = build_services(runtime)
if args.command == "serve":
return int(run_server(app))
if args.command == "status":
return command_status(app)
if args.command == "launch-melee":
return command_action(app, "/launch/melee")
if args.command == "open-slippi":
return command_action(app, "/open/slippi")
if args.command == "stop":
return command_action(app, "/stop")
return 1

151
src/couchd/discovery.py Normal file
View File

@@ -0,0 +1,151 @@
"""Slippi discovery helpers."""
from __future__ import annotations
import re
import subprocess
from dataclasses import dataclass
from pathlib import Path
class AllowlistError(ValueError):
"""Raised when a discovered target is outside the allowlist."""
class DiscoveryError(RuntimeError):
"""Raised when launcher or Dolphin discovery fails."""
def allowlisted_target(
candidate: Path,
allowed_roots: list[Path],
allowed_names: set[str],
) -> Path:
resolved = candidate.resolve(strict=False)
if resolved.name not in allowed_names:
raise AllowlistError(f"disallowed target name: {resolved.name}")
if not any(
resolved == root.resolve(strict=False) or root.resolve(strict=False) in resolved.parents
for root in allowed_roots
):
raise AllowlistError(f"target {resolved} is outside allowed roots")
return resolved
@dataclass(frozen=True)
class DiscoveryConfig:
couch_home: Path
applications_dir: Path
launcher_fallback: list[str] | None = None
class SlippiDiscovery:
def __init__(self, config: DiscoveryConfig) -> None:
self._config = config
self._allowed_roots = [
config.applications_dir,
config.couch_home / ".config",
config.couch_home / ".local" / "share",
]
@property
def allowed_roots(self) -> list[Path]:
return list(self._allowed_roots)
def find_launcher(self) -> Path:
candidates = [self._config.applications_dir / "SlippiLauncher.AppImage"]
candidates.extend(
sorted(self._config.applications_dir.glob("Slippi-Launcher-*.AppImage"))
)
for candidate in reversed(candidates):
if candidate.exists():
return candidate.resolve(strict=False)
raise DiscoveryError("official Slippi Launcher AppImage not found")
def find_dolphin_binary(self) -> Path:
patterns = (
"Slippi Dolphin*",
"slippi-dolphin*",
"dolphin-emu",
"Dolphin*.AppImage",
)
matches: list[Path] = []
for root in self._allowed_roots:
if not root.exists():
continue
for pattern in patterns:
matches.extend(path for path in root.rglob(pattern) if path.is_file())
executable_matches = [path for path in matches if path.stat().st_mode & 0o111]
if not executable_matches:
raise DiscoveryError("no installed Slippi Dolphin target discovered")
executable_matches.sort(key=lambda item: item.stat().st_mtime, reverse=True)
return executable_matches[0].resolve(strict=False)
def resolve_melee_command(self, iso_path: Path) -> list[str]:
dolphin = self.find_dolphin_binary()
flags = self.inspect_supported_flags(dolphin)
exec_flag = "--exec" if "--exec" in flags else "-e" if "-e" in flags else None
if exec_flag is None:
raise DiscoveryError("installed Slippi Dolphin has no supported direct launch flag")
command = [str(dolphin), exec_flag, str(iso_path)]
if "--fullscreen" in flags:
command.append("--fullscreen")
return command
def discover_launcher_shortcuts(self) -> list[list[str]]:
shortcuts: list[list[str]] = []
seen: set[tuple[str, ...]] = set()
desktop_roots = [
self._config.couch_home / ".local" / "share" / "applications",
Path("/usr/share/applications"),
]
for root in desktop_roots:
if not root.exists():
continue
for desktop_file in root.rglob("*.desktop"):
try:
text = desktop_file.read_text(encoding="utf-8")
except OSError:
continue
lowered = text.lower()
if "slippi" not in lowered:
continue
if desktop_file.is_relative_to(root):
desktop_id = desktop_file.relative_to(root).with_suffix("")
else:
desktop_id = Path(desktop_file.stem)
identifier = "-".join(desktop_id.parts)
command = ["/usr/bin/gtk-launch", identifier]
key = tuple(command)
if key not in seen:
seen.add(key)
shortcuts.append(command)
return shortcuts
def validate_launcher_shortcut(self, shortcut: list[str]) -> list[str]:
if len(shortcut) != 2:
raise DiscoveryError("launcher_shortcut must be a discovered gtk-launch desktop entry")
if shortcut[0] not in {"gtk-launch", "/usr/bin/gtk-launch"}:
raise DiscoveryError("launcher_shortcut must use gtk-launch for a discovered desktop entry")
normalized = ["/usr/bin/gtk-launch", shortcut[1]]
if normalized not in self.discover_launcher_shortcuts():
raise DiscoveryError("configured launcher_shortcut is not among discovered launcher entries")
return normalized
def inspect_supported_flags(self, binary: Path) -> set[str]:
for flag in ("--help", "-h"):
try:
completed = subprocess.run( # noqa: S603
[str(binary), flag],
capture_output=True,
check=False,
text=True,
timeout=5,
)
except (FileNotFoundError, subprocess.TimeoutExpired):
continue
text = "\n".join([completed.stdout, completed.stderr])
found = set(re.findall(r"(?<!\w)(--[a-z0-9-]+|-{1}[A-Za-z])(?!\w)", text))
if found:
return found
return set()

317
src/couchd/launcher.py Normal file
View File

@@ -0,0 +1,317 @@
"""Launch management for CouchOS."""
from __future__ import annotations
import json
import os
import signal
import subprocess
import time
from dataclasses import dataclass
from json import JSONDecodeError
from pathlib import Path
from typing import Callable
from couchd.discovery import DiscoveryError, SlippiDiscovery
@dataclass(frozen=True)
class LaunchRequest:
kind: str
argv: list[str]
env: dict[str, str] | None = None
cwd: str | None = None
@dataclass(frozen=True)
class RunningProcess:
pid: int
state: str
identity: str
class ProcessRegistry:
def __init__(
self,
is_pid_running: Callable[[int], bool],
spawn: Callable[[LaunchRequest], int],
terminate: Callable[[int], None] | None = None,
kill: Callable[[int], None] | None = None,
state_file: Path | None = None,
sleep: Callable[[float], None] | None = None,
process_probe: Callable[[], RunningProcess | None] | None = None,
process_identity: Callable[[int], str | None] | None = None,
) -> None:
self._is_pid_running = is_pid_running
self._spawn = spawn
self._terminate = terminate or (lambda pid: None)
self._kill = kill or (lambda pid: None)
self._sleep = sleep or time.sleep
self._state_file = state_file
self._process_probe = process_probe or (lambda: None)
self._process_identity = process_identity or (lambda pid: None)
self._pid: int | None = None
self._state = "idle"
self._identity: str | None = None
self._load_state()
def record_running(self, pid: int, state: str) -> None:
self._pid = pid
self._state = state
self._identity = self._process_identity(pid)
self._persist_state()
def current_state(self) -> dict[str, object]:
if self._pid is not None and self._matches_tracked_process(self._pid):
return {"state": self._state, "pid": self._pid}
adopted = self._process_probe()
if adopted is not None:
self._pid = adopted.pid
self._state = adopted.state
self._identity = adopted.identity
self._persist_state()
return {"state": self._state, "pid": self._pid}
self._pid = None
self._state = "idle"
self._identity = None
self._persist_state()
return {"state": "idle", "pid": None}
def launch(self, request: LaunchRequest) -> dict[str, object]:
current = self.current_state()
if current["pid"] is not None:
return {"state": "already_running", "pid": current["pid"]}
pid = self._spawn(request)
self._pid = pid
self._state = request.kind
self._identity = self._process_identity(pid)
self._persist_state()
return {"state": "launching", "pid": pid}
def stop(self, timeout_seconds: float = 8.0) -> dict[str, object]:
current = self.current_state()
pid = current["pid"]
if pid is None:
return {"state": "stopped", "pid": None}
if not self._matches_tracked_process(int(pid)):
self._clear_state()
return {"state": "stopped", "pid": None}
self._terminate(pid)
deadline = time.monotonic() + timeout_seconds
while time.monotonic() < deadline:
if not self._matches_tracked_process(pid):
self._clear_state()
return {"state": "stopped", "pid": None}
self._sleep(0.1)
if self._matches_tracked_process(pid):
self._kill(pid)
self._clear_state()
return {"state": "stopped", "pid": None}
def _load_state(self) -> None:
if self._state_file is None or not self._state_file.exists():
return
try:
data = json.loads(self._state_file.read_text(encoding="utf-8"))
except (OSError, JSONDecodeError):
self._clear_state()
return
pid = data.get("pid")
self._pid = pid if isinstance(pid, int) and pid > 0 else None
self._state = str(data.get("state", "idle"))
identity = data.get("identity")
self._identity = identity if isinstance(identity, str) and identity else None
def _persist_state(self) -> None:
if self._state_file is None:
return
self._state_file.parent.mkdir(parents=True, exist_ok=True)
payload = {"pid": self._pid, "state": self._state, "identity": self._identity}
temp_file = self._state_file.with_name(f"{self._state_file.name}.tmp")
temp_file.write_text(json.dumps(payload), encoding="utf-8")
temp_file.replace(self._state_file)
def _matches_tracked_process(self, pid: int) -> bool:
if not self._is_pid_running(pid):
return False
identity = self._process_identity(pid)
if identity is None:
return False
if self._identity is None:
self._identity = identity
self._persist_state()
return True
return identity == self._identity
def _clear_state(self) -> None:
self._pid = None
self._state = "idle"
self._identity = None
self._persist_state()
@dataclass(frozen=True)
class LaunchConfig:
couch_home: Path
iso_path: Path | None
display: str = ":0"
xdg_runtime_dir: str | None = None
launcher_shortcut: list[str] | None = None
class LaunchService:
def __init__(
self,
discovery: SlippiDiscovery,
registry: ProcessRegistry,
config: LaunchConfig,
) -> None:
self._discovery = discovery
self._registry = registry
self._config = config
def launch_melee(self) -> dict[str, object]:
request = self._build_melee_request()
return self._registry.launch(request)
def open_slippi(self) -> dict[str, object]:
launcher = self._discovery.find_launcher()
request = LaunchRequest(
kind="launcher",
argv=[str(launcher)],
env=self._session_env(),
cwd=str(self._config.couch_home),
)
return self._registry.launch(request)
def stop(self) -> dict[str, object]:
return self._registry.stop()
def current_state(self) -> dict[str, object]:
return self._registry.current_state()
def _build_melee_request(self) -> LaunchRequest:
if self._config.iso_path is None:
raise DiscoveryError("iso_path is not configured")
if not self._config.iso_path.is_file():
raise DiscoveryError(f"configured ISO does not exist: {self._config.iso_path}")
if not self._config.iso_path.resolve(strict=False).is_relative_to(
self._config.couch_home.resolve(strict=False)
):
raise DiscoveryError("configured ISO must live under /home/couch")
try:
command = self._discovery.resolve_melee_command(self._config.iso_path)
except DiscoveryError:
if not self._config.launcher_shortcut:
raise
command = self._discovery.validate_launcher_shortcut(self._config.launcher_shortcut)
return LaunchRequest(
kind="melee",
argv=command,
env=self._session_env(),
cwd=str(self._config.couch_home),
)
def _session_env(self) -> dict[str, str]:
env = {
"DISPLAY": self._config.display,
"HOME": str(self._config.couch_home),
"XAUTHORITY": str(self._config.couch_home / ".Xauthority"),
}
if self._config.xdg_runtime_dir:
env["XDG_RUNTIME_DIR"] = self._config.xdg_runtime_dir
return env
def default_is_pid_running(pid: int) -> bool:
try:
os.kill(pid, 0)
except ProcessLookupError:
return False
except PermissionError:
return True
return True
def default_spawn(request: LaunchRequest) -> int:
env = os.environ.copy()
if request.env:
env.update(request.env)
process = subprocess.Popen( # noqa: S603
request.argv,
cwd=request.cwd,
env=env,
preexec_fn=os.setsid,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
return process.pid
def default_terminate(pid: int) -> None:
_signal_process(pid, signal.SIGTERM)
def default_kill(pid: int) -> None:
_signal_process(pid, signal.SIGKILL)
def _signal_process(pid: int, sig: signal.Signals) -> None:
try:
process_group = os.getpgid(pid)
except ProcessLookupError:
return
if process_group == pid:
os.killpg(pid, sig)
return
os.kill(pid, sig)
def default_process_identity(
pid: int,
couch_uid: int,
allowed_roots: list[Path],
) -> str | None:
proc_dir = Path("/proc") / str(pid)
try:
stat_info = proc_dir.stat()
if stat_info.st_uid != couch_uid:
return None
exe_path = proc_dir / "exe"
resolved = exe_path.resolve(strict=True)
start_time = (proc_dir / "stat").read_text(encoding="utf-8").rsplit(")", 1)[1].split()[19]
except (FileNotFoundError, OSError):
return None
allowed = [root.resolve(strict=False) for root in allowed_roots]
resolved_root = resolved.resolve(strict=False)
if not any(root == resolved_root or root in resolved_root.parents for root in allowed):
return None
return f"{stat_info.st_uid}:{resolved_root}:{start_time}"
def default_process_probe(couch_uid: int, allowed_roots: list[Path]) -> RunningProcess | None:
matches: list[RunningProcess] = []
for proc_dir in Path("/proc").iterdir():
if not proc_dir.name.isdigit():
continue
pid = int(proc_dir.name)
identity = default_process_identity(pid, couch_uid, allowed_roots)
if identity is None:
continue
_uid, exe_path, _start_time = identity.split(":", 2)
exe_name = Path(exe_path).name.lower()
if "slippi" in exe_name and "launcher" in exe_name:
state = "launcher"
elif "dolphin" in exe_name:
state = "melee"
else:
continue
matches.append(RunningProcess(pid=pid, state=state, identity=identity))
if not matches:
return None
matches.sort(key=lambda item: item.pid, reverse=True)
return matches[0]

168
src/couchd/status.py Normal file
View File

@@ -0,0 +1,168 @@
"""Status collection helpers for couchd."""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from typing import Callable
from couchd.discovery import DiscoveryError, SlippiDiscovery
class CommandError(RuntimeError):
"""Raised when a status probe cannot execute."""
class StatusCollector:
def __init__(
self,
display_probe: Callable[[], dict[str, object]],
slippi_probe: Callable[[], str],
dolphin_probe: Callable[[], str],
controller_probe: Callable[[], dict[str, object]],
) -> None:
self._display_probe = display_probe
self._slippi_probe = slippi_probe
self._dolphin_probe = dolphin_probe
self._controller_probe = controller_probe
def collect(self, state: str, pid: int | None) -> dict[str, object]:
status: dict[str, object] = {"state": state, "pid": pid}
try:
status["display"] = self._display_probe()
except (CommandError, TimeoutError) as exc:
status["display"] = {"mode": "unknown", "error": str(exc)}
try:
status["slippi_version"] = self._slippi_probe()
except (CommandError, TimeoutError) as exc:
status["slippi_version"] = "unknown"
status["slippi_error"] = str(exc)
try:
status["dolphin_version"] = self._dolphin_probe()
except (CommandError, TimeoutError) as exc:
status["dolphin_version"] = "unknown"
status["dolphin_error"] = str(exc)
try:
status["controller"] = self._controller_probe()
except (CommandError, TimeoutError) as exc:
status["controller"] = {"connected": "unknown", "error": str(exc)}
return status
class CommandRunner:
def run(self, argv: list[str], timeout: int = 5) -> subprocess.CompletedProcess[str]:
try:
return subprocess.run( # noqa: S603
argv,
capture_output=True,
check=False,
text=True,
timeout=timeout,
)
except FileNotFoundError as exc:
raise CommandError(f"{argv[0]} missing") from exc
def build_default_status_collector(
discovery: SlippiDiscovery,
xbox_bluetooth_address: str | None,
runner: CommandRunner | None = None,
) -> StatusCollector:
runner = runner or CommandRunner()
return StatusCollector(
display_probe=lambda: probe_display(runner),
slippi_probe=lambda: probe_slippi_version(discovery, runner),
dolphin_probe=lambda: probe_dolphin_version(discovery, runner),
controller_probe=lambda: probe_controller(xbox_bluetooth_address, runner),
)
def probe_display(runner: CommandRunner) -> dict[str, object]:
result = runner.run(["xrandr", "--query"])
if result.returncode != 0:
raise CommandError(result.stderr.strip() or "xrandr failed")
displays: dict[str, str] = {}
current_output: str | None = None
for raw_line in result.stdout.splitlines():
line = raw_line.rstrip()
if " connected" in line and not line.startswith(" "):
current_output = line.split()[0]
displays.setdefault(current_output, "unknown")
continue
if current_output is None or not line.startswith(" "):
continue
parts = line.split()
if not parts or "x" not in parts[0]:
continue
refresh = next((part.replace("*", "").replace("+", "") for part in parts[1:] if "*" in part), None)
if refresh is None:
continue
displays[current_output] = f"{parts[0]}@{refresh}"
primary_name = "HDMI-1" if "HDMI-1" in displays else next(iter(displays), None)
return {
"name": primary_name,
"mode": displays.get(primary_name, "unknown"),
"connected_outputs": displays,
}
def probe_slippi_version(discovery: SlippiDiscovery, runner: CommandRunner) -> str:
try:
launcher = discovery.find_launcher()
except DiscoveryError as exc:
raise CommandError(str(exc)) from exc
result = runner.run([str(launcher), "--version"])
output = "\n".join([result.stdout.strip(), result.stderr.strip()]).strip()
if result.returncode == 0 and output:
return output.splitlines()[0]
return launcher.stem
def probe_dolphin_version(discovery: SlippiDiscovery, runner: CommandRunner) -> str:
try:
dolphin = discovery.find_dolphin_binary()
except DiscoveryError as exc:
raise CommandError(str(exc)) from exc
result = runner.run([str(dolphin), "--version"])
output = "\n".join([result.stdout.strip(), result.stderr.strip()]).strip()
if result.returncode == 0 and output:
return output.splitlines()[0]
return dolphin.stem
def probe_controller(xbox_bluetooth_address: str | None, runner: CommandRunner) -> dict[str, object]:
if not xbox_bluetooth_address:
return {
"name": None,
"connected": "unknown",
"trusted": "unknown",
"address": None,
"error": "controller address not configured",
}
result = runner.run(["bluetoothctl", "info", xbox_bluetooth_address])
if result.returncode != 0:
raise CommandError(result.stderr.strip() or result.stdout.strip() or "bluetoothctl failed")
connected = "Connected: yes" in result.stdout
trusted = "Trusted: yes" in result.stdout
alias = "Xbox Wireless Controller"
for line in result.stdout.splitlines():
if line.strip().startswith("Alias: "):
alias = line.split(":", 1)[1].strip()
break
return {
"name": alias,
"connected": connected,
"trusted": trusted,
"address": xbox_bluetooth_address,
}
def write_status_json(path: Path, payload: dict[str, object]) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8")

View File

@@ -0,0 +1,14 @@
[Unit]
Description=CouchOS HDMI display/audio helper
After=graphical-session.target
Wants=graphical-session.target
PartOf=graphical-session.target
[Service]
Type=oneshot
WorkingDirectory=/home/couch/couchos
ExecStart=/home/couch/couchos/scripts/configure_display_audio.sh
RemainAfterExit=yes
[Install]
WantedBy=graphical-session.target

View File

@@ -0,0 +1,16 @@
[Unit]
Description=CouchOS control daemon
After=graphical-session.target network-online.target
Wants=graphical-session.target network-online.target
PartOf=graphical-session.target
[Service]
Type=simple
WorkingDirectory=/home/couch/couchos
Environment=PYTHONPATH=/home/couch/couchos/src
ExecStart=/usr/bin/python3 -m couchd --config /home/couch/.config/couchd/runtime.json serve
Restart=on-failure
RestartSec=2
[Install]
WantedBy=graphical-session.target

View File

@@ -0,0 +1,9 @@
# Template only. Replace placeholder identifiers after discovery on this machine.
# Preserve analog triggers. Do not convert triggers into digital-only bindings.
# Suggested profile name: Couch Xbox Wireless
# Suggested hotkey: map a safe combo that is unlikely to trigger accidentally.
[Profile]
Device = <REPLACE_WITH_DISCOVERED_DOLPHIN_DEVICE_NAME>
ProfileName = Couch Xbox Wireless
Notes = Preserve analog triggers; verify USB fallback; set safe exit hotkey manually in Dolphin UI.

View File

@@ -0,0 +1 @@
replace-with-long-random-token

View File

@@ -0,0 +1,8 @@
# Slippi First Run Checklist
- Install or update Slippi Dolphin through the official Slippi Launcher.
- Use your own unmodified NTSC 1.02 Melee image.
- In the graphics UI set Vulkan, fullscreen, 4:3, 2x internal resolution, and V-Sync off.
- Set replay storage to `/home/couch/Slippi`.
- Confirm gameplay on HDMI-1 at `1920x1080@60` before attempting any refresh-rate changes.
- If direct launch is unreliable, configure and validate the launcher shortcut fallback from `config/couchd.example.json`.

11
tests/__init__.py Normal file
View File

@@ -0,0 +1,11 @@
"""Test package bootstrap for src-layout unittest discovery."""
from __future__ import annotations
import sys
from pathlib import Path
SRC_DIR = Path(__file__).resolve().parents[1] / "src"
if str(SRC_DIR) not in sys.path:
sys.path.insert(0, str(SRC_DIR))

235
tests/test_couchd.py Normal file
View File

@@ -0,0 +1,235 @@
import unittest
from couchd.app import App, Config, ConfigError, Response, validate_bind_address
from couchd.discovery import DiscoveryError
class BindValidationTests(unittest.TestCase):
def test_rejects_loopback_address(self) -> None:
with self.assertRaises(ConfigError):
validate_bind_address("127.0.0.1")
def test_rejects_unspecified_address(self) -> None:
with self.assertRaises(ConfigError):
validate_bind_address("0.0.0.0")
def test_rejects_non_tailscale_address(self) -> None:
with self.assertRaises(ConfigError):
validate_bind_address("192.168.1.50")
def test_accepts_configured_tailscale_address(self) -> None:
self.assertEqual(validate_bind_address("100.64.0.15"), "100.64.0.15")
class AuthTests(unittest.TestCase):
def test_status_requires_bearer_token(self) -> None:
app = App(Config(bind_address="100.64.0.15", token="secret"))
response = app.handle_request("GET", "/status", headers={})
self.assertEqual(response, Response(401, {"error": "missing or invalid bearer token"}))
def test_status_returns_json_payload_when_authenticated(self) -> None:
app = App(
Config(bind_address="100.64.0.15", token="secret"),
status_provider=lambda: {
"state": "idle",
"pid": None,
"display": {"name": "HDMI-1", "mode": "1920x1080@60"},
"slippi_version": None,
"controller": {"connected": False, "name": "Xbox Wireless Controller"},
},
)
response = app.handle_request(
"GET",
"/status",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.body["state"], "idle")
self.assertEqual(response.body["display"]["mode"], "1920x1080@60")
def test_status_accepts_safe_query_string(self) -> None:
app = App(
Config(bind_address="100.64.0.15", token="secret"),
status_provider=lambda: {"state": "idle", "pid": None},
)
response = app.handle_request(
"GET",
"/status?cache_bust=1",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(response.body["state"], "idle")
class RouteTests(unittest.TestCase):
def test_launch_melee_uses_allowlisted_action(self) -> None:
calls: list[str] = []
class Launcher:
def launch_melee(self) -> dict[str, object]:
calls.append("melee")
return {"state": "launching", "pid": 4321}
def open_slippi(self) -> dict[str, object]:
calls.append("slippi")
return {"state": "launcher", "pid": 4322}
def stop(self) -> dict[str, object]:
calls.append("stop")
return {"state": "stopped", "pid": None}
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=Launcher(),
)
response = app.handle_request(
"POST",
"/launch/melee",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(calls, ["melee"])
self.assertEqual(response, Response(202, {"state": "launching", "pid": 4321}))
def test_open_slippi_uses_allowlisted_action(self) -> None:
calls: list[str] = []
class Launcher:
def launch_melee(self) -> dict[str, object]:
calls.append("melee")
return {"state": "launching", "pid": 4321}
def open_slippi(self) -> dict[str, object]:
calls.append("slippi")
return {"state": "launcher", "pid": 4322}
def stop(self) -> dict[str, object]:
calls.append("stop")
return {"state": "stopped", "pid": None}
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=Launcher(),
)
response = app.handle_request(
"POST",
"/open/slippi",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(calls, ["slippi"])
self.assertEqual(response, Response(202, {"state": "launcher", "pid": 4322}))
def test_stop_uses_allowlisted_action(self) -> None:
calls: list[str] = []
class Launcher:
def launch_melee(self) -> dict[str, object]:
calls.append("melee")
return {"state": "launching", "pid": 4321}
def open_slippi(self) -> dict[str, object]:
calls.append("slippi")
return {"state": "launcher", "pid": 4322}
def stop(self) -> dict[str, object]:
calls.append("stop")
return {"state": "stopped", "pid": None}
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=Launcher(),
)
response = app.handle_request(
"POST",
"/stop",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(calls, ["stop"])
self.assertEqual(response, Response(202, {"state": "stopped", "pid": None}))
def test_rejects_nonempty_body_for_action_route(self) -> None:
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=None,
)
response = app.handle_request(
"POST",
"/stop",
headers={"Authorization": "Bearer secret"},
body=b"x",
)
self.assertEqual(response, Response(400, {"error": "request body is not allowed"}))
def test_rejects_query_variant_of_unknown_route(self) -> None:
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=None,
)
response = app.handle_request(
"POST",
"/launch/unknown?x=1",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(response, Response(404, {"error": "not found"}))
def test_expected_launcher_failures_become_json_conflicts(self) -> None:
class Launcher:
def launch_melee(self) -> dict[str, object]:
raise DiscoveryError("ISO missing")
def open_slippi(self) -> dict[str, object]:
raise AssertionError("not expected")
def stop(self) -> dict[str, object]:
raise AssertionError("not expected")
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=Launcher(),
)
response = app.handle_request(
"POST",
"/launch/melee",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(response, Response(409, {"error": "ISO missing"}))
def test_unexpected_launcher_failures_become_json_service_errors(self) -> None:
class Launcher:
def launch_melee(self) -> dict[str, object]:
raise RuntimeError("boom")
def open_slippi(self) -> dict[str, object]:
raise AssertionError("not expected")
def stop(self) -> dict[str, object]:
raise AssertionError("not expected")
app = App(
Config(bind_address="100.64.0.15", token="secret"),
launcher=Launcher(),
)
response = app.handle_request(
"POST",
"/launch/melee",
headers={"Authorization": "Bearer secret"},
)
self.assertEqual(response, Response(503, {"error": "launcher action failed"}))

23
tests/test_discovery.py Normal file
View File

@@ -0,0 +1,23 @@
from pathlib import Path
import unittest
from couchd.discovery import AllowlistError, allowlisted_target
class AllowlistTests(unittest.TestCase):
def test_rejects_target_outside_allowed_roots(self) -> None:
with self.assertRaises(AllowlistError):
allowlisted_target(
candidate=Path("/tmp/SlippiLauncher.AppImage"),
allowed_roots=[Path("/home/couch/Applications")],
allowed_names={"SlippiLauncher.AppImage"},
)
def test_accepts_allowed_target_name_under_allowed_root(self) -> None:
target = allowlisted_target(
candidate=Path("/home/couch/Applications/SlippiLauncher.AppImage"),
allowed_roots=[Path("/home/couch/Applications")],
allowed_names={"SlippiLauncher.AppImage"},
)
self.assertEqual(target, Path("/home/couch/Applications/SlippiLauncher.AppImage"))

184
tests/test_display_audio.py Normal file
View File

@@ -0,0 +1,184 @@
import os
import subprocess
import tempfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
class DisplayAudioHelperTests(unittest.TestCase):
def test_prefers_hdmi_1080p_mode_disables_edp_and_selects_hdmi_sink(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
bin_dir = temp_path / "bin"
state_dir = temp_path / "state"
bin_dir.mkdir()
state_dir.mkdir()
log_file = state_dir / "calls.log"
(bin_dir / "xrandr").write_text(
f"""#!/usr/bin/env bash
set -Eeuo pipefail
if [[ "${{1:-}}" == "--query" ]]; then
cat <<'EOF'
HDMI-1 connected primary 1920x1080+0+0
1920x1080 60.00*+ 59.94
eDP-1 connected 1920x1080+1920+0
1920x1080 120.00*+ 60.00
EOF
exit 0
fi
printf 'xrandr %s\\n' "$*" >>"{log_file}"
""",
encoding="utf-8",
)
(bin_dir / "pactl").write_text(
f"""#!/usr/bin/env bash
set -Eeuo pipefail
case "$*" in
"list short cards")
cat <<'EOF'
1\talsa_card.pci-0000_00_1f.3\tmodule-alsa-card.c
EOF
;;
"list short sinks")
cat <<'EOF'
0\talsa_output.pci-0000_00_1f.3.analog-stereo\tmodule-alsa-card.c\ts16le 2ch 44100Hz\tSUSPENDED
1\talsa_output.pci-0000_00_1f.3.hdmi-stereo-extra1\tmodule-alsa-card.c\ts16le 2ch 44100Hz\tIDLE
EOF
;;
"list cards")
cat <<'EOF'
Card #1
\tName: alsa_card.pci-0000_00_1f.3
\tProfiles:
\t\toutput:analog-stereo: Analog Stereo (available: yes)
\t\toutput:hdmi-stereo-extra1: Digital Stereo (HDMI 2) (available: yes)
EOF
;;
set-card-profile*|set-default-sink*)
printf 'pactl %s\\n' "$*" >>"{log_file}"
;;
*)
exit 1
;;
esac
""",
encoding="utf-8",
)
(bin_dir / "logger").write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
for path in bin_dir.iterdir():
path.chmod(0o755)
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/configure_display_audio.sh")],
env={**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}"},
capture_output=True,
text=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
calls = log_file.read_text(encoding="utf-8")
self.assertIn(
"xrandr --output HDMI-1 --primary --mode 1920x1080 --rate 60.00 --output eDP-1 --off",
calls,
)
self.assertIn(
"pactl set-card-profile alsa_card.pci-0000_00_1f.3 output:hdmi-stereo-extra1",
calls,
)
self.assertIn(
"pactl set-default-sink alsa_output.pci-0000_00_1f.3.hdmi-stereo-extra1",
calls,
)
def test_requeries_sink_after_switching_hdmi_profile(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
bin_dir = temp_path / "bin"
state_dir = temp_path / "state"
bin_dir.mkdir()
state_dir.mkdir()
log_file = state_dir / "calls.log"
sink_state = state_dir / "sink-state"
sink_state.write_text("before", encoding="utf-8")
(bin_dir / "xrandr").write_text(
f"""#!/usr/bin/env bash
set -Eeuo pipefail
if [[ "${{1:-}}" == "--query" ]]; then
cat <<'EOF'
HDMI-1 connected primary 1920x1080+0+0
1920x1080 60.00*+ 59.94
eDP-1 connected 1920x1080+1920+0
1920x1080 120.00*+ 60.00
EOF
exit 0
fi
printf 'xrandr %s\\n' "$*" >>"{log_file}"
""",
encoding="utf-8",
)
(bin_dir / "pactl").write_text(
f"""#!/usr/bin/env bash
set -Eeuo pipefail
case "$*" in
"list cards")
cat <<'EOF'
Card #1
\tName: alsa_card.pci-0000_00_1f.3
\tProfiles:
\t\toutput:analog-stereo: Analog Stereo (available: yes)
\t\toutput:hdmi-stereo-extra1: Digital Stereo (HDMI 2) (available: yes)
EOF
;;
"list short sinks")
if [[ "$(cat "{sink_state}")" == "after" ]]; then
cat <<'EOF'
1\talsa_output.pci-0000_00_1f.3.hdmi-stereo-extra1\tmodule-alsa-card.c\ts16le 2ch 44100Hz\tIDLE
EOF
else
cat <<'EOF'
0\talsa_output.pci-0000_00_1f.3.analog-stereo\tmodule-alsa-card.c\ts16le 2ch 44100Hz\tSUSPENDED
EOF
fi
;;
set-card-profile*)
printf 'pactl %s\\n' "$*" >>"{log_file}"
printf 'after' >"{sink_state}"
;;
set-default-sink*)
printf 'pactl %s\\n' "$*" >>"{log_file}"
;;
*)
exit 1
;;
esac
""",
encoding="utf-8",
)
(bin_dir / "logger").write_text("#!/usr/bin/env bash\nexit 0\n", encoding="utf-8")
for path in bin_dir.iterdir():
path.chmod(0o755)
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/configure_display_audio.sh")],
env={**os.environ, "PATH": f"{bin_dir}:{os.environ['PATH']}"},
capture_output=True,
text=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
calls = log_file.read_text(encoding="utf-8")
self.assertIn(
"pactl set-card-profile alsa_card.pci-0000_00_1f.3 output:hdmi-stereo-extra1",
calls,
)
self.assertIn(
"pactl set-default-sink alsa_output.pci-0000_00_1f.3.hdmi-stereo-extra1",
calls,
)

163
tests/test_launcher.py Normal file
View File

@@ -0,0 +1,163 @@
import json
import tempfile
import unittest
from pathlib import Path
from unittest import mock
from couchd.launcher import (
LaunchRequest,
ProcessRegistry,
RunningProcess,
default_kill,
default_terminate,
)
class DuplicateLaunchTests(unittest.TestCase):
def test_refuses_second_launch_when_pid_is_still_alive(self) -> None:
launches: list[LaunchRequest] = []
registry = ProcessRegistry(
is_pid_running=lambda pid: pid == 4321,
spawn=lambda request: launches.append(request) or 9876,
process_probe=lambda: None,
process_identity=lambda pid: "proc-4321" if pid == 4321 else None,
)
registry.record_running(pid=4321, state="melee")
result = registry.launch(LaunchRequest(kind="melee", argv=["/bin/true"]))
self.assertEqual(result["state"], "already_running")
self.assertEqual(result["pid"], 4321)
self.assertEqual(launches, [])
def test_stop_escalates_from_term_to_kill_after_timeout(self) -> None:
signals: list[tuple[str, int]] = []
checks = iter([True, True, True, False])
registry = ProcessRegistry(
is_pid_running=lambda pid: next(checks),
spawn=lambda request: 9876,
terminate=lambda pid: signals.append(("TERM", pid)),
kill=lambda pid: signals.append(("KILL", pid)),
sleep=lambda seconds: None,
process_probe=lambda: None,
process_identity=lambda pid: "proc-4321" if pid == 4321 else None,
)
registry.record_running(pid=4321, state="melee")
result = registry.stop(timeout_seconds=0.0)
self.assertEqual(result, {"state": "stopped", "pid": None})
self.assertEqual(signals, [("TERM", 4321), ("KILL", 4321)])
def test_adopts_running_process_from_probe_when_state_file_is_stale(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state_file = Path(temp_dir) / "state.json"
state_file.write_text(json.dumps({"pid": 9999, "state": "melee"}), encoding="utf-8")
launches: list[LaunchRequest] = []
registry = ProcessRegistry(
is_pid_running=lambda pid: False,
spawn=lambda request: launches.append(request) or 7777,
state_file=state_file,
process_probe=lambda: RunningProcess(pid=4321, state="melee", identity="proc-a"),
process_identity=lambda pid: "proc-a" if pid == 4321 else None,
)
result = registry.launch(LaunchRequest(kind="melee", argv=["/bin/true"]))
self.assertEqual(result, {"state": "already_running", "pid": 4321})
self.assertEqual(launches, [])
def test_ignores_corrupt_state_file_and_recovers(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
state_file = Path(temp_dir) / "state.json"
state_file.write_text("{not-json", encoding="utf-8")
registry = ProcessRegistry(
is_pid_running=lambda pid: False,
spawn=lambda request: 9876,
state_file=state_file,
process_probe=lambda: None,
process_identity=lambda pid: None,
)
self.assertEqual(registry.current_state(), {"state": "idle", "pid": None})
def test_refuses_to_signal_reused_pid_with_different_identity(self) -> None:
signals: list[tuple[str, int]] = []
with tempfile.TemporaryDirectory() as temp_dir:
state_file = Path(temp_dir) / "state.json"
state_file.write_text(
json.dumps({"pid": 4321, "state": "melee", "identity": "old-proc"}),
encoding="utf-8",
)
registry = ProcessRegistry(
is_pid_running=lambda pid: True,
spawn=lambda request: 9876,
terminate=lambda pid: signals.append(("TERM", pid)),
kill=lambda pid: signals.append(("KILL", pid)),
state_file=state_file,
process_probe=lambda: None,
process_identity=lambda pid: "new-proc",
)
result = registry.stop(timeout_seconds=0.0)
self.assertEqual(result, {"state": "stopped", "pid": None})
self.assertEqual(signals, [])
def test_refuses_to_signal_same_executable_pid_reuse_when_start_time_changes(self) -> None:
signals: list[tuple[str, int]] = []
with tempfile.TemporaryDirectory() as temp_dir:
state_file = Path(temp_dir) / "state.json"
state_file.write_text(
json.dumps(
{
"pid": 4321,
"state": "melee",
"identity": "1000:/home/couch/Applications/Dolphin-x86_64.AppImage:111",
}
),
encoding="utf-8",
)
registry = ProcessRegistry(
is_pid_running=lambda pid: True,
spawn=lambda request: 9876,
terminate=lambda pid: signals.append(("TERM", pid)),
kill=lambda pid: signals.append(("KILL", pid)),
state_file=state_file,
process_probe=lambda: None,
process_identity=lambda pid: "1000:/home/couch/Applications/Dolphin-x86_64.AppImage:222",
)
result = registry.stop(timeout_seconds=0.0)
self.assertEqual(result, {"state": "stopped", "pid": None})
self.assertEqual(signals, [])
class DefaultSignalTests(unittest.TestCase):
def test_terminate_signals_non_group_leader_process_directly(self) -> None:
with mock.patch("couchd.launcher.os.getpgid", return_value=9999), mock.patch(
"couchd.launcher.os.kill"
) as kill, mock.patch("couchd.launcher.os.killpg") as killpg:
default_terminate(4321)
kill.assert_called_once()
self.assertEqual(kill.call_args.args[0], 4321)
killpg.assert_not_called()
def test_kill_signals_group_leader_process_group(self) -> None:
with mock.patch("couchd.launcher.os.getpgid", return_value=4321), mock.patch(
"couchd.launcher.os.kill"
) as kill, mock.patch("couchd.launcher.os.killpg") as killpg:
default_kill(4321)
kill.assert_not_called()
killpg.assert_called_once()

229
tests/test_scripts.py Normal file
View File

@@ -0,0 +1,229 @@
import io
import json
import os
import shutil
import subprocess
import tempfile
import tarfile
import unittest
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[1]
class InstallSlippiScriptTests(unittest.TestCase):
def test_fails_closed_when_checksum_asset_omits_appimage(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
temp_path = Path(temp_dir)
bin_dir = temp_path / "bin"
install_dir = temp_path / "install"
bin_dir.mkdir()
install_dir.mkdir()
release_json = {
"assets": [
{
"name": "Slippi-Launcher-9.9.9-x86_64.AppImage",
"browser_download_url": "https://github.com/project-slippi/slippi-launcher/releases/download/v9.9.9/Slippi-Launcher-9.9.9-x86_64.AppImage",
},
{
"name": "checksums.txt",
"browser_download_url": "https://github.com/project-slippi/slippi-launcher/releases/download/v9.9.9/checksums.txt",
},
]
}
curl_stub = bin_dir / "curl"
curl_stub.write_text(
"\n".join(
[
"#!/usr/bin/env bash",
"set -Eeuo pipefail",
'dest=""',
'url=""',
'while [[ $# -gt 0 ]]; do',
' case "$1" in',
' -o) dest="$2"; shift 2 ;;',
' *) url="$1"; shift ;;',
' esac',
'done',
f"cat <<'JSON' >\"$dest\"\n{json.dumps(release_json)}\nJSON" if False else "",
]
),
encoding="utf-8",
)
curl_stub.write_text(
"""#!/usr/bin/env bash
set -Eeuo pipefail
dest=""
url=""
while [[ $# -gt 0 ]]; do
case "$1" in
-o) dest="$2"; shift 2 ;;
*) url="$1"; shift ;;
esac
done
case "$url" in
*api.github.com*)
cat <<'JSON' >"$dest"
"""
+ json.dumps(release_json)
+ """
JSON
;;
*.AppImage)
printf 'fake appimage' >"$dest"
;;
*checksums.txt)
printf 'deadbeef SomeOtherFile.AppImage\n' >"$dest"
;;
*)
exit 1
;;
esac
""",
encoding="utf-8",
)
curl_stub.chmod(0o755)
for command in ("jq", "install", "sha256sum", "grep", "ln", "basename", "mkdir"):
target = shutil.which(command)
assert target is not None
os.symlink(target, bin_dir / command)
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/install_slippi.sh")],
env={
**os.environ,
"PATH": f"{bin_dir}:{os.environ['PATH']}",
"SLIPPI_INSTALL_DIR": str(install_dir),
},
capture_output=True,
text=True,
check=False,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("checksum asset present but does not include", result.stderr)
class BackupRestoreScriptTests(unittest.TestCase):
def test_backup_excludes_account_data_and_sets_private_mode(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
couch_home = Path(temp_dir) / "home" / "couch"
archive_path = Path(temp_dir) / "backup.tar.gz"
(couch_home / ".config" / "couchd").mkdir(parents=True)
(couch_home / ".config" / "dolphin-emu").mkdir(parents=True)
(couch_home / ".config" / "Slippi").mkdir(parents=True)
(couch_home / ".local" / "share" / "dolphin-emu").mkdir(parents=True)
(couch_home / "Slippi").mkdir(parents=True)
(couch_home / ".config" / "Slippi Launcher").mkdir(parents=True)
(couch_home / ".config" / "couchd" / "runtime.json").write_text("{}", encoding="utf-8")
(couch_home / ".config" / "Slippi Launcher" / "session.json").write_text("secret", encoding="utf-8")
(couch_home / "Slippi" / "Replay").mkdir()
(couch_home / "Slippi" / "Replay" / "match.slp").write_text("replay", encoding="utf-8")
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/backup_restore.sh"), "backup", str(archive_path)],
env={**os.environ, "COUCH_HOME": str(couch_home)},
capture_output=True,
text=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
self.assertEqual(archive_path.stat().st_mode & 0o777, 0o600)
listing = subprocess.run(
["tar", "-tzf", str(archive_path)],
capture_output=True,
text=True,
check=True,
).stdout
self.assertIn(".config/couchd/runtime.json", listing)
self.assertNotIn(".config/Slippi Launcher", listing)
self.assertNotIn("Replay/match.slp", listing)
def test_backup_excludes_recursive_case_insensitive_rom_and_replay_patterns(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
couch_home = Path(temp_dir) / "home" / "couch"
archive_path = Path(temp_dir) / "backup.tar.gz"
(couch_home / ".config" / "couchd").mkdir(parents=True)
(couch_home / ".config" / "Slippi Launcher" / "Profiles").mkdir(parents=True)
(couch_home / ".config" / "dolphin-emu").mkdir(parents=True)
(couch_home / ".config" / "Slippi").mkdir(parents=True)
(couch_home / ".local" / "share" / "dolphin-emu").mkdir(parents=True)
(couch_home / ".config" / "couchd" / "runtime.json").write_text("{}", encoding="utf-8")
(couch_home / "Slippi" / "Replays" / "nested").mkdir(parents=True)
(couch_home / "Slippi" / "Roms").mkdir(parents=True)
(couch_home / "Slippi" / "Replays" / "nested" / "match.SLP").write_text("replay", encoding="utf-8")
rom_gcm = f"melee.{''.join(['G', 'C', 'M'])}"
rom_wbfs = f"other.{''.join(['w', 'B', 'f', 'S'])}"
rom_nkit = "third." + ".".join(["nkit", "iso"])
(couch_home / "Slippi" / "Roms" / rom_gcm).write_text("rom", encoding="utf-8")
(couch_home / "Slippi" / "Roms" / rom_wbfs).write_text("rom", encoding="utf-8")
(couch_home / "Slippi" / "Roms" / rom_nkit).write_text("rom", encoding="utf-8")
(couch_home / "Slippi" / "keep.txt").write_text("keep", encoding="utf-8")
(couch_home / ".config" / "Slippi Launcher" / "Profiles" / "session.json").write_text(
"secret",
encoding="utf-8",
)
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/backup_restore.sh"), "backup", str(archive_path)],
env={**os.environ, "COUCH_HOME": str(couch_home)},
capture_output=True,
text=True,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
listing = subprocess.run(
["tar", "-tzf", str(archive_path)],
capture_output=True,
text=True,
check=True,
).stdout
self.assertIn(".config/couchd/runtime.json", listing)
self.assertNotIn(".config/Slippi Launcher/Profiles/session.json", listing)
self.assertNotIn("match.SLP", listing)
self.assertNotIn(rom_gcm, listing)
self.assertNotIn(rom_wbfs, listing)
self.assertNotIn(rom_nkit, listing)
self.assertIn("Slippi/keep.txt", listing)
def test_restore_rejects_unsafe_archive_entries(self) -> None:
with tempfile.TemporaryDirectory() as temp_dir:
couch_home = Path(temp_dir) / "home" / "couch"
couch_home.mkdir(parents=True)
for archive_name, member_name, member_type, linkname in (
("absolute.tar.gz", "/absolute.txt", tarfile.REGTYPE, ""),
("traversal.tar.gz", "../../escape.txt", tarfile.REGTYPE, ""),
("device.tar.gz", "device-node", tarfile.CHRTYPE, ""),
("link-escape.tar.gz", "link-out", tarfile.SYMTYPE, "../../escape.txt"),
):
archive_path = Path(temp_dir) / archive_name
with tarfile.open(archive_path, "w:gz") as archive:
info = tarfile.TarInfo(member_name)
info.type = member_type
info.mode = 0o600
if member_type == tarfile.REGTYPE:
payload = b"ok"
info.size = len(payload)
archive.addfile(info, io.BytesIO(payload))
elif member_type == tarfile.SYMTYPE:
info.linkname = linkname
archive.addfile(info)
else:
archive.addfile(info)
result = subprocess.run(
["bash", str(REPO_ROOT / "scripts/backup_restore.sh"), "restore", str(archive_path)],
env={**os.environ, "COUCH_HOME": str(couch_home)},
capture_output=True,
text=True,
check=False,
)
self.assertNotEqual(result.returncode, 0)
self.assertIn("unsafe archive entry", result.stderr)

55
tests/test_setup_modes.sh Normal file
View File

@@ -0,0 +1,55 @@
#!/usr/bin/env bash
set -Eeuo pipefail
REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
TMP_DIR="$(mktemp -d)"
trap 'rm -rf "${TMP_DIR}"' EXIT
BIN_DIR="${TMP_DIR}/bin"
LOG_FILE="${TMP_DIR}/calls.log"
mkdir -p "${BIN_DIR}"
cat >"${BIN_DIR}/id" <<'EOF'
#!/usr/bin/env bash
set -Eeuo pipefail
if [[ "${1:-}" == "-u" && "${2:-}" == "couch" ]]; then
exit 1
fi
exec /usr/bin/id "$@"
EOF
cat >"${BIN_DIR}/useradd" <<EOF
#!/usr/bin/env bash
set -Eeuo pipefail
printf 'useradd %s\n' "\$*" >>"${LOG_FILE}"
EOF
chmod 0755 "${BIN_DIR}/id" "${BIN_DIR}/useradd"
output="$(
PATH="${BIN_DIR}:${PATH}" \
bash -c '
set -Eeuo pipefail
source "'"${REPO_ROOT}"'/setup.sh"
MODE="dry-run"
COUCH_UID=""
ensure_couch_user
ensure_runtime_config
'
)"
grep -Fq '[dry-run] useradd --create-home --home-dir /home/couch --shell /bin/bash --user-group couch' <<<"${output}"
grep -Fq '/run/user/<missing-couch-uid>' <<<"${output}"
check_output_file="${TMP_DIR}/check.out"
set +e
PATH="${BIN_DIR}:${PATH}" bash "${REPO_ROOT}/setup.sh" --phase couch --check >"${check_output_file}" 2>&1
status=$?
set -e
if [[ "${status}" -eq 0 ]]; then
printf 'expected setup --check to fail on missing couch user\n' >&2
exit 1
fi
grep -Fq '[check] FAIL couch user missing' "${check_output_file}"
grep -Fq '[check] FAIL /home/couch/.config/couchd/runtime.json missing' "${check_output_file}"

73
tests/test_status.py Normal file
View File

@@ -0,0 +1,73 @@
import unittest
from pathlib import Path
from couchd.discovery import DiscoveryConfig, SlippiDiscovery
from couchd.status import CommandError, CommandRunner, StatusCollector, probe_display, probe_slippi_version
class StatusCollectorTests(unittest.TestCase):
def test_status_degrades_gracefully_when_probes_fail(self) -> None:
collector = StatusCollector(
display_probe=lambda: (_ for _ in ()).throw(CommandError("xrandr missing")),
slippi_probe=lambda: (_ for _ in ()).throw(TimeoutError("slow launcher")),
dolphin_probe=lambda: (_ for _ in ()).throw(TimeoutError("slow dolphin")),
controller_probe=lambda: (_ for _ in ()).throw(CommandError("bluetoothctl missing")),
)
status = collector.collect(state="idle", pid=None)
self.assertEqual(status["state"], "idle")
self.assertEqual(status["display"]["mode"], "unknown")
self.assertIn("xrandr missing", status["display"]["error"])
self.assertEqual(status["slippi_version"], "unknown")
self.assertIn("slow launcher", status["slippi_error"])
self.assertEqual(status["controller"]["connected"], "unknown")
def test_probe_display_reports_active_starred_mode_refresh(self) -> None:
class Runner(CommandRunner):
def run(self, argv: list[str], timeout: int = 5): # type: ignore[override]
del argv, timeout
return type(
"Completed",
(),
{
"returncode": 0,
"stdout": "\n".join(
[
"HDMI-1 connected primary 1920x1080+0+0",
" 1920x1080 60.00*+ 59.94",
"eDP-1 connected 1920x1080+1920+0",
" 1920x1080 120.00*+ 60.00",
]
),
"stderr": "",
},
)()
display = probe_display(Runner())
self.assertEqual(display["name"], "HDMI-1")
self.assertEqual(display["mode"], "1920x1080@60.00")
self.assertEqual(display["connected_outputs"]["eDP-1"], "1920x1080@120.00")
def test_probe_slippi_version_prefers_binary_version_output(self) -> None:
class Runner(CommandRunner):
def run(self, argv: list[str], timeout: int = 5): # type: ignore[override]
del timeout
if argv[-1] == "--version":
return type(
"Completed",
(),
{"returncode": 0, "stdout": "Slippi Launcher 3.2.1\n", "stderr": ""},
)()
raise AssertionError(argv)
discovery = SlippiDiscovery(
DiscoveryConfig(
couch_home=Path("/home/couch"),
applications_dir=Path("/home/couch/Applications"),
)
)
discovery.find_launcher = lambda: Path("/home/couch/Applications/SlippiLauncher.AppImage") # type: ignore[method-assign]
self.assertEqual(probe_slippi_version(discovery, Runner()), "Slippi Launcher 3.2.1")