Admit service certificates and typed MCP actors at Actions

agentplane/action_service/README.md

@@ -496,6 +496,21 @@ the service its own `actions` database and login role on the namespace's shared
 CNPG cluster `postgres` (`cluster/k8s/agentplane-staging/agentplane-staging.k8s.yaml`),
 separate from the integration app's database.
 
+### Service certificate admission during the cutover
+
+Optional `service_readers` grants exact service SPIFFE IDs the existing read/catalog methods.
+These principals remain read-only and carry no Kubernetes account or Pod proof. `operator_services`
+names transport actors allowed to carry a separately verified operator OIDC token; a certificate
+alone grants no operator authority, and a Kubernetes bearer cannot supply that proof.
+`connection_delegation_service` names the exact MCP edge certificate actor for the existing
+Connection delegation methods. The live effective Identity's PolicyBindings remain its only Action
+grants. An actor's own bindings or reader role never become the effective caller's grants.
+
+All three settings require the listener's SPIFFE client trust. They default to no service grants;
+production settings follow receiver image pins. Existing bearer admission remains during the staged
+cutover. Mixed service/bearer credentials fail closed except the explicit service-plus-operator-OIDC
+path. The listener verifies every presented certificate before method authorization.
+
 ### Schema initialization
 
 Goose version 7 creates the final typed schema on an empty database. A database that completed

agentplane/action_service/cmd/BUILD.bazel

