"""Tests for headscale-ddns.sh's Cloudflare calls (2026-09-23). Three of the four failures in the week to 2026-09-23 (09-19, 09-21, 09-23) had one signature: the Cloudflare zone lookup came back EMPTY after its 15s cap, json.load raised, and the script carried on with empty IDs all the way to a PATCH against `zones//dns_records/`. These pin the fix: Cloudflare calls retry like the WAN lookup does, and no write is attempted without both IDs. python3 -m unittest services/headscale-ddns/test_headscale_ddns.py curl, the vault CLI and sleep are stubbed through PATH / env; nothing leaves the box. """ import os import pathlib import subprocess import tempfile import unittest SCRIPT = pathlib.Path(__file__).with_name("headscale-ddns.sh") # The stub answers by URL. $CF_ZONE_FAILS makes the zone lookup return an empty # body that many times first (the observed failure). Every call is logged. CURL_STUB = r"""#!/bin/bash url=""; for a in "$@"; do case "$a" in http*) url="$a";; esac; done method=GET; prev=""; for a in "$@"; do [ "$prev" = "-X" ] && method="$a"; prev="$a"; done echo "$method $url" >> "$STUB_DIR/calls" echo "$1" >> "$STUB_DIR/first_args" case "$url" in *icanhazip.com*) echo "${WAN_IP:-1.1.1.1}";; *"/zones?name="*) n=$(cat "$STUB_DIR/zone_n" 2>/dev/null || echo 0); n=$((n + 1)); echo "$n" > "$STUB_DIR/zone_n" [ "$n" -le "${CF_ZONE_FAILS:-0}" ] && exit 28 # empty body, curl timeout if [ -n "${CF_ZONE_BODY:-}" ]; then echo "$CF_ZONE_BODY" else echo '{"success":true,"result":[{"id":"Z1","name":"phasefinal.com"}]}'; fi;; *"/dns_records?"*) if [ -n "${CF_RECORDS_BODY:-}" ]; then echo "$CF_RECORDS_BODY" elif [ -n "${CF_NO_RECORD:-}" ]; then echo '{"success":true,"result":[]}' else echo '{"success":true,"result":[{"id":"R1","type":"A","name":"headscale.phasefinal.com","content":"'"${DNS_IP:-1.1.1.1}"'"}]}'; fi;; *"/dns_records/"*) if [ -n "${CF_PATCH_BODY:-}" ]; then echo "$CF_PATCH_BODY" else echo '{"success":true,"result":{"content":"'"${WAN_IP:-1.1.1.1}"'"}}'; fi;; *) echo '{"success":false,"errors":[{"message":"stub: unexpected url"}],"result":null}';; esac """ class CloudflareCallTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.addCleanup(self.tmp.cleanup) d = pathlib.Path(self.tmp.name) bindir = d / "bin" bindir.mkdir() vault = d / "vault cli" # a space in the path, on purpose vault.mkdir() for p, body in ((bindir / "curl", CURL_STUB), (bindir / "sleep", "#!/bin/sh\nexit 0\n"), (vault / "secret", "#!/bin/sh\necho test-token\n")): p.write_text(body) p.chmod(0o755) self.d = d self.env = dict(os.environ, PATH=f"{bindir}:{os.environ['PATH']}", STUB_DIR=str(d), HEADSCALE_DDNS_SECRET_CLI=str(vault / "secret")) def run_script(self, **env): return subprocess.run(["bash", str(SCRIPT)], env=dict(self.env, **env), capture_output=True, text=True, timeout=60) def calls(self): f = self.d / "calls" return f.read_text().splitlines() if f.exists() else [] def test_transient_empty_zone_lookup_is_retried(self): r = self.run_script(CF_ZONE_FAILS="2") self.assertEqual(r.returncode, 0, r.stderr) self.assertIn("unchanged 1.1.1.1", r.stdout) self.assertEqual(r.stderr.count("WARN: Cloudflare"), 2, r.stderr) def test_persistent_empty_zone_lookup_fails_loudly_and_never_writes(self): r = self.run_script(CF_ZONE_FAILS="99") self.assert_clean_fatal(r, "zone") self.assertFalse(any(c.startswith("PATCH") for c in self.calls()), self.calls()) self.assertFalse(any("/zones//" in c for c in self.calls()), self.calls()) def test_missing_a_record_fails_loudly_and_never_writes(self): r = self.run_script(CF_NO_RECORD="1", DNS_IP="198.51.100.1") self.assert_clean_fatal(r, "A record") self.assertFalse(any(c.startswith("PATCH") for c in self.calls()), self.calls()) def test_changed_ip_is_patched_once(self): r = self.run_script(DNS_IP="198.51.100.1", WAN_IP="1.1.1.1") self.assertEqual(r.returncode, 0, r.stderr) patches = [c for c in self.calls() if c.startswith("PATCH")] self.assertEqual(patches, [ "PATCH https://api.cloudflare.com/client/v4/zones/Z1/dns_records/R1"]) self.assertIn("updated 1.1.1.1", r.stdout) def patched(self): return any(c.startswith("PATCH") for c in self.calls()) def assert_clean_fatal(self, r, why): """Failed, with a one-line FATAL naming `why` — so a test cannot pass just because the run died earlier for some unrelated reason.""" self.assertEqual(r.returncode, 1, r.stdout + r.stderr) self.assertNotIn("Traceback", r.stderr) fatal = [l for l in r.stderr.splitlines() if l.startswith("FATAL")] self.assertEqual(len(fatal), 1, r.stderr) self.assertIn(why, fatal[0]) def test_success_true_patch_with_no_result_is_a_failure(self): # The update must be confirmed from the response, and a confirmation # that cannot be read must fail the run, not print "updated" and exit 0. r = self.run_script(DNS_IP="198.51.100.1", CF_PATCH_BODY='{"success":true,"result":null}') self.assert_clean_fatal(r, "update") self.assertNotIn("updated", r.stdout) def test_patch_response_showing_another_value_is_a_failure(self): r = self.run_script(DNS_IP="198.51.100.1", CF_PATCH_BODY='{"success":true,"result":{"content":"198.51.100.1"}}') self.assert_clean_fatal(r, "update") def test_two_a_records_are_refused_not_half_updated(self): two = ('{"success":true,"result":[' '{"id":"R1","type":"A","name":"headscale.phasefinal.com","content":"198.51.100.1"},' '{"id":"R2","type":"A","name":"headscale.phasefinal.com","content":"198.51.100.2"}]}') r = self.run_script(CF_RECORDS_BODY=two) self.assert_clean_fatal(r, "A record") self.assertFalse(self.patched()) def test_empty_record_id_cannot_shift_content_into_the_id(self): body = ('{"success":true,"result":[{"id":"","type":"A",' '"name":"headscale.phasefinal.com","content":"198.51.100.1"}]}') r = self.run_script(CF_RECORDS_BODY=body) self.assert_clean_fatal(r, "A record") self.assertFalse(self.patched()) def test_misshapen_success_bodies_fail_with_one_line_not_a_traceback(self): for var, body in (("CF_ZONE_BODY", '{"success":true,"result":{"id":"Z1"}}'), ("CF_ZONE_BODY", '{"success":true,"result":[{}]}'), ("CF_RECORDS_BODY", '{"success":true,"result":[{}]}'), ("CF_RECORDS_BODY", '{"success":true,"result":"x"}')): with self.subTest(var=var, body=body): (self.d / "calls").unlink(missing_ok=True) r = self.run_script(**{var: body}) self.assert_clean_fatal(r, "zone" if var == "CF_ZONE_BODY" else "A record") self.assertFalse(self.patched()) def test_non_public_wan_answer_is_never_published(self): for junk in ("127.0.0.1", "10.1.2.3", "192.168.1.5", "169.254.1.1", "100.64.0.9", "999.1.1.1"): with self.subTest(ip=junk): (self.d / "calls").unlink(missing_ok=True) r = self.run_script(WAN_IP=junk, DNS_IP="198.51.100.1") self.assert_clean_fatal(r, "WAN") self.assertFalse(self.patched()) def test_curl_ignores_curlrc_so_a_verbose_config_cannot_log_the_token(self): self.run_script() firsts = (self.d / "first_args").read_text().split() self.assertTrue(firsts and all(a == "-q" for a in firsts), firsts) def test_unchanged_ip_is_not_patched(self): r = self.run_script() self.assertEqual(r.returncode, 0, r.stderr) self.assertFalse(any(c.startswith("PATCH") for c in self.calls())) if __name__ == "__main__": unittest.main()