fix(dashboard): validate WebSocket Host and Origin

This commit is contained in:
hinotoi-agent 2026-05-22 12:29:48 +08:00 committed by Teknium
parent 186bf25cb1
commit 2e66eefbc3
2 changed files with 106 additions and 4 deletions

View file

@ -146,3 +146,72 @@ class TestHostHeaderMiddleware:
resp = client.get("/api/status")
# Should get through to the status endpoint, not a 400
assert resp.status_code != 400
class TestWebSocketHostOriginGuard:
"""WebSocket upgrades must enforce the same dashboard boundary as HTTP."""
def test_rebinding_websocket_host_is_rejected(self, monkeypatch):
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
import hermes_cli.web_server as ws
monkeypatch.setattr(ws.app.state, "bound_host", "127.0.0.1", raising=False)
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
client = TestClient(ws.app)
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
with pytest.raises(WebSocketDisconnect) as exc:
with client.websocket_connect(
url,
headers={
"Host": "evil.example",
"Origin": "http://evil.example",
},
):
pass
assert exc.value.code == 4403
def test_rebinding_websocket_origin_is_rejected(self, monkeypatch):
from fastapi.testclient import TestClient
from starlette.websockets import WebSocketDisconnect
import hermes_cli.web_server as ws
monkeypatch.setattr(ws.app.state, "bound_host", "127.0.0.1", raising=False)
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
client = TestClient(ws.app)
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
with pytest.raises(WebSocketDisconnect) as exc:
with client.websocket_connect(
url,
headers={
"Host": "localhost:9119",
"Origin": "http://evil.example",
},
):
pass
assert exc.value.code == 4403
def test_loopback_websocket_host_and_origin_are_accepted(self, monkeypatch):
from fastapi.testclient import TestClient
import hermes_cli.web_server as ws
monkeypatch.setattr(ws.app.state, "bound_host", "127.0.0.1", raising=False)
monkeypatch.setattr(ws, "_DASHBOARD_EMBEDDED_CHAT_ENABLED", True)
client = TestClient(ws.app)
url = f"/api/events?token={ws._SESSION_TOKEN}&channel=security-test"
with client.websocket_connect(
url,
headers={
"Host": "localhost:9119",
"Origin": "http://localhost:9119",
},
):
pass