diff --git a/plugins/lookup/ca_injected.py b/plugins/lookup/ca_injected.py index 70aec51840..de474ed8ac 100644 --- a/plugins/lookup/ca_injected.py +++ b/plugins/lookup/ca_injected.py @@ -32,6 +32,7 @@ from ansible.plugins.loader import lookup_loader from ansible.plugins.lookup import LookupBase from plugins.filter.has.domain import has_domain +from utils.tls_common import AVAILABLE_FLAVORS, as_str, require SELF_SIGNED_MODE = "self_signed" @@ -58,19 +59,22 @@ class LookupModule(LookupBase): args, variables=variables or {} )[0] - services = _run("config", [application_id, "services", {}]) or {} - if any( - _to_bool(service.get("ca_client", False)) - for service in services.values() - if isinstance(service, dict) - ): - mode = (variables or {}).get("TLS_MODE", "") - if templar is not None: - mode = templar.template(mode) - return [str(mode).strip().lower() == SELF_SIGNED_MODE] - if not has_domain(_run("domains", []), application_id): - return [False] + services = _run("config", [application_id, "services", {}]) or {} + if not any( + _to_bool(service.get("ca_client", False)) + for service in services.values() + if isinstance(service, dict) + ): + return [False] + mode = as_str(require(variables or {}, "TLS_MODE", str)) + if mode not in AVAILABLE_FLAVORS: + raise AnsibleError( + f"ca_injected: TLS_MODE must be one of " + f"{sorted(AVAILABLE_FLAVORS)}, got '{mode}'" + ) + return [mode == SELF_SIGNED_MODE] + if not _to_bool(_run("tls", [application_id, "enabled"])): return [False] return [_run("tls", [application_id, "mode"]) == SELF_SIGNED_MODE] diff --git a/plugins/lookup/compose_file_args.py b/plugins/lookup/compose_file_args.py index e935d8f6c6..7b9e0de235 100644 --- a/plugins/lookup/compose_file_args.py +++ b/plugins/lookup/compose_file_args.py @@ -8,8 +8,7 @@ # - Always include base compose.yml # - Include compose.override.yml ONLY when the ROLE (application_id) provides one # (same logic as tasks/04_files.yml with_first_found) -# - Include compose.ca.override.yml ONLY when: -# the app has a domain AND TLS is enabled AND TLS mode == "self_signed" +# - Include compose.ca.override.yml ONLY when lookup('ca_injected') is true # # Optional kwargs: # - include_ca (bool, default True): @@ -59,27 +58,6 @@ def _maybe_template(templar: Any, value: Any) -> Any: return value -def _value_has_domain(v: Any) -> bool: - if v is None: - return False - if isinstance(v, str): - return v.strip() != "" - if isinstance(v, (list, tuple, set)): - return any(_value_has_domain(x) for x in v) - if isinstance(v, dict): - return any(_value_has_domain(x) for x in v.values()) - return False - - -def _has_domain(domains: Any, application_id: str) -> bool: - """ - Dependency-free domain existence check that matches the intent of the old filter. - """ - if isinstance(domains, dict): - return _value_has_domain(domains.get(application_id)) - return _value_has_domain(domains) - - def _role_provides_override(*, application_id: str, templar: Any) -> bool: """ Mirror tasks/04_files.yml "with_first_found" logic: @@ -183,35 +161,15 @@ class LookupModule(LookupBase): parts.append(f"{flag} {override}") if include_ca: - domains = lookup_loader.get( - "domains", loader=self._loader, templar=templar - ).run([], variables=variables)[0] - if _has_domain(domains, application_id): - tlsr = lookup_loader.get("tls", self._loader, self._templar) - tls = tlsr.run([application_id], variables=variables)[0] - - if not isinstance(tls, dict): + injected = lookup_loader.get( + "ca_injected", loader=self._loader, templar=templar + ).run([application_id], variables=variables)[0] + if injected: + if not _as_str(ca_override): raise AnsibleError( - f"compose_file_args: tls returned non-dict: {type(tls)}" + "compose_file_args: compose.files.compose_ca_override is " + "required when the CA is injected" ) - if "enabled" not in tls: - raise AnsibleError( - "compose_file_args: tls did not return 'enabled'" - ) - if "mode" not in tls: - raise AnsibleError("compose_file_args: tls did not return 'mode'") - - enabled = bool(tls["enabled"]) - mode = _as_str(tls["mode"]) - if not mode: - raise AnsibleError("compose_file_args: tls returned empty 'mode'") - - if enabled and mode == "self_signed": - if not _as_str(ca_override): - raise AnsibleError( - "compose_file_args: compose.files.compose_ca_override is required " - "when TLS is enabled and mode is self_signed" - ) - parts.append(f"{flag} {ca_override}") + parts.append(f"{flag} {ca_override}") return [" ".join(parts)] diff --git a/tests/unit/python/plugins/lookup/test_ca_injected.py b/tests/unit/python/plugins/lookup/test_ca_injected.py index ef53960525..ce658492f2 100644 --- a/tests/unit/python/plugins/lookup/test_ca_injected.py +++ b/tests/unit/python/plugins/lookup/test_ca_injected.py @@ -74,7 +74,8 @@ class TestCaInjectedLookup(unittest.TestCase): plugin._loader = None plugin._templar = _DummyTemplar() with mock.patch.object(self.mod.lookup_loader, "get", side_effect=fake_get): - result = plugin.run([app], variables={"TLS_MODE": tls_mode}) + variables = {} if tls_mode is None else {"TLS_MODE": tls_mode} + result = plugin.run([app], variables=variables) return result[0], calls def test_a_self_signed_app_with_a_domain_gets_the_ca(self): @@ -104,7 +105,7 @@ class TestCaInjectedLookup(unittest.TestCase): def test_an_app_without_a_domain_never_asks_about_tls(self): verdict, calls = self._resolve(domains={}, enabled=True, mode="self_signed") self.assertIs(verdict, False) - self.assertEqual([name for name, _ in calls], ["config", "domains"]) + self.assertEqual([name for name, _ in calls], ["domains", "config"]) def test_a_domainless_ca_client_gets_the_ca_on_the_deployment_mode(self): verdict, calls = self._resolve( @@ -120,7 +121,24 @@ class TestCaInjectedLookup(unittest.TestCase): "of its own, and without the CA every outbound call fails the " "handshake", ) - self.assertEqual([name for name, _ in calls], ["config"]) + self.assertEqual([name for name, _ in calls], ["domains", "config"]) + + def test_a_ca_client_with_a_domain_follows_its_own_tls_mode(self): + verdict, calls = self._resolve( + domains={"web-app-moodle": "moodle.example"}, + enabled=True, + mode="letsencrypt", + services={"agent-broker": {"ca_client": True}}, + tls_mode="self_signed", + ) + self.assertIs( + verdict, + False, + "an app that has a domain is served over its own resolved mode, which " + "may override the deployment default; the ca_client shortcut must not " + "answer for it on the global TLS_MODE", + ) + self.assertEqual([name for name, _ in calls], ["domains", "tls", "tls"]) def test_a_ca_client_gets_nothing_when_the_ca_is_not_self_signed(self): verdict, _ = self._resolve( @@ -132,6 +150,20 @@ class TestCaInjectedLookup(unittest.TestCase): ) self.assertIs(verdict, False) + def test_a_ca_client_refuses_an_absent_tls_mode(self): + with self.assertRaises( + AnsibleError, + msg="a missing mode read as 'not self signed' withholds the CA in " + "silence, and every handshake then fails far from here", + ): + self._resolve( + domains={}, + enabled=True, + mode="self_signed", + services={"agent-broker": {"ca_client": True}}, + tls_mode=None, + ) + def test_the_term_is_required(self): plugin = self.mod.LookupModule() plugin._templar = _DummyTemplar() diff --git a/tests/unit/python/plugins/lookup/test_compose_file_args.py b/tests/unit/python/plugins/lookup/test_compose_file_args.py index 8ad1a27a86..38116e0d0b 100644 --- a/tests/unit/python/plugins/lookup/test_compose_file_args.py +++ b/tests/unit/python/plugins/lookup/test_compose_file_args.py @@ -30,6 +30,14 @@ class _DomainsResolveStub: return [(variables or {}).get("domains", {})] +class _CaInjectedStub: + def __init__(self, injected: bool): + self._injected = injected + + def run(self, terms, variables=None, **kwargs): + return [self._injected] + + class TestComposeFArgs(unittest.TestCase): def setUp(self): self.m = _load_module( @@ -48,18 +56,48 @@ class TestComposeFArgs(unittest.TestCase): }, } - def _loader_dispatch(self, tls_stub): - # "domains" resolves through variables['domains'] to keep tests hermetic; - # "tls" resolves through the per-test stub. + def _loader_dispatch(self, tls_stub, ca_injected=None): + verdict = tls_stub.run(None)[0] + injected = ( + (verdict["enabled"] and verdict["mode"] == "self_signed") + if ca_injected is None + else ca_injected + ) + def _get(name, *args, **kwargs): if name == "domains": return _DomainsResolveStub() if name == "tls": return tls_stub + if name == "ca_injected": + return _CaInjectedStub(injected) raise AssertionError(f"unexpected lookup name: {name}") return _get + def test_a_domainless_ca_client_still_gets_the_ca_override_file(self): + with ( + patch.object( + self.m.lookup_loader, + "get", + side_effect=self._loader_dispatch( + _TlsResolveStub(False, "off"), ca_injected=True + ), + ), + patch.object( + self.m, "get_docker_paths", side_effect=self._stub_get_docker_paths + ), + patch.object(self.m, "_role_provides_override", return_value=False), + ): + out = self.lookup.run(["web-app-a"], variables={**self.vars, "domains": {}}) + self.assertIn( + "/x/compose.ca.override.yml", + out[0], + "ca_injected grants a domainless ca_client the override and the " + "handler writes it; a domain gate here leaves that file on disk " + "while the container runs without a trust store", + ) + def _stub_get_docker_paths(self, application_id: str, base_dir: str) -> dict: self.assertEqual(application_id, "web-app-a") self.assertEqual(base_dir, "/x/")