fix(security): remediate Dependabot alerts #93, #94, #86

hermes_cli/web_server.py

@@ -3314,6 +3314,48 @@ def _ws_client_is_allowed(ws: "WebSocket") -> bool:
         return True
     return client_host in _LOOPBACK_HOSTS
 
+
+def _ws_host_origin_is_allowed(ws: "WebSocket") -> bool:
+    """Apply the dashboard Host/Origin guard to WebSocket upgrades.
+
+    FastAPI HTTP middleware (``host_header_middleware``) does not run for
+    WebSocket routes, so the DNS-rebinding Host check used for normal
+    dashboard HTTP requests must be repeated here before accepting the
+    upgrade. Browsers also send an Origin header on WebSocket handshakes;
+    when present, require it to target the same bound dashboard host.
+
+    Mirrors the HTTP-layer ``_is_accepted_host`` defence so a victim browser
+    tricked into a TTL-flipped attacker hostname (DNS rebinding) cannot reach
+    the WS endpoints even though its connection peer is 127.0.0.1.
+
+    See GHSA-4pqm-j46f-795x.
+    """
+    bound_host = getattr(app.state, "bound_host", None)
+    if not bound_host:
+        return True
+
+    host_header = ws.headers.get("host", "")
+    if not _is_accepted_host(host_header, bound_host):
+        return False
+
+    origin = ws.headers.get("origin", "")
+    if not origin:
+        # No Origin header (non-browser client, e.g. the spawned PTY child).
+        # The session-token / client-IP gates remain the auth boundary.
+        return True
+
+    parsed = urllib.parse.urlparse(origin)
+    if parsed.scheme not in {"http", "https"}:
+        # Non-web origin (packaged Electron: file://, null, app://). The
+        # token credential is the real auth boundary for those clients.
+        return True
+
+    if not parsed.netloc:
+        return False
+
+    return _is_accepted_host(parsed.netloc, bound_host)
+
+
 # Per-channel subscriber registry used by /api/pub (PTY-side gateway → dashboard)
 # and /api/events (dashboard → browser sidebar).  Keyed by an opaque channel id
 # the chat tab generates on mount; entries auto-evict when the last subscriber
@@ -3415,6 +3457,10 @@ async def pty_ws(ws: WebSocket) -> None:
         await ws.close(code=4401)
         return
 
+    if not _ws_host_origin_is_allowed(ws):
+        await ws.close(code=4403)
+        return
+
     if not _ws_client_is_allowed(ws):
         await ws.close(code=4403)
         return
@@ -3534,6 +3580,10 @@ async def gateway_ws(ws: WebSocket) -> None:
         await ws.close(code=4401)
         return
 
+    if not _ws_host_origin_is_allowed(ws):
+        await ws.close(code=4403)
+        return
+
     if not _ws_client_is_allowed(ws):
         await ws.close(code=4403)
         return
@@ -3566,6 +3616,10 @@ async def pub_ws(ws: WebSocket) -> None:
         await ws.close(code=4401)
         return
 
+    if not _ws_host_origin_is_allowed(ws):
+        await ws.close(code=4403)
+        return
+
     if not _ws_client_is_allowed(ws):
         await ws.close(code=4403)
         return
@@ -3595,6 +3649,10 @@ async def events_ws(ws: WebSocket) -> None:
         await ws.close(code=4401)
         return
 
+    if not _ws_host_origin_is_allowed(ws):
+        await ws.close(code=4403)
+        return
+
     if not _ws_client_is_allowed(ws):
         await ws.close(code=4403)
         return

tests/hermes_cli/test_web_server.py

@@ -2285,7 +2285,14 @@ def fake_resolve(resume=None, sidecar_url=None):
             self.ws_module.app.state, "bound_port", 9119, raising=False
         )
 
