diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 9c78b6775a..014a938e07 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -2882,6 +2882,25 @@ _VALID_CHANNEL_RE = re.compile(r"^[A-Za-z0-9._-]{1,128}$") # loopback so tests don't need to rewrite request scope. _LOOPBACK_HOSTS = frozenset({"127.0.0.1", "::1", "localhost", "testclient"}) + +def _is_public_bind() -> bool: + """True when bound to all-interfaces (operator used --insecure).""" + return getattr(app.state, "bound_host", "") in ("0.0.0.0", "::") + + +def _ws_client_is_allowed(ws: "WebSocket") -> bool: + """Check if the WebSocket client IP is acceptable. + + Allows loopback always; allows any IP when bound to all-interfaces + (--insecure mode, guarded by session token auth). + """ + if _is_public_bind(): + return True + client_host = ws.client.host if ws.client else "" + if not client_host: + return True + return client_host in _LOOPBACK_HOSTS + # 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 @@ -2972,8 +2991,7 @@ async def pty_ws(ws: WebSocket) -> None: await ws.close(code=4401) return - client_host = ws.client.host if ws.client else "" - if client_host and client_host not in _LOOPBACK_HOSTS: + if not _ws_client_is_allowed(ws): await ws.close(code=4403) return @@ -3080,8 +3098,7 @@ async def gateway_ws(ws: WebSocket) -> None: await ws.close(code=4401) return - client_host = ws.client.host if ws.client else "" - if client_host and client_host not in _LOOPBACK_HOSTS: + if not _ws_client_is_allowed(ws): await ws.close(code=4403) return @@ -3113,8 +3130,7 @@ async def pub_ws(ws: WebSocket) -> None: await ws.close(code=4401) return - client_host = ws.client.host if ws.client else "" - if client_host and client_host not in _LOOPBACK_HOSTS: + if not _ws_client_is_allowed(ws): await ws.close(code=4403) return @@ -3143,8 +3159,7 @@ async def events_ws(ws: WebSocket) -> None: await ws.close(code=4401) return - client_host = ws.client.host if ws.client else "" - if client_host and client_host not in _LOOPBACK_HOSTS: + if not _ws_client_is_allowed(ws): await ws.close(code=4403) return