breakglass: in-cluster emergency-recovery UI for the devvm

Viktor wanted a web UI on the claude service to act as his breakglass when
the devvm is down: open it, have Claude SSH in to diagnose/repair, and
power-cycle the VM via the Proxmox host if needed. This is the app half
(the infra stack + host bootstrap live in the infra repo).

New, ISOLATED ASGI app under app/breakglass/ (never imports app.main, so the
untrusted-input agents — recruiter-triage, nextcloud-todos — can't share a
process with the root-on-devvm / PVE-reset SSH key):
- pve.py: the LLM-independent power-verb path (status|forensics|reset|stop|
  start|cycle on VM 102), whitelist-validated client-side, executed over the
  forced-command SSH key (list argv, no shell).
- agent_session.py: multi-turn streamed chat — claude -p --session-id /
  --resume with --output-format stream-json, translated to a small SSE
  vocabulary (session/text/tool/result/error/done).
- auth.py: edge Authentik header OR bearer; fail-closed.
- server.py: FastAPI (session/chat-SSE/pve-verb routes) + serves the Svelte UI.
- Svelte SPA (frontend/, built into app/breakglass/static/ and committed — no
  in-cluster build, per ADR-0002): streamed chat + danger-styled manual VM
  controls with confirm-on-mutate.
- agents/breakglass.md: narrow tools (Bash/Read/Grep/Glob, no web), taught the
  ssh devvm / ssh pve aliases and cycle-vs-reset.
- docker-entrypoint-breakglass.sh: ssh-agent bootstrap from the mounted key +
  ssh aliases, then uvicorn app.breakglass.server. The breakglass Deployment
  overrides the image CMD with this; the existing service is untouched.

26 new tests (verb whitelist incl. injection attempts, stream-json→SSE
translation, auth gating, route behaviour); full suite 58 green.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

tests/test_breakglass.py

