Provision and mount role-specific service SVIDs before caller cutover

agentplane/docs/system_overview.md

@@ -182,6 +182,14 @@ through the existing Kyverno injection after a live check that the record has ac
 launcher's VMI for the VM's credential. Staging
 and testing use the SVID hop; TLS mode never falls back to token mode.
 
+The Gateway mounts separate caller-role SVIDs for Sessions (`spiffe://agentplane/sessions`), Runtime
+proxy (`spiffe://agentplane/runtime-proxy`) and Environments (`spiffe://agentplane/environments`). Its
+current listener keeps the `spiffe://agentplane/session-dispatcher` server URI until the API split; the
+`spiffe://agentplane/runner-gateway` Runtime-control client certificate remains separate. These mounts
+do not enable new traffic. In the planned binary split, Sessions takes the Dispatcher and its queue;
+Environments takes the Sandbox Controller, EnvironmentService, Runtime proxy and Executor. The
+Sessions-to-Runtime-proxy hop remains a network edge with its own exact caller grant.
+
 Firecracker Sandboxes use a controller-owned launcher Pod, a separate SVID-authenticated relay and a
 NixOS guest. Their supervisor checkpoints RAM and all disks, then restores the same Runtime and
 processes in the same Pod. Its launcher serves SPIFFE Workload API credentials over vsock to the

agentplane/net/peertls/config.py

@@ -7,11 +7,16 @@
 from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 ACTIONS_URI: Final = "spiffe://agentplane/actions"
+APP_URI: Final = "spiffe://agentplane/app"
 POLICY_CONTROLLER_URI: Final = "spiffe://agentplane/policy-controller"
 CENTRAL_SAMPLER_URI: Final = "spiffe://agentplane/central-sampler"
 EGRESS_PROXY_URI: Final = "spiffe://agentplane/egress-proxy"
+GATEWAY_SESSIONS_URI: Final = "spiffe://agentplane/sessions"
+GATEWAY_RUNTIME_PROXY_URI: Final = "spiffe://agentplane/runtime-proxy"
+GATEWAY_ENVIRONMENTS_URI: Final = "spiffe://agentplane/environments"
 HISTORY_SERVICE_URI: Final = "spiffe://agentplane/history-ingress"
 LLM_INGRESS_URI: Final = "spiffe://agentplane/llm-ingress"
+MCP_EDGE_URI: Final = "spiffe://agentplane/mcp-edge"
 NOTIFICATIONS_URI: Final = "spiffe://agentplane/notifications"
 SESSION_DISPATCHER_URI: Final = "spiffe://agentplane/session-dispatcher"
 SANDBOX_CONTROLLER_URI: Final = "spiffe://agentplane/sandbox-controller"

cluster/cdk8s/agentplane/BUILD.bazel

