fix(ca): give the CA to a service that only calls the self-signed endpoint

ca_injected asked three questions, the first being whether the application
has a domain. That decides whether it is SERVED over the self-signed CA. An
application that only CALLS one has no domain of its own, so the predicate
answered false, sys-svc-compose-ca skipped its stack, and the container ran
with an empty trust store.

svc-ai-agent-broker is such a caller: it authorizes every request against
Keycloak group paths over the public HTTPS base. Without the CA every
outbound call died in the TLS handshake, the broker answered 503 "access
check unavailable", Open WebUI turned that into HTTP 400, and the spec
failed on a model it had just listed. Its container was the only one in the
rescue artifact without SSL_CERT_FILE, while litellm and open-webui carried
it.

A service may now declare ca_client: true, and the predicate then answers on
the deployment's TLS mode alone, because the domain question is meaningless
for a client. The three served-side conditions are untouched, so no
application that already received the CA stops receiving it.

The alternative of pointing the broker at an internal Keycloak URL was
rejected: the role declares no keycloak service, so the URL would render
without a route.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 5 committed 2026-09-24 13:52:12 +02:00
1 parent 8e535f2ccf
commit 1254e27c2d
3 files changed
+58 -3

No files matched your search

+17
View File
@@ -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"])):
@@ -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"
@@ -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()