@@ -0,0 +1,174 @@
+"""Tests for the breakglass app: verb whitelist, SSE translation, auth, routes."""
+import os
+
+os.environ.setdefault("API_BEARER_TOKEN", "test-token")
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.breakglass import agent_session, pve
+from app.breakglass.server import app
+
+
+# --------------------------------------------------------------------------- #
+# PVE verb whitelist — the security boundary mirrored client-side.
+# --------------------------------------------------------------------------- #
+
+def test_allowed_verbs_match_host_script():
+    assert pve.ALLOWED_VERBS == {
+        "status", "forensics", "reset", "stop", "start", "cycle"
+    }
+    assert pve.MUTATING_VERBS == {"reset", "stop", "start", "cycle"}
+    assert pve.MUTATING_VERBS < pve.ALLOWED_VERBS
+
+
+@pytest.mark.parametrize("bad", [
+    "rm -rf /", "status; rm -rf /", "status 103", "shutdown", "", "STATUS",
+    "cycle 999", "$(reboot)", "../start",
+])
+@pytest.mark.asyncio
+async def test_run_verb_rejects_non_whitelisted_without_ssh(bad, monkeypatch):
+    """A bad verb must be rejected locally — never spawning a subprocess."""
+    called = False
+
+    async def _boom(*a, **k):
+        nonlocal called
+        called = True
+        raise AssertionError("ssh must not run for a rejected verb")
+
+    monkeypatch.setattr(pve.asyncio, "create_subprocess_exec", _boom)
+    result = await pve.run_verb(bad)
+    assert result["rejected"] is True
+    assert result["exit_code"] is None
+    assert called is False
+
+
+@pytest.mark.asyncio
+async def test_run_verb_allowed_invokes_ssh_with_bare_verb(monkeypatch):
+    captured = {}
+
+    class _FakeProc:
+        returncode = 0
+
+        async def communicate(self):
+            return (b"status: running\n", b"")
+
+    async def _fake_exec(*argv, **kwargs):
+        captured["argv"] = argv
+        return _FakeProc()
+
+    monkeypatch.setattr(pve.asyncio, "create_subprocess_exec", _fake_exec)
+    result = await pve.run_verb("status")
+    assert result["rejected"] is False
+    assert result["exit_code"] == 0
+    assert "running" in result["stdout"]
+    # The verb is the LAST argv element, passed as a single token (no shell).
+    assert captured["argv"][-1] == "status"
+    assert captured["argv"][0] == "ssh"
+
+
+# --------------------------------------------------------------------------- #
+# stream-json -> UI event translation (pure function).
+# --------------------------------------------------------------------------- #
+
+def test_translate_init_to_session():
+    ev = agent_session.translate_event(
+        {"type": "system", "subtype": "init", "session_id": "abc"}
+    )
+    assert ev == {"kind": "session", "session_id": "abc"}
+
+
+@pytest.mark.parametrize("noise", [
+    {"type": "system", "subtype": "hook_started"},
+    {"type": "system", "subtype": "thinking_tokens", "estimated_tokens": 5},
+    {"type": "user", "message": {"content": []}},
+    {"type": "unknown"},
+])
+def test_translate_drops_noise(noise):
+    assert agent_session.translate_event(noise) is None
+
+
+def test_translate_assistant_text():
+    ev = agent_session.translate_event({
+        "type": "assistant",
+        "message": {"content": [{"type": "text", "text": "checking disk"}]},
+    })
+    assert ev == {"kind": "text", "text": "checking disk"}
+
+
+def test_translate_assistant_tool_use():
+    ev = agent_session.translate_event({
+        "type": "assistant",
+        "message": {"content": [
+            {"type": "tool_use", "name": "Bash", "input": {"command": "df -h"}}
+        ]},
+    })
+    assert ev["kind"] == "tool"
+    assert ev["name"] == "Bash"
+    assert ev["input"]["command"] == "df -h"
+
+
+def test_translate_result():
+    ev = agent_session.translate_event({
+        "type": "result", "is_error": False, "result": "done", "duration_ms": 1234,
+    })
+    assert ev == {"kind": "result", "is_error": False, "result": "done", "duration_ms": 1234}
+
+
+# --------------------------------------------------------------------------- #
+# Routes + auth.
+# --------------------------------------------------------------------------- #
+
+client = TestClient(app)
+AUTH = {"Authorization": "Bearer test-token"}
+
+
+def test_health_no_auth():
+    r = client.get("/health")
+    assert r.status_code == 200
+    assert r.json()["service"] == "claude-breakglass"
+
+
+def test_api_requires_auth():
+    assert client.post("/api/session").status_code == 401
+    assert client.get("/api/pve/verbs").status_code == 401
+
+
+def test_api_accepts_bearer():
+    r = client.post("/api/session", headers=AUTH)
+    assert r.status_code == 200
+    assert "session_id" in r.json()
+
+
+def test_api_accepts_authentik_header():
+    r = client.post("/api/session", headers={"X-authentik-username": "me@viktorbarzin.me"})
+    assert r.status_code == 200
+
+
+def test_pve_verb_route_rejects_unknown():
+    r = client.post("/api/pve/destroy", headers=AUTH)
+    assert r.status_code == 400
+
+
+def test_pve_verbs_listing():
+    r = client.get("/api/pve/verbs", headers=AUTH)
+    assert r.status_code == 200
+    body = r.json()
+    assert set(body["verbs"]) == pve.ALLOWED_VERBS
+    assert set(body["mutating"]) == pve.MUTATING_VERBS
+
+
+def test_chat_streams_sse(monkeypatch):
+    async def _fake_turn(session_id, prompt, model=None):
+        yield {"kind": "session", "session_id": session_id}
+        yield {"kind": "text", "text": "hello"}
+        yield {"kind": "result", "is_error": False, "result": "ok"}
+
+    monkeypatch.setattr(agent_session, "run_turn", _fake_turn)
+    r = client.post("/api/chat", headers=AUTH,
+                    json={"session_id": "s1", "prompt": "diagnose"})
+    assert r.status_code == 200
+    assert "text/event-stream" in r.headers["content-type"]
+    body = r.text
+    assert "hello" in body
+    assert '"kind": "done"' in body  # terminal frame always emitted