@@ -221,6 +221,8 @@ py_library(
     deps = [
         ":environment",
         ":service_workload",
+        ":workload_tls",
+        "//agentplane/net/peertls:config",
         "//agentplane/policy_controller:settings",
         "//cluster/cdk8s:cilium",
         "//cluster/cdk8s:config_format",
@@ -986,6 +988,7 @@ py_test(
         ":conftest",
         ":egress",
         "//:conftest",
+        "//agentplane/net/peertls:config",
         "//cluster/cdk8s:conftest",
         "//cluster/cdk8s/cert_manager:cluster_ca",
         "@pypi//cdk8s",

cluster/cdk8s/agentplane/actions.py

@@ -392,7 +392,7 @@ def __init__(
             )
         settings_name = "actions-settings"
         workload = self._add_workload(settings_name)
-        serving_tls = workload_tls.serving_certificate(
+        serving_tls = workload_tls.service_certificate(
             self, env, container=workload.container, name=NAME, uri=ACTIONS_URI
         )
         assert serving_tls.cert_file is not None

cluster/cdk8s/agentplane/app.py

@@ -30,6 +30,7 @@
 from agentplane.crd_group import GROUP
 from agentplane.net.peertls.config import (
     ACTIONS_URI,
+    APP_URI,
     EGRESS_PROXY_URI,
     HISTORY_SERVICE_URI,
     NOTIFICATIONS_URI,
@@ -244,7 +245,7 @@ def _add_workload(self, config: SettingsFile) -> ServiceWorkload:
             },
         )
         config.mount_into(workload.container, env=Settings.config_file_env)
-        workload_tls.mount_trust_bundle(self, workload.container)
+        workload_tls.service_certificate(self, self.env, container=workload.container, name=NAME, uri=APP_URI)
         public_roots = ConfigMap.from_config_map_name(self, "public-web-pki-ref", "public-web-pki")
         workload.container.mount(
             "/var/run/public-web-pki",

cluster/cdk8s/agentplane/central_sampler_workload.py

@@ -21,7 +21,7 @@
 
 from agentplane.central_sampler.settings import Settings
 from agentplane.identity.auth.audience import WORKLOAD_TOKEN_AUDIENCE
-from agentplane.net.peertls.config import HISTORY_SERVICE_URI, SESSION_DISPATCHER_URI
+from agentplane.net.peertls.config import CENTRAL_SAMPLER_URI, HISTORY_SERVICE_URI, SESSION_DISPATCHER_URI
 from agentplane.session_instructions.platform_instructions import SessionRuntime
 from cluster.cdk8s import cilium, node_scheduling
 from cluster.cdk8s.agentplane import (
@@ -147,7 +147,9 @@ def __init__(
         )
         workload.container.mount(str(Path(_CONFIG_PATH).parent), settings_volume, read_only=True)
         workload.container.mount(_LOGS_PATH, Volume.from_empty_dir(self, "logs-volume", "logs"))
-        workload_tls.mount_trust_bundle(self, workload.container)
+        workload_tls.service_certificate(
+            self, env, container=workload.container, name=central_sampler.NAME, uri=CENTRAL_SAMPLER_URI
+        )
         public_roots = ConfigMap.from_config_map_name(self, "public-web-pki-ref", "public-web-pki")
         workload.container.mount(
             "/var/run/public-web-pki",

cluster/cdk8s/agentplane/egress.py

@@ -542,7 +542,11 @@ def __init__(
             duration="24h",
             renew_before="8h",
             private_key=CertificatePrivateKey.ecdsa_p256(CertificateSpecPrivateKeyRotationPolicy.ALWAYS),
-            usages=[CertificateSpecUsages.DIGITAL_SIGNATURE, CertificateSpecUsages.SERVER_AUTH],
+            usages=[
+                CertificateSpecUsages.DIGITAL_SIGNATURE,
+                CertificateSpecUsages.CLIENT_AUTH,
+                CertificateSpecUsages.SERVER_AUTH,
+            ],
             issuer_ref=cluster_ca.WORKLOAD_ISSUER,
         )
 

cluster/cdk8s/agentplane/history_service.py

@@ -181,7 +181,7 @@ def __init__(
             telemetry_enabled=True,
             registry_secret_name=registry_secret_name,
         )
-        serving_tls = workload_tls.serving_certificate(
+        serving_tls = workload_tls.service_certificate(
             self, env, container=workload.container, name=NAME, uri=HISTORY_SERVICE_URI
         )
         assert serving_tls.cert_file is not None

cluster/cdk8s/agentplane/llm_ingress.py

@@ -194,7 +194,7 @@ def __init__(
                 Volume.from_config_map(self, "codex-models-volume", catalog_map),
                 read_only=True,
             )
-        serving_tls = workload_tls.serving_certificate(
+        serving_tls = workload_tls.service_certificate(
             self, env, container=workload.container, name=_NAME, uri=LLM_INGRESS_URI
         )
         assert serving_tls.cert_file is not None

cluster/cdk8s/agentplane/mcp_edge.py

@@ -27,7 +27,7 @@
 from agentplane.crd_group import GROUP
 from agentplane.identity.auth.audience import WORKLOAD_TOKEN_AUDIENCE
 from agentplane.mcp_edge.settings import Settings
-from agentplane.net.peertls.config import ACTIONS_URI
+from agentplane.net.peertls.config import ACTIONS_URI, MCP_EDGE_URI
 from cluster.cdk8s import cilium, node_scheduling
 from cluster.cdk8s.agentplane import actions, app as app_component, database, workload_tls
 from cluster.cdk8s.agentplane.environment import Environment
@@ -149,7 +149,7 @@ def __init__(
             )
             workload.container.mount(_SECRET_DIR, oauth_volume, read_only=True)
         settings.mount_into(workload.container, env=Settings.config_file_env)
-        workload_tls.mount_trust_bundle(self, workload.container)
+        workload_tls.service_certificate(self, env, container=workload.container, name=NAME, uri=MCP_EDGE_URI)
         if props.oauth_secret_name is not None:
             workload.project_tokens(
                 "actions-token", "/var/run/secrets/agentplane-mcp", {"actions": WORKLOAD_TOKEN_AUDIENCE}

cluster/cdk8s/agentplane/notifications.py

@@ -156,7 +156,7 @@ def __init__(
             telemetry_enabled=True,
             registry_secret_name=registry_secret_name,
         )
-        serving_tls = workload_tls.serving_certificate(
+        serving_tls = workload_tls.service_certificate(
             self, env, container=workload.container, name=NAME, uri=NOTIFICATIONS_URI
         )
         assert serving_tls.cert_file is not None

cluster/cdk8s/agentplane/policy_controller.py

@@ -12,8 +12,10 @@
 from constructs import Construct
 from prometheus_operator_crds.com.coreos.monitoring import ServiceMonitorSpecSelector
 
+from agentplane.net.peertls.config import POLICY_CONTROLLER_URI
 from agentplane.policy_controller.settings import Settings
 from cluster.cdk8s import cilium, node_scheduling
+from cluster.cdk8s.agentplane import workload_tls
 from cluster.cdk8s.agentplane.environment import Environment
 from cluster.cdk8s.agentplane.service_workload import add_service_workload
 from cluster.cdk8s.cert_manager.cluster_ca import WORKLOAD_ISSUER
@@ -111,6 +113,7 @@ def __init__(
             telemetry_enabled=True,
             registry_secret_name=registry_secret_name,
         )
+        workload_tls.service_certificate(self, env, container=workload.container, name=NAME, uri=POLICY_CONTROLLER_URI)
         settings_file.mount_into(workload.container, env=Settings.config_file_env)
         self.account = workload.account
         token_reviewer_cluster_rbac(

cluster/cdk8s/agentplane/runner_gateway_workload.py

@@ -32,7 +32,14 @@
 
 from agentplane.crd_group import GROUP
 from agentplane.identity.auth import listener as workload_listener
-from agentplane.net.peertls.config import MTLS, SESSION_DISPATCHER_URI, TLS
+from agentplane.net.peertls.config import (
+    GATEWAY_ENVIRONMENTS_URI,
+    GATEWAY_RUNTIME_PROXY_URI,
+    GATEWAY_SESSIONS_URI,
+    MTLS,
+    SESSION_DISPATCHER_URI,
+    TLS,
+)
 from agentplane.runner_gateway.settings import CentralSamplerTarget, DelegationSettings, Settings
 from agentplane.subjects import ServiceAccountRef
 from cluster.cdk8s import cilium, node_scheduling
@@ -224,9 +231,26 @@ def __init__(
             registry_secret_name=registry_secret_name,
             placement=placement,
         )
-        serving_tls = workload_tls.serving_certificate(
+        serving_tls = workload_tls.service_certificate(
             self, env, container=workload.container, name=runner_gateway.NAME, uri=SESSION_DISPATCHER_URI
         )
+        workload_tls.client_certificate(
+            self, env, container=workload.container, name=f"{runner_gateway.NAME}-sessions", uri=GATEWAY_SESSIONS_URI
+        )
+        workload_tls.client_certificate(
+            self,
+            env,
+            container=workload.container,
+            name=f"{runner_gateway.NAME}-runtime-proxy",
+            uri=GATEWAY_RUNTIME_PROXY_URI,
+        )
+        workload_tls.client_certificate(
+            self,
+            env,
+            container=workload.container,
+            name=f"{runner_gateway.NAME}-environments",
+            uri=GATEWAY_ENVIRONMENTS_URI,
+        )
         assert serving_tls.cert_file is not None
         assert serving_tls.key_file is not None
         assert serving_tls.ca_file is not None

cluster/cdk8s/agentplane/test_runner_gateway_workload.py

@@ -9,7 +9,12 @@
 from cdk8s import App, Chart, Testing as Cdk8sTesting
 from more_itertools import one
 
-from agentplane.net.peertls.config import SESSION_DISPATCHER_URI
+from agentplane.net.peertls.config import (
+    GATEWAY_ENVIRONMENTS_URI,
+    GATEWAY_RUNTIME_PROXY_URI,
+    GATEWAY_SESSIONS_URI,
+    SESSION_DISPATCHER_URI,
+)
 from agentplane.runner_gateway.settings import Settings
 from agentplane.subjects import ServiceAccountRef
 from cluster.cdk8s.agentplane import (
@@ -76,9 +81,23 @@ def test_environment_provisions_the_gateway_identity_from_the_workload_issuer(
     assert workload_spec["secretName"] == f"{runner_gateway.NAME}-workload-tls"
     assert workload_spec["issuerRef"] == workload_issuer
     assert workload_spec["uris"] == [SESSION_DISPATCHER_URI]
-    assert set(workload_spec["usages"]) == {"digital signature", "server auth"}
+    assert set(workload_spec["usages"]) == {"digital signature", "client auth", "server auth"}
     assert workload_spec["privateKey"]["rotationPolicy"] == "Always"
 
+    for name, uri in (
+        ("sessions", GATEWAY_SESSIONS_URI),
+        ("runtime-proxy", GATEWAY_RUNTIME_PROXY_URI),
+        ("environments", GATEWAY_ENVIRONMENTS_URI),
+    ):
+        client_spec = _resource(staging_manifests, "Certificate", f"{runner_gateway.NAME}-{name}-client-workload")[
+            "spec"
+        ]
+        assert client_spec["secretName"] == f"{runner_gateway.NAME}-{name}-client-workload-tls"
+        assert client_spec["issuerRef"] == workload_issuer
+        assert client_spec["uris"] == [uri]
+        assert set(client_spec["usages"]) == {"digital signature", "client auth", "server auth"}
+        assert client_spec["privateKey"]["rotationPolicy"] == "Always"
+
 
 def test_gateway_preserves_services_and_fences_listener_ingress(scratch_app: App) -> None:
     docs = _gateway_objects(scratch_app)
@@ -288,6 +307,12 @@ def test_the_settings_file_the_gateway_reads_is_mounted_beside_its_token(scratch
     # Runners and the Gateway's own leaf come from the workload CA, so it verifies them alone.
     assert trust_source["name"] == "agentplane-workload-ca-bundle"
     assert trust_source["items"] == [{"key": "ca-certificates.crt", "path": "ca.crt"}]
+    for name in ("sessions", "runtime-proxy", "environments"):
+        mount_path = f"/run/agentplane/{runner_gateway.NAME}-{name}-client-tls"
+        client_certificate_volume = volumes[mounted[mount_path]]
+        assert client_certificate_volume["secret"]["secretName"] == (
+            f"{runner_gateway.NAME}-{name}-client-workload-tls"
+        )
     assert volumes[mounted[str(path.parent)]]["configMap"]["name"] == settings_map["metadata"]["name"]
     assert path.name in settings_map["data"]
     assert "projected" in volumes[mounted["/var/run/secrets/agentplane-runner-gateway"]]

cluster/cdk8s/agentplane/test_sandbox_controller.py

@@ -46,7 +46,7 @@ def test_controller_settings_are_mounted_and_gateway_token_is_projected(
 
     certificate = _resource(staging_manifests, "Certificate", f"{sandbox_controller.NAME}-client-workload")
     assert certificate["spec"]["uris"] == [SANDBOX_CONTROLLER_URI]
-    assert certificate["spec"]["usages"] == ["digital signature", "client auth"]
+    assert certificate["spec"]["usages"] == ["digital signature", "client auth", "server auth"]
     cert_mount = one(
         mount
         for mount in container["volumeMounts"]

cluster/cdk8s/agentplane/test_workload_trust.py

@@ -14,6 +14,22 @@
 from more_itertools import one
 
 from agentplane.egress import sidecar_config
+from agentplane.net.peertls.config import (
+    ACTIONS_URI,
+    APP_URI,
+    CENTRAL_SAMPLER_URI,
+    EGRESS_PROXY_URI,
+    GATEWAY_ENVIRONMENTS_URI,
+    GATEWAY_RUNTIME_PROXY_URI,
+    GATEWAY_SESSIONS_URI,
+    HISTORY_SERVICE_URI,
+    LLM_INGRESS_URI,
+    MCP_EDGE_URI,
+    NOTIFICATIONS_URI,
+    POLICY_CONTROLLER_URI,
+    SANDBOX_CONTROLLER_URI,
+    SESSION_DISPATCHER_URI,
+)
 from cluster.cdk8s.agentplane import egress, runner_gateway, staging
 from cluster.cdk8s.agentplane.sandbox_pod import SIDECAR_SETTINGS_VOLUME_NAME
 from cluster.cdk8s.cert_manager import cluster_ca
@@ -167,6 +183,73 @@ def test_agentplane_svids_come_from_the_workload_ca(staging_manifests: list[Body
     } == set()
 
 
+def test_service_svids_are_mounted_with_client_and_server_usage(staging_manifests: list[Body]) -> None:
+    service_uris = {
+        ACTIONS_URI,
+        APP_URI,
+        CENTRAL_SAMPLER_URI,
+        EGRESS_PROXY_URI,
+        GATEWAY_ENVIRONMENTS_URI,
+        GATEWAY_RUNTIME_PROXY_URI,
+        GATEWAY_SESSIONS_URI,
+        HISTORY_SERVICE_URI,
+        LLM_INGRESS_URI,
+        MCP_EDGE_URI,
+        NOTIFICATIONS_URI,
+        POLICY_CONTROLLER_URI,
+        SANDBOX_CONTROLLER_URI,
+        SESSION_DISPATCHER_URI,
+    }
+    certificates = {
+        uri: doc
+        for doc in staging_manifests
+        if doc["kind"] == "Certificate"
+        for uri in doc["spec"].get("uris", [])
+        if uri in service_uris
+    }
+    assert set(certificates) == service_uris
+
+    deployments = {
+        doc["metadata"]["name"]: doc["spec"]["template"]["spec"]
+        for doc in staging_manifests
+        if doc["kind"] == "Deployment"
+    }
+    for uri, certificate in certificates.items():
+        assert {"digital signature", "client auth", "server auth"} <= set(certificate["spec"]["usages"])
+        secret_name = certificate["spec"]["secretName"]
+        assert any(
+            volume.get("secret", {}).get("secretName") == secret_name
+            and any(
+                mount["name"] == volume["name"] and mount["readOnly"]
+                for container in pod["containers"]
+                for mount in container.get("volumeMounts", [])
+            )
+            for pod in deployments.values()
+            for volume in pod.get("volumes", [])
+        ), f"{uri} Secret {secret_name} is not mounted read-only in a service Deployment"
+
+    gateway = deployments[runner_gateway.NAME]
+    caller_secrets = [
+        certificates[uri]["spec"]["secretName"]
+        for uri in (GATEWAY_SESSIONS_URI, GATEWAY_RUNTIME_PROXY_URI, GATEWAY_ENVIRONMENTS_URI)
+    ]
+    assert len(set(caller_secrets)) == len(caller_secrets)
+    caller_mounts = []
+    for secret_name in caller_secrets:
+        volume = one(
+            volume for volume in gateway["volumes"] if volume.get("secret", {}).get("secretName") == secret_name
+        )
+        caller_mounts.append(
+            one(
+                mount
+                for container in gateway["containers"]
+                for mount in container.get("volumeMounts", [])
+                if mount["name"] == volume["name"] and mount["readOnly"]
+            )
+        )
+    assert len({mount["mountPath"] for mount in caller_mounts}) == len(caller_mounts)
+
+
 def test_the_workload_ca_chains_to_the_cluster_root_and_its_bundle_carries_it_alone(scratch_app: App) -> None:
     documents = cdk8s.Testing.synth(cluster_ca.chart(scratch_app))
     issuer = _resource(documents, "ClusterIssuer", cluster_ca.WORKLOAD_ISSUER.name)