diff --git a/agentplane/docs/system_overview.md b/agentplane/docs/system_overview.md index 904edade0b1..b8bf60bf9a7 100644 --- a/agentplane/docs/system_overview.md +++ b/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 diff --git a/agentplane/runner_gateway/config.go b/agentplane/runner_gateway/config.go index 501bb0f948b..b035da7cb7f 100644 --- a/agentplane/runner_gateway/config.go +++ b/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 { diff --git a/agentplane/runner_gateway/config_test.go b/agentplane/runner_gateway/config_test.go index ad15d1ea366..6983208154e 100644 --- a/agentplane/runner_gateway/config_test.go +++ b/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 { diff --git a/agentplane/runner_gateway/dispatcher/BUILD.bazel b/agentplane/runner_gateway/dispatcher/BUILD.bazel index 978afa8cfa9..04d5c815779 100644 --- a/agentplane/runner_gateway/dispatcher/BUILD.bazel +++ b/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", diff --git a/agentplane/runner_gateway/dispatcher/SPEC.md b/agentplane/runner_gateway/dispatcher/SPEC.md index 88c6ba4261d..c079bc3189f 100644 --- a/agentplane/runner_gateway/dispatcher/SPEC.md +++ b/agentplane/runner_gateway/dispatcher/SPEC.md @@ -25,6 +25,22 @@ It scopes existing Sessions and existing environments; it does not create enviro Pod-bound identity. - Creator/broker identity is retained for provenance only. It is never Session authority. +## Service certificate callers during the cutover + +`dispatcher_service_callers` grants exact service SPIFFE IDs the component Dispatcher role. +These callers have their own typed principal and attribution; no account UID or Pod is fabricated. +`dispatcher_service_delegation` independently grants exact certificate actors and exact Dispatcher +RPCs typed Identity delegation. It requires Action provenance and a fresh effective Identity UID. +Delegated actors never inherit the direct component role, cannot create Sessions, and spend only +the effective Identity's wake authority. Direct services require a separate service wake grant. +Both allowlists default empty and are configured only after receiver image pins. + +Additive migration 18 retains service actor, optional effective Identity and Action correlation in +separate service attribution columns. Existing account and Identity evidence is unchanged, and a +retry never rewrites it. Service wake attempts use their own budget table keyed by SPIFFE ID, never +an account-shaped synthetic key. Old binaries see new actor attribution as absent, so rollback to +an old receiver must precede caller enablement or explicitly assess pending service-attributed work. + ## Delegated callers - An independent `delegation` setting names the Actions ServiceAccount, the namespaces whose caller identity it may assert, diff --git a/agentplane/runner_gateway/dispatcher/api.go b/agentplane/runner_gateway/dispatcher/api.go index 284f3b8ea0a..4d2192fab50 100644 --- a/agentplane/runner_gateway/dispatcher/api.go +++ b/agentplane/runner_gateway/dispatcher/api.go @@ -61,11 +61,12 @@ type Service struct { Store *Store // DispatcherCallers retain the explicit Pod-bound platform-component path. A ServiceAccount // remains an authenticated actor, but never owns a Session by account identity. - DispatcherCallers sets.Set[auth.Account] - Sandboxes SandboxReader - Hub *WatchHub - WatchLease time.Duration - CentralTarget *CentralTarget + DispatcherCallers sets.Set[auth.Account] + DispatcherServices sets.Set[string] + Sandboxes SandboxReader + Hub *WatchHub + WatchLease time.Duration + CentralTarget *CentralTarget } // maxFilterCommands bounds DispatchFilter.command_ids, the size of a Session's queue. @@ -174,6 +175,8 @@ func (s *Service) WatchDispatches(request *dispatcherpb.WatchDispatchesRequest, func terminalVisibleTo(event terminalnotice.Event, caller Caller) bool { switch c := caller.(type) { + case ServiceCaller: + return true case ServiceAccountCaller: return c.Privileged case IdentityCaller: @@ -246,6 +249,17 @@ func (s *Service) caller(ctx context.Context) (Caller, error) { return nil, err } switch identity := admitted.(type) { + case auth.ServiceAdmission: + if !s.DispatcherServices.Has(identity.Caller.SPIFFEID) { + return nil, status.Error(codes.PermissionDenied, "service is not a Dispatcher caller") + } + return ServiceCaller{SPIFFEID: identity.Caller.SPIFFEID}, nil + case auth.DelegatedServiceIdentityAdmission: + caller := IdentityCaller{Identity: &identitypb.Identity{Name: identity.EffectiveCaller.Name, Uid: identity.EffectiveCaller.UID}, Proof: IdentityServiceDelegation{ActorSPIFFEID: identity.Actor.SPIFFEID, Action: ActionCorrelation{RequestID: identity.Provenance.GetAction().GetRequestId(), ExecutionID: identity.Provenance.GetAction().GetExecutionId()}}} + if err := validateCaller(caller, nil); err != nil { + return nil, status.Error(codes.Unauthenticated, "delegated service attribution is incomplete") + } + return caller, nil case auth.IdentityAdmission: return identityCaller(identity.Caller) case auth.DelegatedIdentityAdmission: @@ -335,6 +349,8 @@ func (s *Service) requireLiveExecution(ctx context.Context, session *Session, sa return status.Error(codes.PermissionDenied, "Session target does not match") } switch c := caller.(type) { + case ServiceCaller: + return nil case ServiceAccountCaller: if c.Privileged { return nil diff --git a/agentplane/runner_gateway/dispatcher/delivery_wake.go b/agentplane/runner_gateway/dispatcher/delivery_wake.go index f0a33f6e082..863544b7cf7 100644 --- a/agentplane/runner_gateway/dispatcher/delivery_wake.go +++ b/agentplane/runner_gateway/dispatcher/delivery_wake.go @@ -355,6 +355,11 @@ func (w *DeliveryWorker) prepare(ctx context.Context, work queuedWork, sandbox S // unattributed row. func wakeCaller(attribution Attribution) lifecycle.WakeCaller { switch attribution := attribution.(type) { + case *ServiceAttribution: + if attribution.Identity != nil { + return lifecycle.IdentityWakeCaller{Identity: attribution.Identity} + } + return lifecycle.ServiceWakeCaller{SPIFFEID: attribution.ActorSPIFFEID} case *AdmissionAttribution: return lifecycle.AccountWakeCaller{Account: attribution.EffectiveCaller} case *IdentityAttribution: diff --git a/agentplane/runner_gateway/dispatcher/service_admission_integration_test.go b/agentplane/runner_gateway/dispatcher/service_admission_integration_test.go new file mode 100644 index 00000000000..aa329389dfa --- /dev/null +++ b/agentplane/runner_gateway/dispatcher/service_admission_integration_test.go @@ -0,0 +1,156 @@ +package dispatcher + +import ( + "context" + "net" + "testing" + "time" + + "github.com/agentydragon/ducktape/agentplane/identity/auth" + "github.com/agentydragon/ducktape/agentplane/net/peertls" + transporttesting "github.com/agentydragon/ducktape/agentplane/net/peertls/testing" + delegationpb "github.com/agentydragon/ducktape/agentplane/proto/delegation" + identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity" + sessionpb "github.com/agentydragon/ducktape/agentplane/proto/session" + dispatcherpb "github.com/agentydragon/ducktape/agentplane/runner_gateway/dispatcher/api/session/session_dispatcherpb" + "github.com/agentydragon/ducktape/agentplane/runner_gateway/environments/lifecycle" + "github.com/google/uuid" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "k8s.io/apimachinery/pkg/util/sets" +) + +type serviceIdentityReader map[string]string + +func (r serviceIdentityReader) GetIdentityUID(_ context.Context, name string) (string, bool, error) { + uid, ok := r[name] + return uid, ok, nil +} + +func TestServiceActorsRetainSeparateProofAndIdentityScope(t *testing.T) { + owner := &identitypb.Identity{Name: "owner", Uid: uuid.NewString()} + other := &identitypb.Identity{Name: "other", Uid: uuid.NewString()} + store := testStore(t, 10) + sandbox := SandboxSnapshot{Name: "service-target", Namespace: "test", UID: uuid.NewString()} + inventory := &authorityInventory{sandbox: sandbox, identity: owner} + seed := func(identity *identitypb.Identity) string { + t.Helper() + id := uuid.NewString() + spec := &sessionpb.SessionSpec{} + testPlacement(spec, sandbox.Name, sandbox.UID) + _, err := store.CreateSessionWithPlacementForCaller(t.Context(), id, sandbox.Name, sandbox.UID, spec, nil, nil, SessionExecution{Identity: identity}, ServiceCaller{SPIFFEID: peertls.AppURI}) + if err != nil { + t.Fatal(err) + } + return id + } + owned, foreign := seed(owner), seed(other) + row, err := store.GetSession(t.Context(), owned) + if err != nil { + t.Fatal(err) + } + if attribution, ok := row.Attribution.(*ServiceAttribution); !ok || attribution.ActorSPIFFEID != peertls.AppURI || row.Creator != nil { + t.Fatalf("service creator acquired account proof: %#v", row) + } + rule, err := auth.ReviewedServiceAccountOrIdentity(nil).AllowServiceCallers([]string{peertls.AppURI}) + if err != nil { + t.Fatal(err) + } + methods := Methods(rule) + methods, err = methods.WithServiceDelegation([]string{dispatcherpb.SessionDispatcher_SubmitInput_FullMethodName, dispatcherpb.SessionDispatcher_GetSession_FullMethodName}, []auth.ServiceDelegator{{SPIFFEID: peertls.ActionsURI}}, serviceIdentityReader{owner.Name: owner.Uid, other.Name: other.Uid}) + if err != nil { + t.Fatal(err) + } + ca := transporttesting.NewCA(t) + files := ca.Issue(t, peertls.SessionDispatcherURI) + serverTLS, err := files.WorkloadServerTLSConfigWithServices(peertls.SessionDispatcherURI, true) + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer(append(methods.ServerOptions(time.Second), grpc.Creds(credentials.NewTLS(serverTLS)))...) + dispatcherpb.RegisterSessionDispatcherServer(server, &Service{Store: store, Sandboxes: inventory, DispatcherServices: sets.New(peertls.AppURI)}) + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + client := func(serviceID string) dispatcherpb.SessionDispatcherClient { + t.Helper() + config, err := ca.Issue(t, serviceID).ClientTLSConfig(peertls.SessionDispatcherURI) + if err != nil { + t.Fatal(err) + } + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithNoProxy(), grpc.WithTransportCredentials(credentials.NewTLS(config))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = conn.Close() }) + return dispatcherpb.NewSessionDispatcherClient(conn) + } + app, actions, denied := client(peertls.AppURI), client(peertls.ActionsURI), client(peertls.SessionDispatcherURI) + if _, err := app.GetSession(t.Context(), &dispatcherpb.GetSessionRequest{SessionId: foreign}); err != nil { + t.Fatalf("direct component read: %v", err) + } + if _, err := denied.GetSession(t.Context(), &dispatcherpb.GetSessionRequest{SessionId: owned}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("unlisted combined Gateway service read: %v", err) + } + action := &delegationpb.ActionProvenance{RequestId: uuid.NewString(), ExecutionId: uuid.NewString()} + envelope := &delegationpb.Delegation{EffectiveCaller: &delegationpb.Delegation_Identity{Identity: owner}, Provenance: &delegationpb.Delegation_Action{Action: action}} + encoded, err := auth.MarshalDelegation(envelope) + if err != nil { + t.Fatal(err) + } + delegated := metadata.NewOutgoingContext(t.Context(), metadata.Pairs(auth.DelegationMetadataKey, string(encoded))) + commandID := uuid.NewString() + if _, err := actions.SubmitInput(delegated, &dispatcherpb.SubmitInputRequest{CommandId: commandID, SessionId: owned, Text: "delegated service"}); err != nil { + t.Fatal(err) + } + queued, err := store.GetDispatch(t.Context(), commandID) + if err != nil { + t.Fatal(err) + } + attribution, ok := queued.Attribution.(*ServiceAttribution) + if !ok || attribution.ActorSPIFFEID != peertls.ActionsURI || !sameIdentity(attribution.Identity, owner) || attribution.Action.RequestID != action.RequestId { + t.Fatalf("delegated service proof: %#v", queued.Attribution) + } + if wake, ok := wakeCaller(queued.Attribution).(lifecycle.IdentityWakeCaller); !ok || !sameIdentity(wake.Identity, owner) { + t.Fatalf("actor acquired wake grant: %#v", wakeCaller(queued.Attribution)) + } + if _, err := actions.GetSession(delegated, &dispatcherpb.GetSessionRequest{SessionId: foreign}); status.Code(err) != codes.NotFound { + t.Fatalf("delegator inherited component authority: %v", err) + } + if _, err := actions.CreateSession(delegated, &dispatcherpb.CreateSessionRequest{SessionId: uuid.NewString()}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("ungranted delegated lifecycle: %v", err) + } + ambiguous := metadata.AppendToOutgoingContext(delegated, "authorization", "Bearer legacy") + if _, err := actions.GetSession(ambiguous, &dispatcherpb.GetSessionRequest{SessionId: owned}); status.Code(err) != codes.Unauthenticated { + t.Fatalf("bearer fallback: %v", err) + } +} + +func TestServiceWakeBudgetDoesNotSpendAccountOrOtherServiceBudget(t *testing.T) { + store := testStore(t, 10) + target := uuid.NewString() + key := lifecycle.WakeBudgetKey{ServiceSPIFFEID: peertls.NotificationsURI} + for generation := int64(1); generation <= WakeLimit; generation++ { + if wait, err := store.ReserveWakeAttempt(t.Context(), key, target, generation); err != nil || wait != 0 { + t.Fatalf("reserve %d: %s %v", generation, wait, err) + } + } + if wait, err := store.ReserveWakeAttempt(t.Context(), key, target, WakeLimit+1); err != nil || wait <= 0 { + t.Fatalf("service limit: %s %v", wait, err) + } + if wait, err := store.ReserveWakeAttempt(t.Context(), key, target, 1); err != nil || wait != 0 { + t.Fatalf("retry: %s %v", wait, err) + } + if wait, err := store.ReserveWakeAttempt(t.Context(), lifecycle.WakeBudgetKey{ServiceSPIFFEID: peertls.AppURI}, target, 1); err != nil || wait != 0 { + t.Fatalf("other service: %s %v", wait, err) + } + if wait, err := store.ReserveWakeAttempt(t.Context(), lifecycle.WakeBudgetKey{Namespace: "test", Name: "notifications", UID: "real-account-uid"}, target, 1); err != nil || wait != 0 { + t.Fatalf("account budget: %s %v", wait, err) + } +} diff --git a/agentplane/runner_gateway/dispatcher/service_attribution.go b/agentplane/runner_gateway/dispatcher/service_attribution.go new file mode 100644 index 00000000000..adb04358585 --- /dev/null +++ b/agentplane/runner_gateway/dispatcher/service_attribution.go @@ -0,0 +1,93 @@ +package dispatcher + +import ( + "bytes" + "encoding/json" + "errors" + "io" + + "github.com/agentydragon/ducktape/agentplane/identity/auth" + identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity" +) + +// ServiceCaller is an exact, method-admitted component. It carries no Kubernetes account. +type ServiceCaller struct{ SPIFFEID string } + +func (ServiceCaller) isCaller() {} + +// IdentityServiceDelegation proves an effective Identity through a service certificate and Action envelope. +type IdentityServiceDelegation struct { + ActorSPIFFEID string + Action ActionCorrelation +} + +func (IdentityServiceDelegation) isIdentityProof() {} + +// ServiceAttribution retains certificate actor proof independently of the effective Identity. +// An absent Identity denotes the component's own direct authority. +type ServiceAttribution struct { + ActorSPIFFEID string `json:"actor_spiffe_id"` + Identity *identitypb.Identity `json:"identity,omitempty"` + Action *ActionCorrelation `json:"action,omitempty"` +} + +func (*ServiceAttribution) isAttribution() {} + +func isDelegatedIdentityProof(proof IdentityProof) bool { + switch proof.(type) { + case IdentityDelegation, IdentityServiceDelegation: + return true + default: + return false + } +} + +func validateServiceAttribution(a *ServiceAttribution) error { + if err := auth.ValidateServiceCallers([]string{a.ActorSPIFFEID}); err != nil { + return err + } + if (a.Identity == nil) != (a.Action == nil) { + return errors.New("service delegation requires an Identity and Action together") + } + if a.Identity != nil && (a.Identity.GetName() == "" || a.Identity.GetUid() == "" || a.Action.RequestID == "" || a.Action.ExecutionID == "") { + return errors.New("service delegation attribution is incomplete") + } + return nil +} + +func serviceAttributionBytes(recorded Attribution) (any, error) { + a, ok := recorded.(*ServiceAttribution) + if !ok { + return nil, nil + } + if err := validateServiceAttribution(a); err != nil { + return nil, err + } + data, err := json.Marshal(a) + if err != nil { + return nil, err + } + return string(data), nil +} + +func scanServiceAttribution(existing Attribution, data []byte) (Attribution, error) { + if data == nil { + return existing, nil + } + if existing != nil { + return nil, errors.New("service actor and legacy attribution cannot coexist") + } + a := new(ServiceAttribution) + decoder := json.NewDecoder(bytes.NewReader(data)) + decoder.DisallowUnknownFields() + if err := decoder.Decode(a); err != nil { + return nil, err + } + if err := decoder.Decode(new(any)); err != io.EOF { + return nil, errors.New("trailing service attribution") + } + if err := validateServiceAttribution(a); err != nil { + return nil, err + } + return a, nil +} diff --git a/agentplane/runner_gateway/dispatcher/store.go b/agentplane/runner_gateway/dispatcher/store.go index 1f3298b8af1..37ae813a423 100644 --- a/agentplane/runner_gateway/dispatcher/store.go +++ b/agentplane/runner_gateway/dispatcher/store.go @@ -186,13 +186,21 @@ func (ServiceAccountCaller) isCaller() {} func (IdentityCaller) isCaller() {} func privileged(caller Caller) bool { - account, ok := caller.(ServiceAccountCaller) - return ok && account.Privileged + switch c := caller.(type) { + case ServiceAccountCaller: + return c.Privileged + case ServiceCaller: + return true // only exact configured component callers construct this type + default: + return false + } } // callerAttribution is the Attribution a row records for caller, nil for an internal write. func callerAttribution(caller Caller) Attribution { switch c := caller.(type) { + case ServiceCaller: + return &ServiceAttribution{ActorSPIFFEID: c.SPIFFEID} case ServiceAccountCaller: if c.Attribution == nil { return nil @@ -202,6 +210,8 @@ func callerAttribution(caller Caller) Attribution { switch proof := c.Proof.(type) { case IdentitySVID: return &IdentityAttribution{Identity: c.Identity, Pod: proof.Pod} + case IdentityServiceDelegation: + return &ServiceAttribution{ActorSPIFFEID: proof.ActorSPIFFEID, Identity: c.Identity, Action: &proof.Action} case IdentityDelegation: return &DelegatedIdentityAttribution{Identity: c.Identity, Delegation: proof} default: @@ -218,6 +228,11 @@ func validateCaller(caller Caller, wakeCaller *Account) error { return errors.New("caller attribution is required") } switch c := caller.(type) { + case ServiceCaller: + if wakeCaller != nil { + return errors.New("service caller records no Kubernetes wake account") + } + return validateServiceAttribution(&ServiceAttribution{ActorSPIFFEID: c.SPIFFEID}) case ServiceAccountCaller: return validateAttribution(c.Attribution, wakeCaller) case IdentityCaller: @@ -229,6 +244,10 @@ func validateCaller(caller Caller, wakeCaller *Account) error { if proof.Pod != nil && (proof.Pod.Name == "" || proof.Pod.UID == "") { return errors.New("Identity caller has incomplete Pod provenance") } + case IdentityServiceDelegation: + if err := validateServiceAttribution(&ServiceAttribution{ActorSPIFFEID: proof.ActorSPIFFEID, Identity: c.Identity, Action: &proof.Action}); err != nil { + return err + } case IdentityDelegation: if err := proof.validate(); err != nil { return err @@ -249,6 +268,8 @@ func validateCaller(caller Caller, wakeCaller *Account) error { // may read, with its arguments numbered from next. func ownerScope(caller Caller, prefix string, next int) (string, []any) { switch c := caller.(type) { + case ServiceCaller: + return "TRUE", nil case ServiceAccountCaller: return fmt.Sprintf("$%d", next), []any{c.Privileged} case IdentityCaller: @@ -642,7 +663,7 @@ const dispatchColumns = `position, command_id, session_id, sandbox_name, sandbox actor_identity_name, actor_identity_uid, actor_identity_pod_name, actor_identity_pod_uid, delegated_identity_name, delegated_identity_uid, delegator_account_namespace, delegator_account_name, delegator_account_uid, delegator_pod_name, delegator_pod_uid, - delegated_action_request_id::text, delegated_action_execution_id::text` + delegated_action_request_id::text, delegated_action_execution_id::text, service_attribution` const qualifiedDispatchColumns = `d.position, d.command_id, d.session_id, d.sandbox_name, d.sandbox_uid::text, d.wake_eligible, d.wake_caller_namespace, d.wake_caller_name, d.wake_from_incarnation, @@ -654,7 +675,7 @@ const qualifiedDispatchColumns = `d.position, d.command_id, d.session_id, d.sand d.actor_identity_name, d.actor_identity_uid, d.actor_identity_pod_name, d.actor_identity_pod_uid, d.delegated_identity_name, d.delegated_identity_uid, d.delegator_account_namespace, d.delegator_account_name, d.delegator_account_uid, d.delegator_pod_name, d.delegator_pod_uid, - d.delegated_action_request_id::text, d.delegated_action_execution_id::text` + d.delegated_action_request_id::text, d.delegated_action_execution_id::text, d.service_attribution` func scanDispatchWithOwner(row interface{ Scan(...any) error }) (*Dispatch, *identitypb.Identity, error) { var identityName, identityUID sql.NullString @@ -680,7 +701,7 @@ func scanDispatch(row interface{ Scan(...any) error }, additional ...any) (*Disp var identityName, identityUID, identityPodName, identityPodUID sql.NullString var delegated [9]sql.NullString var wakeAt sql.NullTime - var commandBytes, admissionBytes, blockerBytes, wakeBytes []byte + var commandBytes, admissionBytes, blockerBytes, wakeBytes, serviceBytes []byte columns := []any{ &d.Position, &d.CommandID, &d.SessionID, &d.SandboxName, &d.SandboxUID, &d.WakeEligible, &callerNS, &callerName, &incarnation, &wakeAt, &d.Operation, @@ -693,6 +714,7 @@ func scanDispatch(row interface{ Scan(...any) error }, additional ...any) (*Disp for i := range delegated { columns = append(columns, &delegated[i]) } + columns = append(columns, &serviceBytes) columns = append(columns, additional...) err := row.Scan(columns...) if err != nil { @@ -754,6 +776,10 @@ func scanDispatch(row interface{ Scan(...any) error }, additional ...any) (*Disp } d.Attribution = delegatedIdentity } + d.Attribution, err = scanServiceAttribution(d.Attribution, serviceBytes) + if err != nil { + return nil, err + } if d.Attribution == nil && d.WakeCaller == nil { return nil, errors.New("dispatch caller attribution is missing") } @@ -850,6 +876,11 @@ func (s *Store) enqueue(ctx context.Context, sessionID, sandboxName, sandboxUID arguments = append(arguments, nil, wakeAssessment) arguments = append(arguments, identityAttributionValues(attribution)...) arguments = append(arguments, delegatedIdentityAttributionValues(attribution)...) + serviceBytes, err := serviceAttributionBytes(attribution) + if err != nil { + return err + } + arguments = append(arguments, serviceBytes) result, err = scanDispatch(tx.QueryRowContext(ctx, `INSERT INTO dispatch (command_id,session_id,sandbox_name,sandbox_uid,wake_eligible,wake_caller_namespace,wake_caller_name, operation,command,state,traceparent,actor_account_namespace,actor_account_name,actor_account_uid,actor_pod_name,actor_pod_uid, @@ -857,9 +888,9 @@ func (s *Store) enqueue(ctx context.Context, sessionID, sandboxName, sandboxUID created_at,updated_at,delivery_blocker,wake_assessment, actor_identity_name,actor_identity_uid,actor_identity_pod_name,actor_identity_pod_uid, delegated_identity_name,delegated_identity_uid,delegator_account_namespace,delegator_account_name,delegator_account_uid, - delegator_pod_name,delegator_pod_uid,delegated_action_request_id,delegated_action_execution_id) + delegator_pod_name,delegator_pod_uid,delegated_action_request_id,delegated_action_execution_id,service_attribution) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,$21,now(),now(),$22,$23,$24,$25,$26,$27, - $28,$29,$30,$31,$32,$33,$34,$35,$36) RETURNING `+dispatchColumns, + $28,$29,$30,$31,$32,$33,$34,$35,$36,$37) RETURNING `+dispatchColumns, arguments...)) if err != nil { return err @@ -997,7 +1028,7 @@ const sessionColumns = `session_id, sandbox_name, sandbox_uid::text, creator_acc wake_caller_namespace, wake_caller_name, wake_from_incarnation, wake_requested_at, spec, setup_script, state, reason, traceparent, created_at, updated_at, delivery_blocker, wake_assessment, environment_deleted_at, execution_identity_name, execution_identity_uid::text, - creator_identity_name, creator_identity_uid, creator_identity_pod_name, creator_identity_pod_uid` + creator_identity_name, creator_identity_uid, creator_identity_pod_name, creator_identity_pod_uid, service_attribution` func scanSession(row interface{ Scan(...any) error }) (*Session, error) { o := &Session{} @@ -1008,13 +1039,13 @@ func scanSession(row interface{ Scan(...any) error }) (*Session, error) { var wakeAt, environmentDeletedAt sql.NullTime var identityName, identityUID sql.NullString var creatorIdentityName, creatorIdentityUID, creatorIdentityPodName, creatorIdentityPodUID sql.NullString - var spec, blockerBytes, wakeBytes []byte + var spec, blockerBytes, wakeBytes, serviceBytes []byte err := row.Scan(&o.SessionID, &o.SandboxName, &o.SandboxUID, &creatorNS, &creatorName, &creatorUID, &effectiveNS, &effectiveName, &effectiveUID, &creatorPodName, &creatorPodUID, &actionRequestID, &actionExecutionID, &o.WakeEligible, &callerNS, &callerName, &incarnation, &wakeAt, &spec, &script, &o.State, &reason, &o.Traceparent, &o.CreatedAt, &o.UpdatedAt, &blockerBytes, &wakeBytes, &environmentDeletedAt, - &identityName, &identityUID, &creatorIdentityName, &creatorIdentityUID, &creatorIdentityPodName, &creatorIdentityPodUID) + &identityName, &identityUID, &creatorIdentityName, &creatorIdentityUID, &creatorIdentityPodName, &creatorIdentityPodUID, &serviceBytes) if err != nil { return nil, err } @@ -1046,6 +1077,10 @@ func scanSession(row interface{ Scan(...any) error }) (*Session, error) { if err != nil { return nil, err } + o.Attribution, err = scanServiceAttribution(o.Attribution, serviceBytes) + if err != nil { + return nil, err + } if o.Attribution == nil { return nil, errors.New("Session caller attribution is missing") } @@ -1103,7 +1138,7 @@ func (s *Store) createSessionWithPlacementForCaller(ctx context.Context, session return nil, errors.New("ServiceAccount actor and creator attribution are required") } // The session table records no delegated Identity creator: CreateSession refuses one. - if _, delegated := callerAttribution(caller).(*DelegatedIdentityAttribution); delegated { + if identity, ok := caller.(IdentityCaller); ok && isDelegatedIdentityProof(identity.Proof) { return nil, errors.New("a delegated Identity caller opens no Session") } if err := validateCaller(caller, wakeCaller); err != nil { @@ -1118,6 +1153,8 @@ func (s *Store) createSessionWithPlacementForCaller(ctx context.Context, session // authorizeRetry uses retained read authority: a receipt retry performs no new execution. func (o *Session) authorizeRetry(caller Caller) error { switch c := caller.(type) { + case ServiceCaller: + return nil case ServiceAccountCaller: if c.Privileged { return nil @@ -1215,18 +1252,22 @@ func (s *Store) openSession(ctx context.Context, sessionID, sandboxName, sandbox } attributionArgs := attributionValues(attribution) identityArgs := identityAttributionValues(attribution) + serviceBytes, err := serviceAttributionBytes(attribution) + if err != nil { + return err + } result, err = scanSession(tx.QueryRowContext(ctx, `INSERT INTO session (session_id,sandbox_name,sandbox_uid,creator_account_namespace,creator_account_name,creator_account_uid,effective_caller_account_namespace,effective_caller_account_name,effective_caller_account_uid, creator_pod_name,creator_pod_uid,action_request_id,action_execution_id,wake_eligible,wake_caller_namespace,wake_caller_name, spec,setup_script,state,traceparent,created_at,updated_at,delivery_blocker,wake_assessment, execution_identity_name,execution_identity_uid, - creator_identity_name,creator_identity_uid,creator_identity_pod_name,creator_identity_pod_uid) - VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,now(),now(),$21,$22,$23,$24,$25,$26,$27,$28) RETURNING `+sessionColumns, + creator_identity_name,creator_identity_uid,creator_identity_pod_name,creator_identity_pod_uid,service_attribution) + VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17,$18,$19,$20,now(),now(),$21,$22,$23,$24,$25,$26,$27,$28,$29) RETURNING `+sessionColumns, sessionID, sandboxName, sandboxUID, creatorNS, creatorName, creatorUID, attributionArgs[5], attributionArgs[6], attributionArgs[7], attributionArgs[3], attributionArgs[4], attributionArgs[8], attributionArgs[9], wakeCaller != nil, ns, name, data, setupScript, SessionWaitingState, observability.Traceparent(ctx), nil, wakeAssessment, identityName, identityUID, - identityArgs[0], identityArgs[1], identityArgs[2], identityArgs[3])) + identityArgs[0], identityArgs[1], identityArgs[2], identityArgs[3], serviceBytes)) if err != nil { return err } diff --git a/agentplane/runner_gateway/dispatcher/wake.go b/agentplane/runner_gateway/dispatcher/wake.go index d23ed251271..c2cb1cccb29 100644 --- a/agentplane/runner_gateway/dispatcher/wake.go +++ b/agentplane/runner_gateway/dispatcher/wake.go @@ -6,6 +6,7 @@ import ( "errors" "time" + "github.com/agentydragon/ducktape/agentplane/identity/auth" environmentpb "github.com/agentydragon/ducktape/agentplane/proto/environmentlifecyclepb" "github.com/agentydragon/ducktape/agentplane/runner_gateway/environments/lifecycle" ) @@ -13,6 +14,12 @@ import ( // ReserveWakeAttempt counts distinct guarded wake generations per authenticated // caller across replicas. A retry of the same generation spends no new budget. func (s *Store) ReserveWakeAttempt(ctx context.Context, caller lifecycle.WakeBudgetKey, sandboxUID string, generation int64) (time.Duration, error) { + if caller.ServiceSPIFFEID != "" { + if caller.Namespace != "" || caller.Name != "" || caller.UID != "" { + return 0, errors.New("service wake budget cannot contain account fields") + } + return s.reserveServiceWakeAttempt(ctx, caller.ServiceSPIFFEID, sandboxUID, generation) + } if caller.Namespace == "" || caller.Name == "" || caller.UID == "" { return 0, errors.New("wake caller namespace, name and UID are required") } @@ -67,6 +74,52 @@ func (s *Store) ReserveWakeAttempt(ctx context.Context, caller lifecycle.WakeBud return retryAfter, err } +// reserveServiceWakeAttempt spends a distinct certificate actor budget, never an account UID. +func (s *Store) reserveServiceWakeAttempt(ctx context.Context, serviceID, sandboxUID string, generation int64) (time.Duration, error) { + if err := auth.ValidateServiceCallers([]string{serviceID}); err != nil { + return 0, err + } + var retryAfter time.Duration + err := transaction(ctx, s.DB, func(tx *sql.Tx) error { + if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1,0))`, "dispatcher-service-wake:"+serviceID); err != nil { + return err + } + var now time.Time + if err := tx.QueryRowContext(ctx, "SELECT clock_timestamp()").Scan(&now); err != nil { + return err + } + windowStart := now.Add(-WakeWindow) + if _, err := tx.ExecContext(ctx, "DELETE FROM service_wake_attempt WHERE attempted_at <= $1", windowStart); err != nil { + return err + } + var exists bool + if err := tx.QueryRowContext(ctx, `SELECT EXISTS (SELECT 1 FROM service_wake_attempt WHERE service_spiffe_id=$1 AND sandbox_uid=$2 AND wake_generation=$3)`, serviceID, sandboxUID, generation).Scan(&exists); err != nil { + return err + } + if exists { + return nil + } + var count int + var oldest sql.NullTime + if err := tx.QueryRowContext(ctx, `SELECT count(*), min(attempted_at) FROM service_wake_attempt WHERE service_spiffe_id=$1 AND attempted_at>$2`, serviceID, windowStart).Scan(&count, &oldest); err != nil { + return err + } + if count >= WakeLimit { + if !oldest.Valid { + return errors.New("service wake budget full without attempt") + } + retryAfter = oldest.Time.Add(WakeWindow).Sub(now) + if retryAfter < time.Millisecond { + retryAfter = time.Millisecond + } + return nil + } + _, err := tx.ExecContext(ctx, `INSERT INTO service_wake_attempt (service_spiffe_id,sandbox_uid,wake_generation,attempted_at) VALUES ($1,$2,$3,$4)`, serviceID, sandboxUID, generation, now) + return err + }) + return retryAfter, err +} + // A null incarnation means no wake cycle; the empty string means a wake before // any incarnation existed. The distinction is retained from the Python schema. func (s *Store) NoteWaking(ctx context.Context, id string, owner Owner, incarnation *string, isSession bool, assessment *environmentpb.WakeAssessment) (bool, error) { diff --git a/agentplane/runner_gateway/environments/BUILD.bazel b/agentplane/runner_gateway/environments/BUILD.bazel index a509b258d1e..313048bf9f5 100644 --- a/agentplane/runner_gateway/environments/BUILD.bazel +++ b/agentplane/runner_gateway/environments/BUILD.bazel @@ -46,6 +46,7 @@ go_test( name = "environments_test", srcs = [ "environment_grants_test.go", + "service_svid_test.go", "service_test.go", "watch_test.go", ], @@ -54,6 +55,8 @@ go_test( "//agentplane/environments/v1:environment_service_go_proto", "//agentplane/identity/auth", "//agentplane/identity/credentials/api/v1alpha1", + "//agentplane/net/peertls", + "//agentplane/net/peertls/testing", "//agentplane/policy/api/v1alpha1", "//agentplane/proto:delegation_go_proto", "//agentplane/proto:environment_lifecycle_go_proto", @@ -78,6 +81,7 @@ go_test( "@io_k8s_sigs_controller_runtime//pkg/client/interceptor", "@org_golang_google_grpc//:grpc", "@org_golang_google_grpc//codes", + "@org_golang_google_grpc//credentials", "@org_golang_google_grpc//metadata", "@org_golang_google_grpc//status", "@org_golang_google_protobuf//proto", diff --git a/agentplane/runner_gateway/environments/README.md b/agentplane/runner_gateway/environments/README.md index 78546ed0df1..9fd2d112eb5 100644 --- a/agentplane/runner_gateway/environments/README.md +++ b/agentplane/runner_gateway/environments/README.md @@ -62,8 +62,8 @@ are separate current facts. Explicit-resume eligibility includes a denial reason against the caller's live UID and its explicit Identity rule or configured direct operator grant; it does not depend on `wakeOnSend`. Automatic Identity wake has its own assessment and uses the shared lifecycle manager under the `wakeOnSend` and existing -`resumePolicy.allowOwnServiceAccount` opt-in. Configured `wake_callers` remains an automatic-only -ServiceAccount broker grant. +`resumePolicy.allowOwnServiceAccount` opt-in. Configured `wake_callers` and `wake_service_callers` +remain automatic-only broker grants; service wake budgets are keyed by the caller's exact SPIFFE ID. Call `EnvironmentService.ResumeEnvironment` explicitly, then observe the same target again. `REQUESTED` confirms acceptance of desired running state, not readiness; no Session or input is @@ -97,3 +97,12 @@ omit the execution ServiceAccount and automatic-wake eligibility. They expose re wake generation and a token only while an explicit rule authorizes resume. A delegated Action's component reader grant is never inherited by the effective Identity. An idle stream stays open while its credential remains valid and closes when revalidation fails. + +Direct service SVIDs have independent, exact-SPIFFE-ID grants through +`environment_service_readers`, `environment_service_managers`, +`environment_service_launchers`, and `environment_service_operators`. Service operators may +explicitly resume, including a requested stop; delegated callers never inherit those roles. +`environment_service_delegation` admits a service actor only for configured read methods, +using the live effective Identity. `wake_service_callers` grants automatic broker wake only, +with budgets keyed by exact service SPIFFE ID. Create receipts retain the typed service actor +separately from the Action origin. diff --git a/agentplane/runner_gateway/environments/SPEC.md b/agentplane/runner_gateway/environments/SPEC.md index 80318d45e0f..5201591fc88 100644 --- a/agentplane/runner_gateway/environments/SPEC.md +++ b/agentplane/runner_gateway/environments/SPEC.md @@ -204,3 +204,12 @@ shared budget and unknown outcome handling. List/get/watch remain read-only; res evaluated independently. `REQUESTED` means the desired running state was accepted, not that the environment is ready. The Sandbox Controller remains responsible for startup and readiness. + +Direct service SVIDs have independent, exact-SPIFFE-ID grants through +`environment_service_readers`, `environment_service_managers`, +`environment_service_launchers`, and `environment_service_operators`. Service operators may +explicitly resume, including a requested stop; delegated callers never inherit those roles. +`environment_service_delegation` admits a service actor only for configured read methods, +using the live effective Identity. `wake_service_callers` grants automatic broker wake only, +with budgets keyed by exact service SPIFFE ID. Create receipts retain the typed service actor +separately from the Action origin. diff --git a/agentplane/runner_gateway/environments/collection.go b/agentplane/runner_gateway/environments/collection.go index 324c0a25ab2..dff2c506156 100644 --- a/agentplane/runner_gateway/environments/collection.go +++ b/agentplane/runner_gateway/environments/collection.go @@ -46,7 +46,7 @@ func (s *Service) WatchEnvironments(_ *emptypb.Empty, stream grpc.ServerStreamin if !identity.valid() { return status.Error(codes.Unauthenticated, "effective environment identity is incomplete") } - broad := identity.serviceAccount != nil && s.broadReader(ctx, identity.serviceAccount) + broad := s.broadReader(ctx, identity) if identity.identity == nil && !broad { return status.Error(codes.NotFound, "environment not found") } diff --git a/agentplane/runner_gateway/environments/internal/management/BUILD.bazel b/agentplane/runner_gateway/environments/internal/management/BUILD.bazel index dcc404b9a26..3a67b6562ae 100644 --- a/agentplane/runner_gateway/environments/internal/management/BUILD.bazel +++ b/agentplane/runner_gateway/environments/internal/management/BUILD.bazel @@ -11,6 +11,7 @@ go_library( importpath = "github.com/agentydragon/ducktape/agentplane/runner_gateway/environments/internal/management", deps = [ "//agentplane/environments/v1:environment_service_go_proto", + "//agentplane/identity/auth", "//agentplane/policy/api/v1alpha1", "//agentplane/proto:identity_go_proto", "//agentplane/runner_gateway/environments/internal/view", @@ -35,6 +36,7 @@ go_test( embed = [":management"], deps = [ "//agentplane/environments/v1:environment_service_go_proto", + "//agentplane/identity/auth", "//agentplane/policy/api/v1alpha1", "//agentplane/proto:identity_go_proto", "//agentplane/runner_gateway/environments/internal/view", @@ -46,6 +48,7 @@ go_test( "@io_k8s_apimachinery//pkg/types", "@io_k8s_sigs_controller_runtime//pkg/client", "@io_k8s_sigs_controller_runtime//pkg/client/fake", + "@org_golang_google_protobuf//encoding/protojson", "@org_golang_google_protobuf//proto", "@org_golang_google_protobuf//types/known/durationpb", ], diff --git a/agentplane/runner_gateway/environments/internal/management/access.go b/agentplane/runner_gateway/environments/internal/management/access.go index b282985c4ee..872b6bc7cf3 100644 --- a/agentplane/runner_gateway/environments/internal/management/access.go +++ b/agentplane/runner_gateway/environments/internal/management/access.go @@ -14,6 +14,7 @@ import ( identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity" environmentpb "github.com/agentydragon/ducktape/agentplane/environments/v1" + "github.com/agentydragon/ducktape/agentplane/identity/auth" policyapi "github.com/agentydragon/ducktape/agentplane/policy/api/v1alpha1" "github.com/agentydragon/ducktape/agentplane/runner_gateway/environments/internal/view" sandboxapi "github.com/agentydragon/ducktape/agentplane/sandboxes/api" @@ -37,7 +38,14 @@ type Access struct { namespace string objects client.Client reader *sandboxreader.Reader - caller *identitypb.ServiceAccount + actor Actor +} + +// Actor identifies the direct principal whose idempotent creation receipt is recorded. Exactly +// one ServiceAccount or service SPIFFE ID is present; neither is converted into the other. +type Actor struct { + ServiceAccount *identitypb.ServiceAccount + ServiceSPIFFEID string } // New creates access over one Sandbox namespace and caller identity. @@ -46,16 +54,31 @@ func New( objects client.Client, caller *identitypb.ServiceAccount, ) (*Access, error) { - if namespace == "" || objects == nil || caller == nil || caller.Namespace == "" || caller.Name == "" { - return nil, errors.New("Sandbox namespace, Kubernetes client, and caller ServiceAccount are required") + return NewWithActor(namespace, objects, Actor{ServiceAccount: caller}) +} + +// NewWithActor creates access for a directly admitted ServiceAccount or service principal. +func NewWithActor(namespace string, objects client.Client, actor Actor) (*Access, error) { + validAccount := actor.ServiceAccount != nil && actor.ServiceAccount.Namespace != "" && actor.ServiceAccount.Name != "" + validService := actor.ServiceSPIFFEID != "" + if namespace == "" || objects == nil || validAccount == validService { + return nil, errors.New("Sandbox namespace, Kubernetes client, and exactly one complete direct actor are required") + } + if validService { + if err := auth.ValidateServiceCallers([]string{actor.ServiceSPIFFEID}); err != nil { + return nil, err + } } reader, err := sandboxreader.New(namespace, objects) if err != nil { return nil, err } + if actor.ServiceAccount != nil { + actor.ServiceAccount = &identitypb.ServiceAccount{Namespace: actor.ServiceAccount.Namespace, Name: actor.ServiceAccount.Name} + } return &Access{ namespace: namespace, objects: objects, reader: reader, - caller: &identitypb.ServiceAccount{Namespace: caller.Namespace, Name: caller.Name}, + actor: actor, }, nil } @@ -79,7 +102,7 @@ func (a *Access) Create(ctx context.Context, request *environmentpb.CreateEnviro if err := validateCreate(request); err != nil { return nil, err } - intent, err := encodeCreateIntent(request, a.caller) + intent, err := encodeCreateIntent(request, a.actor) if err != nil { return nil, &InvalidSandboxError{Reason: "create request cannot be recorded"} } @@ -296,7 +319,7 @@ func (a *Access) retry(ctx context.Context, request *environmentpb.CreateEnviron } return nil, err } - if resource.DeletionTimestamp != nil || !sameIntent(resource.Annotations[sandboxapi.CreateIntentAnnotation], request, a.caller) { + if resource.DeletionTimestamp != nil || !sameIntent(resource.Annotations[sandboxapi.CreateIntentAnnotation], request, a.actor) { return nil, &SandboxConflictError{Name: name} } return resource, nil @@ -423,25 +446,35 @@ func isAlphaNumeric(value byte) bool { // createIntent is the receipt a create records on its Sandbox: the caller and the request it made. // The encoding needs only to be decodable; retries are matched by sameIntent, never by its bytes. type createIntent struct { - Caller intentCaller `json:"caller"` + Actor *intentActor `json:"actor,omitempty"` + Caller *intentCaller `json:"caller,omitempty"` Request json.RawMessage `json:"request"` } +type intentActor struct { + Kind string `json:"kind"` + ServiceSPIFFEID string `json:"service_spiffe_id,omitempty"` +} + type intentCaller struct { Namespace string `json:"namespace"` Name string `json:"name"` } -func encodeCreateIntent(request *environmentpb.CreateEnvironmentRequest, caller *identitypb.ServiceAccount) (string, error) { +func encodeCreateIntent(request *environmentpb.CreateEnvironmentRequest, actor Actor) (string, error) { // Proto field names, as receipts have always spelled them and readers of the annotation expect. requestJSON, err := (protojson.MarshalOptions{UseProtoNames: true}).Marshal(request) if err != nil { return "", err } - encoded, err := json.Marshal(createIntent{ - Caller: intentCaller{Namespace: caller.GetNamespace(), Name: caller.GetName()}, - Request: requestJSON, - }) + intent := createIntent{Request: requestJSON} + if actor.ServiceAccount != nil { + // Keep account receipts readable by old replicas throughout the receiver rollout. + intent.Caller = &intentCaller{Namespace: actor.ServiceAccount.GetNamespace(), Name: actor.ServiceAccount.GetName()} + } else { + intent.Actor = &intentActor{Kind: "service", ServiceSPIFFEID: actor.ServiceSPIFFEID} + } + encoded, err := json.Marshal(intent) if err != nil { return "", err } @@ -450,12 +483,16 @@ func encodeCreateIntent(request *environmentpb.CreateEnvironmentRequest, caller // sameIntent reports whether recorded is a create by caller of request. A receipt that does not decode is // someone else's. -func sameIntent(recorded string, request *environmentpb.CreateEnvironmentRequest, caller *identitypb.ServiceAccount) bool { +func sameIntent(recorded string, request *environmentpb.CreateEnvironmentRequest, actor Actor) bool { var intent createIntent if err := json.Unmarshal([]byte(recorded), &intent); err != nil { return false } - if intent.Caller.Namespace != caller.GetNamespace() || intent.Caller.Name != caller.GetName() { + if actor.ServiceAccount != nil { + if intent.Actor != nil || intent.Caller == nil || intent.Caller.Namespace != actor.ServiceAccount.GetNamespace() || intent.Caller.Name != actor.ServiceAccount.GetName() { + return false + } + } else if intent.Caller != nil || intent.Actor == nil || intent.Actor.Kind != "service" || intent.Actor.ServiceSPIFFEID != actor.ServiceSPIFFEID { return false } var recordedRequest environmentpb.CreateEnvironmentRequest diff --git a/agentplane/runner_gateway/environments/internal/management/access_test.go b/agentplane/runner_gateway/environments/internal/management/access_test.go index 04d2871f716..bba81f4d318 100644 --- a/agentplane/runner_gateway/environments/internal/management/access_test.go +++ b/agentplane/runner_gateway/environments/internal/management/access_test.go @@ -15,6 +15,7 @@ import ( environmentpb "github.com/agentydragon/ducktape/agentplane/environments/v1" policyapi "github.com/agentydragon/ducktape/agentplane/policy/api/v1alpha1" sandboxapi "github.com/agentydragon/ducktape/agentplane/sandboxes/api" + "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "google.golang.org/protobuf/types/known/durationpb" corev1 "k8s.io/api/core/v1" @@ -127,6 +128,51 @@ func TestCreateNamesTheIdentityAndCallerIntentAndIsIdempotent(t *testing.T) { } } +func TestCreateIntentKeepsTypedServiceActorsAndLegacyAccountRetries(t *testing.T) { + request := &environmentpb.CreateEnvironmentRequest{Name: "created-sandbox", Template: "pod", Identity: "agent"} + serviceActor := Actor{ServiceSPIFFEID: "spiffe://agentplane/actions"} + receipt, err := encodeCreateIntent(request, serviceActor) + if err != nil { + t.Fatal(err) + } + if !sameIntent(receipt, request, serviceActor) { + t.Fatal("typed service actor could not retry its create") + } + if sameIntent(receipt, request, Actor{ServiceSPIFFEID: "spiffe://agentplane/notifications"}) || + sameIntent(receipt, request, Actor{ServiceAccount: &identitypb.ServiceAccount{Namespace: namespace, Name: "app"}}) { + t.Fatal("typed service receipt was adopted by a different principal") + } + + accountActor := Actor{ServiceAccount: &identitypb.ServiceAccount{Namespace: namespace, Name: "app"}} + requestJSON, err := (protojson.MarshalOptions{UseProtoNames: true}).Marshal(request) + if err != nil { + t.Fatal(err) + } + legacy, err := json.Marshal(createIntent{ + Caller: &intentCaller{Namespace: namespace, Name: "app"}, Request: requestJSON, + }) + if err != nil { + t.Fatal(err) + } + if !sameIntent(string(legacy), request, accountActor) { + t.Fatal("legacy ServiceAccount create receipt no longer supports retry") + } + if sameIntent(string(legacy), request, serviceActor) { + t.Fatal("legacy ServiceAccount receipt was adopted by a service actor") + } + accountReceipt, err := encodeCreateIntent(request, accountActor) + if err != nil { + t.Fatal(err) + } + var decoded createIntent + if err := json.Unmarshal([]byte(accountReceipt), &decoded); err != nil { + t.Fatal(err) + } + if decoded.Actor != nil || decoded.Caller == nil || *decoded.Caller != (intentCaller{Namespace: namespace, Name: "app"}) { + t.Fatal("new account receipt is unreadable by old receivers") + } +} + func TestCreateRejectsAMissingIdentityAndInvalidRequests(t *testing.T) { access, _ := newAccess(t) for _, identity := range []string{"", "missing-agent"} { diff --git a/agentplane/runner_gateway/environments/lifecycle/README.md b/agentplane/runner_gateway/environments/lifecycle/README.md index 84f2ade0ddf..ba9aecf010c 100644 --- a/agentplane/runner_gateway/environments/lifecycle/README.md +++ b/agentplane/runner_gateway/environments/lifecycle/README.md @@ -29,3 +29,12 @@ readiness and execution-incarnation transitions; lifecycle acceptance does not i It has no gRPC registration or separate process. See [SPEC.md](SPEC.md) for the RPC and authority contract. + +Direct service SVIDs have independent, exact-SPIFFE-ID grants through +`environment_service_readers`, `environment_service_managers`, +`environment_service_launchers`, and `environment_service_operators`. Service operators may +explicitly resume, including a requested stop; delegated callers never inherit those roles. +`environment_service_delegation` admits a service actor only for configured read methods, +using the live effective Identity. `wake_service_callers` grants automatic broker wake only, +with budgets keyed by exact service SPIFFE ID. Create receipts retain the typed service actor +separately from the Action origin. diff --git a/agentplane/runner_gateway/environments/lifecycle/SPEC.md b/agentplane/runner_gateway/environments/lifecycle/SPEC.md index 7116d8f8c34..89e0dac8e95 100644 --- a/agentplane/runner_gateway/environments/lifecycle/SPEC.md +++ b/agentplane/runner_gateway/environments/lifecycle/SPEC.md @@ -55,7 +55,8 @@ token returns `STATE_CHANGED` with current eligibility and generation rather tha later suspension. It bounds conflict retries. A patch error may have uncertain effect, so the manager returns a typed `OUTCOME_UNKNOWN` status and does not replay the patch. The shared wake budget is reserved before patching and is keyed by effective caller namespace/name/UID (an Identity by the -manager's namespace, its name and UID), target Sandbox UID and next generation. +manager's namespace, its name and UID, or a service by exact SPIFFE ID), target Sandbox UID and next +generation. `REQUESTED` means the desired running state and next wake generation were accepted. It does not mean the environment is ready. `ALREADY_RUNNING` means no wake mutation was needed. A current-generation @@ -85,3 +86,12 @@ wake. Only `EnvironmentService` is registered as a public environment API. Its l remain read-only; explicit resume uses the policy and mutation contract above. The public service exposes no suspend operation. Any future public suspend API requires a separately reviewed authority and retry contract. + +Direct service SVIDs have independent, exact-SPIFFE-ID grants through +`environment_service_readers`, `environment_service_managers`, +`environment_service_launchers`, and `environment_service_operators`. Service operators may +explicitly resume, including a requested stop; delegated callers never inherit those roles. +`environment_service_delegation` admits a service actor only for configured read methods, +using the live effective Identity. `wake_service_callers` grants automatic broker wake only, +with budgets keyed by exact service SPIFFE ID. Create receipts retain the typed service actor +separately from the Action origin. diff --git a/agentplane/runner_gateway/environments/lifecycle/manager.go b/agentplane/runner_gateway/environments/lifecycle/manager.go index 87fe618346e..7aa0de31b68 100644 --- a/agentplane/runner_gateway/environments/lifecycle/manager.go +++ b/agentplane/runner_gateway/environments/lifecycle/manager.go @@ -41,13 +41,15 @@ var ( // honored for direct, Pod-bound admissions. EnvironmentRules and PolicyReady provide Identity // explicit-resume authority; either being absent fails closed for Identity callers. type Options struct { - Namespace string - Objects client.Client - WakeBudget WakeBudget - WakeCallers []auth.Account - OperatorCallers []auth.Account - EnvironmentRules *reader.Index[environmentaccess.Rule] - PolicyReady func() error + Namespace string + Objects client.Client + WakeBudget WakeBudget + WakeCallers []auth.Account + WakeServiceCallers []string + OperatorServiceCallers []string + OperatorCallers []auth.Account + EnvironmentRules *reader.Index[environmentaccess.Rule] + PolicyReady func() error } // Manager owns permission checks and optimistic Sandbox desired-state changes for explicit and @@ -60,6 +62,8 @@ type Manager struct { sandboxes *sandboxreader.Reader wakeBudget WakeBudget wakeCallers sets.Set[auth.Account] + wakeServices sets.Set[string] + operatorServices sets.Set[string] operatorCallers sets.Set[auth.Account] environmentRules *reader.Index[environmentaccess.Rule] policyReady func() error @@ -69,6 +73,12 @@ func NewManager(options Options) (*Manager, error) { if options.Namespace == "" || options.Objects == nil || options.WakeBudget == nil { return nil, errors.New("Sandbox namespace, uncached Kubernetes client, and shared wake budget are required") } + if err := auth.ValidateServiceCallers(options.WakeServiceCallers); err != nil { + return nil, fmt.Errorf("wake service callers: %w", err) + } + if err := auth.ValidateServiceCallers(options.OperatorServiceCallers); err != nil { + return nil, fmt.Errorf("operator service callers: %w", err) + } sandboxes, err := sandboxreader.New(options.Namespace, options.Objects) if err != nil { return nil, err @@ -77,6 +87,7 @@ func NewManager(options Options) (*Manager, error) { namespace: options.Namespace, objects: options.Objects, sandboxes: sandboxes, wakeBudget: options.WakeBudget, wakeCallers: sets.New(options.WakeCallers...), operatorCallers: sets.New(options.OperatorCallers...), environmentRules: options.EnvironmentRules, policyReady: options.PolicyReady, + wakeServices: sets.New(options.WakeServiceCallers...), operatorServices: sets.New(options.OperatorServiceCallers...), }, nil } @@ -312,6 +323,11 @@ func (s *Manager) automaticWakeAssessment(ctx context.Context, view *sandboxapi. return s.accountWakeAssessment(ctx, view, caller.Account) case IdentityWakeCaller: return s.identityWakeAssessment(ctx, view, caller.Identity) + case ServiceWakeCaller: + if !s.wakeServices.Has(caller.SPIFFEID) { + return assessmentFor(environmentlifecyclepb.WakeState_WAKE_STATE_DENIED, environmentlifecyclepb.WakeReason_WAKE_REASON_CALLER_NOT_AUTHORIZED, generation), false + } + return assessmentFor(environmentlifecyclepb.WakeState_WAKE_STATE_ELIGIBLE, environmentlifecyclepb.WakeReason_WAKE_REASON_ALLOWED_WAKE_BROKER, generation), true default: panic(fmt.Sprintf("unknown wake caller %T", caller)) } @@ -383,6 +399,8 @@ func (s *Manager) budgetKey(caller WakeCaller) WakeBudgetKey { return WakeBudgetKey{Namespace: account.GetNamespace(), Name: account.GetName(), UID: caller.Account.GetUid()} case IdentityWakeCaller: return WakeBudgetKey{Namespace: s.namespace, Name: caller.Identity.GetName(), UID: caller.Identity.GetUid()} + case ServiceWakeCaller: + return WakeBudgetKey{ServiceSPIFFEID: caller.SPIFFEID} default: panic(fmt.Sprintf("unknown wake caller %T", caller)) } @@ -407,6 +425,11 @@ func (s *Manager) eligibilityFor(ctx context.Context, view *sandboxapi.Sandbox, return denied(environmentlifecyclepb.ResumeEligibilityReason_RESUME_ELIGIBILITY_REASON_CALLER_NOT_EXECUTION_ACCOUNT), nil case IdentityWakeCaller: return s.identityEligibility(ctx, view, caller.Identity) + case ServiceWakeCaller: + if operator { + return allowed(environmentlifecyclepb.ResumeEligibilityReason_RESUME_ELIGIBILITY_REASON_ALLOWED_TRUSTED_OPERATOR), nil + } + return denied(environmentlifecyclepb.ResumeEligibilityReason_RESUME_ELIGIBILITY_REASON_CALLER_NOT_EXECUTION_ACCOUNT), nil default: return nil, errInvalidCaller } @@ -493,18 +516,33 @@ func requestedStop(view *sandboxapi.Sandbox) bool { return !view.Spec.Running || (view.Status.SuspendedBy != nil && *view.Status.SuspendedBy == sandboxapi.SuspendedByRequested) } -// isWakeBroker reports whether caller is a configured automatic-wake broker, which only a -// ServiceAccount can be. +// isWakeBroker reports whether caller is a configured automatic-wake broker. func (s *Manager) isWakeBroker(caller WakeCaller) bool { - accountCaller, ok := caller.(AccountWakeCaller) - if !ok { + switch caller := caller.(type) { + case AccountWakeCaller: + account := caller.Account.GetAccount() + return account != nil && s.wakeCallers.Has(auth.Account{Namespace: account.GetNamespace(), Name: account.GetName()}) + case ServiceWakeCaller: + return s.wakeServices.Has(caller.SPIFFEID) + default: return false } - account := accountCaller.Account.GetAccount() - if account == nil { +} + +func (s *Manager) explicitOperator(ctx context.Context, caller WakeCaller) bool { + switch caller := caller.(type) { + case AccountWakeCaller: + return s.directOperator(ctx, caller.Account) + case ServiceWakeCaller: + admission, err := auth.Admitted(ctx) + if err != nil { + return false + } + service, ok := admission.(auth.ServiceAdmission) + return ok && service.Caller.SPIFFEID == caller.SPIFFEID && s.operatorServices.Has(caller.SPIFFEID) + default: return false } - return s.wakeCallers.Has(auth.Account{Namespace: account.GetNamespace(), Name: account.GetName()}) } func (s *Manager) directOperator(ctx context.Context, caller *ServiceAccountIdentity) bool { @@ -532,12 +570,24 @@ func (s *Manager) admittedResumeCaller(ctx context.Context) (WakeCaller, bool, e return nil, false, status.Error(codes.Unauthenticated, errInvalidCaller.Error()) } return AccountWakeCaller{Account: caller.Caller}, s.directOperator(ctx, caller.Caller), nil + case auth.ServiceAdmission: + if err := auth.ValidateServiceCallers([]string{caller.Caller.SPIFFEID}); err != nil { + return nil, false, status.Error(codes.Unauthenticated, errInvalidCaller.Error()) + } + wakeCaller := ServiceWakeCaller{SPIFFEID: caller.Caller.SPIFFEID} + return wakeCaller, s.explicitOperator(ctx, wakeCaller), nil case auth.IdentityAdmission: identity := &identitypb.Identity{Name: caller.Caller.Identity.Name, Uid: caller.Caller.Identity.UID} if !validIdentity(identity) { return nil, false, status.Error(codes.Unauthenticated, errInvalidCaller.Error()) } return IdentityWakeCaller{Identity: identity}, false, nil + case auth.DelegatedServiceIdentityAdmission: + identity := &identitypb.Identity{Name: caller.EffectiveCaller.Name, Uid: caller.EffectiveCaller.UID} + if !validIdentity(identity) { + return nil, false, status.Error(codes.Unauthenticated, errInvalidCaller.Error()) + } + return IdentityWakeCaller{Identity: identity}, false, nil case auth.DelegatedIdentityAdmission: identity := &identitypb.Identity{Name: caller.EffectiveCaller.Name, Uid: caller.EffectiveCaller.UID} if !validIdentity(identity) { @@ -642,6 +692,8 @@ func validWakeCaller(caller WakeCaller) bool { return validServiceAccount(caller.Account) case IdentityWakeCaller: return caller.Identity.GetName() != "" && caller.Identity.GetUid() != "" + case ServiceWakeCaller: + return auth.ValidateServiceCallers([]string{caller.SPIFFEID}) == nil default: return false } diff --git a/agentplane/runner_gateway/environments/lifecycle/service_test.go b/agentplane/runner_gateway/environments/lifecycle/service_test.go index 0406fb8b4db..33589f8ed6f 100644 --- a/agentplane/runner_gateway/environments/lifecycle/service_test.go +++ b/agentplane/runner_gateway/environments/lifecycle/service_test.go @@ -878,6 +878,30 @@ func TestAutomaticWakeSharesPolicyBudgetAndReportsRetryFacts(t *testing.T) { }) } +func TestConfiguredServiceWakeUsesItsSPIFFEIDBudget(t *testing.T) { + serviceID := "spiffe://agentplane/notifications" + objects := testObjects(t, suspendedSandbox(&sandboxapi.ResumePolicy{}, true, sandboxapi.SuspendedByIdle, 7), executionUID) + budget := &testWakeBudget{} + manager, err := lifecycle.NewManager(lifecycle.Options{ + Namespace: testNamespace, Objects: objects, WakeBudget: budget, WakeServiceCallers: []string{serviceID}, + }) + if err != nil { + t.Fatal(err) + } + target := &lifecycle.Target{Namespace: testNamespace, Name: testSandbox, Uid: testSandboxUID} + assessment, err := manager.RequestWake(t.Context(), target, lifecycle.ServiceWakeCaller{SPIFFEID: serviceID}) + if err != nil || assessment.GetState() != environmentlifecyclepb.WakeState_WAKE_STATE_REQUESTED { + t.Fatalf("configured service RequestWake = %+v, %v", assessment, err) + } + if len(budget.calls) != 1 || budget.calls[0].caller != (lifecycle.WakeBudgetKey{ServiceSPIFFEID: serviceID}) { + t.Fatalf("service wake budget keys = %+v, want exact SPIFFE caller", budget.calls) + } + assessment, err = manager.RequestWake(t.Context(), target, lifecycle.ServiceWakeCaller{SPIFFEID: "spiffe://agentplane/actions"}) + if err != nil || assessment.GetState() != environmentlifecyclepb.WakeState_WAKE_STATE_DENIED || len(budget.calls) != 1 { + t.Fatalf("unlisted service wake = %+v, %v, budget calls=%+v", assessment, err, budget.calls) + } +} + func TestAutomaticWakeBrokerDoesNotOverrideRequestedStop(t *testing.T) { caller := &lifecycle.ServiceAccountIdentity{Account: &identitypb.ServiceAccount{Namespace: testNamespace, Name: "notifications"}, Uid: "notifications-uid"} sandbox := suspendedSandbox(&sandboxapi.ResumePolicy{}, false, sandboxapi.SuspendedByRequested, 1) diff --git a/agentplane/runner_gateway/environments/lifecycle/types.go b/agentplane/runner_gateway/environments/lifecycle/types.go index 095e5651e78..4bf9a1ab7e7 100644 --- a/agentplane/runner_gateway/environments/lifecycle/types.go +++ b/agentplane/runner_gateway/environments/lifecycle/types.go @@ -18,8 +18,8 @@ type Target = identitypb.EnvironmentTarget // ServiceAccountIdentity pins the effective caller to its live Kubernetes account incarnation. type ServiceAccountIdentity = identitypb.ServiceAccountIdentity -// WakeCaller is the caller a wake is assessed for: an explicitly configured ServiceAccount broker -// or an Identity acting for its bound Sandbox. +// WakeCaller is the caller a wake is assessed for: an explicitly configured ServiceAccount broker, +// a first-party service broker, or an Identity acting for its bound Sandbox. type WakeCaller interface{ isWakeCaller() } // AccountWakeCaller is a ServiceAccount incarnation. It can wake only through the explicit broker @@ -30,12 +30,20 @@ type AccountWakeCaller struct{ Account *ServiceAccountIdentity } // it. It wakes only a Sandbox acting as it, under that Sandbox's existing resume policy opt-in. type IdentityWakeCaller struct{ Identity *identitypb.Identity } +// ServiceWakeCaller is a verified first-party service. Its SPIFFE ID stays distinct from +// Kubernetes ServiceAccount and Identity provenance throughout wake authorization and budgeting. +type ServiceWakeCaller struct{ SPIFFEID string } + func (AccountWakeCaller) isWakeCaller() {} func (IdentityWakeCaller) isWakeCaller() {} +func (ServiceWakeCaller) isWakeCaller() {} -// WakeBudgetKey is whose budget a wake attempt spends: the namespace, name and UID of the caller -// incarnation. An Identity's namespace is the one the manager reads Identities in. -type WakeBudgetKey struct{ Namespace, Name, UID string } +// WakeBudgetKey is whose budget a wake attempt spends. ServiceAccounts and Identities use their +// namespace, name, and UID; service callers use their exact SPIFFE ID. +type WakeBudgetKey struct { + Namespace, Name, UID string + ServiceSPIFFEID string +} // WakeBudget reserves a bounded wake attempt without depending on Dispatcher storage. // Implementations share reservations across replicas and explicit and automatic wake paths. diff --git a/agentplane/runner_gateway/environments/management.go b/agentplane/runner_gateway/environments/management.go index ce922a122fa..16eca71c852 100644 --- a/agentplane/runner_gateway/environments/management.go +++ b/agentplane/runner_gateway/environments/management.go @@ -17,44 +17,89 @@ import ( apierrors "k8s.io/apimachinery/pkg/api/errors" ) -// manager requires an explicit direct, Pod-bound grant. In particular, delegation and read access -// do not confer write authority. Create records the authenticated account as the original caller. +// manager requires an explicit direct, Pod-bound workload grant or a separately configured direct +// service role. Delegation and read access do not confer write authority. func (s *Service) manager(ctx context.Context) (*management.Access, error) { return s.managementAccess(ctx, nil) } func (s *Service) managementAccess(ctx context.Context, origin *environmentpb.EnvironmentLaunchOrigin) (*management.Access, error) { - identity, err := caller(ctx) + admission, err := auth.Admitted(ctx) if err != nil { return nil, err } + switch admitted := admission.(type) { + case auth.ServiceAdmission: + if origin == nil { + if !s.serviceManagers.Has(admitted.Caller.SPIFFEID) { + return nil, status.Error(codes.PermissionDenied, "service caller cannot manage environments") + } + } else { + if !s.serviceLaunchers.Has(admitted.Caller.SPIFFEID) { + return nil, status.Error(codes.PermissionDenied, "service caller cannot launch environments") + } + if err := validateLaunchOrigin(origin); err != nil { + return nil, err + } + } + return management.NewWithActor(s.namespace, s.objects, management.Actor{ServiceSPIFFEID: admitted.Caller.SPIFFEID}) + case auth.DirectAdmission: + if admitted.Pod == nil || !admitted.Pod.IsPodBound() { + return nil, status.Error(codes.PermissionDenied, "environment management requires a direct Pod-bound caller") + } + identity, err := caller(ctx) + if err != nil { + return nil, err + } + account := serviceAccountReference(identity) + if origin == nil { + if !s.managers.Has(account) { + return nil, status.Error(codes.PermissionDenied, "caller cannot manage environments") + } + } else { + if !s.launchers.Has(account) { + return nil, status.Error(codes.PermissionDenied, "caller cannot launch environments on behalf of an Action caller") + } + if err := validateLaunchOrigin(origin); err != nil { + return nil, err + } + } + return management.New(s.namespace, s.objects, identitypbAccount(account)) + default: + return nil, status.Error(codes.PermissionDenied, "environment management requires a direct caller") + } +} + +func identitypbAccount(account auth.Account) *identitypb.ServiceAccount { + return &identitypb.ServiceAccount{Namespace: account.Namespace, Name: account.Name} +} + +func validateLaunchOrigin(origin *environmentpb.EnvironmentLaunchOrigin) error { + account, identity := origin.GetCaller(), origin.GetCallerIdentity() + validAccount := account != nil && account.GetNamespace() != "" && account.GetName() != "" + validIdentity := identity != nil && identity.GetName() != "" && identity.GetUid() != "" + if origin.GetActionGroup() == "" || !(validAccount && identity == nil || validIdentity && account == nil) { + return status.Error(codes.InvalidArgument, "launch origin requires exactly one complete caller and Action group") + } + return nil +} + +func (s *Service) templateAccess(ctx context.Context) (*management.Access, error) { admission, err := auth.Admitted(ctx) if err != nil { return nil, err } - direct, ok := admission.(auth.DirectAdmission) - if !ok || direct.Pod == nil || !direct.Pod.IsPodBound() { - return nil, status.Error(codes.PermissionDenied, "environment management requires a direct Pod-bound caller") - } - account := serviceAccountReference(identity) - if origin == nil { - if _, ok := s.managers[account]; !ok { - return nil, status.Error(codes.PermissionDenied, "caller cannot manage environments") - } - } else { - if _, ok := s.launchers[account]; !ok { - return nil, status.Error(codes.PermissionDenied, "caller cannot launch environments on behalf of an Action caller") - } - account, identity := origin.GetCaller(), origin.GetCallerIdentity() - validAccount := account != nil && account.GetNamespace() != "" && account.GetName() != "" - validIdentity := identity != nil && identity.GetName() != "" && identity.GetUid() != "" - if origin.GetActionGroup() == "" || !(validAccount && identity == nil || validIdentity && account == nil) { - return nil, status.Error(codes.InvalidArgument, "launch origin requires exactly one complete caller and Action group") + switch admitted := admission.(type) { + case auth.ServiceAdmission: + if !s.serviceLaunchers.Has(admitted.Caller.SPIFFEID) && !s.serviceManagers.Has(admitted.Caller.SPIFFEID) { + return nil, status.Error(codes.PermissionDenied, "service caller cannot list environment templates") } + return management.NewWithActor(s.namespace, s.objects, management.Actor{ServiceSPIFFEID: admitted.Caller.SPIFFEID}) + case auth.DirectAdmission: + return s.managementAccess(ctx, nil) + default: + return nil, status.Error(codes.PermissionDenied, "caller cannot list environment templates") } - return management.New(s.namespace, s.objects, &identitypb.ServiceAccount{ - Namespace: account.Namespace, Name: account.Name, - }) } func (s *Service) mutationTarget(target *identitypb.EnvironmentTarget) (string, string, error) { @@ -68,7 +113,7 @@ func (s *Service) mutationTarget(target *identitypb.EnvironmentTarget) (string, } func (s *Service) ListEnvironmentTemplates(ctx context.Context, _ *emptypb.Empty) (*environmentpb.ListEnvironmentTemplatesResponse, error) { - access, err := s.manager(ctx) + access, err := s.templateAccess(ctx) if err != nil { return nil, err } diff --git a/agentplane/runner_gateway/environments/service.go b/agentplane/runner_gateway/environments/service.go index 2c219792b98..155c350a7f7 100644 --- a/agentplane/runner_gateway/environments/service.go +++ b/agentplane/runner_gateway/environments/service.go @@ -45,6 +45,9 @@ type Service struct { readers sets.Set[auth.Account] managers sets.Set[auth.Account] launchers sets.Set[auth.Account] + serviceReaders sets.Set[string] + serviceManagers sets.Set[string] + serviceLaunchers sets.Set[string] identityChanges IdentityChangeGuard collectionAuthInterval time.Duration } @@ -68,6 +71,9 @@ func New( namespace: namespace, objects: objects, sandboxes: sandboxes, runtime: runtime, lifecycle: lifecycleManager, readers: readers, managers: management.Managers, launchers: management.Launchers, + serviceReaders: management.ServiceReaders, + serviceManagers: management.ServiceManagers, + serviceLaunchers: management.ServiceLaunchers, identityChanges: management.IdentityChanges, collectionAuthInterval: auth.DefaultVerdictTTL, }, nil @@ -75,9 +81,12 @@ func New( // ManagementOptions grants mutation authority independently of inventory and resume grants. type ManagementOptions struct { - Managers sets.Set[auth.Account] - Launchers sets.Set[auth.Account] - IdentityChanges IdentityChangeGuard + Managers sets.Set[auth.Account] + Launchers sets.Set[auth.Account] + ServiceReaders sets.Set[string] + ServiceManagers sets.Set[string] + ServiceLaunchers sets.Set[string] + IdentityChanges IdentityChangeGuard } // IdentityChangeGuard serializes a Sandbox patch with Dispatcher Session admission and rejects @@ -109,6 +118,53 @@ func Methods(resolver *auth.Resolver) auth.Methods { } } +// WithServiceCallers adds the EnvironmentService's independent direct component roles. A service +// allowlist only admits the certificate at the RPC boundary; handlers still check the matching +// role before returning inventory or making a mutation. +func WithServiceCallers( + methods auth.Methods, + readers, managers, launchers, operators []string, +) (auth.Methods, error) { + var err error + templateCallers := append(slices.Clone(managers), launchers...) + templateCallers = slices.Compact(slices.Sorted(slices.Values(templateCallers))) + methods, err = methods.WithServiceCallers([]string{ + environmentpb.EnvironmentService_ListEnvironments_FullMethodName, + environmentpb.EnvironmentService_WatchEnvironments_FullMethodName, + environmentpb.EnvironmentService_GetEnvironment_FullMethodName, + environmentpb.EnvironmentService_GetEnvironmentIdentityRecord_FullMethodName, + environmentpb.EnvironmentService_WatchEnvironment_FullMethodName, + }, readers) + if err != nil { + return nil, err + } + methods, err = methods.WithServiceCallers([]string{ + environmentpb.EnvironmentService_SuspendEnvironment_FullMethodName, + environmentpb.EnvironmentService_DeleteEnvironment_FullMethodName, + environmentpb.EnvironmentService_SetEnvironmentWakeOnSend_FullMethodName, + environmentpb.EnvironmentService_SetEnvironmentIdleSuspend_FullMethodName, + environmentpb.EnvironmentService_SetEnvironmentIdentity_FullMethodName, + }, managers) + if err != nil { + return nil, err + } + methods, err = methods.WithServiceCallers([]string{ + environmentpb.EnvironmentService_ListEnvironmentTemplates_FullMethodName, + }, templateCallers) + if err != nil { + return nil, err + } + methods, err = methods.WithServiceCallers([]string{ + environmentpb.EnvironmentService_CreateEnvironment_FullMethodName, + }, launchers) + if err != nil { + return nil, err + } + return methods.WithServiceCallers([]string{ + environmentpb.EnvironmentService_ResumeEnvironment_FullMethodName, + }, operators) +} + // ResumeEnvironment rechecks lifecycle authority and requests desired running state. // Acceptance does not imply readiness and never starts a Session or replays execution. func (s *Service) ResumeEnvironment(ctx context.Context, request *environmentpb.ResumeEnvironmentRequest) (*environmentpb.ResumeEnvironmentResponse, error) { @@ -140,6 +196,7 @@ func caller(ctx context.Context) (*auth.ServiceAccountPrincipal, error) { // role never becomes authority for the effective Identity. type environmentCaller struct { serviceAccount *auth.ServiceAccountPrincipal + service *auth.ServicePrincipal identity *auth.Identity } @@ -157,6 +214,11 @@ func environmentSubject(ctx context.Context) (environmentCaller, error) { case auth.DelegatedIdentityAdmission: identity := admitted.EffectiveCaller return environmentCaller{identity: &identity}, nil + case auth.ServiceAdmission: + return environmentCaller{service: &admitted.Caller}, nil + case auth.DelegatedServiceIdentityAdmission: + identity := admitted.EffectiveCaller + return environmentCaller{identity: &identity}, nil default: return environmentCaller{}, status.Error(codes.Unauthenticated, "environment caller is not admitted") } @@ -167,6 +229,9 @@ func (caller environmentCaller) valid() bool { account := caller.serviceAccount.GetAccount() return account != nil && account.GetNamespace() != "" && account.GetName() != "" && caller.serviceAccount.GetUid() != "" } + if caller.service != nil { + return caller.service.SPIFFEID != "" + } return caller.identity != nil && caller.identity.Name != "" && caller.identity.UID != "" } @@ -185,16 +250,27 @@ func (s *Service) ownsSandbox(ctx context.Context, identity *auth.Identity, targ } // broadReader applies only to direct credentials; delegated actors never lend their grants to a caller. -func (s *Service) broadReader(ctx context.Context, identity *auth.ServiceAccountPrincipal) bool { +func (s *Service) broadReader(ctx context.Context, caller environmentCaller) bool { admission, err := auth.Admitted(ctx) if err != nil { return false } - if _, ok := admission.(auth.DirectAdmission); !ok { + switch admitted := admission.(type) { + case auth.DirectAdmission: + if caller.serviceAccount == nil || !sameAccountPrincipal(admitted.Caller, caller.serviceAccount) { + return false + } + return s.readers.Has(serviceAccountReference(caller.serviceAccount)) + case auth.ServiceAdmission: + return caller.service != nil && admitted.Caller == *caller.service && s.serviceReaders.Has(admitted.Caller.SPIFFEID) + default: return false } - ok := s.readers.Has(serviceAccountReference(identity)) - return ok +} + +func sameAccountPrincipal(left, right *auth.ServiceAccountPrincipal) bool { + return left != nil && right != nil && left.GetUid() == right.GetUid() && + serviceAccountReference(left) == serviceAccountReference(right) } func (s *Service) ListEnvironments(ctx context.Context, request *environmentpb.ListEnvironmentsRequest) (*environmentpb.ListEnvironmentsResponse, error) { @@ -208,8 +284,8 @@ func (s *Service) ListEnvironments(ctx context.Context, request *environmentpb.L if !identity.valid() { return nil, status.Error(codes.Unauthenticated, "effective environment identity is incomplete") } - if identity.serviceAccount != nil && s.broadReader(ctx, identity.serviceAccount) { - return s.listAll(ctx, request, identity.serviceAccount) + if s.broadReader(ctx, identity) { + return s.listAll(ctx, request, identity) } if identity.identity == nil { if request.GetPageToken() != "" { @@ -272,8 +348,7 @@ func (s *Service) listIdentityEnvironments( // listAll is used only by a direct, explicitly granted inventory reader. Its Kubernetes // continuation may include names only that principal is authorized to discover. -func (s *Service) listAll(ctx context.Context, request *environmentpb.ListEnvironmentsRequest, identity *auth.ServiceAccountPrincipal) (*environmentpb.ListEnvironmentsResponse, error) { - caller := environmentCaller{serviceAccount: identity} +func (s *Service) listAll(ctx context.Context, request *environmentpb.ListEnvironmentsRequest, caller environmentCaller) (*environmentpb.ListEnvironmentsResponse, error) { var list api.SandboxList opts := []client.ListOption{client.InNamespace(s.namespace), client.Limit(int64(request.GetLimit()))} if request.GetPageToken() != "" { @@ -364,7 +439,7 @@ func (s *Service) getEnvironment( slog.ErrorContext(ctx, "Environment discovery Sandbox read failed", "error", err) return nil, "", status.Error(codes.Unavailable, "Sandbox inventory is unavailable") } - broad := identity.serviceAccount != nil && s.broadReader(ctx, identity.serviceAccount) + broad := s.broadReader(ctx, identity) if view.Identity.Uid != target.GetUid() || (deleting(view.Resource) && !broad) { return nil, "", status.Error(codes.NotFound, "environment not found") } @@ -521,7 +596,13 @@ func (s *Service) describe( return nil, err } if !caller.isIdentity() && object.Status.OperatingMode == api.OperatingModeSuspended { - wake, err := s.lifecycle.AssessAutomaticWake(ctx, target, lifecycle.AccountWakeCaller{Account: caller.serviceAccount}) + var wakeCaller lifecycle.WakeCaller + if caller.serviceAccount != nil { + wakeCaller = lifecycle.AccountWakeCaller{Account: caller.serviceAccount} + } else if caller.service != nil { + wakeCaller = lifecycle.ServiceWakeCaller{SPIFFEID: caller.service.SPIFFEID} + } + wake, err := s.lifecycle.AssessAutomaticWake(ctx, target, wakeCaller) if err != nil { return nil, status.Error(codes.Unavailable, "automatic wake eligibility is unavailable") } diff --git a/agentplane/runner_gateway/environments/service_svid_test.go b/agentplane/runner_gateway/environments/service_svid_test.go new file mode 100644 index 00000000000..72e6e895bf0 --- /dev/null +++ b/agentplane/runner_gateway/environments/service_svid_test.go @@ -0,0 +1,301 @@ +package environments + +import ( + "context" + "encoding/json" + "net" + "testing" + "time" + + environmentpb "github.com/agentydragon/ducktape/agentplane/environments/v1" + "github.com/agentydragon/ducktape/agentplane/identity/auth" + "github.com/agentydragon/ducktape/agentplane/net/peertls" + peertlstesting "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/agentydragon/ducktape/agentplane/runner_gateway/environments/lifecycle" + sandboxapi "github.com/agentydragon/ducktape/agentplane/sandboxes/api" + "github.com/agentydragon/ducktape/agentplane/sandboxes/runtimeconfig" + "github.com/google/uuid" + "google.golang.org/grpc" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/credentials" + "google.golang.org/grpc/metadata" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/proto" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/types" + "k8s.io/apimachinery/pkg/util/sets" + "sigs.k8s.io/controller-runtime/pkg/client" +) + +type serviceCreateWithUIDClient struct{ client.WithWatch } + +func (c serviceCreateWithUIDClient) Create(ctx context.Context, object client.Object, options ...client.CreateOption) error { + if sandbox, ok := object.(*sandboxapi.Sandbox); ok && sandbox.UID == "" { + sandbox.UID = types.UID("created-" + sandbox.Name + "-uid") + } + return c.WithWatch.Create(ctx, object, options...) +} + +type serviceRoleWakeBudget struct { + caller lifecycle.WakeBudgetKey +} + +func (b *serviceRoleWakeBudget) ReserveWakeAttempt(_ context.Context, caller lifecycle.WakeBudgetKey, _ string, _ int64) (time.Duration, error) { + b.caller = caller + return 0, nil +} + +func TestEnvironmentServiceServiceRolesStayMethodScoped(t *testing.T) { + const ( + readerID = peertls.CentralSamplerURI + managerID = peertls.SessionsURI + operatorID = peertls.ActionsURI + ) + sandbox := sandboxObject(sandboxUID) + sandbox.Spec.WakeGeneration = 3 + sandbox.Status.OperatingMode = sandboxapi.OperatingModeSuspended + sandbox.Status.StopStage = sandboxapi.StopStageStopped + idle := sandboxapi.SuspendedByIdle + sandbox.Status.SuspendedBy = &idle + sandbox.Status.SuspendedAtWakeGeneration = proto.Int64(3) + objects := serviceCreateWithUIDClient{WithWatch: readerWithObjects(t, sandbox, executionAccountObject(sandbox.UID), &policyapi.Identity{ + ObjectMeta: metav1.ObjectMeta{Namespace: discoveryNamespace, Name: "agent", UID: "agent-uid"}, + })} + budget := &serviceRoleWakeBudget{} + lifecycleManager, err := lifecycle.NewManager(lifecycle.Options{ + Namespace: discoveryNamespace, Objects: objects, WakeBudget: budget, + OperatorServiceCallers: []string{operatorID}, + }) + if err != nil { + t.Fatal(err) + } + service, err := New(discoveryNamespace, objects, runtimeconfig.Catalog{}, lifecycleManager, nil, ManagementOptions{ + ServiceReaders: sets.New(readerID), ServiceManagers: sets.New(managerID), ServiceLaunchers: sets.New(launcherID), + }) + if err != nil { + t.Fatal(err) + } + methods, err := WithServiceCallers(Methods(nil), []string{readerID}, []string{managerID}, []string{launcherID}, []string{operatorID}) + if err != nil { + t.Fatal(err) + } + ca := peertlstesting.NewCA(t) + serverFiles := ca.Issue(t, peertls.SessionDispatcherURI) + serverTLS, err := serverFiles.WorkloadServerTLSConfigWithServices(peertls.SessionDispatcherURI, true) + if err != nil { + t.Fatal(err) + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer(append(methods.ServerOptions(time.Second), grpc.Creds(credentials.NewTLS(serverTLS)))...) + environmentpb.RegisterEnvironmentServiceServer(server, service) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + + dial := func(serviceID string) environmentpb.EnvironmentServiceClient { + t.Helper() + files := ca.Issue(t, serviceID) + clientTLS, err := files.ClientTLSConfig(peertls.SessionDispatcherURI) + if err != nil { + t.Fatal(err) + } + connection, err := grpc.NewClient(listener.Addr().String(), grpc.WithNoProxy(), grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = connection.Close() }) + return environmentpb.NewEnvironmentServiceClient(connection) + } + target := &identitypb.EnvironmentTarget{Namespace: discoveryNamespace, Name: sandboxName, Uid: sandboxUID} + reader := dial(readerID) + listed, err := reader.ListEnvironments(t.Context(), &environmentpb.ListEnvironmentsRequest{Limit: 10}) + if err != nil || len(listed.GetItems()) != 1 { + t.Fatalf("service reader list = %v, %v", listed, err) + } + if _, err := reader.SetEnvironmentWakeOnSend(t.Context(), &environmentpb.SetEnvironmentWakeOnSendRequest{Target: target, WakeOnSend: true}); status.Code(err) == 0 { + t.Fatal("service reader mutated Sandbox wake policy") + } + manager := dial(managerID) + if _, err := manager.ListEnvironments(t.Context(), &environmentpb.ListEnvironmentsRequest{Limit: 10}); status.Code(err) == 0 { + t.Fatal("service manager inherited the reader role") + } + if _, err := manager.SetEnvironmentWakeOnSend(t.Context(), &environmentpb.SetEnvironmentWakeOnSendRequest{Target: target, WakeOnSend: true}); err != nil { + t.Fatalf("service manager could not change wake policy: %v", err) + } + launcher := dial(launcherID) + if _, err := launcher.SetEnvironmentWakeOnSend(t.Context(), &environmentpb.SetEnvironmentWakeOnSendRequest{Target: target, WakeOnSend: true}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("service launcher acquired manager authority: %v", err) + } + launchRequest := &environmentpb.CreateEnvironmentRequest{ + Name: "service-created", Template: "runner-vm", Identity: "agent", + LaunchOrigin: &environmentpb.EnvironmentLaunchOrigin{ + CallerIdentity: &identitypb.Identity{Name: "agent", Uid: "agent-uid"}, ActionGroup: "group-a", + }, + } + created, err := launcher.CreateEnvironment(t.Context(), launchRequest) + if err != nil || created.GetUid() != "created-service-created-uid" { + t.Fatalf("service launcher create = %v, %v", created, err) + } + var createdSandbox sandboxapi.Sandbox + if err := objects.Get(t.Context(), types.NamespacedName{Namespace: discoveryNamespace, Name: "service-created"}, &createdSandbox); err != nil { + t.Fatal(err) + } + var receipt struct { + Actor struct { + Kind string `json:"kind"` + ServiceSPIFFEID string `json:"service_spiffe_id"` + } `json:"actor"` + } + if err := json.Unmarshal([]byte(createdSandbox.Annotations[sandboxapi.CreateIntentAnnotation]), &receipt); err != nil { + t.Fatal(err) + } + if receipt.Actor.Kind != "service" || receipt.Actor.ServiceSPIFFEID != launcherID { + t.Fatalf("service create receipt actor = %+v", receipt.Actor) + } + retried, err := launcher.CreateEnvironment(t.Context(), launchRequest) + if err != nil || !proto.Equal(created, retried) { + t.Fatalf("service launcher retry = %v, %v; want exact original target %v", retried, err, created) + } + operator := dial(operatorID) + response, err := operator.ResumeEnvironment(t.Context(), &environmentpb.ResumeEnvironmentRequest{Target: target}) + if err != nil || response.GetOutcome() != environmentpb.ResumeOutcome_RESUME_OUTCOME_REQUESTED { + t.Fatalf("service operator resume = %v, %v", response, err) + } + if budget.caller != (lifecycle.WakeBudgetKey{ServiceSPIFFEID: operatorID}) { + t.Fatalf("operator wake budget caller = %+v, want SPIFFE identity only", budget.caller) + } + var updated sandboxapi.Sandbox + if err := objects.Get(t.Context(), types.NamespacedName{Namespace: discoveryNamespace, Name: sandboxName}, &updated); err != nil { + t.Fatal(err) + } + if !updated.Spec.Running || updated.Spec.WakeGeneration != 4 { + t.Fatalf("service operator did not update desired state: running=%t generation=%d", updated.Spec.Running, updated.Spec.WakeGeneration) + } +} + +const ( + launcherID = peertls.RuntimeProxyURI + actionsID = peertls.ActionsURI +) + +func serviceDelegationContext(t *testing.T, identity auth.Identity) context.Context { + t.Helper() + envelope, err := auth.MarshalDelegation(&delegationpb.Delegation{ + EffectiveCaller: &delegationpb.Delegation_Identity{Identity: &identitypb.Identity{Name: identity.Name, Uid: identity.UID}}, + Provenance: &delegationpb.Delegation_Action{Action: &delegationpb.ActionProvenance{ + RequestId: uuid.NewString(), ExecutionId: uuid.NewString(), + }}, + }) + if err != nil { + t.Fatal(err) + } + return metadata.NewOutgoingContext(t.Context(), metadata.Pairs(auth.DelegationMetadataKey, string(envelope))) +} + +func TestEnvironmentServiceActionDelegationKeepsIdentityScopeAndNoLifecycleAuthority(t *testing.T) { + const ( + identityName = "agent" + identityUID = "agent-uid-current" + otherName = "other-agent" + otherUID = "other-agent-uid" + ) + current := sandboxObject(sandboxUID) + current.Spec.Identity = &sandboxapi.IdentityRef{Name: identityName} + other := sandboxObject("other-sandbox-uid") + other.Name = "other-workspace" + other.Spec.Identity = &sandboxapi.IdentityRef{Name: otherName} + otherAccount := executionAccountObject(other.UID) + otherAccount.Name = other.Name + otherAccount.OwnerReferences[0].Name = other.Name + objects := readerWithObjects( + t, + current, executionAccountObject(current.UID), other, otherAccount, + &policyapi.Identity{ObjectMeta: metav1.ObjectMeta{Namespace: discoveryNamespace, Name: identityName, UID: identityUID}}, + &policyapi.Identity{ObjectMeta: metav1.ObjectMeta{Namespace: discoveryNamespace, Name: otherName, UID: otherUID}}, + ) + lifecycleManager, err := lifecycle.NewManager(lifecycle.Options{ + Namespace: discoveryNamespace, Objects: objects, WakeBudget: &serviceRoleWakeBudget{}, OperatorServiceCallers: []string{actionsID}, + }) + if err != nil { + t.Fatal(err) + } + service, err := New(discoveryNamespace, objects, runtimeconfig.Catalog{}, lifecycleManager, nil, ManagementOptions{ + ServiceReaders: sets.New(actionsID), + }) + if err != nil { + t.Fatal(err) + } + methods, err := WithServiceCallers(Methods(nil), []string{actionsID}, nil, nil, nil) + if err != nil { + t.Fatal(err) + } + identities, err := auth.KubernetesIdentityUIDs(objects, discoveryNamespace) + if err != nil { + t.Fatal(err) + } + methods, err = methods.WithServiceDelegation([]string{ + environmentpb.EnvironmentService_ListEnvironments_FullMethodName, + environmentpb.EnvironmentService_GetEnvironment_FullMethodName, + }, []auth.ServiceDelegator{{SPIFFEID: actionsID}}, identities) + if err != nil { + t.Fatal(err) + } + ca := peertlstesting.NewCA(t) + serverFiles := ca.Issue(t, peertls.SessionDispatcherURI) + serverTLS, err := serverFiles.WorkloadServerTLSConfigWithServices(peertls.SessionDispatcherURI, true) + if err != nil { + t.Fatal(err) + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer(append(methods.ServerOptions(time.Second), grpc.Creds(credentials.NewTLS(serverTLS)))...) + environmentpb.RegisterEnvironmentServiceServer(server, service) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + files := ca.Issue(t, actionsID) + clientTLS, err := files.ClientTLSConfig(peertls.SessionDispatcherURI) + if err != nil { + t.Fatal(err) + } + connection, err := grpc.NewClient(listener.Addr().String(), grpc.WithNoProxy(), grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = connection.Close() }) + client := environmentpb.NewEnvironmentServiceClient(connection) + target := &identitypb.EnvironmentTarget{Namespace: discoveryNamespace, Name: sandboxName, Uid: sandboxUID} + + direct, err := client.ListEnvironments(t.Context(), &environmentpb.ListEnvironmentsRequest{Limit: 10}) + if err != nil || len(direct.GetItems()) != 2 { + t.Fatalf("direct Actions component reader = %v, %v; want both Environments", direct, err) + } + delegated := serviceDelegationContext(t, auth.Identity{Name: identityName, UID: identityUID}) + listed, err := client.ListEnvironments(delegated, &environmentpb.ListEnvironmentsRequest{Limit: 10}) + if err != nil || len(listed.GetItems()) != 1 || listed.GetItems()[0].GetTarget().GetName() != sandboxName { + t.Fatalf("delegated Actions list = %v, %v; want only effective Identity's Environment", listed, err) + } + if _, err := client.GetEnvironment(delegated, &environmentpb.GetEnvironmentRequest{Target: target}); err != nil { + t.Fatalf("delegated Actions GetEnvironment = %v", err) + } + foreignTarget := &identitypb.EnvironmentTarget{Namespace: discoveryNamespace, Name: other.Name, Uid: string(other.UID)} + if _, err := client.GetEnvironment(delegated, &environmentpb.GetEnvironmentRequest{Target: foreignTarget}); status.Code(err) != codes.NotFound { + t.Fatalf("delegated Actions read crossed Identity binding: %v", err) + } + stale := serviceDelegationContext(t, auth.Identity{Name: identityName, UID: "agent-uid-replaced"}) + if _, err := client.ListEnvironments(stale, &environmentpb.ListEnvironmentsRequest{Limit: 10}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("stale delegated Identity UID was admitted: %v", err) + } + if _, err := client.SetEnvironmentWakeOnSend(delegated, &environmentpb.SetEnvironmentWakeOnSendRequest{Target: target, WakeOnSend: true}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("delegated Action inherited lifecycle manager authority: %v", err) + } + if _, err := client.ResumeEnvironment(delegated, &environmentpb.ResumeEnvironmentRequest{Target: target}); status.Code(err) != codes.PermissionDenied { + t.Fatalf("delegated Action inherited operator authority: %v", err) + } +} diff --git a/agentplane/runner_gateway/environments/watch.go b/agentplane/runner_gateway/environments/watch.go index 176d46e9aaf..e501c2e7978 100644 --- a/agentplane/runner_gateway/environments/watch.go +++ b/agentplane/runner_gateway/environments/watch.go @@ -34,7 +34,7 @@ func (s *Service) WatchEnvironment(request *environmentpb.WatchEnvironmentReques if !identity.valid() { return status.Error(codes.Unauthenticated, "effective environment identity is incomplete") } - broad := identity.serviceAccount != nil && s.broadReader(ctx, identity.serviceAccount) + broad := s.broadReader(ctx, identity) if identity.identity == nil && !broad { return status.Error(codes.NotFound, "environment not found") } diff --git a/agentplane/runner_gateway/internal/gateway/BUILD.bazel b/agentplane/runner_gateway/internal/gateway/BUILD.bazel index 22df13ec8ab..c561b64eb86 100644 --- a/agentplane/runner_gateway/internal/gateway/BUILD.bazel +++ b/agentplane/runner_gateway/internal/gateway/BUILD.bazel @@ -103,6 +103,7 @@ go_test( "@org_golang_google_grpc//:grpc", "@org_golang_google_grpc//codes", "@org_golang_google_grpc//connectivity", + "@org_golang_google_grpc//credentials", "@org_golang_google_grpc//credentials/insecure", "@org_golang_google_grpc//metadata", "@org_golang_google_grpc//status", diff --git a/agentplane/runner_gateway/internal/gateway/SPEC.md b/agentplane/runner_gateway/internal/gateway/SPEC.md index 11849062255..21b86506d7f 100644 --- a/agentplane/runner_gateway/internal/gateway/SPEC.md +++ b/agentplane/runner_gateway/internal/gateway/SPEC.md @@ -62,3 +62,13 @@ must still pass Runtime's admission gate. Runtime's initial lameduck permits exa its final Session cutoff refuses new command IDs, including interrupts. Existing command IDs retain journal reconciliation. Removing finalizers or using foreground deletion bypasses this managed protocol and supplies no such proof. + +## Service certificate callers during the cutover + +The Runtime proxy has independent exact-service allowlists for diagnostics readers, Sessions +operations, and Controller lifecycle operations. Executor has its own service allowlist. No shared +Gateway service identity grants all four roles. The existing combined serving identity remains the +server pin; Sessions, runtime-proxy and Environments client identities follow those roles across +the planned binary split. Verified service certificates carry neither Kubernetes account nor Pod +proof, and method denial never falls back to bearer verification. All new lists default empty; +manifest settings and caller switches follow receiver image pins. diff --git a/agentplane/runner_gateway/internal/gateway/fixture_test.go b/agentplane/runner_gateway/internal/gateway/fixture_test.go index 81e7ba7541d..0d8bf5c98f5 100644 --- a/agentplane/runner_gateway/internal/gateway/fixture_test.go +++ b/agentplane/runner_gateway/internal/gateway/fixture_test.go @@ -39,6 +39,15 @@ var ( controllerAccount = auth.Account{Namespace: "test-system", Name: "test-sandbox-controller"} ) +func gatewayMethods(t *testing.T, g *Gateway) auth.Methods { + t.Helper() + methods, err := g.Methods() + if err != nil { + t.Fatal(err) + } + return methods +} + func useAPIServer(server *kubeapiserver.Server) error { api = server return nil diff --git a/agentplane/runner_gateway/internal/gateway/gateway.go b/agentplane/runner_gateway/internal/gateway/gateway.go index 91f776c7c30..9abb75de28b 100644 --- a/agentplane/runner_gateway/internal/gateway/gateway.go +++ b/agentplane/runner_gateway/internal/gateway/gateway.go @@ -21,9 +21,13 @@ import ( // Config contains the per-process Gateway role wiring. Public callers use TokenReview; // runner connections use explicitly configured transport security and incarnation-fenced metadata. type Config struct { - RunnerSecurity peertls.Security - Namespace string - CallerAccounts []auth.Account + RuntimeReaders []string + SessionServices []string + ControllerServices []string + ExecutorServices []string + RunnerSecurity peertls.Security + Namespace string + CallerAccounts []auth.Account // DispatcherAccount is the combined Gateway/Dispatcher workload. Its Pod-bound // projected token is the only public route to Runtime Session creation and commands. DispatcherAccount auth.Account @@ -144,7 +148,7 @@ func New( // Methods admits callers of the Environment Runtime proxy and the typed execution API. Only the Sandbox // controller may quiesce or change Firecracker suspension admission; with no controller configured, nobody may. -func (g *Gateway) Methods() auth.Methods { +func (g *Gateway) Methods() (auth.Methods, error) { callers := auth.Accounts(g.callerAuth, g.config.CallerAccounts) dispatcher := auth.Accounts(g.callerAuth, []auth.Account{g.config.DispatcherAccount}) var controllers []auth.Account @@ -152,6 +156,19 @@ func (g *Gateway) Methods() auth.Methods { controllers = []auth.Account{*g.config.SandboxControllerAccount} } executors := auth.Accounts(g.callerAuth, g.config.ExecutorCallers) + controllerRule := auth.Accounts(g.callerAuth, controllers) + var err error + for _, entry := range []struct { + rule *auth.Rule + services []string + }{ + {&callers, g.config.RuntimeReaders}, {&dispatcher, g.config.SessionServices}, {&controllerRule, g.config.ControllerServices}, {&executors, g.config.ExecutorServices}, + } { + *entry.rule, err = entry.rule.AllowServiceCallers(entry.services) + if err != nil { + return nil, err + } + } return auth.Methods{ environmentruntimepb.EnvironmentRuntime_ListSessions_FullMethodName: callers, environmentruntimepb.EnvironmentRuntime_GetStatus_FullMethodName: callers, @@ -160,9 +177,9 @@ func (g *Gateway) Methods() auth.Methods { environmentruntimepb.EnvironmentRuntime_ChangeModel_FullMethodName: dispatcher, environmentruntimepb.EnvironmentRuntime_ChangeReasoningEffort_FullMethodName: dispatcher, environmentruntimepb.EnvironmentRuntime_InterruptTurn_FullMethodName: dispatcher, - environmentruntimepb.EnvironmentRuntime_Quiesce_FullMethodName: auth.Accounts(g.callerAuth, controllers), - environmentruntimepb.EnvironmentRuntime_PrepareSuspend_FullMethodName: auth.Accounts(g.callerAuth, controllers), - environmentruntimepb.EnvironmentRuntime_ResumeAdmission_FullMethodName: auth.Accounts(g.callerAuth, controllers), + environmentruntimepb.EnvironmentRuntime_Quiesce_FullMethodName: controllerRule, + environmentruntimepb.EnvironmentRuntime_PrepareSuspend_FullMethodName: controllerRule, + environmentruntimepb.EnvironmentRuntime_ResumeAdmission_FullMethodName: controllerRule, executorpb.Executor_OpenSession_FullMethodName: executors, executorpb.Executor_StartProcess_FullMethodName: executors, executorpb.Executor_ReadProcess_FullMethodName: executors, @@ -173,7 +190,7 @@ func (g *Gateway) Methods() auth.Methods { executorpb.Executor_OpenFile_FullMethodName: executors, executorpb.Executor_ReadFile_FullMethodName: executors, executorpb.Executor_CloseFile_FullMethodName: executors, - } + }, nil } // RegisterProxy installs the per-Sandbox Environment Runtime proxy. @@ -184,7 +201,7 @@ func (g *Gateway) RegisterProxy(server *grpc.Server) { // RegisterExecutor enables the typed execution path only when its own caller allowlist is // configured. Existing Gateway deployments therefore keep the RPC absent by default. func (g *Gateway) RegisterExecutor(server *grpc.Server) { - if len(g.config.ExecutorCallers) == 0 { + if len(g.config.ExecutorCallers) == 0 && len(g.config.ExecutorServices) == 0 { return } executorpb.RegisterExecutorServer(server, &executorServer{gateway: g}) diff --git a/agentplane/runner_gateway/internal/gateway/methods_test.go b/agentplane/runner_gateway/internal/gateway/methods_test.go index c850fb18ee1..210782b7b49 100644 --- a/agentplane/runner_gateway/internal/gateway/methods_test.go +++ b/agentplane/runner_gateway/internal/gateway/methods_test.go @@ -3,9 +3,15 @@ package gateway import ( "context" "maps" + "net" + "strings" "testing" "time" + "github.com/agentydragon/ducktape/agentplane/net/peertls" + transporttesting "github.com/agentydragon/ducktape/agentplane/net/peertls/testing" + "google.golang.org/grpc/credentials" + identitypb "github.com/agentydragon/ducktape/agentplane/proto/identity" environmentruntimepb "github.com/agentydragon/ducktape/agentplane/env/runtime/control/environmentruntimepb" @@ -63,7 +69,7 @@ func TestEveryRPCAdmitsOnlyItsCallers(t *testing.T) { }}, } { t.Run(test.name, func(t *testing.T) { - methods := test.gateway.Methods() + methods := gatewayMethods(t, test.gateway) connection := grpcmethods.Serve(t, methods, registerStandIns) for _, desc := range []struct { service grpc.ServiceDesc @@ -131,7 +137,7 @@ func TestUnauthorizedExecutorCallerCannotReadEnvironmentFailureState(t *testing. t.Fatal(err) } g := f.newGateway(t, "test-runner-gateway", "192.0.2.30", executorAccount) - connection := grpcmethods.Serve(t, g.Methods(), func(server *grpc.Server) { + connection := grpcmethods.Serve(t, gatewayMethods(t, g), func(server *grpc.Server) { executorpb.RegisterExecutorServer(server, &executorServer{gateway: g}) }) stream, err := executorpb.NewExecutorClient(connection).OpenSession(bearing(t, ""), &executorpb.OpenSessionRequest{ @@ -164,3 +170,65 @@ func TestCheckpointedLauncherWithIncarnationRoutesAsSuspended(t *testing.T) { t.Fatalf("checkpointed launcher phase = %s, want Suspended", phase) } } + +func TestRuntimeProxyServiceRolesStayMethodScoped(t *testing.T) { + f := newSandboxFixture(t) + g := f.newGateway(t, "test-service-proxy", "192.0.2.44", executorAccount) + g.config.RuntimeReaders = []string{peertls.AppURI} + g.config.SessionServices = []string{peertls.SessionsURI} + g.config.ControllerServices = []string{peertls.SandboxControllerURI} + g.config.ExecutorServices = []string{peertls.CentralSamplerURI} + ca := transporttesting.NewCA(t) + serverTLS, err := ca.Issue(t, peertls.SessionDispatcherURI).WorkloadServerTLSConfigWithServices(peertls.SessionDispatcherURI, true) + if err != nil { + t.Fatal(err) + } + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + server := grpc.NewServer(append(gatewayMethods(t, g).ServerOptions(0), grpc.Creds(credentials.NewTLS(serverTLS)))...) + registerStandIns(server) + go func() { _ = server.Serve(listener) }() + t.Cleanup(server.Stop) + for _, actor := range []string{peertls.AppURI, peertls.SessionsURI, peertls.SandboxControllerURI, peertls.CentralSamplerURI, peertls.SessionDispatcherURI} { + t.Run(actor, func(t *testing.T) { + clientTLS, err := ca.Issue(t, actor).ClientTLSConfig(peertls.SessionDispatcherURI) + if err != nil { + t.Fatal(err) + } + conn, err := grpc.NewClient(listener.Addr().String(), grpc.WithNoProxy(), grpc.WithTransportCredentials(credentials.NewTLS(clientTLS))) + if err != nil { + t.Fatal(err) + } + defer conn.Close() + for _, desc := range []grpc.ServiceDesc{environmentruntimepb.EnvironmentRuntime_ServiceDesc, executorpb.Executor_ServiceDesc} { + for _, method := range grpcmethods.Of(desc) { + allowed := "" + switch method.Name { + case environmentruntimepb.EnvironmentRuntime_ListSessions_FullMethodName, environmentruntimepb.EnvironmentRuntime_GetStatus_FullMethodName: + allowed = peertls.AppURI + case environmentruntimepb.EnvironmentRuntime_CreateSession_FullMethodName, environmentruntimepb.EnvironmentRuntime_SubmitInput_FullMethodName, environmentruntimepb.EnvironmentRuntime_ChangeModel_FullMethodName, environmentruntimepb.EnvironmentRuntime_ChangeReasoningEffort_FullMethodName, environmentruntimepb.EnvironmentRuntime_InterruptTurn_FullMethodName: + allowed = peertls.SessionsURI + case environmentruntimepb.EnvironmentRuntime_Quiesce_FullMethodName, environmentruntimepb.EnvironmentRuntime_PrepareSuspend_FullMethodName, environmentruntimepb.EnvironmentRuntime_ResumeAdmission_FullMethodName: + allowed = peertls.SandboxControllerURI + default: + if strings.HasPrefix(method.Name, "/"+executorpb.Executor_ServiceDesc.ServiceName+"/") { + allowed = peertls.CentralSamplerURI + } + } + want := codes.PermissionDenied + if allowed == actor { + want = codes.Unimplemented + } + ctx, cancel := context.WithTimeout(t.Context(), 5*time.Second) + err = method.Call(ctx, conn) + cancel() + if status.Code(err) != want { + t.Fatalf("%s actor=%s: %v,want=%s", method.Name, actor, err, want) + } + } + } + }) + } +} diff --git a/agentplane/runner_gateway/internal/gateway/proxy_test.go b/agentplane/runner_gateway/internal/gateway/proxy_test.go index 7ef46540d7d..dc9b42638dd 100644 --- a/agentplane/runner_gateway/internal/gateway/proxy_test.go +++ b/agentplane/runner_gateway/internal/gateway/proxy_test.go @@ -66,7 +66,7 @@ func TestRunnerProxyAuthenticatesCallerAndSandboxIdentity(t *testing.T) { if err != nil { t.Fatal(err) } - proxyServer := grpc.NewServer(gateway.Methods().ServerOptions(0)...) + proxyServer := grpc.NewServer(gatewayMethods(t, gateway).ServerOptions(0)...) gateway.RegisterProxy(proxyServer) go func() { _ = proxyServer.Serve(proxyListener) }() t.Cleanup(proxyServer.Stop) diff --git a/agentplane/runner_gateway/internal/gateway/queue_runner_integration_test.go b/agentplane/runner_gateway/internal/gateway/queue_runner_integration_test.go index d4686ce97b6..c3fd5a2334c 100644 --- a/agentplane/runner_gateway/internal/gateway/queue_runner_integration_test.go +++ b/agentplane/runner_gateway/internal/gateway/queue_runner_integration_test.go @@ -189,7 +189,7 @@ func TestQueuedDeliveryThroughGatewayRunnerRPC(t *testing.T) { if err != nil { t.Fatal(err) } - proxyServer := grpc.NewServer(gateway.Methods().ServerOptions(0)...) + proxyServer := grpc.NewServer(gatewayMethods(t, gateway).ServerOptions(0)...) gateway.RegisterProxy(proxyServer) go func() { _ = proxyServer.Serve(proxyListener) }() defer proxyServer.Stop() diff --git a/agentplane/runner_gateway/internal/gateway/runner_observer_test.go b/agentplane/runner_gateway/internal/gateway/runner_observer_test.go index 40249a93a61..132f788abe5 100644 --- a/agentplane/runner_gateway/internal/gateway/runner_observer_test.go +++ b/agentplane/runner_gateway/internal/gateway/runner_observer_test.go @@ -468,7 +468,7 @@ func testRunnerObserver(t *testing.T, mode peertls.SecurityMode) { if err != nil { t.Fatal(err) } - publicServer := grpc.NewServer(g.Methods().ServerOptions(0)...) + publicServer := grpc.NewServer(gatewayMethods(t, g).ServerOptions(0)...) g.RegisterExecutor(publicServer) go func() { _ = publicServer.Serve(publicListener) }() t.Cleanup(publicServer.Stop) diff --git a/agentplane/runner_gateway/main.go b/agentplane/runner_gateway/main.go index 44d9624356b..c265a8d2157 100644 --- a/agentplane/runner_gateway/main.go +++ b/agentplane/runner_gateway/main.go @@ -168,7 +168,7 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { if err := c.RunnerSecurity.Validate(); err != nil { return err } - rpcCredentials, err := c.RPC.ServerCredentials(peertls.SessionDispatcherURI) + rpcCredentials, err := c.RPC.ServerCredentialsWithServices(peertls.SessionDispatcherURI) if err != nil { return fmt.Errorf("Gateway RPC listener TLS: %w", err) } @@ -253,7 +253,8 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { } lifecycleManager, err := lifecycle.NewManager(lifecycle.Options{ Namespace: c.Namespace, Objects: objects, WakeBudget: lifecycleWakeBudget{store: store}, - WakeCallers: c.WakeCallers, OperatorCallers: c.OperatorCallers, + WakeCallers: c.WakeCallers, WakeServiceCallers: c.WakeServiceCallers, + OperatorCallers: c.OperatorCallers, OperatorServiceCallers: c.EnvironmentServiceOperators, EnvironmentRules: environmentRules, PolicyReady: policyReady, }) if err != nil { @@ -265,7 +266,9 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { } environmentService, err := environments.New(c.Namespace, objects, c.RuntimeCatalog, lifecycleManager, sets.New(c.EnvironmentReaders...), environments.ManagementOptions{ Managers: sets.New(c.EnvironmentManagers...), Launchers: sets.New(c.EnvironmentLaunchers...), - IdentityChanges: store, + ServiceReaders: sets.New(c.EnvironmentServiceReaders...), ServiceManagers: sets.New(c.EnvironmentServiceManagers...), + ServiceLaunchers: sets.New(c.EnvironmentServiceLaunchers...), + IdentityChanges: store, }) if err != nil { return err @@ -304,7 +307,8 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { } g, err := gateway.New(gateway.Config{ RunnerSecurity: c.RunnerSecurity, - Namespace: c.Namespace, CallerAccounts: c.GatewayCallers, ExecutorCallers: c.ExecutorCallers, + RuntimeReaders: c.RuntimeProxyReaders, SessionServices: c.RuntimeProxySessionCallers, ControllerServices: c.RuntimeProxyControllerCallers, ExecutorServices: c.ExecutorServiceCallers, + Namespace: c.Namespace, CallerAccounts: c.GatewayCallers, ExecutorCallers: c.ExecutorCallers, DispatcherAccount: deliveryAccount, SandboxControllerAccount: c.ControllerAccount, HolderName: c.PodName, HolderIP: c.PodIP, ProxyPort: int32(c.ProxyPort), @@ -359,14 +363,33 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { return ctx.Err() } service := &dispatcher.Service{ - Store: store, DispatcherCallers: sets.New(c.DispatcherCallers...), + Store: store, DispatcherCallers: sets.New(c.DispatcherCallers...), DispatcherServices: sets.New(c.DispatcherServiceCallers...), Sandboxes: catalog, CentralTarget: c.CentralSampler, Hub: hub, WatchLease: time.Duration(c.WatchLeaseSeconds) * time.Second, } - dispatcherMethods := dispatcher.Methods(auth.ReviewedServiceAccountOrIdentity(dispatcherAuth)) + dispatcherRule, err := auth.ReviewedServiceAccountOrIdentity(dispatcherAuth).AllowServiceCallers(c.DispatcherServiceCallers) + if err != nil { + return err + } + dispatcherMethods := dispatcher.Methods(dispatcherRule) methods := auth.HealthMethods() maps.Copy(methods, dispatcherMethods) - maps.Copy(methods, environments.Methods(dispatcherAuth)) + environmentMethods, err := environments.WithServiceCallers(environments.Methods(dispatcherAuth), + c.EnvironmentServiceReaders, c.EnvironmentServiceManagers, c.EnvironmentServiceLaunchers, c.EnvironmentServiceOperators) + if err != nil { + return fmt.Errorf("configure EnvironmentService service roles: %w", err) + } + if c.EnvironmentServiceDelegation != nil { + identityReader, err := auth.KubernetesIdentityUIDs(objects, c.PolicyNamespace) + if err != nil { + return err + } + environmentMethods, err = environmentMethods.WithServiceDelegation(c.EnvironmentServiceDelegation.Methods, c.EnvironmentServiceDelegation.Actors, identityReader) + if err != nil { + return fmt.Errorf("configure EnvironmentService service delegation: %w", err) + } + } + maps.Copy(methods, environmentMethods) if c.Delegation != nil { delegator := auth.Delegator{ Account: c.Delegation.Actor, IdentityCallers: c.Delegation.IdentityCallers, @@ -380,9 +403,24 @@ func serve(ctx context.Context, c config, rpcListener net.Listener) error { return fmt.Errorf("configure Runner Gateway delegation: %w", err) } } + + if c.DispatcherServiceDelegation != nil { + identityReader, err := auth.KubernetesIdentityUIDs(objects, c.PolicyNamespace) + if err != nil { + return err + } + methods, err = methods.WithServiceDelegation(c.DispatcherServiceDelegation.Methods, c.DispatcherServiceDelegation.Actors, identityReader) + if err != nil { + return fmt.Errorf("configure service delegation: %w", err) + } + } // Delegation remains scoped to the configured Dispatcher and Environment methods. The Gateway proxy and Executor - // continue using only their separate Gateway-audience bearer resolver. - maps.Copy(methods, g.Methods()) + // retain their separate method-scoped grants. + proxyMethods, err := g.Methods() + if err != nil { + return fmt.Errorf("configure Runtime proxy service admission: %w", err) + } + maps.Copy(methods, proxyMethods) serverOptions := methods.ServerOptions(0) if rpcCredentials != nil { serverOptions = append(serverOptions, grpc.Creds(rpcCredentials)) diff --git a/agentplane/runner_gateway/migrations/dispatcher/00019_service_attribution.sql b/agentplane/runner_gateway/migrations/dispatcher/00019_service_attribution.sql new file mode 100644 index 00000000000..d39c6a08fcf --- /dev/null +++ b/agentplane/runner_gateway/migrations/dispatcher/00019_service_attribution.sql @@ -0,0 +1,20 @@ +-- +goose Up +-- Certificate actors never populate account or Pod proof. Additive columns keep overlapping +-- old writers valid; old readers see service-attributed rows as unattributed. +ALTER TABLE dispatch ADD COLUMN service_attribution JSONB, + ADD CONSTRAINT ck_dispatch_service_attribution CHECK (service_attribution IS NULL OR + (jsonb_typeof(service_attribution) = 'object' AND + num_nonnulls(actor_account_namespace, actor_identity_name, delegated_identity_name) = 0)); +ALTER TABLE session ADD COLUMN service_attribution JSONB, + ADD CONSTRAINT ck_session_service_attribution CHECK (service_attribution IS NULL OR + (jsonb_typeof(service_attribution) = 'object' AND + num_nonnulls(creator_account_namespace, creator_identity_name) = 0)); + +CREATE TABLE service_wake_attempt ( + service_spiffe_id VARCHAR NOT NULL, + sandbox_uid UUID NOT NULL, + wake_generation BIGINT NOT NULL, + attempted_at TIMESTAMPTZ NOT NULL, + PRIMARY KEY (service_spiffe_id, sandbox_uid, wake_generation) +); +CREATE INDEX ix_service_wake_attempt_time ON service_wake_attempt (attempted_at); diff --git a/agentplane/runner_gateway/migrations/test_migrations.py b/agentplane/runner_gateway/migrations/test_migrations.py index 6cf2c583aa9..67915c234f0 100644 --- a/agentplane/runner_gateway/migrations/test_migrations.py +++ b/agentplane/runner_gateway/migrations/test_migrations.py @@ -41,7 +41,7 @@ def test_fresh_database_and_repeat_are_idempotent(postgres_admin_url: str, datab try: with engine.connect() as conn: assert conn.scalar(text("SELECT to_regclass('dispatch') IS NOT NULL")) - assert conn.scalar(text(f"SELECT max(version_id) FROM {_VERSION}")) == 18 + assert conn.scalar(text(f"SELECT max(version_id) FROM {_VERSION}")) == 19 finally: engine.dispose() diff --git a/agentplane/runner_gateway/migrations/upgrade_integration_test.go b/agentplane/runner_gateway/migrations/upgrade_integration_test.go index 6aa5801e1b5..abd9df35bd6 100644 --- a/agentplane/runner_gateway/migrations/upgrade_integration_test.go +++ b/agentplane/runner_gateway/migrations/upgrade_integration_test.go @@ -4,8 +4,11 @@ import ( "bytes" "database/sql" "fmt" + "io/fs" + "strconv" "strings" "testing" + "testing/fstest" _ "github.com/lib/pq" ) @@ -26,7 +29,29 @@ func TestBaselineCreatesFinalSchema(t *testing.T) { func TestCutoverRemovesPreIdentityAndUnattributedRows(t *testing.T) { db := openMigrationDatabase(t) - if err := Up(t.Context(), db); err != nil { + // Start from the actual pre-cutover schema so later migrations remain unapplied. + prior := fstest.MapFS{} + names, err := fs.Glob(scripts, "dispatcher/*.sql") + if err != nil { + t.Fatal(err) + } + for _, name := range names { + version, err := strconv.Atoi(strings.Split(strings.TrimPrefix(name, "dispatcher/"), "_")[0]) + if err != nil { + t.Fatal(err) + } + if version >= 18 { + continue + } + data, err := scripts.ReadFile(name) + if err != nil { + t.Fatal(err) + } + prior[name] = &fstest.MapFile{Data: data} + } + previous := schema + previous.Files = prior + if err := previous.Up(t.Context(), db); err != nil { t.Fatal("create final schema:", err) } const sandboxUID = "00000000-0000-0000-0000-000000000018" @@ -136,7 +161,7 @@ func TestCutoverRemovesPreIdentityAndUnattributedRows(t *testing.T) { insertSession("identity-attributed-session", "execution", executionIdentityUID, "identity") insertDispatch("identity-attributed-command", "identity-attributed-session", "identity") insertSession("effective-caller-session", "execution", executionIdentityUID, "none") - _, err := db.ExecContext(t.Context(), `UPDATE session + _, err = db.ExecContext(t.Context(), `UPDATE session SET effective_caller_account_namespace='test',effective_caller_account_name='effective',effective_caller_account_uid='effective-uid' WHERE session_id='effective-caller-session'`) if err != nil { @@ -151,9 +176,6 @@ func TestCutoverRemovesPreIdentityAndUnattributedRows(t *testing.T) { insertDispatch("standalone-action-command", "missing-session", "action") insertDispatch("standalone-wake-caller-command", "missing-session", "wake_caller") - if _, err := db.ExecContext(t.Context(), "DELETE FROM "+gooseTable+" WHERE version_id=18"); err != nil { - t.Fatal("reset migration 18 version:", err) - } if err := Up(t.Context(), db); err != nil { t.Fatal("apply cutover migration:", err) } diff --git a/agentplane/runner_gateway/settings.py b/agentplane/runner_gateway/settings.py index b9ae3808ccd..c57b2d14e65 100644 --- a/agentplane/runner_gateway/settings.py +++ b/agentplane/runner_gateway/settings.py @@ -12,7 +12,7 @@ from pydantic import BaseModel, ConfigDict, Field, StringConstraints, field_serializer, field_validator, model_validator from agentplane.identity.auth.listener import Listener -from agentplane.net.peertls.config import ClientTLS, Security +from agentplane.net.peertls.config import ClientTLS, Security, ServiceSPIFFEID from agentplane.sandboxes.runtimeconfig.model_catalog import Catalog type ServiceAccountName = Annotated[str, StringConstraints(pattern=r"^[^/]+/[^/]+$")] @@ -36,6 +36,18 @@ class DelegationSettings(BaseModel): ) +class ServiceDelegator(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + spiffe_id: ServiceSPIFFEID + connection_provenance: Literal[False] = False + + +class ServiceDelegation(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + actors: tuple[ServiceDelegator, ...] = Field(min_length=1) + methods: tuple[str, ...] = Field(min_length=1) + + class Settings(BaseModel): """Complete non-secret static settings authored in the mounted YAML file.""" @@ -71,7 +83,33 @@ class Settings(BaseModel): ) dispatcher_audience: str = Field(min_length=1, description="Dispatcher caller TokenReview audience.") gateway_audience: str = Field(min_length=1, description="Gateway caller TokenReview audience.") + runtime_proxy_readers: list[ServiceSPIFFEID] = Field( + default_factory=list, + exclude_if=lambda value: not value, + description="Exact service SPIFFE IDs admitted to Runtime diagnostics.", + ) + runtime_proxy_session_callers: list[ServiceSPIFFEID] = Field( + default_factory=list, + exclude_if=lambda value: not value, + description="Exact Sessions-role service IDs admitted to Runtime Session operations.", + ) + runtime_proxy_controller_callers: list[ServiceSPIFFEID] = Field( + default_factory=list, + exclude_if=lambda value: not value, + description="Exact controller service IDs admitted to Runtime lifecycle operations.", + ) + executor_service_callers: list[ServiceSPIFFEID] = Field( + default_factory=list, + exclude_if=lambda value: not value, + description="Exact service SPIFFE IDs admitted to Executor methods.", + ) runner_security: Security + dispatcher_service_delegation: ServiceDelegation | None = Field(default=None, exclude_if=lambda value: not value) + dispatcher_service_callers: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact service SVIDs granted Dispatcher component authority.", + ) dispatcher_callers: Set[ServiceAccountName] = Field( min_length=1, description="ServiceAccounts admitted as Dispatcher callers, as namespace/name." ) @@ -84,6 +122,31 @@ class Settings(BaseModel): environment_launchers: Set[ServiceAccountName] = Field( default_factory=frozenset, description="Direct Pod-bound Action brokers authorized to create Environments." ) + environment_service_readers: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact service SVIDs granted broad EnvironmentService inventory reads.", + ) + environment_service_managers: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact service SVIDs granted EnvironmentService management mutations.", + ) + environment_service_launchers: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact service SVIDs granted creation with a complete Action origin.", + ) + environment_service_operators: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact service SVIDs granted explicit EnvironmentService resume authority.", + ) + environment_service_delegation: ServiceDelegation | None = Field( + default=None, + exclude_if=lambda value: not value, + description="Independent certificate actors allowed to call exact read-only EnvironmentService methods as a live Identity.", + ) delegation: DelegationSettings | None = Field( default=None, description="Independent Actions delegation trust; does not grant direct or privileged caller access.", @@ -99,6 +162,11 @@ class Settings(BaseModel): "like omission, grants no wake eligibility." ), ) + wake_service_callers: Set[ServiceSPIFFEID] = Field( + default_factory=frozenset, + exclude_if=lambda value: not value, + description="Exact Dispatcher service SVIDs allowed to request automatic Sandbox wake, subject to wakeOnSend.", + ) operator_callers: Set[ServiceAccountName] | None = Field( default=None, description=( @@ -130,6 +198,57 @@ def validate_operator_callers(self) -> Settings: raise ValueError("operator_callers must be a subset of dispatcher_callers") return self + @model_validator(mode="after") + def validate_service_caller_scopes(self) -> Settings: + service_callers_present = any( + ( + self.runtime_proxy_readers, + self.runtime_proxy_session_callers, + self.runtime_proxy_controller_callers, + self.executor_service_callers, + self.dispatcher_service_callers, + self.environment_service_readers, + self.environment_service_managers, + self.environment_service_launchers, + self.environment_service_operators, + self.wake_service_callers, + ) + ) + service_delegation_present = ( + self.dispatcher_service_delegation is not None or self.environment_service_delegation is not None + ) + spiffe_listener = any(method.mode == "spiffe" for method in self.rpc.auth) + if (service_callers_present or service_delegation_present) and not spiffe_listener: + raise ValueError("service caller settings require rpc.auth to include spiffe") + if not self.wake_service_callers <= self.dispatcher_service_callers: + raise ValueError("wake_service_callers must be a subset of dispatcher_service_callers") + if self.dispatcher_service_delegation is not None: + dispatcher_methods = { + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/ChangeModel", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/ChangeReasoningEffort", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/InterruptTurn", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/CreateSession", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/GetSession", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/ListSessions", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/GetDispatch", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/WatchDispatches", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/ListDispatches", + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/CancelDispatch", + } + if not set(self.dispatcher_service_delegation.methods) <= dispatcher_methods: + raise ValueError("dispatcher_service_delegation methods must belong to SessionDispatcher") + if self.environment_service_delegation is not None: + read_methods = { + "/agentplane.environments.v1.EnvironmentService/ListEnvironments", + "/agentplane.environments.v1.EnvironmentService/WatchEnvironments", + "/agentplane.environments.v1.EnvironmentService/GetEnvironment", + "/agentplane.environments.v1.EnvironmentService/WatchEnvironment", + } + if not set(self.environment_service_delegation.methods) <= read_methods: + raise ValueError("environment_service_delegation methods must be read-only EnvironmentService methods") + return self + @model_validator(mode="after") def rpc_audiences_are_method_scoped(self) -> Settings: audiences = { @@ -162,6 +281,17 @@ def validate_kubernetes_client_qps(cls, value: float) -> float: def sorted_accounts(self, value: Set[str]) -> list[str]: return sorted(value, key=lambda account: tuple(account.split("/", maxsplit=1))) + @field_serializer( + "environment_service_readers", + "environment_service_managers", + "environment_service_launchers", + "environment_service_operators", + "wake_service_callers", + "dispatcher_service_callers", + ) + def sorted_service_callers(self, value: Set[str]) -> list[str]: + return sorted(value) + @field_serializer("wake_callers", "operator_callers") def sorted_optional_accounts(self, value: Set[str] | None) -> list[str] | None: return None if value is None else sorted(value, key=lambda account: tuple(account.split("/", maxsplit=1))) diff --git a/agentplane/runner_gateway/test_settings.py b/agentplane/runner_gateway/test_settings.py index bcb0d6a2040..312e395227b 100644 --- a/agentplane/runner_gateway/test_settings.py +++ b/agentplane/runner_gateway/test_settings.py @@ -99,6 +99,98 @@ def test_control_service_tls_pins_session_dispatcher_identity() -> None: Settings.model_validate(source) +def test_service_settings_require_spiffe_listener_authentication() -> None: + source = _settings({"mode": "insecure"}).model_dump(mode="json") + source["rpc"]["auth"] = [source["rpc"]["auth"][1]] + source["dispatcher_service_callers"] = ["spiffe://agentplane/actions"] + with pytest.raises(ValidationError, match=r"service caller settings require rpc\.auth to include spiffe"): + Settings.model_validate(source) + + source = _settings({"mode": "insecure"}).model_dump(mode="json") + source["rpc"]["auth"] = [source["rpc"]["auth"][1]] + source["environment_service_delegation"] = { + "actors": [{"spiffe_id": "spiffe://agentplane/actions"}], + "methods": ["/agentplane.environments.v1.EnvironmentService/ListEnvironments"], + } + with pytest.raises(ValidationError, match=r"service caller settings require rpc\.auth to include spiffe"): + Settings.model_validate(source) + + +def test_empty_service_settings_are_omitted_from_deployment_output() -> None: + source = _settings({"mode": "insecure"}).model_dump(mode="json") + new_optional_fields = { + "runtime_proxy_readers", + "runtime_proxy_session_callers", + "runtime_proxy_controller_callers", + "executor_service_callers", + "dispatcher_service_delegation", + "dispatcher_service_callers", + "environment_service_readers", + "environment_service_managers", + "environment_service_launchers", + "environment_service_operators", + "environment_service_delegation", + "wake_service_callers", + } + assert new_optional_fields.isdisjoint(source) + + +def test_dispatcher_service_delegation_is_scoped_to_dispatcher_and_action_provenance() -> None: + source = _settings({"mode": "insecure"}).model_dump(mode="json") + source["dispatcher_service_delegation"] = { + "actors": [{"spiffe_id": "spiffe://agentplane/actions"}], + "methods": ["/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput"], + } + assert Settings.model_validate(source).dispatcher_service_delegation is not None + + source["dispatcher_service_delegation"]["methods"] = [ + "/agentplane.environments.v1.EnvironmentService/ListEnvironments" + ] + with pytest.raises(ValidationError, match="dispatcher_service_delegation methods must belong to SessionDispatcher"): + Settings.model_validate(source) + + source["dispatcher_service_delegation"]["methods"] = [ + "/agentplane.runner_gateway.dispatcher.SessionDispatcher/SubmitInput" + ] + source["dispatcher_service_delegation"]["actors"][0]["connection_provenance"] = True + with pytest.raises(ValidationError): + Settings.model_validate(source) + + source["dispatcher_service_delegation"]["actors"] = [ + {"spiffe_id": "spiffe://agentplane/test/sandbox/credential-id"} + ] + with pytest.raises(ValidationError): + Settings.model_validate(source) + + +def test_environment_service_delegation_is_read_only_and_action_only() -> None: + source = _settings({"mode": "insecure"}).model_dump(mode="json") + source["environment_service_delegation"] = { + "actors": [{"spiffe_id": "spiffe://agentplane/actions"}], + "methods": ["/agentplane.environments.v1.EnvironmentService/ListEnvironments"], + } + assert Settings.model_validate(source).environment_service_delegation is not None + + source["environment_service_delegation"]["methods"] = [ + "/agentplane.environments.v1.EnvironmentService/CreateEnvironment" + ] + with pytest.raises(ValidationError, match="environment_service_delegation methods must be read-only"): + Settings.model_validate(source) + + source["environment_service_delegation"]["methods"] = [ + "/agentplane.environments.v1.EnvironmentService/ListEnvironments" + ] + source["environment_service_delegation"]["actors"][0]["connection_provenance"] = True + with pytest.raises(ValidationError): + Settings.model_validate(source) + + source["environment_service_delegation"]["actors"] = [ + {"spiffe_id": "spiffe://agentplane/test/sandbox/credential-id"} + ] + with pytest.raises(ValidationError): + Settings.model_validate(source) + + def test_static_settings_reject_runtime_inputs_and_legacy_namespace_authority() -> None: source = _settings({"mode": "insecure"}).model_dump(mode="json") source["dispatcher_database_url"] = "postgresql://secret.invalid/dispatcher"