add kubernetes waiting container backend service and API (#39317)

* add kubernetes waiting container backend service and API

* remove other resource type

* add unit test

* fix after rebasing

* address feedback

* remove KubeWaitingContainerRequest

* check for nil in validation func

* remove UnwrapResource153

* fix test

* address feedback

* address linter issues

* fix cache watch kind test

api/client/client.go

@@ -48,6 +48,7 @@ import (
 	"github.com/gravitational/teleport/api/client/accesslist"
 	"github.com/gravitational/teleport/api/client/discoveryconfig"
 	"github.com/gravitational/teleport/api/client/externalauditstorage"
+	kubewaitingcontainerclient "github.com/gravitational/teleport/api/client/kubewaitingcontainer"
 	"github.com/gravitational/teleport/api/client/okta"
 	"github.com/gravitational/teleport/api/client/proto"
 	"github.com/gravitational/teleport/api/client/scim"
@@ -65,6 +66,7 @@ import (
 	externalauditstoragev1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/externalauditstorage/v1"
 	integrationpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/integration/v1"
 	kubeproto "github.com/gravitational/teleport/api/gen/proto/go/teleport/kube/v1"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	loginrulepb "github.com/gravitational/teleport/api/gen/proto/go/teleport/loginrule/v1"
 	machineidv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/machineid/v1"
 	oktapb "github.com/gravitational/teleport/api/gen/proto/go/teleport/okta/v1"
@@ -3271,6 +3273,40 @@ func (c *Client) DeleteAllKubernetesClusters(ctx context.Context) error {
 	return trace.Wrap(err)
 }
 
+// GetKubernetesWaitingContainerClient an unadorned KubeWaitingContainers
+// client, using the underlying Auth gRPC connection.
+func (c *Client) GetKubernetesWaitingContainerClient() *kubewaitingcontainerclient.Client {
+	return kubewaitingcontainerclient.NewClient(kubewaitingcontainerpb.NewKubeWaitingContainersServiceClient(c.conn))
+}
+
+// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (c *Client) ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error) {
+	return c.GetKubernetesWaitingContainerClient().ListKubernetesWaitingContainers(ctx, pageSize, pageToken)
+}
+
+// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (c *Client) GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	return c.GetKubernetesWaitingContainerClient().GetKubernetesWaitingContainer(ctx, req)
+}
+
+// CreateKubernetesWaitingContainer creates a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (c *Client) CreateKubernetesWaitingContainer(ctx context.Context, waitingPod *kubewaitingcontainerpb.KubernetesWaitingContainer) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	return c.GetKubernetesWaitingContainerClient().CreateKubernetesWaitingContainer(ctx, waitingPod)
+}
+
+// DeleteKubernetesWaitingContainer deletes a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (c *Client) DeleteKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest) error {
+	return c.GetKubernetesWaitingContainerClient().DeleteKubernetesWaitingContainer(ctx, req)
+}
+
 // CreateDatabase creates a new database resource.
 func (c *Client) CreateDatabase(ctx context.Context, database types.Database) error {
 	databaseV3, ok := database.(*types.DatabaseV3)

api/client/events.go

@@ -18,6 +18,7 @@ import (
 	"github.com/gravitational/trace"
 
 	"github.com/gravitational/teleport/api/client/proto"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
 	accesslistv1conv "github.com/gravitational/teleport/api/types/accesslist/convert/v1"
@@ -49,6 +50,13 @@ func EventToGRPC(in types.Event) (*proto.Event, error) {
 		return &out, nil
 	}
 	switch r := in.Resource.(type) {
+	case types.Resource153Unwrapper:
+		switch r := r.Unwrap().(type) {
+		case *kubewaitingcontainerpb.KubernetesWaitingContainer:
+			out.Resource = &proto.Event_KubernetesWaitingContainer{
+				KubernetesWaitingContainer: r,
+			}
+		}
 	case *types.ResourceHeader:
 		out.Resource = &proto.Event_ResourceHeader{
 			ResourceHeader: r,
@@ -453,6 +461,9 @@ func EventFromGRPC(in *proto.Event) (*types.Event, error) {
 			return nil, trace.Wrap(err)
 		}
 		return &out, nil
+	} else if r := in.GetKubernetesWaitingContainer(); r != nil {
+		out.Resource = types.Resource153ToLegacy(r)
+		return &out, nil
 	} else {
 		return nil, trace.BadParameter("received unsupported resource %T", in.Resource)
 	}

api/client/kubewaitingcontainer/kubewaitingcontainer.go

@@ -0,0 +1,129 @@
+// Copyright 2023 Gravitational, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kubewaitingcontainer
+
+import (
+	"context"
+
+	"github.com/gravitational/trace"
+
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+)
+
+// Client is a KubeWaitingContainers client.
+type Client struct {
+	grpcClient kubewaitingcontainerpb.KubeWaitingContainersServiceClient
+}
+
+// NewClient creates a new KubeWaitingContainers client.
+func NewClient(grpcClient kubewaitingcontainerpb.KubeWaitingContainersServiceClient) *Client {
+	return &Client{
+		grpcClient: grpcClient,
+	}
+}
+
+// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (c *Client) ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error) {
+	resp, err := c.grpcClient.ListKubernetesWaitingContainers(ctx, &kubewaitingcontainerpb.ListKubernetesWaitingContainersRequest{
+		PageSize:  int32(pageSize),
+		PageToken: pageToken,
+	})
+	if err != nil {
+		return nil, "", trace.Wrap(err)
+	}
+
+	return resp.WaitingContainers, resp.NextPageToken, nil
+}
+
+// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that is waiting to be created until moderated
+// session conditions are met.
+func (c *Client) GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	if req.Username == "" {
+		return nil, trace.BadParameter("missing Username")
+	}
+	if req.Cluster == "" {
+		return nil, trace.BadParameter("missing Cluster")
+	}
+	if req.Namespace == "" {
+		return nil, trace.BadParameter("missing Namespace")
+	}
+	if req.PodName == "" {
+		return nil, trace.BadParameter("missing PodName")
+	}
+	if req.ContainerName == "" {
+		return nil, trace.BadParameter("missing ContainerName")
+	}
+
+	out, err := c.grpcClient.GetKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest{
+		Username:      req.Username,
+		Cluster:       req.Cluster,
+		Namespace:     req.Namespace,
+		PodName:       req.PodName,
+		ContainerName: req.ContainerName,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// CreateKubernetesWaitingContainer creates a Kubernetes ephemeral
+// container that is waiting to be created until moderated
+// session conditions are met.
+func (c *Client) CreateKubernetesWaitingContainer(ctx context.Context, waitingPod *kubewaitingcontainerpb.KubernetesWaitingContainer) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	out, err := c.grpcClient.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+		WaitingContainer: waitingPod,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// DeleteKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that is waiting to be created until moderated
+// session conditions are met.
+func (c *Client) DeleteKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest) error {
+	if req.Username == "" {
+		return trace.BadParameter("missing Username")
+	}
+	if req.Cluster == "" {
+		return trace.BadParameter("missing Cluster")
+	}
+	if req.Namespace == "" {
+		return trace.BadParameter("missing Namespace")
+	}
+	if req.PodName == "" {
+		return trace.BadParameter("missing PodName")
+	}
+	if req.ContainerName == "" {
+		return trace.BadParameter("missing ContainerName")
+	}
+
+	_, err := c.grpcClient.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+		Username:      req.Username,
+		Cluster:       req.Cluster,
+		Namespace:     req.Namespace,
+		PodName:       req.PodName,
+		ContainerName: req.ContainerName,
+	})
+
+	return trace.Wrap(err)
+}

