cadprox/tests/test_app.py

# Copyright 2026 Umbra Tools (umbratools.com)
# SPDX-License-Identifier: Apache-2.0

"""Tests for cadprox's Caddyfile parsing and rendering.

These cover the parts that can silently destroy the user's configuration:
round-tripping a hand-written Caddyfile, preserving blocks cadprox does not
model, and making sure password hashes never leave the server.

Run with:  python3 -m unittest discover -s tests -v
"""

import json
import os
import subprocess
import sys
import unittest

sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))

import app  # noqa: E402

REAL_CADDYFILE = """\
bdmurphy.me {
    encode gzip
    reverse_proxy 127.0.0.1:8080
}

vault.bdmurphy.me {
    encode gzip
    reverse_proxy 127.0.0.1:8082
}

umbraudio.bdmurphy.me {
    encode gzip
    reverse_proxy 192.168.123.12:8420
}
"""


def targets(state):
    return {", ".join(e["hosts"]): e["target"] for e in state["entries"]}


def caddy_available() -> bool:
    try:
        return subprocess.run(
            [app.CADDY_BIN, "version"], capture_output=True, timeout=30
        ).returncode == 0
    except (OSError, subprocess.SubprocessError):
        return False


class CaddyBackedTest(unittest.TestCase):
    """Base class for tests that need the real Caddy binary to adjudicate."""

    @classmethod
    def setUpClass(cls):
        if not caddy_available():
            raise unittest.SkipTest(
                "caddy binary not reachable; set CADPROX_CADDY_BIN to test against a real Caddy"
            )

    def validate_with_real_caddy(self, config: str) -> tuple[bool, str]:
        return app.validate_caddyfile(config)