-        with self.client.websocket_connect(self._url(channel="abc-123")) as conn:
+        # bound_host is set above, so the WS Host/Origin guard
+        # (_ws_host_origin_is_allowed, GHSA-4pqm-j46f-795x) now requires the
+        # handshake Host header to match the bound interface. The TestClient
+        # defaults to "Host: testserver"; send a loopback Host so the upgrade
+        # is accepted.
+        with self.client.websocket_connect(
+            self._url(channel="abc-123"), headers={"host": "127.0.0.1:9119"}
+        ) as conn:
             try:
                 conn.receive_bytes()
             except Exception:

tests/hermes_cli/test_web_server_host_header.py

@@ -146,3 +146,90 @@ def test_no_bound_host_skips_validation(self):
         resp = client.get("/api/status")
         # Should get through to the status endpoint, not a 400
         assert resp.status_code != 400
+
+
+class _FakeWS:
+    """Minimal WebSocket stand-in exposing just the .headers mapping the
+    Host/Origin guard reads. Avoids spinning up the full ASGI app."""
+
+    def __init__(self, headers: dict[str, str]):
+        # Header lookups in the guard are lowercase; mirror that.
+        self.headers = {k.lower(): v for k, v in headers.items()}
+
+
+class TestWebSocketHostOriginGuard:
+    """Tests for GHSA-4pqm-j46f-795x — DNS-rebinding bypass via WebSocket
+    endpoints. FastAPI HTTP middleware does NOT run on WS upgrades, so the
+    Host (and, when present, Origin) header must be validated inside the WS
+    handlers. _ws_host_origin_is_allowed is that guard."""
+
+    def setup_method(self):
+        from hermes_cli.web_server import app
+
+        app.state.bound_host = "127.0.0.1"
+
+    def teardown_method(self):
+        from hermes_cli.web_server import app
+
+        if hasattr(app.state, "bound_host"):
+            del app.state.bound_host
+
+    def test_unbound_host_skips_guard(self):
+        from hermes_cli.web_server import app, _ws_host_origin_is_allowed
+
+        if hasattr(app.state, "bound_host"):
+            del app.state.bound_host
+        # No bound_host → nothing to compare against → allow (HTTP layer
+        # behaves the same).
+        assert _ws_host_origin_is_allowed(_FakeWS({"host": "evil.example"}))
+
+    def test_loopback_host_allowed(self):
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        assert _ws_host_origin_is_allowed(_FakeWS({"host": "localhost:9119"}))
+        assert _ws_host_origin_is_allowed(_FakeWS({"host": "127.0.0.1:9119"}))
+
+    def test_rebinding_host_rejected(self):
+        """An attacker hostname that TTL-flips to 127.0.0.1 passes the peer-IP
+        check but must be rejected by the Host guard on the WS upgrade."""
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        assert not _ws_host_origin_is_allowed(_FakeWS({"host": "evil.example"}))
+        assert not _ws_host_origin_is_allowed(
+            _FakeWS({"host": "rebind.attacker.test:9119"})
+        )
+
+    def test_cross_origin_rejected(self):
+        """Browser sends Origin on the WS handshake; a mismatched web Origin
+        (the cross-site rebinding driver) must be rejected even if Host is
+        spoofed to loopback."""
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        assert not _ws_host_origin_is_allowed(
+            _FakeWS({"host": "127.0.0.1:9119", "origin": "https://evil.example"})
+        )
+
+    def test_matching_origin_allowed(self):
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        assert _ws_host_origin_is_allowed(
+            _FakeWS({"host": "127.0.0.1:9119", "origin": "http://127.0.0.1:9119"})
+        )
+
+    def test_non_web_origin_allowed(self):
+        """Packaged Electron / native clients send file://, null, or app://
+        origins. The token credential is the auth boundary there — don't
+        reject on Origin scheme."""
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        for origin in ("file://", "null", "app://hermes"):
+            assert _ws_host_origin_is_allowed(
+                _FakeWS({"host": "127.0.0.1:9119", "origin": origin})
+            )
+
+    def test_missing_origin_allowed(self):
+        """Non-browser WS clients omit Origin entirely — allowed (Host still
+        validated)."""
+        from hermes_cli.web_server import _ws_host_origin_is_allowed
+
+        assert _ws_host_origin_is_allowed(_FakeWS({"host": "127.0.0.1:9119"}))