From 2ba3a14b985bb31f8c4cf2eb9e1fbb275fe2b5fb Mon Sep 17 00:00:00 2001 From: Kevin Veen-Birkenbach Date: Sun, 9 Aug 2026 22:30:45 +0200 Subject: [PATCH] refactor(svc-db): gate the Tor credential tests on one shared lookup The three svc-db-* credential tests each hand-rolled the same transport gate in their test.env.j2: (lookup('config', application_id, 'services.tor.enabled') | bool and lookup('config', application_id, 'services..exposed') | bool) Three copies of one predicate, each naming its own entity key, and each free to drift from the onion mapping that actually decides reachability. Extract it into plugins/lookup/tor_reachable.py, split the same way as its sibling tor_ports.py: a plain is_reachable() for callers holding the merged applications view, plus a thin LookupModule resolving it through lookup('applications'). Because collect_exposed_ports() scans services.values(), the entity name disappears from the templates entirely -- they collapse to TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }} Rework the integration guard accordingly: it now asserts that every svc-db-* template delegates to the shared lookup, and drives its base/variant agreement table through the real is_reachable() instead of re-implementing the predicate. Add unit coverage for the new lookup. Co-Authored-By: Claude Opus 5 (1M context) --- plugins/lookup/tor_reachable.py | 67 ++++++++++++ roles/svc-db-mariadb/files/test/test.sh | 2 +- roles/svc-db-mariadb/templates/test.env.j2 | 2 +- roles/svc-db-openldap/files/test/test.sh | 2 +- roles/svc-db-openldap/templates/test.env.j2 | 2 +- roles/svc-db-postgres/files/test/test.sh | 2 +- roles/svc-db-postgres/templates/test.env.j2 | 2 +- .../services/test_db_tor_transport_gate.py | 76 +++++++++++++ .../unit/plugins/lookup/test_tor_reachable.py | 101 ++++++++++++++++++ 9 files changed, 250 insertions(+), 6 deletions(-) create mode 100644 plugins/lookup/tor_reachable.py create mode 100644 tests/integration/roles/meta/services/test_db_tor_transport_gate.py create mode 100644 tests/unit/plugins/lookup/test_tor_reachable.py diff --git a/plugins/lookup/tor_reachable.py b/plugins/lookup/tor_reachable.py new file mode 100644 index 0000000000..870083d3cb --- /dev/null +++ b/plugins/lookup/tor_reachable.py @@ -0,0 +1,67 @@ +from __future__ import annotations + +from typing import Any + +from ansible.errors import AnsibleError +from ansible.plugins.loader import lookup_loader +from ansible.plugins.lookup import LookupBase + +from plugins.lookup.tor_ports import _as_bool, collect_exposed_ports + + +def is_reachable(applications: dict[str, Any], application_id: str) -> bool: + """True when *application_id* is reachable over the node onion, i.e. when + svc-net-tor is deployed for it (``services.tor.enabled``) AND at least one + of its services opts in with ``exposed: true``, which is what earns a + ``HiddenServicePort`` (``collect_exposed_ports``). + + Both halves are required: ``tor.enabled`` is the dependency edge, not the + port forward, so a client picking the Tor transport from that flag alone + dials a port with no HiddenServicePort and no host publish. + """ + config = (applications or {}).get(application_id) or {} + services = config.get("services") if isinstance(config, dict) else None + if not isinstance(services, dict): + return False + tor = services.get("tor") + if not (isinstance(tor, dict) and _as_bool(tor.get("enabled"))): + return False + return bool(collect_exposed_ports(applications, [application_id])) + + +class LookupModule(LookupBase): + """ + Usage: + {{ lookup('tor_reachable', application_id) }} + + Returns whether *application_id* is reachable over the node onion in the + current (variant-merged) applications view -- see ``is_reachable``. + + - parameters: + 1) application_id + """ + + def run( + self, + terms: list[Any] | None, + variables: dict[str, Any] | None = None, + **kwargs: Any, + ) -> list[bool]: + if not terms or len(terms) != 1: + raise AnsibleError( + "lookup('tor_reachable', application_id) expects exactly 1 term." + ) + application_id = str(terms[0]).strip() + if not application_id: + raise AnsibleError( + "lookup('tor_reachable', application_id): application_id is empty." + ) + + variables = variables or getattr(self._templar, "available_variables", {}) or {} + applications = lookup_loader.get( + "applications", + loader=getattr(self, "_loader", None), + templar=getattr(self, "_templar", None), + ).run([], variables=variables, roles_dir=kwargs.get("roles_dir"))[0] + + return [is_reachable(applications, application_id)] diff --git a/roles/svc-db-mariadb/files/test/test.sh b/roles/svc-db-mariadb/files/test/test.sh index 4539732f54..ba11c65992 100644 --- a/roles/svc-db-mariadb/files/test/test.sh +++ b/roles/svc-db-mariadb/files/test/test.sh @@ -11,7 +11,7 @@ # container over TCP (forces password auth). # # Env (rendered into test.env from templates/test.env.j2): -# TOR_ENABLED true|false (services.tor.enabled, variant-aware) +# TOR_ENABLED true|false (lookup('tor_reachable', application_id)) # TOR_SOCKS SOCKS proxy (set in test.env.j2) # ONION_HOST node onion host (svc-net-tor services.tor.node) # DB_PORT onion-forwarded database port diff --git a/roles/svc-db-mariadb/templates/test.env.j2 b/roles/svc-db-mariadb/templates/test.env.j2 index 176db27ea0..4845066e6c 100644 --- a/roles/svc-db-mariadb/templates/test.env.j2 +++ b/roles/svc-db-mariadb/templates/test.env.j2 @@ -3,5 +3,5 @@ DB_PASSWORD_B64={{ MARIADB_ROOT_PWD | b64encode }} DB_PORT={{ MARIADB_PORT }} DB_USER=root ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }} -TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }} +TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }} TOR_SOCKS={{ lookup('tor_socks') }} diff --git a/roles/svc-db-openldap/files/test/test.sh b/roles/svc-db-openldap/files/test/test.sh index 4352c33f55..88d56e4935 100644 --- a/roles/svc-db-openldap/files/test/test.sh +++ b/roles/svc-db-openldap/files/test/test.sh @@ -11,7 +11,7 @@ # (ldapwhoami over TCP). # # Env (rendered into test.env from templates/test.env.j2): -# TOR_ENABLED true|false (services.tor.enabled, variant-aware) +# TOR_ENABLED true|false (lookup('tor_reachable', application_id)) # TOR_SOCKS SOCKS proxy (set in test.env.j2) # ONION_HOST node onion host (svc-net-tor services.tor.node) # DB_PORT onion-forwarded LDAP port diff --git a/roles/svc-db-openldap/templates/test.env.j2 b/roles/svc-db-openldap/templates/test.env.j2 index 0c35aac81c..b947f17d9f 100644 --- a/roles/svc-db-openldap/templates/test.env.j2 +++ b/roles/svc-db-openldap/templates/test.env.j2 @@ -3,5 +3,5 @@ DB_CONTAINER={{ OPENLDAP_CONTAINER_ADDRESS }} DB_PASSWORD_B64={{ OPENLDAP_BIND_PW | b64encode }} DB_PORT={{ OPENLDAP_PORT }} ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }} -TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }} +TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }} TOR_SOCKS={{ lookup('tor_socks') }} diff --git a/roles/svc-db-postgres/files/test/test.sh b/roles/svc-db-postgres/files/test/test.sh index ad1375b751..f9d890c525 100644 --- a/roles/svc-db-postgres/files/test/test.sh +++ b/roles/svc-db-postgres/files/test/test.sh @@ -11,7 +11,7 @@ # container over TCP (forces password auth via pg_hba). # # Env (rendered into test.env from templates/test.env.j2): -# TOR_ENABLED true|false (services.tor.enabled, variant-aware) +# TOR_ENABLED true|false (lookup('tor_reachable', application_id)) # TOR_SOCKS SOCKS proxy (set in test.env.j2) # ONION_HOST node onion host (svc-net-tor services.tor.node) # DB_PORT onion-forwarded database port diff --git a/roles/svc-db-postgres/templates/test.env.j2 b/roles/svc-db-postgres/templates/test.env.j2 index 85577ad317..473bbe5ad6 100644 --- a/roles/svc-db-postgres/templates/test.env.j2 +++ b/roles/svc-db-postgres/templates/test.env.j2 @@ -4,5 +4,5 @@ DB_PASSWORD_B64={{ POSTGRES_PASSWORD | b64encode }} DB_PORT={{ POSTGRES_PORT }} DB_USER=postgres ONION_HOST={{ lookup('config', 'svc-net-tor', 'services.tor.node') }} -TOR_ENABLED={{ lookup('config', application_id, 'services.tor.enabled') | bool | lower }} +TOR_ENABLED={{ lookup('tor_reachable', application_id) | lower }} TOR_SOCKS={{ lookup('tor_socks') }} diff --git a/tests/integration/roles/meta/services/test_db_tor_transport_gate.py b/tests/integration/roles/meta/services/test_db_tor_transport_gate.py new file mode 100644 index 0000000000..fcc4c8e4e4 --- /dev/null +++ b/tests/integration/roles/meta/services/test_db_tor_transport_gate.py @@ -0,0 +1,76 @@ +"""Integration guard: for every ``svc-db-*`` role shipping an over-Tor +credential test, assert that the transport its ``test.env.j2`` selects agrees +with the onion mapping ``plugins.lookup.tor_ports.collect_exposed_ports`` +produces, in the base config and in every declared variant. +""" + +from __future__ import annotations + +import re +import unittest +from pathlib import Path + +from plugins.lookup.tor_ports import collect_exposed_ports +from plugins.lookup.tor_reachable import is_reachable +from utils import PROJECT_ROOT +from utils.cache.applications import get_merged_applications, get_variants +from utils.cache.files import read_text + +ROLES_DIR = PROJECT_ROOT / "roles" + +_GATE_RE = re.compile( + r"TOR_ENABLED=\{\{\s*" + r"lookup\('tor_reachable',\s*application_id\)" + r"\s*\|\s*lower\s*\}\}" +) + + +def _roles_with_tor_credential_test() -> dict[str, Path]: + found = {} + for template in sorted(ROLES_DIR.glob("svc-db-*/templates/test.env.j2")): + if "TOR_ENABLED=" in read_text(str(template)): + found[template.parent.parent.name] = template + return found + + +class TestDbTorTransportGate(unittest.TestCase): + def setUp(self) -> None: + self.templates = _roles_with_tor_credential_test() + self.assertTrue(self.templates, "no svc-db-* role renders TOR_ENABLED") + + def test_template_delegates_to_the_shared_lookup(self) -> None: + for role, template in self.templates.items(): + with self.subTest(role=role): + self.assertRegex( + read_text(str(template)), + _GATE_RE, + f"{role} must select its transport with " + f"lookup('tor_reachable', application_id) rather than a " + f"hand-rolled flag expression, so the credential test and " + f"the onion mapping (plugins/lookup/tor_ports.py) cannot " + f"drift apart.", + ) + + def test_gate_agrees_with_the_onion_mapping(self) -> None: + base = get_merged_applications(roles_dir=str(ROLES_DIR)) + variants = get_variants(roles_dir=str(ROLES_DIR)) + + for role in self.templates: + configs = [("base", base[role])] + configs += [(f"variant {i}", v) for i, v in enumerate(variants.get(role) or [])] + + for label, config in configs: + with self.subTest(role=role, config=label): + gate = is_reachable({role: config}, role) + ports = collect_exposed_ports({role: config}, [role]) + self.assertEqual( + gate, + bool(ports), + f"{role} [{label}]: the credential test would use " + f"{'Tor' if gate else 'local'} transport while the deploy " + f"forwards {ports or 'no'} port(s) over the onion", + ) + + +if __name__ == "__main__": # pragma: no cover + unittest.main() diff --git a/tests/unit/plugins/lookup/test_tor_reachable.py b/tests/unit/plugins/lookup/test_tor_reachable.py new file mode 100644 index 0000000000..2c86e9462d --- /dev/null +++ b/tests/unit/plugins/lookup/test_tor_reachable.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from ansible.errors import AnsibleError + +from plugins.lookup.tor_reachable import LookupModule, is_reachable +from utils.cache import _reset_cache_for_tests +from utils.cache.yaml import dump_yaml_str +from utils.roles.mapping import ROLE_FILE_META_SERVICES + + +def _apps(tor_enabled: object, exposed: object) -> dict: + return { + "svc-db-mariadb": { + "services": { + "tor": {"enabled": tor_enabled}, + "mariadb": {"exposed": exposed, "ports": {"local": {"mariadb": 3306}}}, + } + } + } + + +class TestIsReachable(unittest.TestCase): + def test_tor_enabled_and_exposed(self) -> None: + self.assertTrue(is_reachable(_apps(True, True), "svc-db-mariadb")) + + def test_exposed_without_tor(self) -> None: + self.assertFalse(is_reachable(_apps(False, True), "svc-db-mariadb")) + + def test_tor_without_exposed(self) -> None: + self.assertFalse(is_reachable(_apps(True, False), "svc-db-mariadb")) + + def test_string_truthiness_matches_tor_ports(self) -> None: + self.assertTrue(is_reachable(_apps("true", "yes"), "svc-db-mariadb")) + self.assertFalse(is_reachable(_apps("true", "off"), "svc-db-mariadb")) + + def test_exposed_service_without_local_port_has_no_onion_target(self) -> None: + apps = { + "svc-db-mariadb": { + "services": { + "tor": {"enabled": True}, + "mariadb": {"exposed": True, "ports": {"public": {"tls": 3307}}}, + } + } + } + self.assertFalse(is_reachable(apps, "svc-db-mariadb")) + + def test_unknown_or_malformed_application(self) -> None: + self.assertFalse(is_reachable(_apps(True, True), "svc-db-nope")) + self.assertFalse(is_reachable({"a": None}, "a")) + self.assertFalse(is_reachable({}, "a")) + + +class TestTorReachableLookup(unittest.TestCase): + def setUp(self) -> None: + self._tmpdir = tempfile.TemporaryDirectory() + self.roles_dir = Path(self._tmpdir.name) / "roles" + _reset_cache_for_tests() + for role, exposed in (("svc-db-mariadb", True), ("svc-db-postgres", False)): + path = self.roles_dir / role / ROLE_FILE_META_SERVICES + path.parent.mkdir(parents=True, exist_ok=True) + entity = role.rsplit("-", 1)[-1] + path.write_text( + dump_yaml_str( + { + "tor": {"enabled": True}, + entity: { + "exposed": exposed, + "ports": {"local": {entity: 3306}}, + }, + } + ), + encoding="utf-8", + ) + self.lookup = LookupModule() + self.lookup._templar = None + + def tearDown(self) -> None: + _reset_cache_for_tests() + self._tmpdir.cleanup() + + def _run(self, application_id: str) -> bool: + return self.lookup.run( + [application_id], variables={}, roles_dir=str(self.roles_dir) + )[0] + + def test_run_resolves_against_the_merged_applications_view(self) -> None: + self.assertTrue(self._run("svc-db-mariadb")) + self.assertFalse(self._run("svc-db-postgres")) + + def test_arity_and_empty_term_raise(self) -> None: + for terms in ([], ["a", "b"], [""], [" "]): + with self.subTest(terms=terms), self.assertRaises(AnsibleError): + self.lookup.run(terms, variables={}, roles_dir=str(self.roles_dir)) + + +if __name__ == "__main__": + unittest.main()