class TestParse(CaddyBackedTest):
    def test_parses_real_caddyfile(self):
        state = app.parse_caddyfile(REAL_CADDYFILE)
        self.assertEqual(
            targets(state),
            {
                "bdmurphy.me": "127.0.0.1:8080",
                "vault.bdmurphy.me": "127.0.0.1:8082",
                "umbraudio.bdmurphy.me": "192.168.123.12:8420",
            },
        )
        self.assertTrue(all(e["managed"] for e in state["entries"]))

    def test_header_does_not_accumulate_across_saves(self):
        # Regression: the generated header used to be absorbed as preserved
        # leading text, so every save added another copy and the file grew.
        state = app.parse_caddyfile(REAL_CADDYFILE)
        text = app.render_caddyfile(state)
        for _ in range(5):
            text = app.render_caddyfile(app.parse_caddyfile(text))
        self.assertEqual(
            text.count(app.MANAGED_MARKER), 1,
            "the cadprox header was duplicated across repeated saves",
        )
        self.assertEqual(
            len([ln for ln in text.splitlines() if ln.startswith("#") and not ln.startswith("# ")]),
            1,
        )
        ok, msg = self.validate_with_real_caddy(text)
        self.assertTrue(ok, f"repeated saves produced an invalid config: {msg}")

    def test_operator_comments_at_the_top_are_preserved(self):
        text = """\
# My own note about this file
# second line

a.example.com {
\treverse_proxy 10.0.0.1:80
}
"""
        state = app.parse_caddyfile(text)
        rendered = app.render_caddyfile(state)
        self.assertIn("# My own note about this file", rendered)
        self.assertIn("# second line", rendered)
        self.assertEqual(rendered.count(app.MANAGED_MARKER), 1)

    def test_header_does_not_hide_global_block(self):
        # Regression: cadprox writes its own comment header at the top of the
        # file, and those comments used to stop the global block being found,
        # which turned it into a phantom entry with a blank URL.
        text = """\
# Caddy reverse proxy configuration
#
# Site blocks below are managed by cadprox.

{
\tadmin 127.0.0.1:2019
\tauto_https off
}

a.example.com {
\tencode gzip
\treverse_proxy 10.0.0.1:80
}
"""
        state = app.parse_caddyfile(text)
        self.assertIn("auto_https off", state["globals"])
        self.assertEqual([e["hosts"] for e in state["entries"]], [["a.example.com"]])
        for entry in state["entries"]:
            self.assertTrue(entry["hosts"], "entry with no site address leaked into the UI")
        rendered = app.render_caddyfile(state)
        self.assertLess(rendered.index("auto_https off"), rendered.index("a.example.com"))
        ok, msg = self.validate_with_real_caddy(rendered)
        self.assertTrue(ok, f"rendered config rejected by caddy: {msg}")

    def test_keyless_block_never_becomes_an_entry(self):
        state = app.parse_caddyfile("{\n\tadmin 0.0.0.0:2019\n}\n")
        self.assertEqual(state["entries"], [])
        self.assertIn("admin 0.0.0.0:2019", state["globals"])

    def test_mid_file_keyless_block_preserved_not_listed(self):
        text = """\
a.example.com {
\treverse_proxy 10.0.0.1:80
}

{
\tservers {
\t\ttrusted_proxies static 10.0.0.0/8
\t}
}
"""
        state = app.parse_caddyfile(text)
        self.assertEqual([e["hosts"] for e in state["entries"]], [["a.example.com"]])
        self.assertIn("trusted_proxies", state["extras"])
        self.assertIn("trusted_proxies", app.render_caddyfile(state))

    def test_global_options_block_preserved(self):
        text = """\
{
\tadmin 127.0.0.1:2019
\temail me@example.com
}

a.example.com {
\treverse_proxy 10.0.0.1:80
}
"""
        state = app.parse_caddyfile(text)
        self.assertEqual(len(state["entries"]), 1)
        self.assertIn("email me@example.com", state["globals"])
        self.assertIn("admin 127.0.0.1:2019", state["globals"])
        self.assertEqual(state["extras"], "")
        rendered = app.render_caddyfile(state)
        self.assertIn("email me@example.com", rendered)
        # Caddy rejects a global block that is not first, so it must precede
        # every site block in the rendered output.
        self.assertLess(
            rendered.index("email me@example.com"),
            rendered.index("a.example.com"),
        )

    def test_global_block_alone_round_trips(self):
        text = "{\n\tadmin 0.0.0.0:2019\n}\n"
        state = app.parse_caddyfile(text)
        self.assertEqual(state["entries"], [])
        self.assertIn("admin 0.0.0.0:2019", app.render_caddyfile(state))

    def test_basic_auth_round_trips_hashes(self):
        text = """\
secure.example.com {
\tencode gzip
\tbasic_auth {
\t\talice $2a$14$abcdefghijklmnopqrstuv
\t\tbob $2a$14$zyxwvutsrqponmlkjihg
\t}
\treverse_proxy 10.0.0.2:8080
}
"""
        state = app.parse_caddyfile(text)
        entry = state["entries"][0]
        self.assertTrue(entry["managed"])
        self.assertEqual(entry["auth"]["users"], ["alice", "bob"])
        self.assertEqual(
            entry["auth"]["hashes"],
            ["$2a$14$abcdefghijklmnopqrstuv", "$2a$14$zyxwvutsrqponmlkjihg"],
        )
        # The exact hashes must survive a render.
        rendered = app.render_caddyfile(state)
        self.assertIn("alice $2a$14$abcdefghijklmnopqrstuv", rendered)
        self.assertIn("bob $2a$14$zyxwvutsrqponmlkjihg", rendered)

    def test_hash_password_produces_a_hash_basic_auth_accepts(self):
        # Regression: Caddy's hash-password can emit argon2id, but the HTTP
        # basic auth provider in Caddy 2.11 only verifies bcrypt, so an argon2id
        # hash loads fine and then rejects every login including the right one.
        hashed = app.hash_password("correcthorsebattery")
        self.assertTrue(
            hashed.startswith("$2"),
            f"expected a bcrypt hash, got {hashed[:16]!r}",
        )
        self.assertNotIn("argon2id", hashed)
        self.assertNotIn("correcthorsebattery", hashed)
        # bcrypt salts randomly, so the same password must not hash identically.
        self.assertNotEqual(hashed, app.hash_password("correcthorsebattery"))

    def test_generated_hash_survives_a_render_and_reparse(self):
        hashed = app.hash_password("correcthorsebattery")
        state = {
            "entries": [
                {
                    "hosts": ["s.example.com"],
                    "target": "10.0.0.1:80",
                    "encode": True,
                    "auth": {"users": ["alice"], "hashes": [hashed]},
                    "managed": True,
                    "reason": "",
                    "lines": [],
                }
            ],
            "globals": "",
            "extras": "",
        }
        reparsed = app.parse_caddyfile(app.render_caddyfile(state))
        self.assertEqual(reparsed["entries"][0]["auth"]["hashes"], [hashed])
        ok, msg = self.validate_with_real_caddy(app.render_caddyfile(state))
        self.assertTrue(ok, f"caddy rejected the generated config: {msg}")

    def test_unknown_directive_marks_block_unmanaged_and_preserves_it(self):
        text = """\
fancy.example.com {
\tencode gzip
\theader_up X-Custom abc
\treverse_proxy 10.0.0.3:9000
}
"""
        state = app.parse_caddyfile(text)
        entry = state["entries"][0]
        self.assertFalse(entry["managed"])
        self.assertIn("header_up X-Custom abc", entry["reason"])
        rendered = app.render_caddyfile(state)
        self.assertIn("header_up X-Custom abc", rendered)
        self.assertIn("reverse_proxy 10.0.0.3:9000", rendered)

    def test_multiple_upstreams_marked_unmanaged(self):
        text = """\
lb.example.com {
\treverse_proxy 10.0.0.4:80 10.0.0.5:80
}
"""
        entry = app.parse_caddyfile(text)["entries"][0]
        self.assertFalse(entry["managed"])

    def test_import_preserved_verbatim(self):
        text = """\
import /etc/caddy/snippets/*.caddy

x.example.com {
\treverse_proxy 10.0.0.6:80
}
"""
        state = app.parse_caddyfile(text)
        self.assertIn("import /etc/caddy/snippets/*.caddy", state["extras"])
        self.assertIn("import /etc/caddy/snippets/*.caddy", app.render_caddyfile(state))

    def test_nested_braces_do_not_split_blocks(self):
        text = """\
a.example.com {
\tbasic_auth {
\t\tu $2a$14$abcdefghijklmnopqrstuv
\t}
\treverse_proxy 10.0.0.7:80
}
b.example.com {
\treverse_proxy 10.0.0.8:80
}
"""
        state = app.parse_caddyfile(text)
        self.assertEqual(len(state["entries"]), 2)
        self.assertEqual(
            targets(state), {"a.example.com": "10.0.0.7:80", "b.example.com": "10.0.0.8:80"}
        )

    def test_comments_ignored(self):
        text = """\
# leading comment
a.example.com {
\t# inner comment
\tencode gzip
\treverse_proxy 10.0.0.9:80 # trailing
}
"""
        state = app.parse_caddyfile(text)
        self.assertEqual(len(state["entries"]), 1)
        self.assertTrue(state["entries"][0]["managed"])


