copilot-swe-agent[bot]andGitHub
2cb0c734d8
- Add modules/core/local-domain-loopback.nix: systemd service and activation script that write configured service domains to a Sovran-managed block in /etc/hosts (127.0.0.1 / ::1) so requests originating on this computer reach Caddy without NAT loopback. - Import local-domain-loopback.nix in modules/modules.nix. - server.py: add _validate_domain_value, _is_loopback_address, _resolve_all_addresses, _trigger_hosts_update helpers. - server.py: update _check_domain_reachable to use --resolve so reachability is checked locally via Caddy, not via NAT loopback. - server.py: update _evaluate_domain_checklist, api_services inline DNS check, and api_domains_check to recognise loopback resolution as an intentional local override rather than a DNS mismatch. - server.py: call _trigger_hosts_update from api_domains_set after saving a service domain so the /etc/hosts entry is applied immediately. - Add app/tests/test_loopback_diagnostics.py with 47 tests covering domain validation, loopback detection, diagnostic checklist logic, composite health, and api_domains_check."
400 lines
16 KiB
Python
400 lines
16 KiB
Python
"""Tests for server-local loopback diagnostics and domain validation.
|
|
|
|
Covers:
|
|
- Domain value validation and injection prevention.
|
|
- Loopback address detection (IPv4 and IPv6).
|
|
- _resolve_all_addresses returning multiple addresses.
|
|
- _check_domain_health_fast with loopback resolution.
|
|
- _evaluate_domain_checklist with loopback override — no false dns_mismatch.
|
|
- _evaluate_domain_checklist with genuine DNS mismatch — still reports error.
|
|
- api_services health stays "healthy" when domain resolves to loopback.
|
|
- api_services health stays "needs_attention" when DNS is genuinely wrong.
|
|
- api_domains_check returns "local_override" for loopback-resolved domains.
|
|
"""
|
|
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
import sys
|
|
import types
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Minimal stubs so server.py can be imported without the full FastAPI stack.
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _install_web_stubs():
|
|
if "fastapi" in sys.modules:
|
|
return
|
|
|
|
class _HTTPException(Exception):
|
|
def __init__(self, status_code=None, detail=None):
|
|
super().__init__(detail)
|
|
self.status_code = status_code
|
|
self.detail = detail
|
|
|
|
class _FastAPI:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
def mount(self, *args, **kwargs):
|
|
return None
|
|
|
|
def add_middleware(self, *args, **kwargs):
|
|
return None
|
|
|
|
def __getattr__(self, _name):
|
|
def _decorator_factory(*args, **kwargs):
|
|
def _decorator(func):
|
|
return func
|
|
return _decorator
|
|
return _decorator_factory
|
|
|
|
class _BaseModel:
|
|
pass
|
|
|
|
class _StaticFiles:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
class _Jinja2Templates:
|
|
def __init__(self, *args, **kwargs):
|
|
pass
|
|
|
|
class _BaseHTTPMiddleware:
|
|
pass
|
|
|
|
fastapi_module = types.ModuleType("fastapi")
|
|
fastapi_module.FastAPI = _FastAPI
|
|
fastapi_module.HTTPException = _HTTPException
|
|
sys.modules["fastapi"] = fastapi_module
|
|
|
|
responses_module = types.ModuleType("fastapi.responses")
|
|
responses_module.HTMLResponse = object
|
|
responses_module.JSONResponse = object
|
|
responses_module.RedirectResponse = object
|
|
sys.modules["fastapi.responses"] = responses_module
|
|
|
|
staticfiles_module = types.ModuleType("fastapi.staticfiles")
|
|
staticfiles_module.StaticFiles = _StaticFiles
|
|
sys.modules["fastapi.staticfiles"] = staticfiles_module
|
|
|
|
templating_module = types.ModuleType("fastapi.templating")
|
|
templating_module.Jinja2Templates = _Jinja2Templates
|
|
sys.modules["fastapi.templating"] = templating_module
|
|
|
|
requests_module = types.ModuleType("fastapi.requests")
|
|
requests_module.Request = object
|
|
sys.modules["fastapi.requests"] = requests_module
|
|
|
|
pydantic_module = types.ModuleType("pydantic")
|
|
pydantic_module.BaseModel = _BaseModel
|
|
sys.modules["pydantic"] = pydantic_module
|
|
|
|
starlette_base_module = types.ModuleType("starlette.middleware.base")
|
|
starlette_base_module.BaseHTTPMiddleware = _BaseHTTPMiddleware
|
|
sys.modules["starlette.middleware.base"] = starlette_base_module
|
|
|
|
starlette_middleware_module = types.ModuleType("starlette.middleware")
|
|
starlette_middleware_module.base = starlette_base_module
|
|
sys.modules["starlette.middleware"] = starlette_middleware_module
|
|
|
|
starlette_module = types.ModuleType("starlette")
|
|
starlette_module.middleware = starlette_middleware_module
|
|
sys.modules["starlette"] = starlette_module
|
|
|
|
|
|
_install_web_stubs()
|
|
from sovran_systemsos_web import server # noqa: E402
|
|
|
|
|
|
# ===========================================================================
|
|
# Domain value validation
|
|
# ===========================================================================
|
|
|
|
class TestValidateDomainValue(unittest.TestCase):
|
|
"""_validate_domain_value must reject anything that could corrupt /etc/hosts."""
|
|
|
|
def _v(self, value: str) -> bool:
|
|
return server._validate_domain_value(value)
|
|
|
|
# -- Valid values --------------------------------------------------------
|
|
|
|
def test_simple_domain_valid(self):
|
|
self.assertTrue(self._v("cloud.example.com"))
|
|
|
|
def test_subdomain_valid(self):
|
|
self.assertTrue(self._v("matrix.home.example.org"))
|
|
|
|
def test_single_label_with_tld_valid(self):
|
|
self.assertTrue(self._v("example.com"))
|
|
|
|
def test_hyphen_in_domain_valid(self):
|
|
self.assertTrue(self._v("my-nextcloud.example.com"))
|
|
|
|
# -- Injection / malformed values ----------------------------------------
|
|
|
|
def test_empty_string_invalid(self):
|
|
self.assertFalse(self._v(""))
|
|
|
|
def test_newline_injection_invalid(self):
|
|
self.assertFalse(self._v("evil.com\n127.0.0.1 other.host"))
|
|
|
|
def test_carriage_return_injection_invalid(self):
|
|
self.assertFalse(self._v("evil.com\r127.0.0.1 other.host"))
|
|
|
|
def test_space_injection_invalid(self):
|
|
self.assertFalse(self._v("evil.com 127.0.0.1"))
|
|
|
|
def test_hash_comment_injection_invalid(self):
|
|
self.assertFalse(self._v("evil.com# comment"))
|
|
|
|
def test_bare_hostname_no_dot_invalid(self):
|
|
self.assertFalse(self._v("localhost"))
|
|
|
|
def test_bare_ip_invalid(self):
|
|
self.assertFalse(self._v("192.168.1.1"))
|
|
|
|
def test_too_long_invalid(self):
|
|
self.assertFalse(self._v("a" * 254 + ".com"))
|
|
|
|
def test_leading_dot_invalid(self):
|
|
self.assertFalse(self._v(".example.com"))
|
|
|
|
def test_trailing_dot_invalid(self):
|
|
self.assertFalse(self._v("example.com."))
|
|
|
|
|
|
# ===========================================================================
|
|
# Loopback address detection
|
|
# ===========================================================================
|
|
|
|
class TestIsLoopbackAddress(unittest.TestCase):
|
|
|
|
def test_ipv4_loopback(self):
|
|
self.assertTrue(server._is_loopback_address("127.0.0.1"))
|
|
|
|
def test_ipv4_loopback_other(self):
|
|
self.assertTrue(server._is_loopback_address("127.0.0.2"))
|
|
|
|
def test_ipv4_loopback_high(self):
|
|
self.assertTrue(server._is_loopback_address("127.255.255.255"))
|
|
|
|
def test_ipv6_loopback(self):
|
|
self.assertTrue(server._is_loopback_address("::1"))
|
|
|
|
def test_public_ipv4_not_loopback(self):
|
|
self.assertFalse(server._is_loopback_address("203.0.113.10"))
|
|
|
|
def test_private_ipv4_not_loopback(self):
|
|
self.assertFalse(server._is_loopback_address("192.168.1.50"))
|
|
|
|
def test_ipv6_public_not_loopback(self):
|
|
self.assertFalse(server._is_loopback_address("2001:db8::1"))
|
|
|
|
def test_invalid_string_not_loopback(self):
|
|
self.assertFalse(server._is_loopback_address("not-an-ip"))
|
|
|
|
|
|
# ===========================================================================
|
|
# _check_domain_health_fast
|
|
# ===========================================================================
|
|
|
|
class TestCheckDomainHealthFast(unittest.TestCase):
|
|
"""_check_domain_health_fast returns True when there is an issue,
|
|
False when everything looks fine."""
|
|
|
|
def _fast(self, domain, external_ip, resolved_addrs):
|
|
with patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs):
|
|
return server._check_domain_health_fast(domain, external_ip)
|
|
|
|
def test_no_domain_no_issue(self):
|
|
# None/empty domain: the fast check reports True (handled by checklist).
|
|
result = server._check_domain_health_fast(None, "203.0.113.10")
|
|
self.assertTrue(result)
|
|
|
|
def test_empty_domain_no_issue(self):
|
|
result = server._check_domain_health_fast("", "203.0.113.10")
|
|
self.assertTrue(result)
|
|
|
|
def test_loopback_ipv4_no_issue(self):
|
|
"""Loopback override must not be flagged as a DNS mismatch."""
|
|
result = self._fast("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
|
self.assertFalse(result)
|
|
|
|
def test_loopback_ipv6_no_issue(self):
|
|
result = self._fast("cloud.example.com", "203.0.113.10", ["::1"])
|
|
self.assertFalse(result)
|
|
|
|
def test_matches_external_ip_no_issue(self):
|
|
result = self._fast("cloud.example.com", "203.0.113.10", ["203.0.113.10"])
|
|
self.assertFalse(result)
|
|
|
|
def test_mismatch_is_an_issue(self):
|
|
result = self._fast("cloud.example.com", "203.0.113.10", ["198.51.100.1"])
|
|
self.assertTrue(result)
|
|
|
|
def test_unavailable_external_ip_no_issue(self):
|
|
result = self._fast("cloud.example.com", "unavailable", ["198.51.100.1"])
|
|
self.assertFalse(result)
|
|
|
|
def test_multiple_addresses_one_matches_no_issue(self):
|
|
"""If any resolved address matches external_ip the check should pass."""
|
|
result = self._fast(
|
|
"cloud.example.com", "203.0.113.10",
|
|
["198.51.100.1", "203.0.113.10"],
|
|
)
|
|
self.assertFalse(result)
|
|
|
|
|
|
# ===========================================================================
|
|
# _evaluate_domain_checklist — loopback override path
|
|
# ===========================================================================
|
|
|
|
class TestEvaluateDomainChecklistLoopback(unittest.TestCase):
|
|
|
|
def _eval(self, domain, external_ip, resolved_addrs, reachable_result=None):
|
|
with (
|
|
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
|
patch.object(server, "_check_domain_reachable",
|
|
return_value=reachable_result or {"reachable": True, "status_code": 200}),
|
|
):
|
|
return server._evaluate_domain_checklist(domain, external_ip)
|
|
|
|
def test_loopback_dns_step_is_ok_not_error(self):
|
|
result = self._eval("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
|
dns_step = next(s for s in result["domain_check_steps"] if s["step"] == 2)
|
|
self.assertEqual(dns_step["status"], "ok")
|
|
self.assertNotIn("mismatch", dns_step.get("detail", "").lower())
|
|
|
|
def test_loopback_domain_status_is_local_override(self):
|
|
result = self._eval("cloud.example.com", "203.0.113.10", ["127.0.0.1"])
|
|
self.assertEqual(result["domain_status"]["status"], "local_override")
|
|
|
|
def test_loopback_has_no_issues_when_reachable(self):
|
|
result = self._eval(
|
|
"cloud.example.com", "203.0.113.10", ["127.0.0.1"],
|
|
reachable_result={"reachable": True, "status_code": 200},
|
|
)
|
|
self.assertFalse(result["has_issues"])
|
|
|
|
def test_loopback_has_issues_when_caddy_unreachable(self):
|
|
"""A loopback override with Caddy down should still report an issue."""
|
|
result = self._eval(
|
|
"cloud.example.com", "203.0.113.10", ["127.0.0.1"],
|
|
reachable_result={"reachable": False, "error": "connection refused"},
|
|
)
|
|
self.assertTrue(result["has_issues"])
|
|
|
|
def test_ipv6_loopback_no_issue(self):
|
|
result = self._eval("cloud.example.com", "203.0.113.10", ["::1"])
|
|
self.assertEqual(result["domain_status"]["status"], "local_override")
|
|
self.assertFalse(result["has_issues"])
|
|
|
|
def test_genuine_mismatch_still_reports_error(self):
|
|
result = self._eval("cloud.example.com", "203.0.113.10", ["198.51.100.1"])
|
|
self.assertEqual(result["domain_status"]["status"], "dns_mismatch")
|
|
self.assertTrue(result["has_issues"])
|
|
|
|
def test_correct_public_dns_still_reports_ok(self):
|
|
result = self._eval("cloud.example.com", "203.0.113.10", ["203.0.113.10"])
|
|
self.assertEqual(result["domain_status"]["status"], "connected")
|
|
self.assertFalse(result["has_issues"])
|
|
|
|
def test_no_domain_has_issues(self):
|
|
result = self._eval(None, "203.0.113.10", [])
|
|
self.assertTrue(result["has_issues"])
|
|
|
|
|
|
# ===========================================================================
|
|
# api_services — composite health with loopback
|
|
# ===========================================================================
|
|
|
|
class TestApiServicesLoopbackHealth(unittest.IsolatedAsyncioTestCase):
|
|
|
|
async def _get_health(self, resolved_addrs, cached_reachable):
|
|
"""Return the health value for a single domain-requiring service."""
|
|
service_cfg = {
|
|
"services": [
|
|
{"unit": "caddy.service", "icon": "nextcloud", "enabled": True, "type": "system"}
|
|
]
|
|
}
|
|
with (
|
|
patch.object(server, "load_config", return_value=service_cfg),
|
|
patch.object(server, "_read_hub_overrides", return_value=({}, None, None)),
|
|
patch.object(server.sysctl, "is_active", return_value="active"),
|
|
patch.dict(server.SERVICE_DOMAIN_MAP, {"caddy.service": "nextcloud"}, clear=False),
|
|
patch("builtins.open", mock_open(read_data="cloud.example.com\n")),
|
|
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
|
patch.object(server, "_is_domain_reachable_cached", return_value=cached_reachable),
|
|
patch.object(server, "_get_listening_ports",
|
|
return_value={"tcp": {80, 443}, "udp": set()}),
|
|
patch.object(server, "_get_firewall_allowed_ports",
|
|
return_value={"tcp": set(), "udp": set()}),
|
|
patch.object(server, "_cached_external_ip", "203.0.113.10"),
|
|
):
|
|
results = await server.api_services()
|
|
|
|
return results[0]["health"]
|
|
|
|
async def test_loopback_and_reachable_is_healthy(self):
|
|
"""Loopback override + Caddy reachable → healthy, not needs_attention."""
|
|
health = await self._get_health(["127.0.0.1"], cached_reachable=True)
|
|
self.assertEqual(health, "healthy")
|
|
|
|
async def test_loopback_and_caddy_down_is_needs_attention(self):
|
|
"""Loopback override + Caddy unreachable → needs_attention (genuine issue)."""
|
|
health = await self._get_health(["127.0.0.1"], cached_reachable=False)
|
|
self.assertEqual(health, "needs_attention")
|
|
|
|
async def test_correct_dns_and_reachable_is_healthy(self):
|
|
health = await self._get_health(["203.0.113.10"], cached_reachable=True)
|
|
self.assertEqual(health, "healthy")
|
|
|
|
async def test_dns_mismatch_is_needs_attention(self):
|
|
health = await self._get_health(["198.51.100.1"], cached_reachable=True)
|
|
self.assertEqual(health, "needs_attention")
|
|
|
|
|
|
# ===========================================================================
|
|
# api_domains_check — loopback detection
|
|
# ===========================================================================
|
|
|
|
class TestApiDomainsCheckLoopback(unittest.IsolatedAsyncioTestCase):
|
|
|
|
async def _check(self, resolved_addrs, external_ip="203.0.113.10"):
|
|
with (
|
|
patch.object(server, "_resolve_all_addresses", return_value=resolved_addrs),
|
|
patch.object(server, "_cached_external_ip", external_ip),
|
|
):
|
|
result = await server.api_domains_check(
|
|
MagicMock(domains=["cloud.example.com"])
|
|
)
|
|
return result["domains"][0]
|
|
|
|
async def test_loopback_ipv4_returns_local_override(self):
|
|
result = await self._check(["127.0.0.1"])
|
|
self.assertEqual(result["status"], "local_override")
|
|
|
|
async def test_loopback_ipv6_returns_local_override(self):
|
|
result = await self._check(["::1"])
|
|
self.assertEqual(result["status"], "local_override")
|
|
|
|
async def test_correct_dns_returns_connected(self):
|
|
result = await self._check(["203.0.113.10"])
|
|
self.assertEqual(result["status"], "connected")
|
|
|
|
async def test_mismatch_returns_dns_mismatch(self):
|
|
result = await self._check(["198.51.100.1"])
|
|
self.assertEqual(result["status"], "dns_mismatch")
|
|
|
|
async def test_no_resolution_returns_unresolvable(self):
|
|
result = await self._check([])
|
|
self.assertEqual(result["status"], "unresolvable")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|