Compare commits

...

1 Commits

Author SHA1 Message Date
Alexander Whitestone
c22cdcaa8e fix: add _validate_gateway_config tests and API_SERVER_KEY network binding warning
Some checks failed
Docker Build and Publish / build-and-push (pull_request) Has been skipped
Contributor Attribution Check / check-attribution (pull_request) Failing after 23s
Supply Chain Audit / Scan PR for supply chain risks (pull_request) Successful in 27s
Tests / e2e (pull_request) Successful in 1m51s
Tests / test (pull_request) Failing after 37m0s
Refs #892 - Gateway config debt: missing keys and broken fallbacks

Changes:
- Add `_is_network_accessible()` helper to gateway/config.py (avoids circular
  import with gateway.platforms.base which imports from gateway.config)
- Add API_SERVER_KEY warning in `_validate_gateway_config`: when the API server
  is enabled on a network-accessible address (0.0.0.0, public IP, hostname) but
  no key is configured, log a warning at config-load time so operators see the
  issue before any adapter initialisation runs
- Add `TestValidateGatewayConfig` in tests/gateway/test_config.py covering:
  - idle_minutes <= 0 and None are corrected to 1440 (default)
  - at_hour outside 0-23 is corrected to 4 (default)
  - Boundary hours 0 and 23 are accepted unchanged
  - Empty platform token triggers a warning log
  - Disabled platform with empty token produces no warning
  - API server on 0.0.0.0 without key logs a warning
  - API server on 127.0.0.1 without key is silent (loopback is allowed)
  - API server with a key set logs no warning regardless of bind address

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
2026-04-17 02:18:02 -04:00
2 changed files with 186 additions and 0 deletions

View File

@@ -8,6 +8,7 @@ Handles loading and validating configuration for:
- Delivery preferences
"""
import ipaddress
import logging
import os
import json
@@ -679,6 +680,26 @@ def load_gateway_config() -> GatewayConfig:
return config
def _is_network_accessible(host: str) -> bool:
"""Return True if *host* would expose a server beyond the loopback interface.
Duplicates the logic in ``gateway.platforms.base.is_network_accessible``
without creating a circular import (base.py imports from this module).
"""
try:
addr = ipaddress.ip_address(host)
if addr.is_loopback:
return False
# ::ffff:127.x.x.x — Python's is_loopback returns False for
# IPv4-mapped loopback; unwrap and check the underlying IPv4.
if getattr(addr, "ipv4_mapped", None) and addr.ipv4_mapped.is_loopback:
return False
return True
except ValueError:
# Hostname: assume it could be network-accessible.
return True
def _validate_gateway_config(config: "GatewayConfig") -> None:
"""Validate and sanitize a loaded GatewayConfig in place.
@@ -747,6 +768,22 @@ def _validate_gateway_config(config: "GatewayConfig") -> None:
)
pconfig.enabled = False
# Warn when the API server is enabled on a network-accessible address
# without an auth key. The adapter will refuse to start anyway, but
# surfacing this at config-load time lets operators see the problem in
# the startup log before any platform adapter initialisation runs.
api_cfg = config.platforms.get(Platform.API_SERVER)
if api_cfg and api_cfg.enabled:
key = api_cfg.extra.get("key", "")
host = api_cfg.extra.get("host", "127.0.0.1")
if not key and _is_network_accessible(host):
logger.warning(
"API Server is enabled on %s but API_SERVER_KEY is not set. "
"The adapter will refuse to start on a network-accessible address. "
"Set API_SERVER_KEY or bind to 127.0.0.1 for local-only access.",
host,
)
def _apply_env_overrides(config: GatewayConfig) -> None:
"""Apply environment variable overrides to config."""

View File

