Files
Sovran_SystemsOS/app/tests/test_domain_conflict.py
T

510 lines
23 KiB
Python

"""
Domain conflict and unique-hostname tests.
Tests cover:
1. Wallet Connections initial setup displays unique-hostname guidance.
2. Wallet Connections reconfiguration displays the same guidance.
3. The field example is lightning.yourdomain.com for Wallet Connections.
4. An unused hostname such as lightning.example.com is accepted.
5. Reusing a hostname assigned to Matrix, Nextcloud, WordPress, BTCPay Server,
Vaultwarden, Haven, or Element Calling returns HTTP 409.
6. Comparison is case-insensitive.
7. A hostname with one trailing dot conflicts with the equivalent hostname without it.
8. Re-saving the existing lightning hostname for lightning remains allowed.
9. Invalid hostnames are rejected before mutation.
10. On conflict, the domain file remains unchanged.
11. On conflict, the DDNS script remains unchanged and is not executed.
12. Generic domain flows for unrelated services remain intact (matrix -> matrix, etc.).
13. JavaScript syntax checks pass for features.js and helpers.js.
"""
import json
import os
import subprocess
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
# ── Minimal stubs so server.py can be imported without full FastAPI ──
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, *a, **kw): pass
def mount(self, *a, **kw): return None
def add_middleware(self, *a, **kw): return None
def __getattr__(self, _name):
def _deco_factory(*a, **kw):
def _deco(func): return func
return _deco
return _deco_factory
class _BaseModel: pass
class _JSONResponse:
def __init__(self, content=None, status_code=200):
self.content = content
self.status_code = status_code
self.body = json.dumps(content or {}).encode("utf-8")
fastapi_mod = types.ModuleType("fastapi")
fastapi_mod.FastAPI = _FastAPI
fastapi_mod.HTTPException = _HTTPException
sys.modules["fastapi"] = fastapi_mod
resp_mod = types.ModuleType("fastapi.responses")
resp_mod.HTMLResponse = object
resp_mod.RedirectResponse = object
resp_mod.JSONResponse = _JSONResponse
sys.modules["fastapi.responses"] = resp_mod
sys.modules["fastapi.staticfiles"] = types.ModuleType("fastapi.staticfiles")
class _StaticFiles:
def __init__(self, *args, **kwargs): pass
sys.modules["fastapi.staticfiles"].StaticFiles = _StaticFiles
class _Jinja2Templates:
def __init__(self, *args, **kwargs): pass
tmpl_mod = types.ModuleType("fastapi.templating")
tmpl_mod.Jinja2Templates = _Jinja2Templates
sys.modules["fastapi.templating"] = tmpl_mod
req_mod = types.ModuleType("fastapi.requests")
req_mod.Request = object
sys.modules["fastapi.requests"] = req_mod
pyd_mod = types.ModuleType("pydantic")
pyd_mod.BaseModel = _BaseModel
sys.modules["pydantic"] = pyd_mod
stl_base = types.ModuleType("starlette.middleware.base")
stl_base.BaseHTTPMiddleware = object
sys.modules["starlette.middleware.base"] = stl_base
stl_mw = types.ModuleType("starlette.middleware")
sys.modules["starlette.middleware"] = stl_mw
stl = types.ModuleType("starlette")
sys.modules["starlette"] = stl
_install_web_stubs()
from sovran_systemsos_web import server # noqa: E402
# ── Helpers ───────────────────────────────────────────────────────
def _make_req(domain_name, domain, ddns_url=""):
"""Build a DomainSetRequest-like object using the server's model."""
req = object.__new__(server.DomainSetRequest)
req.domain_name = domain_name
req.domain = domain
req.ddns_url = ddns_url
return req
def _write_domain_file(domains_dir, key, value):
path = os.path.join(domains_dir, key)
with open(path, "w") as fh:
fh.write(value)
# ── Unit tests for server helper functions ─────────────────────────
class NormalizeHostnameTests(unittest.TestCase):
def test_trims_whitespace(self):
self.assertEqual(server._normalize_hostname(" foo.example.com "), "foo.example.com")
def test_lowercases(self):
self.assertEqual(server._normalize_hostname("FOO.Example.COM"), "foo.example.com")
def test_removes_exactly_one_trailing_dot(self):
self.assertEqual(server._normalize_hostname("foo.example.com."), "foo.example.com")
def test_does_not_remove_two_trailing_dots(self):
# Only one trailing dot is removed; two trailing dots leave one.
self.assertEqual(server._normalize_hostname("foo.example.com.."), "foo.example.com.")
def test_no_trailing_dot_unchanged(self):
self.assertEqual(server._normalize_hostname("foo.example.com"), "foo.example.com")
def test_strips_and_lowercases_with_trailing_dot(self):
self.assertEqual(server._normalize_hostname(" Lightning.Example.COM. "), "lightning.example.com")
class ValidateHostnameTests(unittest.TestCase):
def test_valid_simple_domain(self):
self.assertTrue(server._validate_hostname("foo.example.com"))
def test_valid_subdomain(self):
self.assertTrue(server._validate_hostname("lightning.yourdomain.com"))
def test_valid_bare_hostname(self):
self.assertTrue(server._validate_hostname("example"))
def test_valid_with_hyphens(self):
self.assertTrue(server._validate_hostname("my-host.example.com"))
def test_invalid_empty(self):
self.assertFalse(server._validate_hostname(""))
def test_invalid_trailing_dot(self):
# The hostname validator rejects trailing dots directly.
# In the API flow, normalization removes exactly one trailing dot before
# this validator is called, so a single trailing dot in user input is
# handled before reaching validation.
self.assertFalse(server._validate_hostname("foo.example.com."))
def test_invalid_with_underscore(self):
self.assertFalse(server._validate_hostname("foo_bar.example.com"))
def test_invalid_leading_hyphen(self):
self.assertFalse(server._validate_hostname("-foo.example.com"))
def test_invalid_spaces(self):
self.assertFalse(server._validate_hostname("foo example.com"))
class CheckDomainConflictTests(unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
self._orig_domains_dir = server.DOMAINS_DIR
server.DOMAINS_DIR = self.tmpdir
def tearDown(self):
server.DOMAINS_DIR = self._orig_domains_dir
import shutil
shutil.rmtree(self.tmpdir, ignore_errors=True)
def test_no_conflict_when_no_other_files(self):
result = server._check_domain_conflict("lightning", "lightning.example.com")
self.assertIsNone(result)
def test_conflict_when_matrix_has_same_hostname(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "matrix")
def test_conflict_when_nextcloud_has_same_hostname(self):
_write_domain_file(self.tmpdir, "nextcloud", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "nextcloud")
def test_conflict_when_wordpress_has_same_hostname(self):
_write_domain_file(self.tmpdir, "wordpress", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "wordpress")
def test_conflict_when_btcpayserver_has_same_hostname(self):
_write_domain_file(self.tmpdir, "btcpayserver", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "btcpayserver")
def test_conflict_when_vaultwarden_has_same_hostname(self):
_write_domain_file(self.tmpdir, "vaultwarden", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "vaultwarden")
def test_conflict_when_haven_has_same_hostname(self):
_write_domain_file(self.tmpdir, "haven", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "haven")
def test_conflict_when_element_calling_has_same_hostname(self):
_write_domain_file(self.tmpdir, "element-calling", "shared.example.com")
result = server._check_domain_conflict("lightning", "shared.example.com")
self.assertEqual(result, "element-calling")
def test_no_conflict_for_self(self):
# Re-saving lightning's own existing hostname must be allowed.
_write_domain_file(self.tmpdir, "lightning", "lightning.example.com")
result = server._check_domain_conflict("lightning", "lightning.example.com")
self.assertIsNone(result)
def test_symmetric_conflict_matrix_reusing_lightning(self):
# Saving matrix with lightning's existing hostname is also rejected.
_write_domain_file(self.tmpdir, "lightning", "shared.example.com")
result = server._check_domain_conflict("matrix", "shared.example.com")
self.assertEqual(result, "lightning")
def test_no_conflict_unrelated_services_without_lightning(self):
# Two unrelated non-lightning services with the same hostname.
# The rule only applies when lightning is involved.
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
result = server._check_domain_conflict("nextcloud", "shared.example.com")
self.assertIsNone(result)
# ── API endpoint integration-style tests ──────────────────────────
import asyncio
def _run(coro):
"""Run an async coroutine synchronously."""
return asyncio.get_event_loop().run_until_complete(coro)
class ApiDomainsSetConflictTests(unittest.TestCase):
def setUp(self):
self.tmpdir = tempfile.mkdtemp()
self.njalla_dir = tempfile.mkdtemp()
self.njalla_script = os.path.join(self.njalla_dir, "njalla.sh")
self._orig_domains_dir = server.DOMAINS_DIR
self._orig_njalla = server.NJALLA_SCRIPT
server.DOMAINS_DIR = self.tmpdir
server.NJALLA_SCRIPT = self.njalla_script
def tearDown(self):
server.DOMAINS_DIR = self._orig_domains_dir
server.NJALLA_SCRIPT = self._orig_njalla
import shutil
shutil.rmtree(self.tmpdir, ignore_errors=True)
shutil.rmtree(self.njalla_dir, ignore_errors=True)
# ── Test 4: unused hostname is accepted ──────────────────────────
def test_unused_hostname_is_accepted(self):
with patch.object(server, "_trigger_hosts_update"):
result = _run(server.api_domains_set(_make_req("lightning", "lightning.example.com")))
self.assertEqual(result.get("ok"), True)
# Domain file should be written.
with open(os.path.join(self.tmpdir, "lightning")) as fh:
saved = fh.read()
self.assertEqual(saved, "lightning.example.com")
# ── Test 5: conflict with each managed service returns 409 ────────
def _assert_conflict_409(self, target_key, conflicting_key, hostname):
_write_domain_file(self.tmpdir, conflicting_key, hostname)
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req(target_key, hostname)))
exc = ctx.exception
self.assertEqual(getattr(exc, "status_code", None), 409)
detail = getattr(exc, "detail", {})
self.assertEqual(detail.get("error"), "domain_conflict")
self.assertEqual(detail.get("conflicting_domain_key"), conflicting_key)
self.assertIn("message", detail)
def test_conflict_with_matrix_returns_409(self):
self._assert_conflict_409("lightning", "matrix", "shared.example.com")
def test_conflict_with_nextcloud_returns_409(self):
self._assert_conflict_409("lightning", "nextcloud", "shared.example.com")
def test_conflict_with_wordpress_returns_409(self):
self._assert_conflict_409("lightning", "wordpress", "shared.example.com")
def test_conflict_with_btcpayserver_returns_409(self):
self._assert_conflict_409("lightning", "btcpayserver", "shared.example.com")
def test_conflict_with_vaultwarden_returns_409(self):
self._assert_conflict_409("lightning", "vaultwarden", "shared.example.com")
def test_conflict_with_haven_returns_409(self):
self._assert_conflict_409("lightning", "haven", "shared.example.com")
def test_conflict_with_element_calling_returns_409(self):
self._assert_conflict_409("lightning", "element-calling", "shared.example.com")
# ── Test 6: comparison is case-insensitive ───────────────────────
def test_conflict_is_case_insensitive(self):
_write_domain_file(self.tmpdir, "matrix", "Shared.Example.COM")
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
def test_conflict_case_insensitive_reversed(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "SHARED.EXAMPLE.COM")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
# ── Test 7: trailing-dot normalization causes conflict ────────────
def test_trailing_dot_conflicts_with_same_hostname(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "shared.example.com.")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
def test_stored_trailing_dot_conflicts_with_clean_submission(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com.")
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 409)
# ── Test 8: re-saving existing lightning hostname is allowed ──────
def test_resave_existing_lightning_hostname_allowed(self):
_write_domain_file(self.tmpdir, "lightning", "lightning.example.com")
with patch.object(server, "_trigger_hosts_update"):
result = _run(server.api_domains_set(_make_req("lightning", "lightning.example.com")))
self.assertEqual(result.get("ok"), True)
# ── Test 9: invalid hostname is rejected before mutation ──────────
def test_invalid_hostname_rejected(self):
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "not a hostname!")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 400)
def test_hostname_with_underscore_rejected(self):
with self.assertRaises(Exception) as ctx:
_run(server.api_domains_set(_make_req("lightning", "foo_bar.example.com")))
self.assertEqual(getattr(ctx.exception, "status_code", None), 400)
# ── Test 10: on conflict, domain file remains unchanged ───────────
def test_domain_file_unchanged_on_conflict(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
lightning_path = os.path.join(self.tmpdir, "lightning")
# Write a prior value.
_write_domain_file(self.tmpdir, "lightning", "old.example.com")
with self.assertRaises(Exception):
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
# The lightning domain file must still contain the old value.
with open(lightning_path) as fh:
saved = fh.read()
self.assertEqual(saved, "old.example.com")
def test_domain_file_not_created_on_conflict_if_absent(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
lightning_path = os.path.join(self.tmpdir, "lightning")
self.assertFalse(os.path.exists(lightning_path))
with self.assertRaises(Exception):
_run(server.api_domains_set(_make_req("lightning", "shared.example.com")))
self.assertFalse(os.path.exists(lightning_path))
# ── Test 11: on conflict, DDNS script unchanged and not executed ──
def test_njalla_script_not_written_on_conflict(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
with self.assertRaises(Exception):
_run(server.api_domains_set(
_make_req("lightning", "shared.example.com",
ddns_url='curl "https://njal.la/update/?h=shared.example.com&k=key&auto"')
))
self.assertFalse(os.path.exists(self.njalla_script))
def test_njalla_script_not_executed_on_conflict(self):
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
with patch("subprocess.run") as mock_run, self.assertRaises(Exception):
_run(server.api_domains_set(
_make_req("lightning", "shared.example.com",
ddns_url='curl "https://njal.la/update/?h=shared.example.com&k=key&auto"')
))
mock_run.assert_not_called()
# ── Test 12: generic domain flows for unrelated services intact ───
def test_matrix_save_without_lightning_succeeds(self):
with patch.object(server, "_trigger_hosts_update"):
result = _run(server.api_domains_set(_make_req("matrix", "matrix.example.com")))
self.assertEqual(result.get("ok"), True)
with open(os.path.join(self.tmpdir, "matrix")) as fh:
saved = fh.read()
self.assertEqual(saved, "matrix.example.com")
def test_nextcloud_save_without_lightning_succeeds(self):
with patch.object(server, "_trigger_hosts_update"):
result = _run(server.api_domains_set(_make_req("nextcloud", "cloud.example.com")))
self.assertEqual(result.get("ok"), True)
def test_two_non_lightning_services_sharing_hostname_allowed(self):
# The uniqueness rule is only enforced when lightning is involved.
_write_domain_file(self.tmpdir, "matrix", "shared.example.com")
with patch.object(server, "_trigger_hosts_update"):
result = _run(server.api_domains_set(_make_req("nextcloud", "shared.example.com")))
self.assertEqual(result.get("ok"), True)
# ── Test 13: JavaScript syntax checks ─────────────────────────────
class JsSyntaxTests(unittest.TestCase):
def _js_files(self):
static_js = Path(__file__).resolve().parents[1] / "sovran_systemsos_web" / "static" / "js"
return sorted(static_js.glob("*.js"))
def test_js_syntax_no_errors(self):
for js_file in self._js_files():
with self.subTest(file=js_file.name):
result = subprocess.run(
["node", "--check", str(js_file)],
capture_output=True, text=True
)
self.assertEqual(
result.returncode, 0,
msg=f"Syntax error in {js_file.name}:\n{result.stderr}"
)
# ── UI guidance content tests (features.js) ───────────────────────
class FeaturesJsContentTests(unittest.TestCase):
"""Verify the features.js source contains the required Wallet Connections guidance."""
def setUp(self):
self.features_js = (
Path(__file__).resolve().parents[1]
/ "sovran_systemsos_web" / "static" / "js" / "features.js"
).read_text(encoding="utf-8")
# ── Test 1: initial setup displays unique-hostname guidance ───────
def test_setup_modal_contains_nwc_warning(self):
self.assertIn("Wallet Connections requires its own unique hostname", self.features_js)
# ── Test 2: reconfiguration displays the same guidance ────────────
def test_reconfig_modal_contains_nwc_warning(self):
# The warning text must appear in both openDomainSetupModal and
# openDomainReconfigureModal — two occurrences minimum.
count = self.features_js.count("Wallet Connections requires its own unique hostname")
self.assertGreaterEqual(count, 2, "Warning must appear in both setup and reconfigure modals")
# ── Test 3: field example is lightning.yourdomain.com ────────────
def test_lightning_placeholder_is_present(self):
self.assertIn("lightning.yourdomain.com", self.features_js)
# ── Warning references correct services ───────────────────────────
def test_warning_mentions_matrix(self):
self.assertIn("Matrix", self.features_js)
def test_warning_mentions_nextcloud(self):
self.assertIn("Nextcloud", self.features_js)
def test_warning_mentions_btcpay_server(self):
self.assertIn("BTCPay Server", self.features_js)
def test_warning_mentions_vaultwarden(self):
self.assertIn("Vaultwarden", self.features_js)
def test_warning_mentions_haven(self):
self.assertIn("Haven", self.features_js)
def test_warning_mentions_wordpress(self):
self.assertIn("WordPress", self.features_js)
# ── isWalletConnections guard targets correct identifiers ─────────
def test_nwc_wallets_id_check_present(self):
self.assertIn('feat.id === "nwc-wallets"', self.features_js)
def test_lightning_domain_name_check_present(self):
self.assertIn('feat.domain_name === "lightning"', self.features_js)
# ── Error surfacing ───────────────────────────────────────────────
def test_error_message_surfaced_in_setup_modal(self):
# The catch block must use err.message rather than hard-coded string.
self.assertIn("err.message", self.features_js)
if __name__ == "__main__":
unittest.main()