api/types/constants.go

@@ -260,6 +260,11 @@ const (
 	// KindKubeIngress is a Kubernetes Ingress resource type.
 	KindKubeIngress = "ingress"
 
+	// KindKubeWaitingContainer is a Kubernetes ephemeral
+	// container that are waiting to be created until moderated
+	// session conditions are met.
+	KindKubeWaitingContainer = "kube_ephemeral_container"
+
 	// KindToken is a provisioning token resource
 	KindToken = "token"
 

api/types/kubewaitingcontainer/kubewaitingcontainer.go

@@ -0,0 +1,90 @@
+/*
+Copyright 2024 Gravitational, Inc.
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+*/
+
+package kubewaitingcontainer
+
+import (
+	"time"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	headerv1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/header/v1"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+	"github.com/gravitational/teleport/api/types"
+)
+
+// NewKubeWaitingContainer creates a new Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func NewKubeWaitingContainer(name string, spec *kubewaitingcontainerpb.KubernetesWaitingContainerSpec) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	waitingCont := &kubewaitingcontainerpb.KubernetesWaitingContainer{
+		Kind:    types.KindKubeWaitingContainer,
+		Version: types.V1,
+		Metadata: &headerv1.Metadata{
+			Name:    name,
+			Expires: timestamppb.New(time.Now().Add(time.Hour)),
+		},
+		Spec: spec,
+	}
+	if err := ValidateKubeWaitingContainer(waitingCont); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return waitingCont, nil
+}
+
+// ValidateKubeWaitingContainer checks that required parameters are set
+// for the specified KubeWaitingContainer
+func ValidateKubeWaitingContainer(k *kubewaitingcontainerpb.KubernetesWaitingContainer) error {
+	if k == nil {
+		return trace.BadParameter("KubernetesWaitingContainer is nil")
+	}
+	if k.Metadata == nil {
+		return trace.BadParameter("Metadata is nil")
+	}
+	if k.Spec == nil {
+		return trace.BadParameter("Spec is nil")
+	}
+
+	if k.Spec.Username == "" {
+		return trace.BadParameter("Username is unset")
+	}
+	if k.Spec.Cluster == "" {
+		return trace.BadParameter("KubeCluster is unset")
+	}
+	if k.Spec.Namespace == "" {
+		return trace.BadParameter("KubeNamespace is unset")
+	}
+	if k.Spec.PodName == "" {
+		return trace.BadParameter("PodName is unset")
+	}
+	if k.Spec.ContainerName == "" {
+		return trace.BadParameter("ContainerName is unset")
+	}
+	if len(k.Spec.Patch) == 0 {
+		return trace.BadParameter("Patch is unset")
+	}
+
+	if k.Metadata.Name == "" {
+		return trace.BadParameter("Name is unset")
+	}
+	if k.Metadata.Name != k.Spec.ContainerName {
+		return trace.BadParameter("Name must be ContainerName")
+	}
+
+	return nil
+}

api/types/resource_153.go

@@ -125,6 +125,12 @@ func Resource153ToLegacy(r Resource153) Resource {
 	return &resource153ToLegacyAdapter{inner: r}
 }
 
+// Resource153Unwrapper returns a legacy [Resource] type from a wrapped RFD
+// 153 style resource
+type Resource153Unwrapper interface {
+	Unwrap() Resource153
+}
+
 type resource153ToLegacyAdapter struct {
 	inner Resource153
 }

lib/auth/accesspoint/accesspoint.go