@@ -10,6 +10,7 @@ from gateway.config import (
PlatformConfig,
SessionResetPolicy,
_apply_env_overrides,
_validate_gateway_config,
load_gateway_config,
)
@@ -294,3 +295,151 @@ class TestHomeChannelEnvOverrides:
home = config.platforms[platform].home_channel
assert home is not None, f"{platform.value}: home_channel should not be None"
assert (home.chat_id, home.name) == expected, platform.value
class TestValidateGatewayConfig:
"""Tests for _validate_gateway_config — in-place sanitisation of loaded config."""
# -- idle_minutes validation --
def test_idle_minutes_zero_is_corrected_to_default(self):
config = GatewayConfig()
config.default_reset_policy.idle_minutes = 0
_validate_gateway_config(config)
assert config.default_reset_policy.idle_minutes == 1440
def test_idle_minutes_negative_is_corrected_to_default(self):
config = GatewayConfig()
config.default_reset_policy.idle_minutes = -60
_validate_gateway_config(config)
assert config.default_reset_policy.idle_minutes == 1440
def test_idle_minutes_none_is_corrected_to_default(self):
config = GatewayConfig()
config.default_reset_policy.idle_minutes = None # type: ignore[assignment]
_validate_gateway_config(config)
assert config.default_reset_policy.idle_minutes == 1440
def test_valid_idle_minutes_is_unchanged(self):
config = GatewayConfig()
config.default_reset_policy.idle_minutes = 90
_validate_gateway_config(config)
assert config.default_reset_policy.idle_minutes == 90
# -- at_hour validation --
def test_at_hour_too_high_is_corrected_to_default(self):
config = GatewayConfig()
config.default_reset_policy.at_hour = 24
_validate_gateway_config(config)
assert config.default_reset_policy.at_hour == 4
def test_at_hour_negative_is_corrected_to_default(self):
config = GatewayConfig()
config.default_reset_policy.at_hour = -1
_validate_gateway_config(config)
assert config.default_reset_policy.at_hour == 4
def test_valid_at_hour_is_unchanged(self):
config = GatewayConfig()
config.default_reset_policy.at_hour = 3
_validate_gateway_config(config)
assert config.default_reset_policy.at_hour == 3
def test_at_hour_boundary_values_are_valid(self):
for valid_hour in (0, 23):
config = GatewayConfig()
config.default_reset_policy.at_hour = valid_hour
_validate_gateway_config(config)
assert config.default_reset_policy.at_hour == valid_hour
# -- empty-token warning (enabled platforms) --
def test_empty_string_token_logs_warning(self, caplog):
import logging
config = GatewayConfig(
platforms={
Platform.TELEGRAM: PlatformConfig(enabled=True, token=""),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert any(
"TELEGRAM_BOT_TOKEN" in r.message and "empty" in r.message
for r in caplog.records
)
def test_disabled_platform_with_empty_token_no_warning(self, caplog):
import logging
config = GatewayConfig(
platforms={
Platform.TELEGRAM: PlatformConfig(enabled=False, token=""),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert not any("TELEGRAM_BOT_TOKEN" in r.message for r in caplog.records)
# -- API Server key / binding warnings --
def test_api_server_network_binding_without_key_logs_warning(self, caplog):
import logging
config = GatewayConfig(
platforms={
Platform.API_SERVER: PlatformConfig(
enabled=True,
extra={"host": "0.0.0.0"},
),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert any(
"API_SERVER_KEY" in r.message for r in caplog.records
)
def test_api_server_loopback_without_key_no_warning(self, caplog):
import logging
config = GatewayConfig(
platforms={
Platform.API_SERVER: PlatformConfig(
enabled=True,
extra={"host": "127.0.0.1"},
),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert not any(
"API_SERVER_KEY" in r.message for r in caplog.records
)
def test_api_server_network_binding_with_key_no_warning(self, caplog):
import logging
config = GatewayConfig(
platforms={
Platform.API_SERVER: PlatformConfig(
enabled=True,
extra={"host": "0.0.0.0", "key": "sk-real-key-here"},
),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert not any(
"API_SERVER_KEY" in r.message for r in caplog.records
)
def test_api_server_default_loopback_without_key_no_warning(self, caplog):
"""API server with no explicit host defaults to 127.0.0.1 — no warning."""
import logging
config = GatewayConfig(
platforms={
Platform.API_SERVER: PlatformConfig(enabled=True),
}
)
with caplog.at_level(logging.WARNING, logger="gateway.config"):
_validate_gateway_config(config)
assert not any(
"API_SERVER_KEY" in r.message for r in caplog.records
)