@@ -139,6 +139,7 @@ go_test(
         "//agentplane/policy/api/v1alpha1",
         "//agentplane/postgres/probes",
         "//agentplane/postgres/wakeups",
+        "//agentplane/proto:delegation_go_proto",
         "//agentplane/proto:identity_go_proto",
         "//agentplane/testing/kubeapiserver",
         "//agentplane/testing/postgres",

agentplane/action_service/cmd/config.go

@@ -30,8 +30,11 @@ const (
 // config holds the required YAML settings plus the explicitly named runtime Secret inputs.
 // DatabaseURL and the Web Push private key are never read from the settings file.
 type config struct {
-	DatabaseURL string          `mapstructure:"-"`
-	RPC         listener.Config `mapstructure:"rpc"`
+	ServiceReaders              []string        `mapstructure:"service_readers"`
+	OperatorServices            []string        `mapstructure:"operator_services"`
+	ConnectionDelegationService string          `mapstructure:"connection_delegation_service"`
+	DatabaseURL                 string          `mapstructure:"-"`
+	RPC                         listener.Config `mapstructure:"rpc"`
 	// CallerServiceAccountNamespaces hold the ServiceAccounts whose workloads may call; a labeled account outside
 	// them is refused.
 	CallerServiceAccountNamespaces []string `mapstructure:"caller_service_account_namespaces"`
@@ -151,6 +154,19 @@ func (c *config) validate() error {
 			return errors.New(required.name + " is required")
 		}
 	}
+	if (len(c.ServiceReaders)+len(c.OperatorServices) > 0 || c.ConnectionDelegationService != "") && !c.RPC.Has(listener.SPIFFE) {
+		return errors.New("service callers require rpc spiffe client trust")
+	}
+	for _, services := range [][]string{c.ServiceReaders, c.OperatorServices} {
+		if err := auth.ValidateServiceCallers(services); err != nil {
+			return err
+		}
+	}
+	if c.ConnectionDelegationService != "" {
+		if err := auth.ValidateServiceCallers([]string{c.ConnectionDelegationService}); err != nil {
+			return err
+		}
+	}
 	for _, account := range c.ReaderAccounts {
 		if err := new(auth.Account).UnmarshalText([]byte(account.Namespace + "/" + account.Name)); err != nil {
 			return errors.New("reader_accounts requires namespace and name")

agentplane/action_service/cmd/main.go

@@ -337,8 +337,8 @@ func gate(c config, d deps, callers rpc.Callers) (*rpc.Gate, error) {
 	if err != nil {
 		return nil, err
 	}
-	options := rpc.GateOptions{Workloads: workloads, Callers: callers, ReaderAccounts: c.ReaderAccounts}
-	if c.ConnectionDelegationActor != nil {
+	options := rpc.GateOptions{Workloads: workloads, Callers: callers, ReaderAccounts: c.ReaderAccounts, ServiceReaders: c.ServiceReaders, OperatorServices: c.OperatorServices}
+	if c.ConnectionDelegationActor != nil || c.ConnectionDelegationService != "" {
 		scheme := runtime.NewScheme()
 		if err := policyapi.AddToScheme(scheme); err != nil {
 			return nil, fmt.Errorf("register Policy API for delegated Identity checks: %w", err)
@@ -351,7 +351,7 @@ func gate(c config, d deps, callers rpc.Callers) (*rpc.Gate, error) {
 		if err != nil {
 			return nil, fmt.Errorf("configure delegated Identity checks: %w", err)
 		}
-		options.ConnectionDelegation, err = actionConnectionDelegation(workloads, *c.ConnectionDelegationActor, identities)
+		options.ConnectionDelegation, err = actionConnectionDelegation(workloads, c.ConnectionDelegationActor, c.ConnectionDelegationService, identities)
 		if err != nil {
 			return nil, fmt.Errorf("configure MCP Connection delegation: %w", err)
 		}
@@ -374,7 +374,7 @@ func gate(c config, d deps, callers rpc.Callers) (*rpc.Gate, error) {
 // actionConnectionDelegation grants the MCP edge only the Action calls needed to serve a Connection. The actor's
 // own Action-audience token is still reviewed; this does not make it a privileged backend caller.
 func actionConnectionDelegation(
-	workloads *auth.Resolver, actor auth.Account, identities auth.IdentityUIDReader,
+	workloads *auth.Resolver, actor *auth.Account, serviceActor string, identities auth.IdentityUIDReader,
 ) (auth.Methods, error) {
 	methods := []string{
 		action_servicepb.ActionService_SubmitActionRequest_FullMethodName,
@@ -395,9 +395,17 @@ func actionConnectionDelegation(
 	for _, method := range methods {
 		rules[method] = auth.ReviewedServiceAccountOrIdentity(workloads)
 	}
-	return rules.WithDelegation(methods, []auth.Delegator{{
-		Account: actor, IdentityCallers: true, ConnectionProvenance: true,
-	}}, identities)
+	var err error
+	if actor != nil {
+		rules, err = rules.WithDelegation(methods, []auth.Delegator{{Account: *actor, IdentityCallers: true, ConnectionProvenance: true}}, identities)
+		if err != nil {
+			return nil, err
+		}
+	}
+	if serviceActor != "" {
+		rules, err = rules.WithServiceDelegation(methods, []auth.ServiceDelegator{{SPIFFEID: serviceActor, ConnectionProvenance: true}}, identities)
+	}
+	return rules, err
 }
 
 // serve answers gRPC on listener, runs the dispatch worker, supervises the MCP groups' sessions, refreshes the linked
@@ -549,7 +557,7 @@ func serve(ctx context.Context, c config, d deps, listener net.Listener) error {
 		observability.GRPCServerOption(),
 		grpc.ChainUnaryInterceptor(admission.UnaryInterceptor), grpc.ChainStreamInterceptor(admission.StreamInterceptor),
 	}
-	transport, err := c.RPC.ServerCredentials(peertls.ActionsURI)
+	transport, err := c.RPC.ServerCredentialsWithServices(peertls.ActionsURI)
 	if err != nil {
 		return fmt.Errorf("Action Service RPC TLS: %w", err)
 	}

agentplane/action_service/cmd/svid_test.go

@@ -14,11 +14,13 @@ import (
 	"github.com/agentydragon/ducktape/agentplane/action_service/internal/admission/catalog"
 	"github.com/agentydragon/ducktape/agentplane/action_service/internal/admission/policies"
 	"github.com/agentydragon/ducktape/agentplane/action_service/internal/store"
+	"github.com/agentydragon/ducktape/agentplane/identity/auth"
 	"github.com/agentydragon/ducktape/agentplane/identity/auth/listener"
 	workloadauthtesting "github.com/agentydragon/ducktape/agentplane/identity/auth/testing"
 	"github.com/agentydragon/ducktape/agentplane/net/peertls"
 	transporttesting "github.com/agentydragon/ducktape/agentplane/net/peertls/testing"
 	policyapi "github.com/agentydragon/ducktape/agentplane/policy/api/v1alpha1"
+	delegationpb "github.com/agentydragon/ducktape/agentplane/proto/delegation"
 	identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity"
 	"github.com/google/uuid"
 	"google.golang.org/grpc"
@@ -210,6 +212,7 @@ func TestSVIDIdentityCallerActsByItsIdentityBindings(t *testing.T) {
 	ca := transporttesting.NewCA(t)
 	serving := ca.IssueWithUsage(t, peertls.ActionsURI, x509.ExtKeyUsageServerAuth)
 	service := startConfigured(t, kube.Config, groups, func(c *config, _ *deps) {
+		c.ConnectionDelegationService = peertls.MCPEdgeURI
 		c.RPC = listener.Config{
 			ListenAddress: "127.0.0.1:0",
 			TLS:           &listener.TLS{CertFile: serving.CertFile, KeyFile: serving.KeyFile},
@@ -278,4 +281,78 @@ func TestSVIDIdentityCallerActsByItsIdentityBindings(t *testing.T) {
 	if err != nil || !proto.Equal(policy.GetActionPolicy().GetIdentity(), wantCaller) || len(policy.GetActionPolicy().GetAutoApproveIf()) != 1 {
 		t.Errorf("the Identity's own policy: %v, %v", policy, err)
 	}
+	transportCredentials, err := ca.Issue(t, peertls.MCPEdgeURI).ClientCredentials(peertls.ActionsURI)
+	if err != nil {
+		t.Fatal(err)
+	}
+	edgeConn, err := grpc.NewClient(service.address, grpc.WithNoProxy(), grpc.WithTransportCredentials(transportCredentials))
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = edgeConn.Close() })
+	edgeClient := action_servicepb.NewActionServiceClient(edgeConn)
+	for _, subject := range []*identitypb.Identity{wantCaller, {Name: unbound.Identity.Name, Uid: string(unbound.Identity.UID)}, {Name: wantCaller.Name, Uid: uuid.NewString()}} {
+		envelope := &delegationpb.Delegation{EffectiveCaller: &delegationpb.Delegation_Identity{Identity: subject}, Provenance: &delegationpb.Delegation_Connection{Connection: &delegationpb.ConnectionProvenance{ConnectionId: uuid.NewString(), GrantId: uuid.NewString()}}}
+		encoded, err := auth.MarshalDelegation(envelope)
+		if err != nil {
+			t.Fatal(err)
+		}
+		delegated := metadata.AppendToOutgoingContext(ctx, auth.DelegationMetadataKey, string(encoded))
+		result, err := edgeClient.SubmitActionRequest(delegated, &action_servicepb.SubmitActionRequestRequest{IdempotencyKey: uuid.NewString(), Action: &action_servicepb.ActionIdentity{Group: "test-mcp", Name: "test_echo"}, Arguments: &structpb.Struct{}, Title: "delegated service request"})
+		if proto.Equal(subject, wantCaller) {
+			if err != nil || !proto.Equal(result.GetActionRequest().GetCallerIdentity(), wantCaller) || result.GetActionRequest().GetCaller() != nil {
+				t.Fatalf("service actor effective caller=%v,%v", result, err)
+			}
+		} else if status.Code(err) != codes.Unauthenticated && status.Code(err) != codes.PermissionDenied {
+			t.Fatalf("service actor admitted ungranted/stale Identity: %v", err)
+		}
+	}
+}
+
+func TestServiceReaderAdmissionIsReadOnlyAndNotAnOperator(t *testing.T) {
+	ca := transporttesting.NewCA(t)
+	serving := ca.Issue(t, peertls.ActionsURI)
+	service := startConfigured(t, kube.Config, &catalog.Catalog{}, func(c *config, _ *deps) {
+		c.ServiceReaders = []string{peertls.NotificationsURI}
+		c.RPC = listener.Config{ListenAddress: "127.0.0.1:0", TLS: &listener.TLS{CertFile: serving.CertFile, KeyFile: serving.KeyFile}, Auth: []listener.Authentication{&listener.SPIFFEAuth{CAFile: serving.CAFile}, &listener.KubernetesBearerAuth{Audiences: []string{testAudience}}}}
+	})
+	waitReady(t, service)
+	for _, actor := range []string{peertls.NotificationsURI, peertls.MCPEdgeURI} {
+		t.Run(actor, func(t *testing.T) {
+			security, err := ca.Issue(t, actor).ClientCredentials(peertls.ActionsURI)
+			if err != nil {
+				t.Fatal(err)
+			}
+			conn, err := grpc.NewClient(service.address, grpc.WithNoProxy(), grpc.WithTransportCredentials(security))
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer conn.Close()
+			client := action_servicepb.NewActionServiceClient(conn)
+			ctx, cancel := context.WithTimeout(t.Context(), callDeadline)
+			defer cancel()
+			_, err = client.ListActionRequests(ctx, &action_servicepb.ListActionRequestsRequest{})
+			if actor == peertls.NotificationsURI {
+				if err != nil {
+					t.Fatal(err)
+				}
+			} else if status.Code(err) != codes.Unauthenticated {
+				t.Fatalf("unlisted service read: %v", err)
+			}
+			if _, err := client.SubmitActionRequest(ctx, &action_servicepb.SubmitActionRequestRequest{}); status.Code(err) != codes.PermissionDenied && status.Code(err) != codes.Unauthenticated {
+				t.Fatalf("service mutated: %v", err)
+			}
+			mixed := metadata.AppendToOutgoingContext(ctx, "authorization", "Bearer "+readerToken)
+			if _, err := client.ListActionRequests(mixed, &action_servicepb.ListActionRequestsRequest{}); status.Code(err) != codes.Unauthenticated {
+				t.Fatalf("bearer fallback: %v", err)
+			}
+			operatorWatch, err := operator_action_servicepb.NewOperatorActionServiceClient(conn).WatchMcpServers(ctx, &operator_action_servicepb.WatchMcpServersRequest{})
+			if err == nil {
+				_, err = operatorWatch.Recv()
+			}
+			if status.Code(err) != codes.Unauthenticated {
+				t.Fatalf("service acquired operator grant: %v", err)
+			}
+		})
+	}
 }

agentplane/action_service/internal/rpc/BUILD.bazel

@@ -34,6 +34,7 @@ go_library(
         "//agentplane/grpc/streaming",
         "//agentplane/identity/auth",
         "//agentplane/identity/operator",
+        "//agentplane/net/peertls",
         "//agentplane/policy/reader",
         "//agentplane/postgres/wakeups",
         "@com_github_google_uuid//:uuid",
@@ -117,6 +118,8 @@ go_test(
         "//agentplane/grpc/facade",
         "//agentplane/identity/auth",
         "//agentplane/identity/operator",
+        "//agentplane/net/peertls",
+        "//agentplane/net/peertls/testing",
         "//agentplane/policy/api/v1alpha1",
         "//agentplane/policy/reader",
         "//agentplane/postgres/wakeups",
@@ -139,11 +142,13 @@ go_test(
         "@io_k8s_sigs_controller_runtime//pkg/client",
         "@org_golang_google_grpc//:grpc",
         "@org_golang_google_grpc//codes",
+        "@org_golang_google_grpc//credentials",
         "@org_golang_google_grpc//credentials/insecure",
         "@org_golang_google_grpc//health",
         "@org_golang_google_grpc//health/grpc_health_v1",
         "@org_golang_google_grpc//interop/grpc_testing",
         "@org_golang_google_grpc//metadata",
+        "@org_golang_google_grpc//peer",
         "@org_golang_google_grpc//status",
         "@org_golang_google_protobuf//encoding/protojson",
         "@org_golang_google_protobuf//proto",