feat: GET /api/terminal-token endpoint for muxterm HMAC auth

muxplex/main.py

@@ -10,6 +10,7 @@ Background poll loop reconciles tmux session state every POLL_INTERVAL seconds.
 import asyncio
 import contextlib
 import copy
+import hashlib
 import hmac
 import importlib.metadata
 import json
@@ -95,8 +96,11 @@ from muxplex.ttyd import (
 
 POLL_INTERVAL: float = float(os.environ.get("POLL_INTERVAL", "2.0"))
 SERVER_PORT: int = int(os.environ.get("MUXPLEX_PORT", "8088"))
+MUXTERM_PORT: int = int(os.environ.get("MUXTERM_PORT", "7682"))
 SETTINGS_SYNC_INTERVAL: int = 15  # sync every ~30 seconds (15 * 2s poll interval)
 
+_muxterm_secret: str = os.environ.get("MUXTERM_SECRET", "")
+
 _log = logging.getLogger(__name__)
 
 # ---------------------------------------------------------------------------
@@ -973,6 +977,27 @@ async def put_settings_sync(payload: SettingsSyncPayload):
         )
 
 
+def _generate_muxterm_token() -> str:
+    """Generate an HMAC-signed token for muxterm WebSocket authentication.
+
+    Token format: ``hex_signature.timestamp`` (valid ~30 seconds).
+    """
+    ts = str(int(time.time()))
+    sig = hmac.new(_muxterm_secret.encode(), ts.encode(), hashlib.sha256).hexdigest()
+    return f"{sig}.{ts}"
+
+
+@app.get("/api/terminal-token")
+async def get_terminal_token() -> dict:
+    """Return a short-lived HMAC token for muxterm WebSocket auth.
+
+    Raises HTTP 503 if MUXTERM_SECRET is not configured.
+    """
+    if not _muxterm_secret:
+        raise HTTPException(status_code=503, detail="MUXTERM_SECRET not configured")
+    return {"token": _generate_muxterm_token(), "port": MUXTERM_PORT}
+
+
 @app.get("/api/instance-info")
 async def instance_info() -> dict:
     """Return this instance's display name, device identity, and version.

muxplex/tests/test_api.py

@@ -2854,7 +2854,8 @@ async def test_poll_cycle_fires_federation_bell_clear_for_remote_session(
     monkeypatch.setattr("muxplex.main.enumerate_sessions", mock_enumerate)
     monkeypatch.setattr("muxplex.main.snapshot_all", mock_snapshot_all)
     monkeypatch.setattr(
-        "muxplex.main.update_session_cache", lambda names, snapshots, activity=None: None
+        "muxplex.main.update_session_cache",
+        lambda names, snapshots, activity=None: None,
     )
     monkeypatch.setattr("muxplex.main.apply_bell_clear_rule", lambda state: None)
     monkeypatch.setattr("muxplex.main.prune_devices", lambda state: None)
@@ -2966,7 +2967,8 @@ async def test_poll_cycle_fires_federation_bell_clear_for_remote_session_with_uu
     monkeypatch.setattr("muxplex.main.enumerate_sessions", mock_enumerate)
     monkeypatch.setattr("muxplex.main.snapshot_all", mock_snapshot_all)
     monkeypatch.setattr(
-        "muxplex.main.update_session_cache", lambda names, snapshots, activity=None: None
+        "muxplex.main.update_session_cache",
+        lambda names, snapshots, activity=None: None,
     )
     monkeypatch.setattr("muxplex.main.apply_bell_clear_rule", lambda state: None)
     monkeypatch.setattr("muxplex.main.prune_devices", lambda state: None)
@@ -3660,3 +3662,61 @@ def test_federation_connect_device_id_not_found(client, monkeypatch, tmp_path):
     assert response.status_code == 404, (
         f"Expected 404 for unknown device_id, got {response.status_code}: {response.text}"
     )
+
+
+# ---------------------------------------------------------------------------
+# GET /api/terminal-token (muxterm HMAC auth)
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def _set_muxterm_secret(monkeypatch):
+    """Set _muxterm_secret so the terminal-token endpoint returns 200."""
+    import muxplex.main as main_mod
+
+    monkeypatch.setattr(main_mod, "_muxterm_secret", "test-secret-for-tests")
+
+
+def test_terminal_token_returns_200(client, _set_muxterm_secret):
+    """GET /api/terminal-token returns 200 with a token string containing a dot."""
+    response = client.get("/api/terminal-token")
+    assert response.status_code == 200
+    data = response.json()
+    assert "token" in data
+    assert isinstance(data["token"], str)
+    assert "." in data["token"], (
+        f"Token must be in 'hex_signature.timestamp' format, got: {data['token']!r}"
+    )
+
+
+def test_terminal_token_contains_port(client, _set_muxterm_secret):
+    """GET /api/terminal-token returns 200 with an integer port field."""
+    response = client.get("/api/terminal-token")
+    assert response.status_code == 200
+    data = response.json()
+    assert "port" in data
+    assert isinstance(data["port"], int), (
+        f"port must be an int, got: {type(data['port']).__name__}"
+    )
+
+
+def test_terminal_token_is_unique_per_call(client, _set_muxterm_secret, monkeypatch):
+    """GET /api/terminal-token returns valid JSON and unique tokens across calls."""
+    import time
+
+    # Freeze time for first call, then advance for second call so timestamps differ
+    t1 = time.time()
+    monkeypatch.setattr(time, "time", lambda: t1)
+    response1 = client.get("/api/terminal-token")
+    assert response1.status_code == 200
+    data1 = response1.json()
+
+    t2 = t1 + 1.0
+    monkeypatch.setattr(time, "time", lambda: t2)
+    response2 = client.get("/api/terminal-token")
+    assert response2.status_code == 200
+    data2 = response2.json()
+
+    assert data1["token"] != data2["token"], (
+        "Tokens from different calls must be unique"
+    )