diff --git a/muxplex/main.py b/muxplex/main.py index 7e5e8cd..f23aa29 100644 --- a/muxplex/main.py +++ b/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. diff --git a/muxplex/tests/test_api.py b/muxplex/tests/test_api.py index 6971da0..8ce1e71 100644 --- a/muxplex/tests/test_api.py +++ b/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" + )