class TestRenderRoundTrip(unittest.TestCase):
    def test_render_then_parse_is_stable(self):
        state = app.parse_caddyfile(REAL_CADDYFILE)
        first = app.render_caddyfile(state)
        second = app.render_caddyfile(app.parse_caddyfile(first))
        self.assertEqual(first, second, "render is not idempotent")

    def test_real_targets_survive_render(self):
        state = app.parse_caddyfile(REAL_CADDYFILE)
        reparsed = app.parse_caddyfile(app.render_caddyfile(state))
        self.assertEqual(targets(state), targets(reparsed))
        self.assertEqual(
            [", ".join(e["hosts"]) for e in state["entries"]],
            [", ".join(e["hosts"]) for e in reparsed["entries"]],
        )

    def test_render_uses_tabs_and_ends_with_newline(self):
        state = app.parse_caddyfile(REAL_CADDYFILE)
        out = app.render_caddyfile(state)
        self.assertTrue(out.endswith("\n"))
        self.assertIn("\treverse_proxy 127.0.0.1:8080", out)
        self.assertIn("\tencode gzip", out)


class TestValidation(unittest.TestCase):
    def test_host_accepted(self):
        for good in ("bdmurphy.me", "A.Bdmurphy.ME", "localhost", "192.168.1.1", "a-b.c"):
            self.assertEqual(app.validate_host(good), good.lower().rstrip("."))

    def test_host_rejected(self):
        for bad in ("", "http://a.com", "a.com/path", "a.com:80", "has space", "a..b", "-x.com"):
            with self.assertRaises(app.ValidationError, msg=bad):
                app.validate_host(bad)

    def test_target_accepted(self):
        self.assertEqual(app.validate_target(" 192.168.123.14:8425 "), "192.168.123.14:8425")
        self.assertEqual(app.validate_target("host.local:1"), "host.local:1")
        self.assertEqual(app.validate_target("1.2.3.4:65535"), "1.2.3.4:65535")

    def test_target_rejected(self):
        for bad in ("", "http://1.2.3.4:80", "1.2.3.4", "1.2.3.4:80:90", "1.2.3.4:0",
                    "1.2.3.4:70000", "1.2.3.4:http", "1.2.3.4/80"):
            with self.assertRaises(app.ValidationError, msg=bad):
                app.validate_target(bad)

    def test_password_rules(self):
        with self.assertRaises(app.ValidationError):
            app.validate_password("short")
        with self.assertRaises(app.ValidationError):
            app.validate_password("")
        self.assertEqual(app.validate_password("longenough1"), "longenough1")

    def test_username_rules(self):
        with self.assertRaises(app.ValidationError):
            app.validate_username("bad user")
        with self.assertRaises(app.ValidationError):
            app.validate_username("")
        self.assertEqual(app.validate_username("alice"), "alice")


