esh-pfi-infrastructure/services/headscale-ddns/test_headscale_ddns.py

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