"""Tests for the DDNS runner (sovran_systemsos_web.ddns_update). The runner asks Njal.la to use the address the request came from ("&auto"), reads back the address Njal.la recorded and saves it for LiveKit and the Hub. Tests must never: - access the network (curl is replaced by a fake ``run``) - write to system paths (the URL and IP files live in a temp dir) """ import contextlib import io import json import os import subprocess import sys import tempfile import unittest from unittest import mock _REPO_ROOT = os.path.normpath(os.path.join(os.path.dirname(__file__), "..")) _APP_PARENT = os.path.join(_REPO_ROOT, "app") if _APP_PARENT not in sys.path: sys.path.insert(0, _APP_PARENT) from sovran_systemsos_web import ddns_update as d # noqa: E402 KEY = "SECRETKEY123" AUTO_URL = f"https://njal.la/update/?h=sub.example.com&k={KEY}&auto" LEGACY_URL = f"https://njal.la/update/?h=sub.example.com&k={KEY}&a=${{IP}}" PUBLIC_IP = "93.184.216.34" OTHER_IP = "8.8.4.4" def reply(ip=PUBLIC_IP, status=200): return json.dumps({"status": status, "message": "record updated", "value": {"A": ip}}) class FakeRun: """Stands in for subprocess.run; records every command it is given.""" def __init__(self, stdout=None, returncode=0, raises=None): self.stdout = reply() if stdout is None else stdout self.returncode = returncode self.raises = raises self.calls = [] def __call__(self, cmd, **kwargs): self.calls.append(cmd) if self.raises: raise self.raises return subprocess.CompletedProcess(cmd, self.returncode, stdout=self.stdout, stderr="") class NormaliseUrlTests(unittest.TestCase): def test_legacy_placeholder_becomes_auto(self): self.assertEqual(d.normalise_url(LEGACY_URL), AUTO_URL) def test_quiet_is_dropped_so_the_reply_can_be_read(self): self.assertEqual(d.normalise_url(AUTO_URL + "&quiet"), AUTO_URL) def test_plain_auto_is_unchanged(self): self.assertEqual(d.normalise_url(AUTO_URL), AUTO_URL) def test_explicit_address_is_unchanged(self): url = "https://njal.la/update/?h=a.example.com&k=K&a=93.184.216.34" self.assertEqual(d.normalise_url(url), url) class IsPublicIpv4Tests(unittest.TestCase): def test_public_addresses(self): for ip in ("93.184.216.34", "8.8.8.8", " 1.1.1.1\n"): self.assertTrue(d.is_public_ipv4(ip), ip) def test_everything_else_is_rejected(self): for ip in ("10.0.0.1", "192.168.1.5", "172.16.0.9", "127.0.0.1", "169.254.1.1", "100.64.0.1", "0.0.0.0", "224.0.0.1", "::1", "2001:4860:4860::8888", "not-an-ip", "", None): self.assertFalse(d.is_public_ipv4(ip), ip) class ParseReplyTests(unittest.TestCase): def test_success_returns_the_recorded_address(self): self.assertEqual(d.parse_reply(reply()), PUBLIC_IP) def test_status_may_be_a_string(self): self.assertEqual(d.parse_reply(reply(status="200")), PUBLIC_IP) def test_error_status_is_rejected(self): body = json.dumps({"status": 401, "message": "invalid host or key"}) self.assertIsNone(d.parse_reply(body)) def test_garbage_is_rejected(self): for body in ("", "not json", "[]", "null", "{}", json.dumps({"status": 200})): self.assertIsNone(d.parse_reply(body), body) def test_non_public_or_non_ipv4_address_is_rejected(self): for ip in ("10.1.2.3", "100.64.9.9", "127.0.0.1", "::1", "2001:4860:4860::8888", "x"): self.assertIsNone(d.parse_reply(reply(ip)), ip) class IpFileTests(unittest.TestCase): def test_write_then_read(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "secrets", "external-ip") d.write_ip_file(PUBLIC_IP, path) self.assertEqual(d.read_ip_file(path), PUBLIC_IP) self.assertEqual(open(path).read(), PUBLIC_IP) # no trailing newline self.assertEqual(os.stat(path).st_mode & 0o777, 0o644) def test_replace_is_atomic_and_leaves_no_temp_files(self): with tempfile.TemporaryDirectory() as tmp: path = os.path.join(tmp, "external-ip") d.write_ip_file(PUBLIC_IP, path) d.write_ip_file(OTHER_IP, path) self.assertEqual(d.read_ip_file(path), OTHER_IP) self.assertEqual(os.listdir(tmp), ["external-ip"]) def test_missing_file_reads_as_none(self): with tempfile.TemporaryDirectory() as tmp: self.assertIsNone(d.read_ip_file(os.path.join(tmp, "nope"))) class UpdateAllTests(unittest.TestCase): def run_update(self, urls, fake): out = io.StringIO() with contextlib.redirect_stdout(out): result = d.update_all(urls, run=fake) return result, out.getvalue() def test_curl_is_called_directly_with_ipv4_and_no_redirects(self): fake = FakeRun() result, _ = self.run_update([AUTO_URL], fake) self.assertEqual(result, PUBLIC_IP) self.assertEqual(len(fake.calls), 1) cmd = fake.calls[0] self.assertEqual(cmd[0], "curl") for flag in ("--ipv4", "--no-location", "--fail", "--silent"): self.assertIn(flag, cmd) self.assertEqual(cmd[-1], AUTO_URL) def test_legacy_entry_and_its_auto_twin_are_one_call(self): fake = FakeRun() self.run_update([LEGACY_URL, AUTO_URL], fake) self.assertEqual(fake.calls[0][-1], AUTO_URL) self.assertEqual(len(fake.calls), 1) def test_url_for_another_host_is_never_called(self): fake = FakeRun() result, _ = self.run_update([f"https://evil.example/update/?h=x&k={KEY}&auto"], fake) self.assertIsNone(result) self.assertEqual(fake.calls, []) def test_failed_curl_yields_nothing(self): result, _ = self.run_update([AUTO_URL], FakeRun(returncode=22)) self.assertIsNone(result) def test_reply_without_an_address_yields_nothing(self): result, _ = self.run_update([AUTO_URL], FakeRun(stdout=json.dumps({"status": 200}))) self.assertIsNone(result) def test_missing_curl_is_survived(self): result, out = self.run_update([AUTO_URL], FakeRun(raises=FileNotFoundError("curl"))) self.assertIsNone(result) self.assertIn("skipped", out) def test_first_reported_address_wins(self): calls = iter([reply(PUBLIC_IP), reply(OTHER_IP)]) def fake(cmd, **kwargs): return subprocess.CompletedProcess(cmd, 0, stdout=next(calls), stderr="") other = f"https://njal.la/update/?h=other.example.com&k={KEY}&auto" out = io.StringIO() with contextlib.redirect_stdout(out): result = d.update_all([AUTO_URL, other], run=fake) self.assertEqual(result, PUBLIC_IP) def test_the_key_is_never_printed(self): for fake in (FakeRun(), FakeRun(returncode=22), FakeRun(stdout="junk"), FakeRun(raises=FileNotFoundError("curl"))): _, out = self.run_update([AUTO_URL, LEGACY_URL], fake) self.assertNotIn(KEY, out) class MainTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) self.urls_file = os.path.join(self.tmp.name, "ddns_urls.json") self.ip_file = os.path.join(self.tmp.name, "secrets", "external-ip") for patch in (mock.patch.object(d, "URLS_FILE", self.urls_file), mock.patch.object(d, "IP_FILE", self.ip_file)): patch.start() self.addCleanup(patch.stop) def store(self, urls): with open(self.urls_file, "w") as f: json.dump(urls, f) def main(self, fake): out = io.StringIO() with mock.patch.object(d.subprocess, "run", fake), contextlib.redirect_stdout(out): code = d.main() self.assertEqual(code, 0) return out.getvalue() def test_first_update_records_the_address(self): self.store([LEGACY_URL]) out = self.main(FakeRun()) self.assertEqual(d.read_ip_file(self.ip_file), PUBLIC_IP) self.assertIn("now " + PUBLIC_IP, out) self.assertNotIn(KEY, out) def test_unchanged_address_does_not_touch_the_file(self): # A path unit restarts LiveKit whenever the file is written, so an # unchanged address must not rewrite it. self.store([AUTO_URL]) self.main(FakeRun()) before = os.stat(self.ip_file) out = self.main(FakeRun()) after = os.stat(self.ip_file) self.assertEqual((before.st_ino, before.st_mtime_ns), (after.st_ino, after.st_mtime_ns)) self.assertIn("unchanged", out) def test_changed_address_is_recorded(self): self.store([AUTO_URL]) self.main(FakeRun(stdout=reply(PUBLIC_IP))) out = self.main(FakeRun(stdout=reply(OTHER_IP))) self.assertEqual(d.read_ip_file(self.ip_file), OTHER_IP) self.assertIn(f"now {OTHER_IP} (was {PUBLIC_IP})", out) def test_failed_update_keeps_the_last_known_address(self): self.store([AUTO_URL]) self.main(FakeRun(stdout=reply(PUBLIC_IP))) self.main(FakeRun(returncode=7)) self.assertEqual(d.read_ip_file(self.ip_file), PUBLIC_IP) def test_nothing_configured_does_nothing(self): fake = FakeRun() self.main(fake) # no URL file at all self.store([]) self.main(fake) # empty list self.assertEqual(fake.calls, []) self.assertFalse(os.path.exists(self.ip_file)) if __name__ == "__main__": unittest.main()