security(gateway): keep health endpoint local by default

Bind the gateway health listener to localhost by default and reduce the probe response to a minimal status payload so accidental public exposure leaks less information.

Made-with: Cursor
This commit is contained in:
Xubin Ren 2026-04-14 07:19:38 +00:00
parent 4999e2f734
commit e4b3f9bd28
4 changed files with 14 additions and 33 deletions

View File

@ -1727,6 +1727,7 @@ Example config:
}
},
"gateway": {
"host": "127.0.0.1",
"port": 18790
}
}
@ -1739,11 +1740,13 @@ nanobot gateway --config ~/.nanobot-telegram/config.json
nanobot gateway --config ~/.nanobot-discord/config.json
```
Each gateway instance also exposes a lightweight HTTP status endpoint on
`gateway.host:gateway.port`:
Each gateway instance also exposes a lightweight HTTP health endpoint on
`gateway.host:gateway.port`. By default, the gateway binds to `127.0.0.1`,
so the endpoint stays local unless you explicitly set `gateway.host` to a
public or LAN-facing address.
- `GET /` returns `nanobot`
- `GET /health` returns JSON with service metadata, uptime, and enabled channels
- `GET /health` returns `{"status":"ok"}`
- Other paths return `404`
Override workspace for one-off runs when needed:

View File

@ -824,9 +824,6 @@ def gateway(
async def _health_server(host: str, health_port: int):
"""Lightweight HTTP health endpoint on the gateway port."""
import json as _json
import time
start_time = time.monotonic()
async def handle(reader, writer):
try:
@ -842,28 +839,13 @@ def gateway(
method, path = parts[0], parts[1]
if method == "GET" and path == "/health":
uptime_s = int(time.monotonic() - start_time)
body = _json.dumps({
"service": "nanobot",
"version": __version__,
"status": "running",
"uptime_seconds": uptime_s,
"channels": channels.enabled_channels,
})
body = _json.dumps({"status": "ok"})
resp = (
f"HTTP/1.0 200 OK\r\n"
f"Content-Type: application/json\r\n"
f"Content-Length: {len(body)}\r\n"
f"\r\n{body}"
)
elif method == "GET" and path == "/":
body = "nanobot"
resp = (
f"HTTP/1.0 200 OK\r\n"
f"Content-Type: text/plain\r\n"
f"Content-Length: {len(body)}\r\n"
f"\r\n{body}"
)
else:
body = "Not Found"
resp = (

View File

@ -152,7 +152,7 @@ class ApiConfig(Base):
class GatewayConfig(Base):
"""Gateway/server configuration."""
host: str = "0.0.0.0"
host: str = "127.0.0.1" # Safer default: local-only bind.
port: int = 18790
heartbeat: HeartbeatConfig = Field(default_factory=HeartbeatConfig)

View File

@ -1131,7 +1131,6 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
) -> None:
config_file = _write_instance_config(tmp_path)
config = Config()
config.gateway.host = "127.0.0.9"
config.gateway.port = 18791
captured: dict[str, object] = {}
@ -1245,9 +1244,9 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
result = runner.invoke(app, ["gateway", "--config", str(config_file)])
assert result.exit_code == 0
assert captured["host"] == "127.0.0.9"
assert captured["host"] == "127.0.0.1"
assert captured["port"] == 18791
assert "Health endpoint: http://127.0.0.9:18791/health" in result.stdout
assert "Health endpoint: http://127.0.0.1:18791/health" in result.stdout
def _call_handler(path: str) -> tuple[str, _FakeWriter]:
request = f"GET {path} HTTP/1.1\r\nHost: localhost\r\n\r\n".encode()
@ -1259,17 +1258,14 @@ def test_gateway_health_endpoint_binds_and_serves_expected_responses(
root_response, root_writer = _call_handler("/")
assert root_writer.closed is True
assert "HTTP/1.0 200 OK" in root_response
assert root_response.endswith("\r\n\r\nnanobot")
assert "HTTP/1.0 404 Not Found" in root_response
assert root_response.endswith("\r\n\r\nNot Found")
health_response, health_writer = _call_handler("/health")
assert health_writer.closed is True
assert "HTTP/1.0 200 OK" in health_response
health_body = json.loads(health_response.split("\r\n\r\n", 1)[1])
assert health_body["service"] == "nanobot"
assert health_body["status"] == "running"
assert health_body["channels"] == ["telegram", "discord"]
assert health_body["uptime_seconds"] >= 0
assert health_body == {"status": "ok"}
missing_response, missing_writer = _call_handler("/missing")
assert missing_writer.closed is True