@@ -149,6 +149,7 @@ func NewAccessCache(cfg AccessCacheConfig) (*cache.Cache, error) {
 		DiscoveryConfigs:        cfg.Services.DiscoveryConfigClient(),
 		WebSession:              cfg.Services.WebSessions(),
 		WebToken:                cfg.Services.WebTokens(),
+		KubeWaitingContainers:   cfg.Services,
 		Component:               teleport.Component(component...),
 		MetricComponent:         teleport.Component(metricComponent...),
 		Tracer:                  tracer,

lib/auth/api.go

@@ -28,6 +28,7 @@ import (
 
 	"github.com/gravitational/teleport/api/client/proto"
 	integrationpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/integration/v1"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
 	"github.com/gravitational/teleport/api/types/discoveryconfig"
@@ -253,6 +254,16 @@ type ReadProxyAccessPoint interface {
 	// GetKubernetesServers returns a list of kubernetes servers registered in the cluster
 	GetKubernetesServers(context.Context) ([]types.KubeServer, error)
 
+	// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+	// containers that are waiting to be created until moderated
+	// session conditions are met.
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+
+	// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+	// container that are waiting to be created until moderated
+	// session conditions are met.
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+
 	// GetDatabaseServers returns all registered database proxy servers.
 	GetDatabaseServers(ctx context.Context, namespace string, opts ...services.MarshalOption) ([]types.DatabaseServer, error)
 
@@ -450,6 +461,16 @@ type ReadKubernetesAccessPoint interface {
 	// GetKubernetesServers returns a list of kubernetes servers registered in the cluster
 	GetKubernetesServers(context.Context) ([]types.KubeServer, error)
 
+	// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+	// containers that are waiting to be created until moderated
+	// session conditions are met.
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+
+	// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+	// container that are waiting to be created until moderated
+	// session conditions are met.
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+
 	// GetKubernetesClusters returns all kubernetes cluster resources.
 	GetKubernetesClusters(ctx context.Context) ([]types.KubeCluster, error)
 	// GetKubernetesCluster returns the specified kubernetes cluster resource.
@@ -1010,6 +1031,16 @@ type Cache interface {
 	// GetKubernetesServers returns a list of kubernetes servers registered in the cluster
 	GetKubernetesServers(context.Context) ([]types.KubeServer, error)
 
+	// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+	// containers that are waiting to be created until moderated
+	// session conditions are met.
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+
+	// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+	// container that are waiting to be created until moderated
+	// session conditions are met.
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+
 	// GetDatabaseServers returns all registered database proxy servers.
 	GetDatabaseServers(ctx context.Context, namespace string, opts ...services.MarshalOption) ([]types.DatabaseServer, error)
 

lib/auth/auth.go

@@ -347,6 +347,13 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 		return nil, trace.Wrap(err)
 	}
 
+	if cfg.KubeWaitingContainers == nil {
+		cfg.KubeWaitingContainers, err = local.NewKubeWaitingContainerService(cfg.Backend)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+	}
+
 	closeCtx, cancelFunc := context.WithCancel(context.TODO())
 	services := &Services{
 		Trust:                     cfg.Trust,
@@ -381,6 +388,7 @@ func NewServer(cfg *InitConfig, opts ...ServerOption) (*Server, error) {
 		Assistant:                 cfg.Assist,
 		UserPreferences:           cfg.UserPreferences,
 		PluginData:                cfg.PluginData,
+		KubeWaitingContainer:      cfg.KubeWaitingContainers,
 	}
 
 	as := Server{
@@ -535,6 +543,7 @@ type Services struct {
 	types.Events
 	events.AuditLogSessionStreamer
 	services.SecReports
+	services.KubeWaitingContainer
 }
 
 // SecReportsClient returns the security reports client.
@@ -586,6 +595,12 @@ func (r *Services) UserLoginStateClient() services.UserLoginStates {
 	return r
 }
 
+// KubernetesWaitingContainerClient returns the Kubernetes waiting
+// container client.
+func (r *Services) KubernetesWaitingContainerClient() services.KubeWaitingContainer {
+	return r
+}
+
 var (
 	generateRequestsCount = prometheus.NewCounter(
 		prometheus.CounterOpts{

lib/auth/clt.go

@@ -824,6 +824,7 @@ type ClientI interface {
 	services.ConnectionsDiagnostic
 	services.SAMLIdPSession
 	services.Integrations
+	services.KubeWaitingContainer
 	types.Events
 
 	types.WebSessionsGetter

lib/auth/grpcserver.go

@@ -53,6 +53,7 @@ import (
 	dbobjectimportrulev12 "github.com/gravitational/teleport/api/gen/proto/go/teleport/dbobjectimportrule/v1"
 	discoveryconfigpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/discoveryconfig/v1"
 	integrationpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/integration/v1"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	loginrulepb "github.com/gravitational/teleport/api/gen/proto/go/teleport/loginrule/v1"
 	machineidv1pb "github.com/gravitational/teleport/api/gen/proto/go/teleport/machineid/v1"
 	mfav1 "github.com/gravitational/teleport/api/gen/proto/go/teleport/mfa/v1"
@@ -73,6 +74,7 @@ import (
 	"github.com/gravitational/teleport/lib/auth/dbobjectimportrule/dbobjectimportrulev1"
 	"github.com/gravitational/teleport/lib/auth/discoveryconfig/discoveryconfigv1"
 	integrationService "github.com/gravitational/teleport/lib/auth/integration/integrationv1"
+	kubewaitingcontainerv1 "github.com/gravitational/teleport/lib/auth/kubewaitingcontainer"
 	"github.com/gravitational/teleport/lib/auth/loginrule"
 	"github.com/gravitational/teleport/lib/auth/machineid/machineidv1"
 	"github.com/gravitational/teleport/lib/auth/okta"
@@ -5409,6 +5411,17 @@ func NewGRPCServer(cfg GRPCServerConfig) (*GRPCServer, error) {
 
 	clusterconfigpb.RegisterClusterConfigServiceServer(server, clusterConfigService)
 
+	// Initialize and register the Kubernetes waiting container service.
+	kubeWaitingContsServer, err := kubewaitingcontainerv1.NewService(kubewaitingcontainerv1.ServiceConfig{
+		Authorizer: cfg.Authorizer,
+		Backend:    cfg.AuthServer.Services,
+		Cache:      cfg.AuthServer.Cache,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	kubewaitingcontainerpb.RegisterKubeWaitingContainersServiceServer(server, kubeWaitingContsServer)
+
 	// Only register the service if this is an open source build. Enterprise builds
 	// register the actual service via an auth plugin, if we register here then all
 	// Enterprise builds would fail with a duplicate service registered error.

lib/auth/init.go

@@ -286,6 +286,11 @@ type InitConfig struct {
 
 	// CloudClients provides clients for various cloud providers.
 	CloudClients cloud.Clients
+
+	// KubeWaitingContainers is a service that manages
+	// Kubernetes ephemeral containers that are waiting
+	// to be created until moderated session conditions are met.
+	KubeWaitingContainers services.KubeWaitingContainer
 }
 
 // Init instantiates and configures an instance of AuthServer

lib/auth/kubewaitingcontainer/service.go

@@ -0,0 +1,214 @@
+// Copyright 2024 Gravitational, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kubewaitingcontainer
+
+import (
+	"context"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/emptypb"
+
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/authz"
+	"github.com/gravitational/teleport/lib/services"
+)
+
+// ServiceConfig holds configuration options for
+// the Kubernetes waiting container gRPC service.
+type ServiceConfig struct {
+	// Authorizer is the authorizer used to check access to resources.
+	Authorizer authz.Authorizer
+	// Backend is the backend used to store Kubernetes waiting containers.
+	Backend services.KubeWaitingContainer
+	// Cache is the cache used to store Kubernetes waiting containers.
+	Cache Cache
+}
+
+// Cache is responsible for getting Kubernetes
+// ephemeral containers that are waiting to be created until moderated
+// session conditions are met.
+type Cache interface {
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+}
+
+// Service implements the teleport.kubewaitingcontainer.v1.KubernetesWaitingContainer
+// RPC service.
+type Service struct {
+	kubewaitingcontainerpb.UnimplementedKubeWaitingContainersServiceServer
+
+	authorizer authz.Authorizer
+	backend    services.KubeWaitingContainer
+	cache      Cache
+}
+
+// NewService returns a new Kubernetes waiting container gRPC service.
+func NewService(cfg ServiceConfig) (*Service, error) {
+	switch {
+	case cfg.Backend == nil:
+		return nil, trace.BadParameter("backend is required")
+	case cfg.Authorizer == nil:
+		return nil, trace.BadParameter("authorizer is required")
+	case cfg.Cache == nil:
+		return nil, trace.BadParameter("cache is required")
+	}
+
+	return &Service{
+		authorizer: cfg.Authorizer,
+		backend:    cfg.Backend,
+		cache:      cfg.Cache,
+	}, nil
+}
+
+// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (s *Service) ListKubernetesWaitingContainers(ctx context.Context, req *kubewaitingcontainerpb.ListKubernetesWaitingContainersRequest) (*kubewaitingcontainerpb.ListKubernetesWaitingContainersResponse, error) {
+	authCtx, err := s.authorizer.Authorize(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := authCtx.CheckAccessToKind(types.KindKubeWaitingContainer, types.VerbList, types.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if !isKubeSvcOrProxy(authCtx) {
+		return nil, trace.AccessDenied("unauthorized to list Kubernetes waiting container resources")
+	}
+
+	conts, nextToken, err := s.cache.ListKubernetesWaitingContainers(ctx, int(req.PageSize), req.PageToken)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &kubewaitingcontainerpb.ListKubernetesWaitingContainersResponse{
+		WaitingContainers: conts,
+		NextPageToken:     nextToken,
+	}, nil
+}
+
+// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (s *Service) GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	if req.Username == "" {
+		return nil, trace.BadParameter("missing username")
+	}
+	if req.Cluster == "" {
+		return nil, trace.BadParameter("missing cluster")
+	}
+	if req.Namespace == "" {
+		return nil, trace.BadParameter("missing namespace")
+	}
+	if req.PodName == "" {
+		return nil, trace.BadParameter("missing pod name")
+	}
+	if req.ContainerName == "" {
+		return nil, trace.BadParameter("missing container name")
+	}
+
+	authCtx, err := s.authorizer.Authorize(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := authCtx.CheckAccessToKind(types.KindKubeWaitingContainer, types.VerbRead); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if !isKubeSvcOrProxy(authCtx) {
+		return nil, trace.AccessDenied("unauthorized to read Kubernetes waiting container resources")
+	}
+
+	out, err := s.cache.GetKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest{
+		Username:      req.Username,
+		Cluster:       req.Cluster,
+		Namespace:     req.Namespace,
+		PodName:       req.PodName,
+		ContainerName: req.ContainerName,
+	})
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// CreateKubernetesWaitingContainer creates a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (s *Service) CreateKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	authCtx, err := s.authorizer.Authorize(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := authCtx.CheckAccessToKind(types.KindKubeWaitingContainer, types.VerbCreate); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if !isKubeSvcOrProxy(authCtx) {
+		return nil, trace.AccessDenied("unauthorized to create Kubernetes waiting container resources")
+	}
+
+	out, err := s.backend.CreateKubernetesWaitingContainer(ctx, req.WaitingContainer)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// DeleteKubernetesWaitingContainer deletes a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (s *Service) DeleteKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest) (*emptypb.Empty, error) {
+	if req.Username == "" {
+		return nil, trace.BadParameter("missing username")
+	}
+	if req.Cluster == "" {
+		return nil, trace.BadParameter("missing cluster")
+	}
+	if req.Namespace == "" {
+		return nil, trace.BadParameter("missing namespace")
+	}
+	if req.PodName == "" {
+		return nil, trace.BadParameter("missing pod name")
+	}
+	if req.ContainerName == "" {
+		return nil, trace.BadParameter("missing container name")
+	}
+
+	authCtx, err := s.authorizer.Authorize(ctx)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if err := authCtx.CheckAccessToKind(types.KindKubeWaitingContainer, types.VerbDelete); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if !isKubeSvcOrProxy(authCtx) {
+		return nil, trace.AccessDenied("unauthorized to delete Kubernetes waiting container resources")
+	}
+
+	return &emptypb.Empty{}, trace.Wrap(s.backend.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+		Username:      req.Username,
+		Cluster:       req.Cluster,
+		Namespace:     req.Namespace,
+		PodName:       req.PodName,
+		ContainerName: req.ContainerName,
+	}))
+}
+
+// isKubeSvcOrProxy returns true if the given context has the builtin role
+// of "kube" or "proxy".
+func isKubeSvcOrProxy(authCtx *authz.Context) bool {
+	return authz.HasBuiltinRole(*authCtx, string(types.RoleKube)) || authz.HasBuiltinRole(*authCtx, string(types.RoleProxy))
+}

lib/auth/kubewaitingcontainer/service_test.go

@@ -0,0 +1,392 @@
+// Copyright 2024 Gravitational, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package kubewaitingcontainer
+
+import (
+	"context"
+	"testing"
+
+	"github.com/google/uuid"
+	"github.com/gravitational/trace"
+	"github.com/stretchr/testify/require"
+
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/kubewaitingcontainer"
+	"github.com/gravitational/teleport/lib/authz"
+	"github.com/gravitational/teleport/lib/backend/memory"
+	"github.com/gravitational/teleport/lib/services/local"
+	"github.com/gravitational/teleport/lib/tlsca"
+)
+
+func TestKubeWaitingContainerServiceCRUD(t *testing.T) {
+	t.Parallel()
+
+	const username = "user"
+	const cluster = "cluster"
+	const namespace = "default"
+	const podName = "pod"
+
+	sampleKubeWaitingContFn := func(t *testing.T, name string) *kubewaitingcontainerpb.KubernetesWaitingContainer {
+		wc, err := kubewaitingcontainer.NewKubeWaitingContainer(
+			name,
+			&kubewaitingcontainerpb.KubernetesWaitingContainerSpec{
+				Username:      username,
+				Cluster:       cluster,
+				Namespace:     namespace,
+				PodName:       podName,
+				ContainerName: name,
+				Patch:         []byte("patch"),
+			},
+		)
+		require.NoError(t, err)
+		return wc
+	}
+
+	kubeAuthFn := authz.AuthorizerFunc(func(ctx context.Context) (*authz.Context, error) {
+		authzContext, err := authz.ContextForBuiltinRole(authz.BuiltinRole{
+			Role:     types.RoleKube,
+			Username: string(types.RoleKube),
+		}, &types.SessionRecordingConfigV2{})
+		require.NoError(t, err)
+		return authzContext, nil
+	})
+
+	tt := []struct {
+		Name       string
+		Authorizer func(t *testing.T, client localClient) authz.Authorizer
+		Setup      func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string)
+		Test       func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string)
+	}{
+		// List
+		{
+			Name: "allowed list access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Setup: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				for i := 0; i < 10; i++ {
+					_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+						WaitingContainer: sampleKubeWaitingContFn(t, uuid.NewString()),
+					})
+					require.NoError(t, err)
+				}
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.ListKubernetesWaitingContainers(ctx, &kubewaitingcontainerpb.ListKubernetesWaitingContainersRequest{})
+				require.NoError(t, err)
+			},
+		},
+		{
+			Name: "not allowed list access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return authz.AuthorizerFunc(func(ctx context.Context) (*authz.Context, error) {
+					return authorizerForDummyUser(t, ctx, client, []string{types.VerbRead, types.VerbList}), nil
+				})
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.ListKubernetesWaitingContainers(ctx, &kubewaitingcontainerpb.ListKubernetesWaitingContainersRequest{})
+				require.Error(t, err)
+				require.True(t, trace.IsAccessDenied(err))
+			},
+		},
+
+		// Get
+		{
+			Name: "allowed get access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Setup: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.NoError(t, err)
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				out, err := resourceSvc.GetKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.NoError(t, err)
+				require.Equal(t, wcName, out.Metadata.Name)
+				require.Equal(t, username, out.Spec.Username)
+				require.Equal(t, cluster, out.Spec.Cluster)
+				require.Equal(t, namespace, out.Spec.Namespace)
+				require.Equal(t, podName, out.Spec.PodName)
+				require.Equal(t, wcName, out.Spec.ContainerName)
+			},
+		},
+		{
+			Name: "not allowed get access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return authz.AuthorizerFunc(func(ctx context.Context) (*authz.Context, error) {
+					return authorizerForDummyUser(t, ctx, client, []string{types.VerbRead, types.VerbList}), nil
+				})
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.GetKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsAccessDenied(err))
+			},
+		},
+		{
+			Name: "get nonexistent resource",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.GetKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsNotFound(err))
+			},
+		},
+
+		// Create
+		{
+			Name: "allowed create access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				out, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.NoError(t, err)
+				require.Equal(t, wcName, out.Metadata.Name)
+				require.Equal(t, username, out.Spec.Username)
+				require.Equal(t, cluster, out.Spec.Cluster)
+				require.Equal(t, namespace, out.Spec.Namespace)
+				require.Equal(t, podName, out.Spec.PodName)
+				require.Equal(t, wcName, out.Spec.ContainerName)
+			},
+		},
+		{
+			Name: "not allowed create access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return authz.AuthorizerFunc(func(ctx context.Context) (*authz.Context, error) {
+					return authorizerForDummyUser(t, ctx, client, []string{types.VerbRead, types.VerbList}), nil
+				})
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsAccessDenied(err))
+			},
+		},
+		{
+			Name: "create resource twice",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Setup: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.NoError(t, err)
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsAlreadyExists(err))
+			},
+		},
+
+		// Delete
+		{
+			Name: "allowed delete access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Setup: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.CreateKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.CreateKubernetesWaitingContainerRequest{
+					WaitingContainer: sampleKubeWaitingContFn(t, wcName),
+				})
+				require.NoError(t, err)
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.NoError(t, err)
+			},
+		},
+		{
+			Name: "not allowed delete access",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return authz.AuthorizerFunc(func(ctx context.Context) (*authz.Context, error) {
+					return authorizerForDummyUser(t, ctx, client, []string{types.VerbRead, types.VerbList}), nil
+				})
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsAccessDenied(err))
+			},
+		},
+		{
+			Name: "get nonexistent resource",
+			Authorizer: func(t *testing.T, client localClient) authz.Authorizer {
+				return kubeAuthFn
+			},
+			Test: func(t *testing.T, ctx context.Context, resourceSvc *Service, wcName string) {
+				_, err := resourceSvc.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+					Username:      username,
+					Cluster:       cluster,
+					Namespace:     namespace,
+					PodName:       podName,
+					ContainerName: wcName,
+				})
+				require.Error(t, err)
+				require.True(t, trace.IsNotFound(err))
+			},
+		},
+	}
+
+	for _, tc := range tt {
+		tc := tc
+		t.Run(tc.Name, func(t *testing.T) {
+			t.Parallel()
+
+			ctx, _, resourceSvc := initSvc(t, tc.Authorizer)
+
+			wcName := uuid.NewString()
+			if tc.Setup != nil {
+				tc.Setup(t, ctx, resourceSvc, wcName)
+			}
+
+			tc.Test(t, ctx, resourceSvc, wcName)
+		})
+	}
+}
+
+func authorizerForDummyUser(t *testing.T, ctx context.Context, localClient localClient, roleVerbs []string) *authz.Context {
+	const clusterName = "localhost"
+
+	// Create role
+	roleName := "role-" + uuid.NewString()
+	role, err := types.NewRole(roleName, types.RoleSpecV6{
+		Allow: types.RoleConditions{Rules: []types.Rule{{
+			Resources: []string{types.KindKubeWaitingContainer},
+			Verbs:     roleVerbs,
+		}}},
+	})
+	require.NoError(t, err)
+
+	role, err = localClient.CreateRole(ctx, role)
+	require.NoError(t, err)
+
+	// Create user
+	user, err := types.NewUser("user-" + uuid.NewString())
+	require.NoError(t, err)
+	user.AddRole(roleName)
+	user, err = localClient.CreateUser(ctx, user)
+	require.NoError(t, err)
+
+	localUser := authz.LocalUser{
+		Username: user.GetName(),
+		Identity: tlsca.Identity{
+			Username: user.GetName(),
+			Groups:   []string{role.GetName()},
+		},
+	}
+	authCtx, err := authz.ContextForLocalUser(ctx, localUser, localClient, clusterName, true)
+	require.NoError(t, err)
+
+	return authCtx
+}
+
+type localClient interface {
+	authz.AuthorizerAccessPoint
+
+	CreateUser(ctx context.Context, user types.User) (types.User, error)
+	CreateRole(ctx context.Context, role types.Role) (types.Role, error)
+}
+
+func initSvc(t *testing.T, authorizerFn func(t *testing.T, client localClient) authz.Authorizer) (context.Context, localClient, *Service) {
+	ctx := context.Background()
+	backend, err := memory.New(memory.Config{})
+	require.NoError(t, err)
+
+	roleSvc := local.NewAccessService(backend)
+	userSvc := local.NewIdentityService(backend)
+	clusterSrv, err := local.NewClusterConfigurationService(backend)
+	require.NoError(t, err)
+	caSrv := local.NewCAService(backend)
+
+	clusterConfigSvc, err := local.NewClusterConfigurationService(backend)
+	require.NoError(t, err)
+	_, err = clusterConfigSvc.UpsertAuthPreference(ctx, types.DefaultAuthPreference())
+	require.NoError(t, err)
+	_, err = clusterConfigSvc.UpsertClusterAuditConfig(ctx, types.DefaultClusterAuditConfig())
+	require.NoError(t, err)
+	_, err = clusterConfigSvc.UpsertClusterNetworkingConfig(ctx, types.DefaultClusterNetworkingConfig())
+	require.NoError(t, err)
+	_, err = clusterConfigSvc.UpsertSessionRecordingConfig(ctx, types.DefaultSessionRecordingConfig())
+	require.NoError(t, err)
+
+	localResourceService, err := local.NewKubeWaitingContainerService(backend)
+	require.NoError(t, err)
+
+	client := struct {
+		*local.AccessService
+		*local.IdentityService
+		*local.ClusterConfigurationService
+		*local.CA
+	}{
+		AccessService:               roleSvc,
+		IdentityService:             userSvc,
+		ClusterConfigurationService: clusterSrv,
+		CA:                          caSrv,
+	}
+
+	resourceSvc, err := NewService(ServiceConfig{
+		Authorizer: authorizerFn(t, client),
+		Backend:    localResourceService,
+		Cache:      localResourceService,
+	})
+	require.NoError(t, err)
+
+	return ctx, client, resourceSvc
+}

