feat: server-side loopback overrides and Hub diagnostic fixes
- 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."
This commit is contained in:
copilot-swe-agent[bot]
@@ -8,6 +8,7 @@ import contextlib
|
||||
import glob
|
||||
import hashlib
|
||||
import hmac
|
||||
import ipaddress
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
@@ -80,6 +81,15 @@ DOMAINS_DIR = "/var/lib/domains"
|
||||
NOSTR_NPUB_FILE = "/var/lib/secrets/nostr_npub"
|
||||
NJALLA_SCRIPT = "/var/lib/njalla/njalla.sh"
|
||||
|
||||
# Systemd service that rewrites the Sovran-managed /etc/hosts loopback block
|
||||
SOVRAN_HOSTS_SERVICE = "sovran-hosts-update.service"
|
||||
|
||||
# Domain keys that produce a public HTTPS virtual host via Caddy
|
||||
_SERVICE_DOMAIN_KEYS = frozenset([
|
||||
"matrix", "wordpress", "nextcloud", "btcpayserver",
|
||||
"vaultwarden", "haven", "element-calling",
|
||||
])
|
||||
|
||||
INTERNAL_IP_FILE = "/var/lib/secrets/internal-ip"
|
||||
ZEUS_CONNECT_FILE = "/var/lib/secrets/zeus-connect-url"
|
||||
|
||||
@@ -964,18 +974,87 @@ def _check_port_status(
|
||||
return "closed"
|
||||
|
||||
|
||||
|
||||
# Regex for validating domain values written into /etc/hosts. Rejects anything
|
||||
# containing whitespace, newlines, or characters that could escape a hosts entry.
|
||||
_SAFE_DOMAIN_RE = re.compile(
|
||||
r'^(?:[a-zA-Z0-9](?:[a-zA-Z0-9\-]{0,61}[a-zA-Z0-9])?\.)+[a-zA-Z]{2,}$'
|
||||
)
|
||||
|
||||
|
||||
def _validate_domain_value(domain: str) -> bool:
|
||||
"""Return True if *domain* is a valid hostname safe to write into /etc/hosts.
|
||||
|
||||
Rejects values containing whitespace, newlines, or other characters that
|
||||
could inject additional entries or corrupt the hosts file.
|
||||
"""
|
||||
if not domain or len(domain) > 253:
|
||||
return False
|
||||
# Guard against newline / whitespace injection before regex check.
|
||||
if any(c in domain for c in ('\n', '\r', ' ', '\t', '#')):
|
||||
return False
|
||||
return bool(_SAFE_DOMAIN_RE.match(domain))
|
||||
|
||||
|
||||
def _is_loopback_address(ip: str) -> bool:
|
||||
"""Return True if *ip* is a loopback address (127.0.0.0/8 or ::1)."""
|
||||
try:
|
||||
return ipaddress.ip_address(ip).is_loopback
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _resolve_all_addresses(domain: str) -> list[str]:
|
||||
"""Return all IP addresses that *domain* resolves to, or an empty list."""
|
||||
try:
|
||||
results = socket.getaddrinfo(domain, None)
|
||||
seen: list[str] = []
|
||||
for r in results:
|
||||
addr = r[4][0]
|
||||
if addr not in seen:
|
||||
seen.append(addr)
|
||||
return seen
|
||||
except Exception:
|
||||
return []
|
||||
|
||||
|
||||
def _trigger_hosts_update() -> None:
|
||||
"""Start the sovran-hosts-update systemd service (best-effort, no-op if unavailable)."""
|
||||
try:
|
||||
subprocess.run(
|
||||
["systemctl", "start", SOVRAN_HOSTS_SERVICE],
|
||||
timeout=30,
|
||||
check=False,
|
||||
capture_output=True,
|
||||
)
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
|
||||
def _check_domain_reachable(domain: str) -> dict:
|
||||
"""Curl the domain to verify end-to-end HTTPS reachability."""
|
||||
"""Check HTTPS reachability for *domain* via local Caddy (loopback).
|
||||
|
||||
Using ``--resolve`` ensures the request reaches Caddy on this computer
|
||||
without depending on router NAT loopback or the public DNS result.
|
||||
A successful local check is sufficient to confirm that Caddy and the
|
||||
virtual-host configuration are working correctly.
|
||||
"""
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}", "--max-time", "10", f"https://{domain}"],
|
||||
[
|
||||
"curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}",
|
||||
"--max-time", "10",
|
||||
"--resolve", f"{domain}:443:127.0.0.1",
|
||||
"--resolve", f"{domain}:80:127.0.0.1",
|
||||
f"https://{domain}",
|
||||
],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=15,
|
||||
)
|
||||
status_code = result.stdout.strip()
|
||||
if status_code and status_code.isdigit() and int(status_code) > 0:
|
||||
return {"reachable": True, "status_code": int(status_code)}
|
||||
return {"reachable": True, "status_code": int(status_code), "via_loopback": True}
|
||||
return {"reachable": False, "error": result.stderr.strip() or "No response"}
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"reachable": False, "error": "timeout"}
|
||||
@@ -984,25 +1063,28 @@ def _check_domain_reachable(domain: str) -> dict:
|
||||
|
||||
|
||||
def _check_domain_health_fast(domain: str | None, external_ip: str) -> bool:
|
||||
"""Fast domain issue check for tile health (no curl/subprocess calls)."""
|
||||
"""Fast domain issue check for tile health (no curl/subprocess calls).
|
||||
|
||||
Returns ``True`` when a domain issue is detected that warrants
|
||||
``needs_attention``, ``False`` otherwise.
|
||||
Loopback resolution is treated as an intentional server-local override,
|
||||
not a DNS mismatch.
|
||||
"""
|
||||
if not domain:
|
||||
return True
|
||||
|
||||
resolved_ip: str | None = None
|
||||
try:
|
||||
results = socket.getaddrinfo(domain, None)
|
||||
if results:
|
||||
resolved_ip = results[0][4][0]
|
||||
except socket.gaierror:
|
||||
resolved_ip = None
|
||||
except Exception:
|
||||
resolved_ip = None
|
||||
|
||||
if not resolved_ip:
|
||||
addrs = _resolve_all_addresses(domain)
|
||||
if not addrs:
|
||||
return True
|
||||
|
||||
# If every resolved address is loopback the intentional /etc/hosts
|
||||
# override is in place — this is healthy, not a mismatch.
|
||||
if all(_is_loopback_address(a) for a in addrs):
|
||||
return False
|
||||
|
||||
if external_ip == "unavailable":
|
||||
return False
|
||||
return resolved_ip != external_ip
|
||||
return not any(a == external_ip for a in addrs)
|
||||
|
||||
|
||||
def _is_domain_reachable_cached(domain: str) -> bool | None:
|
||||
@@ -1070,15 +1152,8 @@ def _evaluate_domain_checklist(
|
||||
"detail": domain,
|
||||
})
|
||||
|
||||
resolved_ip: str | None = None
|
||||
try:
|
||||
results = socket.getaddrinfo(domain, None)
|
||||
if results:
|
||||
resolved_ip = results[0][4][0]
|
||||
except socket.gaierror:
|
||||
resolved_ip = None
|
||||
except Exception:
|
||||
resolved_ip = None
|
||||
addrs = _resolve_all_addresses(domain)
|
||||
resolved_ip: str | None = addrs[0] if addrs else None
|
||||
|
||||
if not resolved_ip:
|
||||
domain_status = {
|
||||
@@ -1105,7 +1180,29 @@ def _evaluate_domain_checklist(
|
||||
"has_issues": True,
|
||||
}
|
||||
|
||||
if external_ip == "unavailable":
|
||||
# Detect intentional server-local loopback override from /etc/hosts.
|
||||
# When all addresses are loopback the public DNS is not checked via the
|
||||
# system resolver (which would always return the override). We proceed
|
||||
# to the reachability check so Caddy health can still be verified.
|
||||
loopback_override = all(_is_loopback_address(a) for a in addrs)
|
||||
|
||||
if loopback_override:
|
||||
domain_status = {
|
||||
"status": "local_override",
|
||||
"resolved_ip": resolved_ip,
|
||||
"expected_ip": external_ip,
|
||||
}
|
||||
steps.append({
|
||||
"step": 2,
|
||||
"label": "DNS / Local Override",
|
||||
"status": "ok",
|
||||
"detail": (
|
||||
"Server-local loopback override is active — this computer routes the domain "
|
||||
"directly to Caddy. Public DNS verification is skipped on this computer; "
|
||||
"confirm your DNS provider points the domain to your external IP separately."
|
||||
),
|
||||
})
|
||||
elif external_ip == "unavailable":
|
||||
domain_status = {
|
||||
"status": "error",
|
||||
"resolved_ip": resolved_ip,
|
||||
@@ -1117,7 +1214,7 @@ def _evaluate_domain_checklist(
|
||||
"status": "warning",
|
||||
"detail": f"Resolves to {resolved_ip} (external IP unavailable for comparison)",
|
||||
})
|
||||
elif resolved_ip != external_ip:
|
||||
elif not any(a == external_ip for a in addrs):
|
||||
domain_status = {
|
||||
"status": "dns_mismatch",
|
||||
"resolved_ip": resolved_ip,
|
||||
@@ -2749,19 +2846,17 @@ async def api_services():
|
||||
break
|
||||
has_domain_issues = False
|
||||
if needs_domain and domain and enabled:
|
||||
addrs = _resolve_all_addresses(domain)
|
||||
dns_ok = True
|
||||
try:
|
||||
results = socket.getaddrinfo(domain, None)
|
||||
if results:
|
||||
resolved_ip = results[0][4][0]
|
||||
if (
|
||||
_cached_external_ip != "unavailable"
|
||||
and resolved_ip != _cached_external_ip
|
||||
):
|
||||
dns_ok = False
|
||||
else:
|
||||
dns_ok = False
|
||||
except (socket.gaierror, Exception):
|
||||
if not addrs:
|
||||
dns_ok = False
|
||||
elif all(_is_loopback_address(a) for a in addrs):
|
||||
# Intentional server-local /etc/hosts override — not a mismatch.
|
||||
dns_ok = True
|
||||
elif (
|
||||
_cached_external_ip != "unavailable"
|
||||
and not any(a == _cached_external_ip for a in addrs)
|
||||
):
|
||||
dns_ok = False
|
||||
|
||||
if not dns_ok:
|
||||
@@ -3886,6 +3981,12 @@ async def api_domains_set(req: DomainSetRequest):
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
# Regenerate the server-local /etc/hosts loopback entries so the newly
|
||||
# saved domain is immediately reachable on this computer without NAT
|
||||
# loopback support on the router.
|
||||
if req.domain_name in _SERVICE_DOMAIN_KEYS:
|
||||
_trigger_hosts_update()
|
||||
|
||||
return {"ok": True}
|
||||
|
||||
|
||||
@@ -3933,38 +4034,35 @@ async def api_domains_check(req: DomainCheckRequest):
|
||||
external_ip = _cached_external_ip
|
||||
|
||||
def check_domain(domain: str) -> dict:
|
||||
try:
|
||||
results = socket.getaddrinfo(domain, None)
|
||||
if not results:
|
||||
return {
|
||||
"domain": domain, "status": "unresolvable",
|
||||
"resolved_ip": None, "expected_ip": external_ip,
|
||||
}
|
||||
resolved_ip = results[0][4][0]
|
||||
if external_ip == "unavailable":
|
||||
return {
|
||||
"domain": domain, "status": "error",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
if resolved_ip == external_ip:
|
||||
return {
|
||||
"domain": domain, "status": "connected",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
return {
|
||||
"domain": domain, "status": "dns_mismatch",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
except socket.gaierror:
|
||||
addrs = _resolve_all_addresses(domain)
|
||||
if not addrs:
|
||||
return {
|
||||
"domain": domain, "status": "unresolvable",
|
||||
"resolved_ip": None, "expected_ip": external_ip,
|
||||
}
|
||||
except Exception:
|
||||
resolved_ip = addrs[0]
|
||||
# Server-local /etc/hosts loopback override — report as such rather
|
||||
# than as a DNS mismatch. Public DNS cannot be verified from this
|
||||
# computer when the override is active.
|
||||
if all(_is_loopback_address(a) for a in addrs):
|
||||
return {
|
||||
"domain": domain, "status": "local_override",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
if external_ip == "unavailable":
|
||||
return {
|
||||
"domain": domain, "status": "error",
|
||||
"resolved_ip": None, "expected_ip": external_ip,
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
if any(a == external_ip for a in addrs):
|
||||
return {
|
||||
"domain": domain, "status": "connected",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
return {
|
||||
"domain": domain, "status": "dns_mismatch",
|
||||
"resolved_ip": resolved_ip, "expected_ip": external_ip,
|
||||
}
|
||||
|
||||
check_results = await asyncio.gather(*[
|
||||
loop.run_in_executor(None, check_domain, d) for d in req.domains
|
||||
|
||||
@@ -0,0 +1,399 @@
|
||||
"""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()
|
||||
Reference in New Issue
Block a user