fix(webhook): widen INSECURE_NO_AUTH loopback check + tests + docs

Follow-up to the previous commit:
- Add _is_loopback_host() helper covering 127.0.0.1, localhost, ::1,
  ip6-localhost, ip6-loopback (case-insensitive). Empty/None host is
  treated as non-loopback since unset usually means public default bind.
- Fix mixed-indent comment in the safety rail (comment now aligned with
  the if-block) and collapse the nested-if into one condition.
- Add TestInsecureNoAuthSafetyRail covering rejection on 0.0.0.0, a LAN
  IP, and empty host; allowance on 127.0.0.1/localhost; plus unit-level
  parametrized coverage of _is_loopback_host for spellings we can't bind
  in the hermetic test env (::1, ip6-localhost, ip6-loopback).
- Pin test_connect_starts_server + test_webhook_deliver_only defaults
  to 127.0.0.1 so they keep passing under the new rail.
- Document the behavior in website/docs/user-guide/messaging/webhooks.md.
This commit is contained in:
teknium1 2026-05-07 07:34:54 -07:00 committed by Teknium
parent fb4f953569
commit 898b6d7d55
4 changed files with 116 additions and 11 deletions

View file

@ -352,7 +352,7 @@ class TestHTTPHandling:
async def test_connect_starts_server(self):
"""connect() starts the HTTP listener and marks adapter as connected."""
routes = {"r1": {"secret": _INSECURE_NO_AUTH, "prompt": "x"}}
adapter = _make_adapter(routes=routes, port=0)
adapter = _make_adapter(routes=routes, host="127.0.0.1", port=0)
# Use port 0 — the OS picks a free port, but aiohttp requires a real bind.
# We just test that the method completes and marks connected.
# Need to mock TCPSite to avoid actual binding.
@ -758,3 +758,80 @@ class TestDeliverCrossPlatformThreadId:
mock_target.send.assert_awaited_once_with(
"12345", "hello", metadata=None
)
class TestInsecureNoAuthSafetyRail:
"""connect() refuses to start when INSECURE_NO_AUTH is combined with a
non-loopback bind. Guards against accidentally exposing an unauthenticated
webhook endpoint on a public interface."""
@pytest.mark.asyncio
async def test_connect_rejects_insecure_no_auth_on_public_bind(self):
"""INSECURE_NO_AUTH + 0.0.0.0 is refused before the server starts."""
routes = {"r1": {"secret": _INSECURE_NO_AUTH, "prompt": "x"}}
adapter = _make_adapter(routes=routes, host="0.0.0.0", port=0)
with pytest.raises(ValueError, match="INSECURE_NO_AUTH"):
await adapter.connect()
@pytest.mark.asyncio
async def test_connect_rejects_insecure_no_auth_on_lan_ip(self):
"""A LAN IP is treated as public."""
routes = {"r1": {"secret": _INSECURE_NO_AUTH, "prompt": "x"}}
adapter = _make_adapter(routes=routes, host="192.168.1.50", port=0)
with pytest.raises(ValueError, match="non-loopback"):
await adapter.connect()
@pytest.mark.asyncio
async def test_connect_rejects_insecure_no_auth_on_empty_host(self):
"""Empty host is conservatively treated as non-loopback."""
routes = {"r1": {"secret": _INSECURE_NO_AUTH, "prompt": "x"}}
adapter = _make_adapter(routes=routes, host="", port=0)
with pytest.raises(ValueError, match="INSECURE_NO_AUTH"):
await adapter.connect()
@pytest.mark.parametrize(
"host",
["127.0.0.1", "localhost"],
)
@pytest.mark.asyncio
async def test_connect_allows_insecure_no_auth_on_loopback(self, host):
"""Recognised loopback hosts are permitted with INSECURE_NO_AUTH."""
routes = {"r1": {"secret": _INSECURE_NO_AUTH, "prompt": "x"}}
adapter = _make_adapter(routes=routes, host=host, port=0)
try:
with patch.object(adapter, "_reload_dynamic_routes"):
result = await adapter.connect()
assert result is True
finally:
await adapter.disconnect()
@pytest.mark.parametrize(
"host",
["127.0.0.1", "localhost", "Localhost", "::1", "ip6-localhost", "ip6-loopback"],
)
def test_is_loopback_host_accepts(self, host):
"""_is_loopback_host covers all documented loopback spellings."""
from gateway.platforms.webhook import _is_loopback_host
assert _is_loopback_host(host) is True
@pytest.mark.parametrize(
"host",
["0.0.0.0", "192.168.1.5", "10.0.0.1", "example.com", "", None],
)
def test_is_loopback_host_rejects(self, host):
"""_is_loopback_host treats public/LAN/empty as non-loopback."""
from gateway.platforms.webhook import _is_loopback_host
assert _is_loopback_host(host) is False
@pytest.mark.asyncio
async def test_connect_allows_real_secret_on_public_bind(self):
"""A real HMAC secret bound to 0.0.0.0 is the normal production case."""
routes = {"r1": {"secret": "real-secret-abc123", "prompt": "x"}}
adapter = _make_adapter(routes=routes, host="0.0.0.0", port=0)
try:
with patch.object(adapter, "_reload_dynamic_routes"):
result = await adapter.connect()
assert result is True
finally:
await adapter.disconnect()

View file

@ -33,7 +33,7 @@ from gateway.platforms.webhook import WebhookAdapter, _INSECURE_NO_AUTH
# ---------------------------------------------------------------------------
def _make_adapter(routes, **extra_kw) -> WebhookAdapter:
extra = {"host": "0.0.0.0", "port": 0, "routes": routes}
extra = {"host": "127.0.0.1", "port": 0, "routes": routes}
extra.update(extra_kw)
config = PlatformConfig(enabled=True, extra=extra)
return WebhookAdapter(config)