diff --git a/plugins/lookup/ca_injected.py b/plugins/lookup/ca_injected.py index fb7e98f6e6..70aec51840 100644 --- a/plugins/lookup/ca_injected.py +++ b/plugins/lookup/ca_injected.py @@ -10,6 +10,12 @@ lived as four hand-copied expressions - both `sys-svc-compose` handlers, `sys-svc-compose-ca/vars/main.yml` and the msmtp template - which is one copy per place that can drift. +Those three ask whether the application is *served* over the self-signed CA. +An application that only *calls* one has no domain of its own and would be +skipped, so a service may declare ``ca_client: true`` and be answered on the +deployment's TLS mode alone. Without it such a container verifies against an +empty trust store and fails every outbound call at the TLS handshake. + Returns a plain bool, so callers read as a predicate: when: lookup('ca_injected', application_id) @@ -52,6 +58,17 @@ 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] if not _to_bool(_run("tls", [application_id, "enabled"])): diff --git a/roles/svc-ai-agent-broker/meta/services.yml b/roles/svc-ai-agent-broker/meta/services.yml index fa01b96f50..fbb3a75f89 100644 --- a/roles/svc-ai-agent-broker/meta/services.yml +++ b/roles/svc-ai-agent-broker/meta/services.yml @@ -14,6 +14,7 @@ agent-broker: # nocheck: docker-version base of a self-built stdlib-only image, not the served application version: "3.13.15-slim" custom: true + ca_client: true name: agent-broker min_storage: 130MB cpus: "0.5" diff --git a/tests/unit/python/plugins/lookup/test_ca_injected.py b/tests/unit/python/plugins/lookup/test_ca_injected.py index 5680300214..ef53960525 100644 --- a/tests/unit/python/plugins/lookup/test_ca_injected.py +++ b/tests/unit/python/plugins/lookup/test_ca_injected.py @@ -47,12 +47,23 @@ class TestCaInjectedLookup(unittest.TestCase): def setUpClass(cls): cls.mod = _load_lookup() - def _resolve(self, *, domains, enabled, mode, app="web-app-moodle"): + def _resolve( + self, + *, + domains, + enabled, + mode, + app="web-app-moodle", + services=None, + tls_mode="self_signed", + ): calls: list[tuple[str, list]] = [] def fake_get(name, **_kwargs): def run(args, variables=None): calls.append((name, list(args))) + if name == "config": + return [services or {}] if name == "domains": return [domains] return [enabled if args[1] == "enabled" else mode] @@ -63,7 +74,7 @@ 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={}) + result = plugin.run([app], variables={"TLS_MODE": tls_mode}) return result[0], calls def test_a_self_signed_app_with_a_domain_gets_the_ca(self): @@ -93,7 +104,33 @@ 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], ["domains"]) + self.assertEqual([name for name, _ in calls], ["config", "domains"]) + + def test_a_domainless_ca_client_gets_the_ca_on_the_deployment_mode(self): + verdict, calls = self._resolve( + domains={}, + enabled=True, + mode="self_signed", + services={"agent-broker": {"ca_client": True}, "socket-proxy": {}}, + ) + self.assertIs( + verdict, + True, + "a container that only calls the self-signed endpoint has no domain " + "of its own, and without the CA every outbound call fails the " + "handshake", + ) + self.assertEqual([name for name, _ in calls], ["config"]) + + def test_a_ca_client_gets_nothing_when_the_ca_is_not_self_signed(self): + verdict, _ = self._resolve( + domains={}, + enabled=True, + mode="self_signed", + services={"agent-broker": {"ca_client": True}}, + tls_mode="letsencrypt", + ) + self.assertIs(verdict, False) def test_the_term_is_required(self): plugin = self.mod.LookupModule()