class TestPublicState(unittest.TestCase):
    def test_hashes_never_leave_the_server(self):
        text = """\
s.example.com {
\tbasic_auth {
\t\talice $2a$14$abcdefghijklmnopqrstuv
\t}
\treverse_proxy 10.0.0.1:80
}
"""
        public = app.public_state(app.parse_caddyfile(text))
        blob = json.dumps(public)
        self.assertNotIn("$2a$14$", blob)
        self.assertNotIn("abcdefghijklmnopqrstuv", blob)
        self.assertEqual(public["entries"][0]["auth_user"], "alice")
        self.assertEqual(public["entries"][0]["auth_count"], 1)

    def test_open_proxy_reported_as_no_auth(self):
        public = app.public_state(app.parse_caddyfile(REAL_CADDYFILE))
        self.assertTrue(all(e["auth_user"] == "" for e in public["entries"]))


class TestSessions(unittest.TestCase):
    def test_token_lifecycle(self):
        token = app.new_session()
        self.assertTrue(app.session_valid(token))
        self.assertFalse(app.session_valid("nonsense"))
        self.assertFalse(app.session_valid(""))
        app._sessions[token] = 0
        self.assertFalse(app.session_valid(token), "expired token accepted")


class TestSecretsFile(unittest.TestCase):
    """The admin password may live in a file rather than the compose file.

    The file is mounted by Docker, so both the "bare password" and "key=value"
    shapes have to work, and a bind mount whose source is missing has to fail
    loudly rather than silently.
    """

    def setUp(self):
        import tempfile

        self.tmp = tempfile.mkdtemp()

    def tearDown(self):
        import shutil

        shutil.rmtree(self.tmp, ignore_errors=True)

    def write(self, name, text):
        path = os.path.join(self.tmp, name)
        with open(path, "w", encoding="utf-8") as fh:
            fh.write(text)
        return path

    def test_bare_password_with_trailing_newline(self):
        path = self.write("bare", "Fake#Pass@99\n")
        self.assertEqual(app.read_secrets_file(path), "Fake#Pass@99")

    def test_special_characters_survive_verbatim(self):
        # '#' is literal mid-line, not a comment, and the value is not quoted
        # or unescaped on the way in. These are fixtures, never real secrets.
        for password in ("Fake#Pass@99", "a b c", "p@ss w#rd", " leading-ish"):
            path = self.write("s", password + "\n")
            self.assertEqual(app.read_secrets_file(path), password.strip())

    def test_key_value_form(self):
        path = self.write("kv", "CADPROX_ADMIN_PASSWORD=Fake#Pass@99\nOTHER=ignored\n")
        self.assertEqual(app.read_secrets_file(path), "Fake#Pass@99")

    def test_key_value_form_with_comments(self):
        path = self.write(
            "kv", "# a leading comment\n\nCADPROX_ADMIN_PASSWORD=Fake#Pass@99\n"
        )
        self.assertEqual(app.read_secrets_file(path), "Fake#Pass@99")

    def test_unrecognised_key_falls_back_to_whole_file(self):
        # A bare password that happens to contain '=' must not be mangled by
        # the key/value parse, because no key we recognise is present.
        path = self.write("eq", "abc=def\n")
        self.assertEqual(app.read_secrets_file(path), "abc=def")

    def test_empty_file_returns_empty_string(self):
        # load_admin() turns this into a startup error with a clear message.
        self.assertEqual(app.read_secrets_file(self.write("empty", "")), "")

    def test_directory_instead_of_file_is_named_as_such(self):
        # Docker silently creates a directory when a bind source is missing.
        with self.assertRaises(ValueError) as ctx:
            app.read_secrets_file(self.tmp)
        self.assertIn("directory", str(ctx.exception))

    def test_missing_file_raises_oserror(self):
        with self.assertRaises(OSError):
            app.read_secrets_file(os.path.join(self.tmp, "nope"))


if __name__ == "__main__":
    unittest.main(verbosity=2)