Merge pull request #325 from naturallaw777/copilot/implement-server-side-loopback-solution

feat: server-side loopback overrides + Hub diagnostic fixes for NAT loopback

app/sovran_systemsos_web/server.py

@@ -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,94 @@ 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.
+# NOTE: The equivalent pattern in modules/core/local-domain-loopback.nix (shell
+# grep -E) must be kept in sync with this Python regex.
+_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 unique IP addresses that *domain* resolves to, or an empty list.
+
+    The first element is the address that the system resolver would normally
+    use for a connection.  All elements are checked when determining whether
+    any address matches the expected public IP or is a loopback address.
+    """
+    try:
+        results = socket.getaddrinfo(domain, None)
+        unique_addresses: list[str] = []
+        for r in results:
+            addr = r[4][0]
+            if addr not in unique_addresses:
+                unique_addresses.append(addr)
+        return unique_addresses
+    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 +1070,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
+    addrs = _resolve_all_addresses(domain)
-    try:
+    if not addrs:
-        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:
         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 +1159,8 @@ def _evaluate_domain_checklist(
         "detail": domain,
     })
 
-    resolved_ip: str | None = None
+    addrs = _resolve_all_addresses(domain)
-    try:
+    resolved_ip: str | None = addrs[0] if addrs else None
-        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:
         domain_status = {
@@ -1105,7 +1187,31 @@ 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 without going through the router. "
+                "Public DNS cannot be verified from this computer while the override is in place. "
+                "To check your public DNS from outside, use a tool such as "
+                "https://dnschecker.org or run: dig @1.1.1.1 " + domain
+            ),
+        })
+    elif external_ip == "unavailable":
         domain_status = {
             "status": "error",
             "resolved_ip": resolved_ip,
@@ -1117,7 +1223,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 +2855,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:
+            if not addrs:
-                results = socket.getaddrinfo(domain, None)
+                dns_ok = False
-                if results:
+            elif all(_is_loopback_address(a) for a in addrs):
-                    resolved_ip = results[0][4][0]
+                # Intentional server-local /etc/hosts override — not a mismatch.
-                    if (
+                dns_ok = True
-                        _cached_external_ip != "unavailable"
+            elif (
-                        and resolved_ip != _cached_external_ip
+                _cached_external_ip != "unavailable"
-                    ):
+                and not any(a == _cached_external_ip for a in addrs)
-                        dns_ok = False
+            ):
-                else:
-                    dns_ok = False
-            except (socket.gaierror, Exception):
                 dns_ok = False
 
             if not dns_ok:
@@ -3886,6 +3990,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 +4043,35 @@ async def api_domains_check(req: DomainCheckRequest):
     external_ip = _cached_external_ip
 
     def check_domain(domain: str) -> dict:
-        try:
+        addrs = _resolve_all_addresses(domain)
-            results = socket.getaddrinfo(domain, None)
+        if not addrs:
-            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:
             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

app/tests/test_loopback_diagnostics.py

@@ -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()

modules/core/local-domain-loopback.nix

@@ -0,0 +1,145 @@
+{ config, pkgs, lib, ... }:
+
+# ── Server-local domain loopback overrides ────────────────────────────────────
+#
+# Some routers (especially newer ISP-provided devices) do not support NAT
+# loopback (hairpin NAT).  When a request originates on this computer and
+# targets a public domain name that resolves to the router's WAN address, the
+# router may refuse to loop the connection back in — causing Nextcloud, WordPress
+# background jobs, and other server-side callbacks to fail even when the service
+# is fully operational from the internet.
+#
+# This module installs a one-shot systemd service,
+# ``sovran-hosts-update.service``, that reads the configured service domains
+# from ``/var/lib/domains/`` at boot (and whenever triggered by the Hub after a
+# domain is saved) and writes ``127.0.0.1`` entries for them into a dedicated
+# Sovran-managed block in ``/etc/hosts``.
+#
+# With those entries in place:
+#   • Requests originating on this computer resolve the public domain name to
+#     127.0.0.1, reach Caddy directly, and never touch the router.
+#   • Caddy still receives the correct public hostname via TLS SNI so virtual-
+#     host routing and certificate validation continue to work.
+#   • The Sovran Hub can verify Caddy reachability locally without needing NAT
+#     loopback.
+#
+# Limitation: this does not help other devices on your home network (phones,
+# laptops).  Those devices resolve domains via the router's DNS and still depend
+# on NAT loopback (or require manual router DNS overrides).  For now, only
+# server-originated requests benefit from this override.
+#
+# On NixOS, /etc/hosts is normally a symlink into the Nix store and is
+# regenerated by the system activation script.  The ``system.activationScripts``
+# hook below converts it to a writable file each time the system is activated
+# (i.e. after every ``nixos-rebuild switch``) and then injects the Sovran block.
+# The same script is also run by the ``sovran-hosts-update.service`` unit so
+# that the Hub can trigger it immediately after saving a domain without
+# requiring a full rebuild.
+
+{
+  # ── Helper script (stored in the Nix store, never reads /var/lib at eval) ──
+
+  environment.systemPackages = [ pkgs.coreutils ];
+
+  environment.etc."sovran-hosts-update.sh" = {
+    mode = "0755";
+    text = ''
+      #!/bin/sh
+      # Regenerate the Sovran-managed loopback block in /etc/hosts.
+      # Safe to run multiple times — idempotent.
+      set -eu
+
+      DOMAINS_DIR="/var/lib/domains"
+      HOSTS_FILE="/etc/hosts"
+      BEGIN_MARKER="# Sovran managed begin — server-local loopback overrides"
+      END_MARKER="# Sovran managed end"
+
+      # ── Step 1: ensure /etc/hosts is a regular writable file ──────────────
+      # On NixOS /etc/hosts starts as a symlink to the Nix store.  We replace
+      # it with a copy so we can append our block without touching the store.
+      if [ -L "$HOSTS_FILE" ]; then
+        TARGET=$(readlink -f "$HOSTS_FILE")
+        cp --no-preserve=all "$TARGET" "$HOSTS_FILE.sovran-tmp"
+        mv "$HOSTS_FILE.sovran-tmp" "$HOSTS_FILE"
+        chmod 644 "$HOSTS_FILE"
+      fi
+
+      # ── Step 2: remove any existing Sovran block ──────────────────────────
+      # Use a temp file so the operation is atomic.
+      TMP=$(mktemp "$HOSTS_FILE.XXXXXX")
+      trap 'rm -f "$TMP"' EXIT
+      awk "
+        /^$BEGIN_MARKER\$/ { skip=1; next }
+        /^$END_MARKER\$/   { skip=0; next }
+        !skip
+      " "$HOSTS_FILE" > "$TMP"
+
+      # ── Step 3: collect valid configured service domains ──────────────────
+      # NOTE: The hostname validation regex below must stay in sync with
+      # _SAFE_DOMAIN_RE in app/sovran_systemsos_web/server.py.
+      ENTRIES=""
+      for KEY in matrix wordpress nextcloud btcpayserver vaultwarden haven element-calling; do
+        FILE="$DOMAINS_DIR/$KEY"
+        [ -f "$FILE" ] || continue
+        # Read the domain value (strip all whitespace, limit to 253 chars)
+        DOMAIN=$(tr -d '[:space:]' < "$FILE" | head -c 253)
+        [ -z "$DOMAIN" ] && continue
+        # Validate: must match a reasonable hostname pattern (no injection)
+        if ! printf '%s' "$DOMAIN" | grep -qE \
+          '^[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?(\.[a-zA-Z0-9]([a-zA-Z0-9-]{0,61}[a-zA-Z0-9])?)+$'; then
+          echo "sovran-hosts-update: skipping invalid domain value for $KEY: $DOMAIN" >&2
+          continue
+        fi
+        ENTRIES="$ENTRIES
+127.0.0.1 $DOMAIN
+::1       $DOMAIN"
+      done
+
+      # ── Step 4: append the Sovran block if there are any entries ──────────
+      if [ -n "$ENTRIES" ]; then
+        printf '\n%s\n' "$BEGIN_MARKER" >> "$TMP"
+        printf '%s\n' "# These entries route configured service domains to local Caddy." >> "$TMP"
+        printf '%s\n' "# They are managed automatically — do not edit this block." >> "$TMP"
+        printf '%s\n' "$ENTRIES" >> "$TMP"
+        printf '%s\n' "$END_MARKER" >> "$TMP"
+      fi
+
+      # ── Step 5: atomically replace /etc/hosts ─────────────────────────────
+      mv "$TMP" "$HOSTS_FILE"
+      chmod 644 "$HOSTS_FILE"
+    '';
+  };
+
+  # ── Systemd service ────────────────────────────────────────────────────────
+
+  systemd.services.sovran-hosts-update = {
+    description = "Update /etc/hosts with Sovran server-local loopback overrides";
+    documentation = [ "https://github.com/naturallaw777/sovran-systems" ];
+
+    # Run before Caddy so loopback entries are ready when it starts.
+    before = [
+      "caddy.service"
+      "network-online.target"
+    ];
+    wantedBy = [ "multi-user.target" ];
+
+    serviceConfig = {
+      Type = "oneshot";
+      RemainAfterExit = true;
+      ExecStart = "/etc/sovran-hosts-update.sh";
+    };
+  };
+
+  # ── Activation script (runs after every nixos-rebuild switch) ─────────────
+  # This ensures the loopback block survives rebuilds that restore the /etc/hosts
+  # symlink.  The "users" and "etc" scripts must complete first.
+
+  system.activationScripts.sovranDomainLoopback = {
+    text = ''
+      if [ -x /etc/sovran-hosts-update.sh ] && [ -d /var/lib/domains ]; then
+        /etc/sovran-hosts-update.sh || true
+      fi
+    '';
+    deps = [ "etc" "users" ];
+  };
+}

modules/modules.nix

@@ -16,6 +16,7 @@
     ./core/remote-deploy.nix
     ./core/no-sleep.nix
     ./core/cpu-performance.nix
+    ./core/local-domain-loopback.nix
 
     # ── Always on (no flag) ───────────────────────────────────
     ./php.nix