lib/authz/permissions.go

@@ -822,6 +822,7 @@ func roleSpecForProxy(clusterName string) types.RoleSpecV6 {
 				types.NewRule(types.KindWebSession, services.RW()),
 				types.NewRule(types.KindWebToken, services.RW()),
 				types.NewRule(types.KindKubeServer, services.RW()),
+				types.NewRule(types.KindKubeWaitingContainer, services.RW()),
 				types.NewRule(types.KindDatabaseServer, services.RO()),
 				types.NewRule(types.KindLock, services.RO()),
 				types.NewRule(types.KindToken, []string{types.VerbRead, types.VerbDelete}),
@@ -1053,6 +1054,7 @@ func definitionForBuiltinRole(clusterName string, recConfig types.SessionRecordi
 					KubernetesLabels: types.Labels{types.Wildcard: []string{types.Wildcard}},
 					Rules: []types.Rule{
 						types.NewRule(types.KindKubeServer, services.RW()),
+						types.NewRule(types.KindKubeWaitingContainer, services.RW()),
 						types.NewRule(types.KindEvent, services.RW()),
 						types.NewRule(types.KindCertAuthority, services.ReadNoSecrets()),
 						types.NewRule(types.KindClusterName, services.RO()),

lib/cache/cache.go

@@ -37,6 +37,7 @@ import (
 	"github.com/gravitational/teleport"
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/internalutils/stream"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
@@ -166,6 +167,7 @@ func ForAuth(cfg Config) Config {
 		{Kind: types.KindAccessList},
 		{Kind: types.KindAccessListMember},
 		{Kind: types.KindAccessListReview},
+		{Kind: types.KindKubeWaitingContainer},
 	}
 	cfg.QueueSize = defaults.AuthQueueSize
 	// We don't want to enable partial health for auth cache because auth uses an event stream
@@ -217,6 +219,7 @@ func ForProxy(cfg Config) Config {
 		{Kind: types.KindAuditQuery},
 		{Kind: types.KindSecurityReport},
 		{Kind: types.KindSecurityReportState},
+		{Kind: types.KindKubeWaitingContainer},
 	}
 	cfg.QueueSize = defaults.ProxyQueueSize
 	return cfg
@@ -229,7 +232,7 @@ func ForProxy(cfg Config) Config {
 // incompatible changes don't render a remote cluster permanently unhealthy.
 // First, the cfg.Watches of ForOldRemoteProxy must be replaced
 // with the current cfg.Watches of ForRemoteProxy. Next, the
-// version used by `lib/reversetunnel/srv/go` to determine whether
+// version used by `lib/reversetunnel/srv.go` to determine whether
 // to use ForRemoteProxy or ForOldRemoteProxy must be updated
 // to be the release in which the new resource(s) will exist in. Finally,
 // add the new types.WatchKind below. Also note that this only
@@ -340,6 +343,7 @@ func ForKubernetes(cfg Config) Config {
 		{Kind: types.KindNamespace, Name: apidefaults.Namespace},
 		{Kind: types.KindKubeServer},
 		{Kind: types.KindKubernetesCluster},
+		{Kind: types.KindKubeWaitingContainer},
 	}
 	cfg.QueueSize = defaults.KubernetesQueueSize
 	return cfg
@@ -539,6 +543,7 @@ type Cache struct {
 	accessListCache              *simple.AccessListService
 	eventsFanout                 *services.FanoutV2
 	lowVolumeEventsFanout        *utils.RoundRobin[*services.FanoutV2]
+	kubeWaitingContsCache        *local.KubeWaitingContainerService
 
 	// closed indicates that the cache has been closed
 	closed atomic.Bool
@@ -701,6 +706,8 @@ type Config struct {
 	SecReports services.SecReports
 	// AccessLists is the access lists service.
 	AccessLists services.AccessLists
+	// KubeWaitingContainers is the Kubernetes waiting container service.
+	KubeWaitingContainers services.KubeWaitingContainer
 	// Backend is a backend for local cache
 	Backend backend.Backend
 	// MaxRetryPeriod is the maximum period between cache retries on failures
@@ -906,6 +913,12 @@ func New(config Config) (*Cache, error) {
 		lowVolumeFanouts = append(lowVolumeFanouts, services.NewFanoutV2(services.FanoutV2Config{}))
 	}
 
+	kubeWaitingContsCache, err := local.NewKubeWaitingContainerService(config.Backend)
+	if err != nil {
+		cancel()
+		return nil, trace.Wrap(err)
+	}
+
 	cs := &Cache{
 		ctx:                          ctx,
 		cancel:                       cancel,
@@ -941,6 +954,7 @@ func New(config Config) (*Cache, error) {
 		accessListCache:              accessListCache,
 		eventsFanout:                 fanout,
 		lowVolumeEventsFanout:        utils.NewRoundRobin(lowVolumeFanouts),
+		kubeWaitingContsCache:        kubeWaitingContsCache,
 		Logger: log.WithFields(log.Fields{
 			teleport.ComponentKey: config.Component,
 		}),
@@ -973,7 +987,6 @@ func (c *Cache) Start() error {
 		Jitter: retryutils.NewHalfJitter(),
 		Clock:  c.Clock,
 	})
-
 	if err != nil {
 		c.Close()
 		return trace.Wrap(err)
@@ -2229,6 +2242,36 @@ func (c *Cache) GetKubernetesServers(ctx context.Context) ([]types.KubeServer, e
 	return rg.reader.GetKubernetesServers(ctx)
 }
 
+// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (c *Cache) ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error) {
+	ctx, span := c.Tracer.Start(ctx, "cache/ListKubernetesWaitingContainers")
+	defer span.End()
+
+	rg, err := readCollectionCache(c, c.collections.kubeWaitingContainers)
+	if err != nil {
+		return nil, "", trace.Wrap(err)
+	}
+	defer rg.Release()
+	return rg.reader.ListKubernetesWaitingContainers(ctx, pageSize, pageToken)
+}
+
+// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (c *Cache) GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	ctx, span := c.Tracer.Start(ctx, "cache/GetKubernetesWaitingContainer")
+	defer span.End()
+
+	rg, err := readCollectionCache(c, c.collections.kubeWaitingContainers)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	defer rg.Release()
+	return rg.reader.GetKubernetesWaitingContainer(ctx, req)
+}
+
 // GetApplicationServers returns all registered application servers.
 func (c *Cache) GetApplicationServers(ctx context.Context, namespace string) ([]types.AppServer, error) {
 	ctx, span := c.Tracer.Start(ctx, "cache/GetApplicationServers")

lib/cache/cache_test.go

@@ -36,14 +36,18 @@ import (
 	log "github.com/sirupsen/logrus"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+	protobuf "google.golang.org/protobuf/proto"
+	"google.golang.org/protobuf/testing/protocmp"
 
 	"github.com/gravitational/teleport/api/client"
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
 	"github.com/gravitational/teleport/api/types/discoveryconfig"
 	"github.com/gravitational/teleport/api/types/header"
+	"github.com/gravitational/teleport/api/types/kubewaitingcontainer"
 	"github.com/gravitational/teleport/api/types/secreports"
 	"github.com/gravitational/teleport/api/types/trait"
 	"github.com/gravitational/teleport/api/types/userloginstate"
@@ -2923,6 +2927,7 @@ func TestCacheWatchKindExistsInEvents(t *testing.T) {
 		types.KindAccessList:              newAccessList(t, "access-list", clock),
 		types.KindAccessListMember:        newAccessListMember(t, "access-list", "member"),
 		types.KindAccessListReview:        newAccessListReview(t, "access-list", "review"),
+		types.KindKubeWaitingContainer:    newKubeWaitingContainer(t),
 	}
 
 	for name, cfg := range cases {
@@ -2940,7 +2945,23 @@ func TestCacheWatchKindExistsInEvents(t *testing.T) {
 				event, err := client.EventFromGRPC(protoEvent)
 				require.NoError(t, err)
 
-				require.Empty(t, cmp.Diff(resource, event.Resource))
+				// unwrap the RFD 153 resource if necessary
+				switch r := resource.(type) {
+				case types.Resource153Unwrapper:
+					eventResource, ok := event.Resource.(types.Resource153Unwrapper)
+					require.True(t, ok)
+
+					// if the resource is a protobuf message, pass an option so
+					// attempting to compare the messages does not result in a panic
+					switch r := r.Unwrap().(type) {
+					case protobuf.Message:
+						require.Empty(t, cmp.Diff(r, eventResource.Unwrap(), protocmp.Transform()))
+					default:
+						require.Empty(t, cmp.Diff(r, eventResource.Unwrap()))
+					}
+				default:
+					require.Empty(t, cmp.Diff(resource, event.Resource))
+				}
 			}
 		})
 	}
@@ -3144,6 +3165,7 @@ func newDiscoveryConfig(t *testing.T, name string) *discoveryconfig.DiscoveryCon
 	require.NoError(t, err)
 	return discoveryConfig
 }
+
 func newAuditQuery(t *testing.T, name string) *secreports.AuditQuery {
 	t.Helper()
 
@@ -3339,6 +3361,22 @@ func newAccessListReview(t *testing.T, accessList, name string) *accesslist.Revi
 	return review
 }
 
+func newKubeWaitingContainer(t *testing.T) types.Resource {
+	t.Helper()
+
+	waitingCont, err := kubewaitingcontainer.NewKubeWaitingContainer("container", &kubewaitingcontainerpb.KubernetesWaitingContainerSpec{
+		Username:      "user",
+		Cluster:       "cluster",
+		Namespace:     "namespace",
+		PodName:       "pod",
+		ContainerName: "container",
+		Patch:         []byte("patch"),
+	})
+	require.NoError(t, err)
+
+	return types.Resource153ToLegacy(waitingCont)
+}
+
 func withKeepalive[T any](fn func(context.Context, T) (*types.KeepAlive, error)) func(context.Context, T) error {
 	return func(ctx context.Context, resource T) error {
 		_, err := fn(ctx, resource)

lib/cache/collections.go

@@ -28,6 +28,7 @@ import (
 	"github.com/gravitational/teleport/api/client"
 	"github.com/gravitational/teleport/api/client/proto"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/types"
 	"github.com/gravitational/teleport/api/types/accesslist"
 	"github.com/gravitational/teleport/api/types/discoveryconfig"
@@ -54,7 +55,7 @@ type collection interface {
 
 // executor[T, R] is a specific way to run the collector operations that we need
 // for the genericCollector for a generic resource type T and its reader type R.
-type executor[T types.Resource, R any] interface {
+type executor[T any, R any] interface {
 	// getAll returns all of the target resources from the auth server.
 	// For singleton objects, this should be a size-1 slice.
 	getAll(ctx context.Context, cache *Cache, loadSecrets bool) ([]T, error)
@@ -84,7 +85,7 @@ type noReader struct{}
 // genericCollection is a generic collection implementation for resource type T with collection-specific logic
 // encapsulated in executor type E. Type R provides getter methods related to the collection, e.g. GetNodes(),
 // GetRoles().
-type genericCollection[T types.Resource, R any, E executor[T, R]] struct {
+type genericCollection[T any, R any, E executor[T, R]] struct {
 	cache *Cache
 	watch types.WatchKind
 	exec  E
@@ -199,6 +200,7 @@ type cacheCollections struct {
 	installers               collectionReader[installerGetter]
 	integrations             collectionReader[services.IntegrationsGetter]
 	kubeClusters             collectionReader[kubernetesClusterGetter]
+	kubeWaitingContainers    collectionReader[kubernetesWaitingContainerGetter]
 	kubeServers              collectionReader[kubeServerGetter]
 	locks                    collectionReader[services.LockGetter]
 	namespaces               collectionReader[namespaceGetter]
@@ -655,6 +657,15 @@ func setupCollections(c *Cache, watches []types.WatchKind) (*cacheCollections, e
 			}
 			collections.accessListReviews = &genericCollection[*accesslist.Review, accessListReviewsGetter, accessListReviewExecutor]{cache: c, watch: watch}
 			collections.byKind[resourceKind] = collections.accessListReviews
+		case types.KindKubeWaitingContainer:
+			if c.Presence == nil {
+				return nil, trace.BadParameter("missing parameter Presence")
+			}
+			collections.kubeWaitingContainers = &genericCollection[*kubewaitingcontainerpb.KubernetesWaitingContainer, kubernetesWaitingContainerGetter, kubeWaitingContainerExecutor]{
+				cache: c,
+				watch: watch,
+			}
+			collections.byKind[resourceKind] = collections.kubeWaitingContainers
 		default:
 			return nil, trace.BadParameter("resource %q is not supported", watch.Kind)
 		}
@@ -2081,7 +2092,72 @@ type kubernetesClusterGetter interface {
 	GetKubernetesCluster(ctx context.Context, name string) (types.KubeCluster, error)
 }
 
-var _ executor[types.KubeCluster, kubernetesClusterGetter] = kubeClusterExecutor{}
+type kubeWaitingContainerExecutor struct{}
+
+func (kubeWaitingContainerExecutor) getAll(ctx context.Context, cache *Cache, loadSecrets bool) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	var (
+		startKey string
+		allConts []*kubewaitingcontainerpb.KubernetesWaitingContainer
+	)
+	for {
+		conts, nextKey, err := cache.kubeWaitingContsCache.ListKubernetesWaitingContainers(ctx, 0, startKey)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+
+		allConts = append(allConts, conts...)
+
+		if nextKey == "" {
+			break
+		}
+		startKey = nextKey
+	}
+	return allConts, nil
+}
+
+func (kubeWaitingContainerExecutor) upsert(ctx context.Context, cache *Cache, resource *kubewaitingcontainerpb.KubernetesWaitingContainer) error {
+	_, err := cache.kubeWaitingContsCache.UpsertKubernetesWaitingContainer(ctx, resource)
+	return trace.Wrap(err)
+}
+
+func (kubeWaitingContainerExecutor) deleteAll(ctx context.Context, cache *Cache) error {
+	return trace.Wrap(cache.kubeWaitingContsCache.DeleteAllKubernetesWaitingContainers(ctx))
+}
+
+func (kubeWaitingContainerExecutor) delete(ctx context.Context, cache *Cache, resource types.Resource) error {
+	switch r := resource.(type) {
+	case types.Resource153Unwrapper:
+		switch wc := r.Unwrap().(type) {
+		case *kubewaitingcontainerpb.KubernetesWaitingContainer:
+			err := cache.kubeWaitingContsCache.DeleteKubernetesWaitingContainer(ctx, &kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest{
+				Username:      wc.Spec.Username,
+				Cluster:       wc.Spec.Cluster,
+				Namespace:     wc.Spec.Namespace,
+				PodName:       wc.Spec.PodName,
+				ContainerName: wc.Spec.ContainerName,
+			})
+			return trace.Wrap(err)
+		}
+	}
+
+	return trace.BadParameter("unknown KubeWaitingContainer type, expected *kubewaitingcontainerpb.KubernetesWaitingContainer, got %T", resource)
+}
+
+func (kubeWaitingContainerExecutor) isSingleton() bool { return false }
+
+func (kubeWaitingContainerExecutor) getReader(cache *Cache, cacheOK bool) kubernetesWaitingContainerGetter {
+	if cacheOK {
+		return cache.kubeWaitingContsCache
+	}
+	return cache.Config.KubeWaitingContainers
+}
+
+type kubernetesWaitingContainerGetter interface {
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+}
+
+var _ executor[*kubewaitingcontainerpb.KubernetesWaitingContainer, kubernetesWaitingContainerGetter] = kubeWaitingContainerExecutor{}
 
 //nolint:revive // Because we want this to be IdP.
 type samlIdPServiceProvidersExecutor struct{}

lib/services/kubewaitingcontainer.go

@@ -0,0 +1,97 @@
+/*
+ * Teleport
+ * Copyright (C) 2024  Gravitational, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package services
+
+import (
+	"context"
+
+	"github.com/gravitational/trace"
+	"google.golang.org/protobuf/types/known/timestamppb"
+
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+	"github.com/gravitational/teleport/api/types/kubewaitingcontainer"
+	"github.com/gravitational/teleport/lib/utils"
+)
+
+// KubeWaitingContainer is responsible for managing Kubernetes
+// ephemeral containers that are waiting to be created until moderated
+// session conditions are met.
+type KubeWaitingContainer interface {
+	ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error)
+	GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+	CreateKubernetesWaitingContainer(ctx context.Context, in *kubewaitingcontainerpb.KubernetesWaitingContainer) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error)
+	DeleteKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest) error
+}
+
+// MarshalKubeWaitingContainer marshals a KubernetesWaitingContainer resource to JSON.
+func MarshalKubeWaitingContainer(in *kubewaitingcontainerpb.KubernetesWaitingContainer, opts ...MarshalOption) ([]byte, error) {
+	if in == nil {
+		return nil, trace.BadParameter("message is nil")
+	}
+	if err := kubewaitingcontainer.ValidateKubeWaitingContainer(in); err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	cfg, err := CollectOptions(opts)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	if !cfg.PreserveResourceID {
+		//nolint:staticcheck // SA1019. Deprecated, but still needed.
+		in.Metadata.Id = 0
+		in.Metadata.Revision = ""
+	}
+
+	out, err := utils.FastMarshal(in)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	return out, nil
+}
+
+// UnmarshalKubeWaitingContainer unmarshals a KubernetesWaitingContainer resource from JSON.
+func UnmarshalKubeWaitingContainer(data []byte, opts ...MarshalOption) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	if len(data) == 0 {
+		return nil, trace.BadParameter("data is empty")
+	}
+	cfg, err := CollectOptions(opts)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+	var out *kubewaitingcontainerpb.KubernetesWaitingContainer
+	if err := utils.FastUnmarshal(data, &out); err != nil {
+		return nil, trace.BadParameter(err.Error())
+	}
+	if err := kubewaitingcontainer.ValidateKubeWaitingContainer(out); err != nil {
+		return nil, trace.Wrap(err)
+	}
+	if cfg.ID != 0 {
+		//nolint:staticcheck // SA1019. Deprecated, but still needed.
+		out.Metadata.Id = cfg.ID
+	}
+	if cfg.Revision != "" {
+		out.Metadata.Revision = cfg.Revision
+	}
+	if !cfg.Expires.IsZero() {
+		out.Metadata.Expires = timestamppb.New(cfg.Expires)
+	}
+
+	return out, nil
+}

lib/services/local/events.go

@@ -21,6 +21,7 @@ package local
 import (
 	"bytes"
 	"context"
+	"strings"
 
 	"github.com/gravitational/trace"
 	"github.com/jonboulle/clockwork"
@@ -28,7 +29,9 @@ import (
 
 	"github.com/gravitational/teleport"
 	apidefaults "github.com/gravitational/teleport/api/defaults"
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
 	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/api/types/kubewaitingcontainer"
 	"github.com/gravitational/teleport/lib/backend"
 	"github.com/gravitational/teleport/lib/defaults"
 	"github.com/gravitational/teleport/lib/services"
@@ -190,6 +193,8 @@ func (e *EventsService) NewWatcher(ctx context.Context, watch types.Watch) (type
 			parser = newAccessListMemberParser()
 		case types.KindAccessListReview:
 			parser = newAccessListReviewParser()
+		case types.KindKubeWaitingContainer:
+			parser = newKubeWaitingContainerParser()
 		default:
 			if watch.AllowPartialSuccess {
 				continue
@@ -1885,6 +1890,60 @@ func (p *accessListReviewParser) parse(event backend.Event) (types.Resource, err
 	}
 }
 
+func newKubeWaitingContainerParser() *kubeWaitingContainerParser {
+	return &kubeWaitingContainerParser{
+		baseParser: newBaseParser(backend.Key(kubeWaitingContPrefix)),
+	}
+}
+
+type kubeWaitingContainerParser struct {
+	baseParser
+}
+
+func (p *kubeWaitingContainerParser) parse(event backend.Event) (types.Resource, error) {
+	switch event.Type {
+	case types.OpDelete:
+		// remove the first separator so no separated parts should be
+		// empty strings
+		key := string(event.Item.Key)
+		if len(key) > 0 && key[0] == backend.Separator {
+			key = key[1:]
+		}
+		parts := strings.Split(key, string(backend.Separator))
+		if len(parts) != 6 {
+			return nil, trace.BadParameter("malformed key for %s event: %s", types.KindKubeWaitingContainer, event.Item.Key)
+		}
+
+		resource, err := kubewaitingcontainer.NewKubeWaitingContainer(
+			parts[5],
+			&kubewaitingcontainerpb.KubernetesWaitingContainerSpec{
+				Username:      parts[1],
+				Cluster:       parts[2],
+				Namespace:     parts[3],
+				PodName:       parts[4],
+				ContainerName: parts[5],
+			},
+		)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		return types.Resource153ToLegacy(resource), nil
+	case types.OpPut:
+		resource, err := services.UnmarshalKubeWaitingContainer(
+			event.Item.Value,
+			services.WithResourceID(event.Item.ID),
+			services.WithExpires(event.Item.Expires),
+			services.WithRevision(event.Item.Revision),
+		)
+		if err != nil {
+			return nil, trace.Wrap(err)
+		}
+		return types.Resource153ToLegacy(resource), nil
+	default:
+		return nil, trace.BadParameter("event %v is not supported", event.Type)
+	}
+}
+
 func resourceHeader(event backend.Event, kind, version string, offset int) (types.Resource, error) {
 	name, err := base(event.Item.Key, offset)
 	if err != nil {

lib/services/local/kubewaitingcontainer.go

@@ -0,0 +1,124 @@
+/*
+ * Teleport
+ * Copyright (C) 2024  Gravitational, Inc.
+ *
+ * This program is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU Affero General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU Affero General Public License for more details.
+ *
+ * You should have received a copy of the GNU Affero General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package local
+
+import (
+	"context"
+
+	"github.com/gravitational/trace"
+
+	kubewaitingcontainerpb "github.com/gravitational/teleport/api/gen/proto/go/teleport/kubewaitingcontainer/v1"
+	"github.com/gravitational/teleport/api/types"
+	"github.com/gravitational/teleport/lib/backend"
+	"github.com/gravitational/teleport/lib/services"
+	"github.com/gravitational/teleport/lib/services/local/generic"
+)
+
+const (
+	kubeWaitingContPrefix = "k8s_waiting_container"
+)
+
+// KubeWaitingContainerService manages Kubernetes ephemeral containers
+// that are waiting to be created until moderated session conditions are met.
+type KubeWaitingContainerService struct {
+	svc *generic.ServiceWrapper[*kubewaitingcontainerpb.KubernetesWaitingContainer]
+}
+
+// NewKubeWaitingContainerService returns a new Kubernetes waiting
+// container service.
+func NewKubeWaitingContainerService(backend backend.Backend) (*KubeWaitingContainerService, error) {
+	svc, err := generic.NewServiceWrapper(
+		backend,
+		types.KindKubeWaitingContainer,
+		kubeWaitingContPrefix,
+		services.MarshalKubeWaitingContainer,
+		services.UnmarshalKubeWaitingContainer,
+	)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return &KubeWaitingContainerService{
+		svc: svc,
+	}, nil
+}
+
+// ListKubernetesWaitingContainers lists Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) ListKubernetesWaitingContainers(ctx context.Context, pageSize int, pageToken string) ([]*kubewaitingcontainerpb.KubernetesWaitingContainer, string, error) {
+	out, nextToken, err := k.svc.ListResources(ctx, pageSize, pageToken)
+	if err != nil {
+		return nil, "", trace.Wrap(err)
+	}
+
+	return out, nextToken, nil
+}
+
+// GetKubernetesWaitingContainer returns a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) GetKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.GetKubernetesWaitingContainerRequest) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	out, err := k.svc.WithPrefix(req.Username, req.Cluster, req.Namespace, req.PodName).GetResource(ctx, req.ContainerName)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// CreateKubernetesWaitingContainer creates a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) CreateKubernetesWaitingContainer(ctx context.Context, in *kubewaitingcontainerpb.KubernetesWaitingContainer) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	svc := k.svc.WithPrefix(in.Spec.Username, in.Spec.Cluster, in.Spec.Namespace, in.Spec.PodName)
+	out, err := svc.CreateResource(ctx, in)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// UpsertKubernetesWaitingContainer upserts a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) UpsertKubernetesWaitingContainer(ctx context.Context, in *kubewaitingcontainerpb.KubernetesWaitingContainer) (*kubewaitingcontainerpb.KubernetesWaitingContainer, error) {
+	svc := k.svc.WithPrefix(in.Spec.Username, in.Spec.Cluster, in.Spec.Namespace, in.Spec.PodName)
+	out, err := svc.UpsertResource(ctx, in)
+	if err != nil {
+		return nil, trace.Wrap(err)
+	}
+
+	return out, nil
+}
+
+// DeleteKubernetesWaitingContainer deletes a Kubernetes ephemeral
+// container that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) DeleteKubernetesWaitingContainer(ctx context.Context, req *kubewaitingcontainerpb.DeleteKubernetesWaitingContainerRequest) error {
+	return trace.Wrap(k.svc.WithPrefix(req.Username, req.Cluster, req.Namespace, req.PodName).DeleteResource(ctx, req.ContainerName))
+}
+
+// DeleteAllKubernetesWaitingContainers deletes all Kubernetes ephemeral
+// containers that are waiting to be created until moderated
+// session conditions are met.
+func (k *KubeWaitingContainerService) DeleteAllKubernetesWaitingContainers(ctx context.Context) error {
+	return trace.Wrap(k.svc.DeleteAllResources(ctx))
+}

lib/services/services.go

@@ -47,6 +47,7 @@ type Services interface {
 	SAMLIdPServiceProviders
 	UserGroups
 	Integrations
+	KubeWaitingContainer
 
 	OktaClient() Okta
 	AccessListClient() AccessLists