From 6f908513e38408f7074eafc3623967ee8183a803 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:14:53 +0000 Subject: [PATCH 1/3] Initial plan From 2cb0c734d83310d2ab839217f1756ab16898cef2 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:24:16 +0000 Subject: [PATCH 2/3] 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." --- app/sovran_systemsos_web/server.py | 226 ++++++++++---- app/tests/test_loopback_diagnostics.py | 399 +++++++++++++++++++++++++ modules/core/local-domain-loopback.nix | 143 +++++++++ modules/modules.nix | 1 + 4 files changed, 705 insertions(+), 64 deletions(-) create mode 100644 app/tests/test_loopback_diagnostics.py create mode 100644 modules/core/local-domain-loopback.nix diff --git a/app/sovran_systemsos_web/server.py b/app/sovran_systemsos_web/server.py index 8ba366d..d6660f9 100644 --- a/app/sovran_systemsos_web/server.py +++ b/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,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 diff --git a/app/tests/test_loopback_diagnostics.py b/app/tests/test_loopback_diagnostics.py new file mode 100644 index 0000000..a229d90 --- /dev/null +++ b/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() diff --git a/modules/core/local-domain-loopback.nix b/modules/core/local-domain-loopback.nix new file mode 100644 index 0000000..4ac94e6 --- /dev/null +++ b/modules/core/local-domain-loopback.nix @@ -0,0 +1,143 @@ +{ 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 -euf + + 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 ────────────────── + 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 + 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" ]; + }; +} diff --git a/modules/modules.nix b/modules/modules.nix index aecd688..f348a55 100755 --- a/modules/modules.nix +++ b/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 From de9b069a88c2205d1a7b1e8326866a19845ac820 Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Wed, 15 Jul 2026 15:26:43 +0000 Subject: [PATCH 3/3] fix: address code review feedback - Use set -eu (not set -euf) in sovran-hosts-update.sh - Add sync note for domain validation regex between shell and Python - Rename `seen` to `unique_addresses` in _resolve_all_addresses - Improve loopback override UI message with external DNS check guidance - Add docstring note about first-address display in _resolve_all_addresses" --- app/sovran_systemsos_web/server.py | 23 ++++++++++++++++------- modules/core/local-domain-loopback.nix | 6 ++++-- 2 files changed, 20 insertions(+), 9 deletions(-) diff --git a/app/sovran_systemsos_web/server.py b/app/sovran_systemsos_web/server.py index d6660f9..76674f3 100644 --- a/app/sovran_systemsos_web/server.py +++ b/app/sovran_systemsos_web/server.py @@ -977,6 +977,8 @@ def _check_port_status( # 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,}$' ) @@ -1005,15 +1007,20 @@ def _is_loopback_address(ip: str) -> bool: def _resolve_all_addresses(domain: str) -> list[str]: - """Return all IP addresses that *domain* resolves to, or an empty list.""" + """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) - seen: list[str] = [] + unique_addresses: list[str] = [] for r in results: addr = r[4][0] - if addr not in seen: - seen.append(addr) - return seen + if addr not in unique_addresses: + unique_addresses.append(addr) + return unique_addresses except Exception: return [] @@ -1198,8 +1205,10 @@ def _evaluate_domain_checklist( "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." + "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": diff --git a/modules/core/local-domain-loopback.nix b/modules/core/local-domain-loopback.nix index 4ac94e6..ffe8292 100644 --- a/modules/core/local-domain-loopback.nix +++ b/modules/core/local-domain-loopback.nix @@ -47,7 +47,7 @@ #!/bin/sh # Regenerate the Sovran-managed loopback block in /etc/hosts. # Safe to run multiple times — idempotent. - set -euf + set -eu DOMAINS_DIR="/var/lib/domains" HOSTS_FILE="/etc/hosts" @@ -75,6 +75,8 @@ " "$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" @@ -82,7 +84,7 @@ # 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 + # 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