Admit independent service SVID roles on Gateway receivers

agentplane/docs/system_overview.md

@@ -326,6 +326,24 @@ Workload SVID and verifies `HISTORY_SERVICE_URI`; the central proxy carries that
 opaquely. History authorizes the read against the exact Identity name and UID in the SVID. This path
 uses no History bearer credential or projected audience token and adds no `reader_accounts` grant.
 
+### Service admission in the current Gateway
+
+The co-hosted receivers can additionally admit exact service SVIDs when configured.
+Their reader, manager, launcher, operator, Executor and proxy grants remain separate.
+The Sessions, Environments and Runtime-proxy roles can move with the planned binary split.
+No caller or production listener configuration switches in this stage.
+
+```mermaid
+flowchart LR
+    app[App service] -->|service mTLS: independent grants| environments[EnvironmentService]
+    actions[Action service] -->|service mTLS + typed Identity envelope| dispatcher[Dispatcher / Sessions role]
+    actions -->|service mTLS + typed launch origin| environments
+    sessions[Sessions service role] -->|service mTLS: session methods| proxy[Runtime proxy role]
+    controller[Sandbox Controller service] -->|service mTLS: controller methods| proxy
+    sampler[Central sampler service] -->|service mTLS: Executor grant| executor[Executor]
+    proxy -->|existing Gateway client SVID until caller cutover| runtime[Native Runtime]
+```
+
 ## Central code-mode sampling (opt in)
 
 ```mermaid

agentplane/runner_gateway/config.go

@@ -35,40 +35,58 @@ const (
 // The database URL is a Secret and the Pod IP comes from the Downward API, so both are environment
 // values. Kubernetes sets the Pod hostname to its name unless an explicit hostname is configured.
 type config struct {
-	RunnerSecurity peertls.Security `mapstructure:"runner_security"`
-	Namespace      string           `mapstructure:"sandbox_namespace"`
+	RuntimeProxyReaders           []string         `mapstructure:"runtime_proxy_readers"`
+	RuntimeProxySessionCallers    []string         `mapstructure:"runtime_proxy_session_callers"`
+	RuntimeProxyControllerCallers []string         `mapstructure:"runtime_proxy_controller_callers"`
+	ExecutorServiceCallers        []string         `mapstructure:"executor_service_callers"`
+	RunnerSecurity                peertls.Security `mapstructure:"runner_security"`
+	Namespace                     string           `mapstructure:"sandbox_namespace"`
 	// PolicyNamespace holds the Identities a workload SVID may name.
-	PolicyNamespace             string             `mapstructure:"policy_namespace"`
-	DispatcherDatabaseURL       string             `mapstructure:"-"`
-	PodName                     string             `mapstructure:"-"`
-	PodIP                       string             `mapstructure:"-"`
-	RunnerTokenFile             string             `mapstructure:"runner_token_file"`
-	Kubeconfig                  string             `mapstructure:"kubeconfig"`
-	KubernetesClientQPS         float32            `mapstructure:"kubernetes_client_qps"`
-	RPC                         listener.Config    `mapstructure:"rpc"`
-	GatewayTLS                  *peertls.ClientTLS `mapstructure:"gateway_tls"`
-	ProxyPort                   int                `mapstructure:"proxy_port"`
-	RunnerControlTargetTemplate string             `mapstructure:"runner_control_target_template"`
-	DispatcherAudience          string             `mapstructure:"dispatcher_audience"`
-	GatewayAudience             string             `mapstructure:"gateway_audience"`
-	EnvironmentReaders          []auth.Account     `mapstructure:"environment_readers"`
-	EnvironmentManagers         []auth.Account     `mapstructure:"environment_managers"`
-	EnvironmentLaunchers        []auth.Account     `mapstructure:"environment_launchers"`
-	DispatcherCallers           []auth.Account     `mapstructure:"dispatcher_callers"`
-	Delegation                  *delegationConfig  `mapstructure:"delegation"`
-	GatewayCallers              []auth.Account     `mapstructure:"gateway_callers"`
-	WakeCallers                 []auth.Account     `mapstructure:"wake_callers"`
-	OperatorCallers             []auth.Account     `mapstructure:"operator_callers"`
-	ControllerAccount           *auth.Account      `mapstructure:"sandbox_controller_account"`
-	QueueCap                    int                `mapstructure:"max_queued_per_session"`
-	WatchLeaseSeconds           int                `mapstructure:"dispatcher_watch_lease_seconds"`
-	RunnerTimeoutSeconds        int                `mapstructure:"runner_request_timeout_seconds"`
+	PolicyNamespace              string                   `mapstructure:"policy_namespace"`
+	DispatcherDatabaseURL        string                   `mapstructure:"-"`
+	PodName                      string                   `mapstructure:"-"`
+	PodIP                        string                   `mapstructure:"-"`
+	RunnerTokenFile              string                   `mapstructure:"runner_token_file"`
+	Kubeconfig                   string                   `mapstructure:"kubeconfig"`
+	KubernetesClientQPS          float32                  `mapstructure:"kubernetes_client_qps"`
+	RPC                          listener.Config          `mapstructure:"rpc"`
+	GatewayTLS                   *peertls.ClientTLS       `mapstructure:"gateway_tls"`
+	ProxyPort                    int                      `mapstructure:"proxy_port"`
+	RunnerControlTargetTemplate  string                   `mapstructure:"runner_control_target_template"`
+	DispatcherAudience           string                   `mapstructure:"dispatcher_audience"`
+	GatewayAudience              string                   `mapstructure:"gateway_audience"`
+	EnvironmentReaders           []auth.Account           `mapstructure:"environment_readers"`
+	EnvironmentManagers          []auth.Account           `mapstructure:"environment_managers"`
+	EnvironmentLaunchers         []auth.Account           `mapstructure:"environment_launchers"`
+	EnvironmentServiceReaders    []string                 `mapstructure:"environment_service_readers"`
+	EnvironmentServiceManagers   []string                 `mapstructure:"environment_service_managers"`
+	EnvironmentServiceLaunchers  []string                 `mapstructure:"environment_service_launchers"`
+	EnvironmentServiceOperators  []string                 `mapstructure:"environment_service_operators"`
+	WakeServiceCallers           []string                 `mapstructure:"wake_service_callers"`
+	DispatcherServiceCallers     []string                 `mapstructure:"dispatcher_service_callers"`
+	DispatcherCallers            []auth.Account           `mapstructure:"dispatcher_callers"`
+	DispatcherServiceDelegation  *serviceDelegationConfig `mapstructure:"dispatcher_service_delegation"`
+	EnvironmentServiceDelegation *serviceDelegationConfig `mapstructure:"environment_service_delegation"`
+	Delegation                   *delegationConfig        `mapstructure:"delegation"`
+	GatewayCallers               []auth.Account           `mapstructure:"gateway_callers"`
+	WakeCallers                  []auth.Account           `mapstructure:"wake_callers"`
+	OperatorCallers              []auth.Account           `mapstructure:"operator_callers"`
+	ControllerAccount            *auth.Account            `mapstructure:"sandbox_controller_account"`
+	QueueCap                     int                      `mapstructure:"max_queued_per_session"`
+	WatchLeaseSeconds            int                      `mapstructure:"dispatcher_watch_lease_seconds"`
+	RunnerTimeoutSeconds         int                      `mapstructure:"runner_request_timeout_seconds"`
 
 	CentralSampler  *dispatcher.CentralTarget `mapstructure:"central_sampler"`
 	ExecutorCallers []auth.Account            `mapstructure:"executor_callers"`
 	RuntimeCatalog  runtimeconfig.Catalog     `mapstructure:"runtime_catalog"`
 }
 
+// serviceDelegationConfig is a separate certificate actor grant scoped to exact methods.
+type serviceDelegationConfig struct {
+	Actors  []auth.ServiceDelegator `mapstructure:"actors"`
+	Methods []string                `mapstructure:"methods"`
+}
+
 // delegationConfig is an independent trust grant. It does not inherit the direct caller or wake
 // allowlists, and its method list contains exact full gRPC method names served by this process.
 type delegationConfig struct {
@@ -139,6 +157,76 @@ func loadConfig() (config, error) {
 }
 
 func (c *config) validate() error {
+	serviceCallers := map[string][]string{
+		"dispatcher_service_callers":       c.DispatcherServiceCallers,
+		"runtime_proxy_readers":            c.RuntimeProxyReaders,
+		"runtime_proxy_session_callers":    c.RuntimeProxySessionCallers,
+		"runtime_proxy_controller_callers": c.RuntimeProxyControllerCallers,
+		"executor_service_callers":         c.ExecutorServiceCallers,
+		"environment_service_readers":      c.EnvironmentServiceReaders,
+		"environment_service_managers":     c.EnvironmentServiceManagers,
+		"environment_service_launchers":    c.EnvironmentServiceLaunchers,
+		"environment_service_operators":    c.EnvironmentServiceOperators,
+		"wake_service_callers":             c.WakeServiceCallers,
+	}
+	for name, callers := range serviceCallers {
+		if err := auth.ValidateServiceCallers(callers); err != nil {
+			return fmt.Errorf("%s: %w", name, err)
+		}
+	}
+	if hasServiceCallers(serviceCallers) || c.DispatcherServiceDelegation != nil || c.EnvironmentServiceDelegation != nil {
+		if !c.RPC.Has(listener.SPIFFE) {
+			return errors.New("service caller settings require rpc.auth to include spiffe")
+		}
+	}
+	if c.DispatcherServiceDelegation != nil {
+		if len(c.DispatcherServiceDelegation.Actors) == 0 || len(c.DispatcherServiceDelegation.Methods) == 0 {
+			return errors.New("dispatcher_service_delegation requires actors and methods")
+		}
+		dispatcherMethods := dispatcher.Methods(auth.Rule{})
+		for _, method := range c.DispatcherServiceDelegation.Methods {
+			if _, ok := dispatcherMethods[method]; !ok {
+				return fmt.Errorf("dispatcher_service_delegation method %q must belong to SessionDispatcher", method)
+			}
+		}
+		var actors []string
+		for _, actor := range c.DispatcherServiceDelegation.Actors {
+			if actor.ConnectionProvenance {
+				return errors.New("Dispatcher service delegation requires Action provenance")
+			}
+			actors = append(actors, actor.SPIFFEID)
+		}
+		if err := auth.ValidateServiceCallers(actors); err != nil {
+			return err
+		}
+	}
+
+	if c.EnvironmentServiceDelegation != nil {
+		if len(c.EnvironmentServiceDelegation.Actors) == 0 || len(c.EnvironmentServiceDelegation.Methods) == 0 {
+			return errors.New("environment_service_delegation requires actors and methods")
+		}
+		allowed := map[string]struct{}{
+			"/agentplane.environments.v1.EnvironmentService/ListEnvironments":  {},
+			"/agentplane.environments.v1.EnvironmentService/WatchEnvironments": {},
+			"/agentplane.environments.v1.EnvironmentService/GetEnvironment":    {},
+			"/agentplane.environments.v1.EnvironmentService/WatchEnvironment":  {},
+		}
+		for _, method := range c.EnvironmentServiceDelegation.Methods {
+			if _, ok := allowed[method]; !ok {
+				return fmt.Errorf("environment_service_delegation method %q is not a read-only EnvironmentService method", method)
+			}
+		}
+		var actors []string
+		for _, actor := range c.EnvironmentServiceDelegation.Actors {
+			if actor.ConnectionProvenance {
+				return errors.New("EnvironmentService delegation does not admit Connection provenance")
+			}
+			actors = append(actors, actor.SPIFFEID)
+		}
+		if err := auth.ValidateServiceCallers(actors); err != nil {
+			return err
+		}
+	}
 	for _, required := range []struct{ name, value string }{
 		{"sandbox_namespace", c.Namespace},
 		{"policy_namespace", c.PolicyNamespace},
@@ -229,6 +317,11 @@ func (c *config) validate() error {
 			return fmt.Errorf("wake caller %s/%s is not a Dispatcher caller", account.Namespace, account.Name)
 		}
 	}
+	for _, service := range c.WakeServiceCallers {
+		if !slices.Contains(c.DispatcherServiceCallers, service) {
+			return fmt.Errorf("wake service caller %s is not a Dispatcher service caller", service)
+		}
+	}
 	for _, account := range c.OperatorCallers {
 		if !slices.Contains(c.DispatcherCallers, account) {
 			return fmt.Errorf("operator caller %s/%s is not a Dispatcher caller", account.Namespace, account.Name)
@@ -249,6 +342,15 @@ func (c *config) validate() error {
 	return nil
 }
 
+func hasServiceCallers(callers map[string][]string) bool {
+	for _, values := range callers {
+		if len(values) > 0 {
+			return true
+		}
+	}
+	return false
+}
+
 func (c config) rpcBearerAudiences() []string {
 	for _, method := range c.RPC.Auth {
 		if bearer, ok := method.(*listener.KubernetesBearerAuth); ok && bearer != nil {

agentplane/runner_gateway/config_test.go

@@ -148,6 +148,98 @@ func TestSharedListenerRequiresSVIDTLSAndBothBearerAudiences(t *testing.T) {
 	}
 }
 
+func TestServiceCallerSettingsRequireSPIFFEListener(t *testing.T) {
+	withoutSPIFFE := strings.Replace(requiredFile, "    - {mode: spiffe, ca_file: /tmp/workload-ca}\n", "", 1)
+	for name, settings := range map[string]string{
+		"runtime proxy reader":            "runtime_proxy_readers: [\"spiffe://agentplane/actions\"]\n",
+		"runtime proxy session caller":    "runtime_proxy_session_callers: [\"spiffe://agentplane/sessions\"]\n",
+		"runtime proxy controller caller": "runtime_proxy_controller_callers: [\"spiffe://agentplane/sandbox-controller\"]\n",
+		"executor caller":                 "executor_service_callers: [\"spiffe://agentplane/actions\"]\n",
+		"dispatcher caller":               "dispatcher_service_callers: [\"spiffe://agentplane/actions\"]\n",
+		"environment reader":              "environment_service_readers: [\"spiffe://agentplane/actions\"]\n",
+		"environment manager":             "environment_service_managers: [\"spiffe://agentplane/actions\"]\n",
+		"environment launcher":            "environment_service_launchers: [\"spiffe://agentplane/actions\"]\n",
+		"environment operator":            "environment_service_operators: [\"spiffe://agentplane/actions\"]\n",
+		"wake caller":                     "wake_service_callers: [\"spiffe://agentplane/actions\"]\n",
+		"Dispatcher delegation": `dispatcher_service_delegation:
+  actors: [{spiffe_id: "spiffe://agentplane/actions"}]
+  methods: [/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput]
+`,
+		"EnvironmentService delegation": `environment_service_delegation:
+  actors: [{spiffe_id: "spiffe://agentplane/actions"}]
+  methods: [/agentplane.environments.v1.EnvironmentService/ListEnvironments]
+`,
+	} {
+		t.Run(name, func(t *testing.T) {
+			settings := configtest.WriteSettings(t, withoutSPIFFE+settings)
+			_, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+settings))
+			if err == nil || !strings.Contains(err.Error(), "service caller settings require rpc.auth to include spiffe") {
+				t.Fatalf("loadConfig error = %v, want missing-SPIFFE service-caller error", err)
+			}
+		})
+	}
+}
+
+func TestDispatcherServiceDelegationIsScopedToServiceMethodsAndActionActors(t *testing.T) {
+	base := `dispatcher_service_delegation:
+  actors: [{spiffe_id: "spiffe://agentplane/actions"}]
+  methods: [/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput]
+`
+	path := configtest.WriteSettings(t, requiredFile+base)
+	c, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+path))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.DispatcherServiceDelegation == nil || len(c.DispatcherServiceDelegation.Actors) != 1 ||
+		c.DispatcherServiceDelegation.Actors[0].SPIFFEID != peertls.ActionsURI ||
+		!slices.Equal(c.DispatcherServiceDelegation.Methods, []string{"/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput"}) {
+		t.Fatalf("Dispatcher service delegation = %+v", c.DispatcherServiceDelegation)
+	}
+
+	for name, settings := range map[string]string{
+		"EnvironmentService method": strings.Replace(base, "/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput", "/agentplane.environments.v1.EnvironmentService/ListEnvironments", 1),
+		"unknown method":            strings.Replace(base, "/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput", "/agentplane.runner_gateway.dispatcher.SessionDispatcher/NotAnRPC", 1),
+		"Connection provenance":     strings.Replace(base, `spiffe_id: "spiffe://agentplane/actions"`, `spiffe_id: "spiffe://agentplane/actions", connection_provenance: true`, 1),
+		"Identity URI actor":        strings.Replace(base, "spiffe://agentplane/actions", "spiffe://agentplane/test/sandbox/credential-id", 1),
+	} {
+		t.Run(name, func(t *testing.T) {
+			path := configtest.WriteSettings(t, requiredFile+settings)
+			if _, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+path)); err == nil {
+				t.Fatal("invalid Dispatcher service delegation was accepted")
+			}
+		})
+	}
+}
+
+func TestEnvironmentServiceDelegationIsReadOnlyAndDoesNotAdmitConnectionProvenance(t *testing.T) {
+	base := `environment_service_delegation:
+  actors: [{spiffe_id: "spiffe://agentplane/actions"}]
+  methods: [/agentplane.environments.v1.EnvironmentService/ListEnvironments]
+`
+	path := configtest.WriteSettings(t, requiredFile+base)
+	c, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+path))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if c.EnvironmentServiceDelegation == nil || len(c.EnvironmentServiceDelegation.Actors) != 1 ||
+		c.EnvironmentServiceDelegation.Actors[0].SPIFFEID != peertls.ActionsURI {
+		t.Fatalf("EnvironmentService service delegation = %+v", c.EnvironmentServiceDelegation)
+	}
+
+	for name, settings := range map[string]string{
+		"mutation method":       strings.Replace(base, "/agentplane.environments.v1.EnvironmentService/ListEnvironments", "/agentplane.environments.v1.EnvironmentService/CreateEnvironment", 1),
+		"Connection provenance": strings.Replace(base, `spiffe_id: "spiffe://agentplane/actions"`, `spiffe_id: "spiffe://agentplane/actions", connection_provenance: true`, 1),
+		"Identity URI actor":    strings.Replace(base, "spiffe://agentplane/actions", "spiffe://agentplane/test/sandbox/credential-id", 1),
+	} {
+		t.Run(name, func(t *testing.T) {
+			path := configtest.WriteSettings(t, requiredFile+settings)
+			if _, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+path)); err == nil {
+				t.Fatal("invalid EnvironmentService delegation was accepted")
+			}
+		})
+	}
+}
+
 func TestOperatorCallersMustAlsoBeAuthenticatedDispatcherCallers(t *testing.T) {
 	path := configtest.WriteSettings(t, requiredFile+"operator_callers: [agentplane/unknown]\n")
 	if _, err := configtest.Load(t, settingPrefix, loadConfig, required(t, settingPrefix+"CONFIG_FILE="+path)); err == nil {

agentplane/runner_gateway/dispatcher/BUILD.bazel

@@ -10,6 +10,7 @@ go_library(
         "delivery_wake.go",
         "ownership.go",
         "runner_client.go",
+        "service_attribution.go",
         "snapshots.go",
         "store.go",
         "terminal.go",
@@ -115,6 +116,7 @@ go_test(
         "execution_identity_integration_test.go",
         "identity_change_test.go",
         "placement_test.go",
+        "service_admission_integration_test.go",
         "store_integration_test.go",
         "store_integration_testmain_test.go",
         "wake_integration_test.go",