Require auth for WebSocket token issuance

nanobot/channels/websocket.py

@@ -649,14 +649,14 @@ class WebSocketChannel(BaseChannel):
         return True
 
     def _handle_token_issue_http(self, connection: Any, request: Any) -> Any:
-        secret = self.config.token_issue_secret.strip()
+        secret = self.config.token_issue_secret.strip() or self.config.token.strip()
         if secret:
             if not _issue_route_secret_matches(request.headers, secret):
                 return connection.respond(401, "Unauthorized")
         else:
             self.logger.warning(
-                "token_issue_path is set but token_issue_secret is empty; "
+                "token_issue_path is set but no token_issue_secret or static token is configured; "
-                "any client can obtain connection tokens — set token_issue_secret for production."
+                "any client can obtain connection tokens — set a secret for production."
             )
         self._purge_expired_issued_tokens()
         if len(self._issued_tokens) >= self._MAX_ISSUED_TOKENS:

tests/channels/test_websocket_channel.py

@@ -202,6 +202,35 @@ def test_issue_route_secret_matches_empty_secret() -> None:
     assert _issue_route_secret_matches(Headers([("Authorization", "Bearer anything")]), "") is True
 
 
+@pytest.mark.asyncio
+async def test_token_issue_route_requires_secret_when_static_token_configured(bus: MagicMock) -> None:
+    port = 29882
+    channel = _ch(
+        bus,
+        port=port,
+        token="static-token",
+        tokenIssuePath="/auth/token",
+        websocketRequiresToken=True,
+    )
+
+    server_task = asyncio.create_task(channel.start())
+    await asyncio.sleep(0.3)
+
+    try:
+        denied = await _http_get(f"http://127.0.0.1:{port}/auth/token")
+        assert denied.status_code == 401
+
+        allowed = await _http_get(
+            f"http://127.0.0.1:{port}/auth/token",
+            headers={"Authorization": "Bearer static-token"},
+        )
+        assert allowed.status_code == 200
+        assert allowed.json()["token"].startswith("nbwt_")
+    finally:
+        await channel.stop()
+        await server_task
+
+
 @pytest.mark.asyncio
 async def test_webui_message_envelope_marks_inbound_metadata(bus: MagicMock) -> None:
     channel = _ch(bus)