atunnel: broker actor certificates through atelet

cmd/ateapi/internal/actoridentity/actoridentity.go

@@ -85,9 +85,10 @@ func New(clientJWTIssuer, clientJWTAudience, actorIDJWTPoolFile, actorIDCAPoolFi
 // imported so that this package does not depend on controlapi for three
 // strings; if a third pkg that need these constants appears, they should move to a shared package.
 const (
-	ateletTrustDomain = "cluster.local"
-	ateletNamespace   = "ate-system"
-	ateletSA          = "atelet"
+	ateletTrustDomain        = "cluster.local"
+	ateletNamespace          = "ate-system"
+	ateletSA                 = "atelet"
+	actorCertificateLifetime = time.Hour
 )
 
 func (s *Server) MintJWT(ctx context.Context, req *ateapipb.MintJWTRequest) (*ateapipb.MintJWTResponse, error) {
@@ -125,7 +126,6 @@ func (s *Server) MintJWT(ctx context.Context, req *ateapipb.MintJWTRequest) (*at
 	if err != nil {
 		return nil, fmt.Errorf("while unmarshaling signing pool: %w", err)
 	}
-
 	// We only issue tokens with audience bindings.
 	if len(req.GetAudience()) == 0 {
 		return nil, fmt.Errorf("at least one audience must be requested")
@@ -179,7 +179,10 @@ func (s *Server) MintCert(ctx context.Context, req *ateapipb.MintCertRequest) (*
 	}
 
 	actorRef := resources.ActorRef{Atespace: atespace, Name: actorName}
-	actor, err := s.authorizeActor(ctx, caller, actorRef)
+	if req.GetWorkerPodUid() == "" {
+		return nil, status.Error(codes.InvalidArgument, "worker_pod_uid is required")
+	}
+	actor, err := s.authorizeActor(ctx, caller, actorRef, req.GetWorkerPodUid())
 	if err != nil {
 		return nil, err
 	}
@@ -232,7 +235,7 @@ func (s *Server) MintCert(ctx context.Context, req *ateapipb.MintCertRequest) (*
 	template := &x509.Certificate{
 		URIs:                  []*url.URL{spiffeURI},
 		NotBefore:             time.Now().Add(-5 * time.Minute),
-		NotAfter:              time.Now().Add(15 * time.Minute),
+		NotAfter:              time.Now().Add(actorCertificateLifetime),
 		KeyUsage:              x509.KeyUsageDigitalSignature,
 		ExtKeyUsage:           []x509.ExtKeyUsage{x509.ExtKeyUsageClientAuth},
 		BasicConstraintsValid: true,
@@ -269,7 +272,7 @@ func (s *Server) MintCert(ctx context.Context, req *ateapipb.MintCertRequest) (*
 	}, nil
 }
 
-// ateletCaller is the verified identity of an atelet that called MintCert.
+// ateletCaller is the verified identity of an atelet requesting an actor credential.
 type ateletCaller struct {
 	podName  string
 	nodeName string
@@ -298,7 +301,7 @@ func authenticateAtelet(ctx context.Context) (*ateletCaller, error) {
 	}
 	leaf := tlsInfo.State.PeerCertificates[0]
 
-	// Only atelet may mint actor certificates. Everything else with a valid
+	// Only atelet may mint actor credentials. Everything else with a valid
 	// pod-identity certificate — including the actor workloads themselves — is
 	// rejected here.
 	expected := (&url.URL{
@@ -307,19 +310,19 @@ func authenticateAtelet(ctx context.Context) (*ateletCaller, error) {
 		Path:   path.Join("ns", ateletNamespace, "sa", ateletSA),
 	}).String()
 	if len(leaf.URIs) == 0 || leaf.URIs[0].String() != expected {
-		slog.WarnContext(ctx, "MintCert denied: caller is not atelet",
+		slog.WarnContext(ctx, "ActorIdentity denied: caller is not atelet",
 			slog.Any("uris", leaf.URIs), slog.String("expected", expected))
-		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor certificates")
+		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor credentials")
 	}
 
 	identity, err := substratex509.PodIdentityFromCertificate(leaf)
 	if err != nil {
-		slog.WarnContext(ctx, "MintCert denied: malformed PodIdentity extension", slog.Any("err", err))
-		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor certificates")
+		slog.WarnContext(ctx, "ActorIdentity denied: malformed PodIdentity extension", slog.Any("err", err))
+		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor credentials")
 	}
 	if identity == nil {
-		slog.WarnContext(ctx, "MintCert denied: certificate has no PodIdentity extension")
-		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor certificates")
+		slog.WarnContext(ctx, "ActorIdentity denied: certificate has no PodIdentity extension")
+		return nil, status.Errorf(codes.PermissionDenied, "caller is not permitted to mint actor credentials")
 	}
 
 	return &ateletCaller{podName: identity.PodName, nodeName: identity.NodeName}, nil
@@ -333,30 +336,31 @@ func authenticateAtelet(ctx context.Context) (*ateletCaller, error) {
 // An atelet is therefore confined to the actors it is actually hosting, and an
 // actor that has been suspended, paused or migrated elsewhere can no longer
 // have credentials minted for it.
-func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, actorRef resources.ActorRef) (*ateapipb.Actor, error) {
+// expectedWorkerPodUID binds the request to the exact worker incarnation.
+func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, actorRef resources.ActorRef, expectedWorkerPodUID string) (*ateapipb.Actor, error) {
 	// Denials are deliberately indistinguishable from each other: a caller that
 	// is not entitled to an actor should not be able to use this RPC to learn
 	// whether that actor exists, or where it is running.
 	deny := func(reason string, args ...any) error {
-		slog.WarnContext(ctx, "MintCert denied: "+reason,
+		slog.WarnContext(ctx, "ActorIdentity denied: "+reason,
 			append([]any{slog.Any("actor", actorRef), slog.String("callerPod", caller.podName), slog.String("callerNode", caller.nodeName)}, args...)...)
-		return status.Errorf(codes.PermissionDenied, "caller is not permitted to mint certificates for this actor")
+		return status.Errorf(codes.PermissionDenied, "caller is not permitted to mint credentials for this actor")
 	}
 
 	actor, err := s.store.GetActor(ctx, actorRef)
 	if err != nil {
 		if errors.Is(err, store.ErrNotFound) {
 			return nil, deny("actor not found")
 		}
-		slog.ErrorContext(ctx, "MintCert: failed to read actor", slog.Any("actor", actorRef), slog.Any("err", err))
+		slog.ErrorContext(ctx, "ActorIdentity: failed to read actor", slog.Any("actor", actorRef), slog.Any("err", err))
 		return nil, status.Errorf(codes.Internal, "failed to look up actor")
 	}
 
 	// Deletion is only entered from SUSPENDED or CRASHED, both of which
 	// have already released the worker, so the assignment check below would
 	// reject this too. It is kept because minting for better visbility and logging.
 	if actor.GetStatus() == ateapipb.Actor_STATUS_DELETING {
-		slog.WarnContext(ctx, "MintCert refused: actor is being deleted", slog.Any("actor", actorRef))
+		slog.WarnContext(ctx, "ActorIdentity refused: actor is being deleted", slog.Any("actor", actorRef))
 		return nil, status.Errorf(codes.FailedPrecondition, "actor is being deleted")
 	}
 
@@ -365,7 +369,7 @@ func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, actor
 	// folded into deny().
 	podNamespace, podName, pool := actor.GetAteomPodNamespace(), actor.GetAteomPodName(), actor.GetWorkerPoolName()
 	if podNamespace == "" || podName == "" || pool == "" {
-		slog.ErrorContext(ctx, "MintCert: running actor has incomplete placement",
+		slog.ErrorContext(ctx, "ActorIdentity: running actor has incomplete placement",
 			slog.Any("actor", actorRef), slog.String("podNamespace", podNamespace),
 			slog.String("podName", podName), slog.String("workerPool", pool))
 		return nil, status.Errorf(codes.FailedPrecondition, "actor has no worker assigned")
@@ -376,13 +380,16 @@ func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, actor
 		if errors.Is(err, store.ErrNotFound) {
 			return nil, deny("worker hosting the actor not found", slog.String("workerPod", podNamespace+"/"+podName))
 		}
-		slog.ErrorContext(ctx, "MintCert: failed to read worker", slog.Any("actor", actorRef), slog.Any("err", err))
+		slog.ErrorContext(ctx, "ActorIdentity: failed to read worker", slog.Any("actor", actorRef), slog.Any("err", err))
 		return nil, status.Errorf(codes.Internal, "failed to look up worker")
 	}
 
 	if worker.GetNodeName() != caller.nodeName {
 		return nil, deny("actor is hosted on a different node", slog.String("actorNode", worker.GetNodeName()))
 	}
+	if worker.GetWorkerPodUid() != expectedWorkerPodUID {
+		return nil, deny("worker Pod UID does not match", slog.String("workerPodUID", expectedWorkerPodUID))
+	}
 
 	// The worker must still agree that it is hosting this actor.
 	if assigned := worker.GetAssignment().GetActor(); resources.ActorRefFromObjectRef(assigned) != actorRef {

cmd/ateapi/internal/actoridentity/actoridentity_test.go

@@ -247,9 +247,10 @@ func TestMintCertAuthorization(t *testing.T) {
 
 		fixture actorFixture
 
-		// atespace and actorName override the request fields when non-nil.
-		atespace  *string
-		actorName *string
+		// Request fields override their defaults when non-nil.
+		atespace     *string
+		actorName    *string
+		workerPodUID *string
 
 		wantCode codes.Code
 	}{
@@ -308,6 +309,11 @@ func TestMintCertAuthorization(t *testing.T) {
 			fixture:  runningOnNode(testOtherNode),
 			wantCode: codes.PermissionDenied,
 		},
+		"worker Pod UID does not match": {
+			fixture:      runningOnNode(testNode),
+			workerPodUID: ptr("sibling-worker-uid"),
+			wantCode:     codes.PermissionDenied,
+		},
 		"worker is assigned to a different actor": {
 			fixture: actorFixture{
 				status:     ateapipb.Actor_STATUS_RUNNING,
@@ -376,10 +382,15 @@ func TestMintCertAuthorization(t *testing.T) {
 				actorName = *tc.actorName
 			}
 
+			workerPodUID := "worker-uid"
+			if tc.workerPodUID != nil {
+				workerPodUID = *tc.workerPodUID
+			}
 			resp, err := srv.MintCert(ctxWithCert(callerCert), &ateapipb.MintCertRequest{
 				Atespace:                  atespace,
 				ActorName:                 actorName,
 				CertificateSigningRequest: newCSR(t),
+				WorkerPodUid:              workerPodUID,
 			})
 			if got := status.Code(err); got != tc.wantCode {
 				t.Fatalf("MintCert() code = %v (err = %v), want %v", got, err, tc.wantCode)
@@ -448,6 +459,7 @@ func TestMintCertEmbedsActorIdentity(t *testing.T) {
 			Atespace:                  testAtespace,
 			ActorName:                 testActorName,
 			CertificateSigningRequest: newCSR(t),
+			WorkerPodUid:              "worker-uid",
 		}
 	})
 	if err != nil {
@@ -493,6 +505,7 @@ func TestMintCertActorUID(t *testing.T) {
 					ActorName:                 testActorName,
 					ActorUid:                  tc.requestUID(actorUID),
 					CertificateSigningRequest: newCSR(t),
+					WorkerPodUid:              "worker-uid",
 				}
 			})
 			if got := status.Code(err); got != tc.wantCode {
@@ -549,6 +562,7 @@ func TestMintCertActorStatus(t *testing.T) {
 				Atespace:                  testAtespace,
 				ActorName:                 testActorName,
 				CertificateSigningRequest: newCSR(t),
+				WorkerPodUid:              "worker-uid",
 			})
 			if got := status.Code(err); got != wantCode {
 				t.Errorf("MintCert() code = %v (err = %v), want %v", got, err, wantCode)
@@ -584,6 +598,7 @@ func TestMintCertDeniesUnassignedActorWhateverItsStatus(t *testing.T) {
 				Atespace:                  testAtespace,
 				ActorName:                 testActorName,
 				CertificateSigningRequest: newCSR(t),
+				WorkerPodUid:              "worker-uid",
 			})
 			if got := status.Code(err); got != codes.PermissionDenied {
 				t.Errorf("MintCert() code = %v (err = %v), want %v", got, err, codes.PermissionDenied)
@@ -611,6 +626,7 @@ func TestMintCertAuthorizesBeforeSigning(t *testing.T) {
 		Atespace:                  testAtespace,
 		ActorName:                 testActorName,
 		CertificateSigningRequest: []byte("not a CSR"),
+		WorkerPodUid:              "worker-uid",
 	})
 	if got := status.Code(err); got != codes.PermissionDenied {
 		t.Errorf("MintCert() code = %v (err = %v), want %v", got, err, codes.PermissionDenied)

cmd/ateapi/internal/controlapi/functional_test.go

@@ -324,7 +324,7 @@ func setupTest(t *testing.T, ns string) *testContext {
 		return insecure.NewCredentials(), nil
 	}
 
-	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient)
+	service := NewService(persistence, wc, actorTemplateLister, workerPoolLister, sandboxConfigLister, dialer, k8sClient, "")
 
 	// 5. Start REAL gRPC Server for ATE API
 	grpcServer := grpc.NewServer(grpc.UnaryInterceptor(ateinterceptors.ServerUnaryInterceptor))

cmd/ateapi/internal/controlapi/service.go

@@ -43,13 +43,14 @@ func NewService(
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	dialer *AteletDialer,
 	kubeClient kubernetes.Interface,
+	egressGatewayAddress string,
 ) *Service {
 	s := &Service{
 		persistence:         persistence,
 		actorTemplateLister: actorTemplateLister,
 		workerPoolLister:    workerPoolLister,
 		dialer:              dialer,
-		actorWorkflow:       NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient),
+		actorWorkflow:       NewActorWorkflow(persistence, workerCache, dialer, actorTemplateLister, workerPoolLister, sandboxConfigLister, kubeClient, egressGatewayAddress),
 	}
 	return s
 }

cmd/ateapi/internal/controlapi/workflow.go

@@ -130,15 +130,16 @@ func runStep[Params any, Context any](ctx context.Context, params Params, wCtx C
 
 // ActorWorkflow handles the workflows for actor's resume / suspend operations.
 type ActorWorkflow struct {
-	store               store.Interface
-	workerCache         *workercache.Cache
-	scheduler           scheduling.Scheduler
-	dialer              *AteletDialer
-	actorTemplateLister listersv1alpha1.ActorTemplateLister
-	workerPoolLister    listersv1alpha1.WorkerPoolLister
-	sandboxConfigLister listersv1alpha1.SandboxConfigLister
-	kubeClient          kubernetes.Interface
-	secretCache         *envSecretCache
+	store                store.Interface
+	workerCache          *workercache.Cache
+	scheduler            scheduling.Scheduler
+	dialer               *AteletDialer
+	actorTemplateLister  listersv1alpha1.ActorTemplateLister
+	workerPoolLister     listersv1alpha1.WorkerPoolLister
+	sandboxConfigLister  listersv1alpha1.SandboxConfigLister
+	kubeClient           kubernetes.Interface
+	secretCache          *envSecretCache
+	egressGatewayAddress string
 }
 
 // NewActorWorkflow creates a new ActorWorkflow.
@@ -150,17 +151,19 @@ func NewActorWorkflow(
 	workerPoolLister listersv1alpha1.WorkerPoolLister,
 	sandboxConfigLister listersv1alpha1.SandboxConfigLister,
 	kubeClient kubernetes.Interface,
+	egressGatewayAddress string,
 ) *ActorWorkflow {
 	return &ActorWorkflow{
-		store:               store,
-		workerCache:         workerCache,
-		scheduler:           scheduling.New(workerCache),
-		dialer:              dialer,
-		actorTemplateLister: actorTemplateLister,
-		workerPoolLister:    workerPoolLister,
-		sandboxConfigLister: sandboxConfigLister,
-		kubeClient:          kubeClient,
-		secretCache:         newEnvSecretCache(envSecretCacheTTL),
+		store:                store,
+		workerCache:          workerCache,
+		scheduler:            scheduling.New(workerCache),
+		dialer:               dialer,
+		actorTemplateLister:  actorTemplateLister,
+		workerPoolLister:     workerPoolLister,
+		sandboxConfigLister:  sandboxConfigLister,
+		kubeClient:           kubeClient,
+		secretCache:          newEnvSecretCache(envSecretCacheTTL),
+		egressGatewayAddress: egressGatewayAddress,
 	}
 }
 
@@ -183,7 +186,7 @@ func (w *ActorWorkflow) ResumeActor(ctx context.Context, actorRef resources.Acto
 		&CreateVolumesStep{store: w.store},
 		&AssignWorkerStep{store: w.store, workerCache: w.workerCache, scheduler: w.scheduler},
 		&AttachVolumesStep{store: w.store},
-		&CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister, scheduler: w.scheduler},
+		&CallAteletRestoreStep{store: w.store, dialer: w.dialer, kubeClient: w.kubeClient, secretCache: w.secretCache, workerPoolLister: w.workerPoolLister, sandboxConfigLister: w.sandboxConfigLister, scheduler: w.scheduler, egressGatewayAddress: w.egressGatewayAddress},
 		&FinalizeRunningStep{store: w.store},
 	}
 

cmd/ateapi/internal/controlapi/workflow_resume.go

@@ -452,13 +452,14 @@ func (s *AttachVolumesStep) Execute(ctx context.Context, input *ResumeInput, sta
 func (s *AttachVolumesStep) RetryBackoff() *wait.Backoff { return nil }
 
 type CallAteletRestoreStep struct {
-	store               store.Interface
-	dialer              *AteletDialer
-	kubeClient          kubernetes.Interface
-	secretCache         *envSecretCache
-	workerPoolLister    listersv1alpha1.WorkerPoolLister
-	sandboxConfigLister listersv1alpha1.SandboxConfigLister
-	scheduler           scheduling.Scheduler
+	store                store.Interface
+	dialer               *AteletDialer
+	kubeClient           kubernetes.Interface
+	secretCache          *envSecretCache
+	workerPoolLister     listersv1alpha1.WorkerPoolLister
+	sandboxConfigLister  listersv1alpha1.SandboxConfigLister
+	scheduler            scheduling.Scheduler
+	egressGatewayAddress string
 }
 
 func (s *CallAteletRestoreStep) Name() string { return "CallAteletRestore" }
@@ -516,6 +517,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 	if err != nil {
 		return err
 	}
+	egressGateway := s.egressGateway()
 
 	if local := state.Actor.GetLocalSnapshotInfo(); local != nil {
 		slog.InfoContext(ctx, "Actor has snapshot; Restoring from snapshot")
@@ -528,6 +530,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 			ActorTemplateName:      state.Actor.GetActorTemplateName(),
 			Spec:                   workloadSpec,
 			ActorUid:               state.Actor.GetMetadata().Uid,
+			EgressGateway:          egressGateway,
 		}
 		req.Type = ateletpb.CheckpointType_CHECKPOINT_TYPE_LOCAL
 		req.Config = &ateletpb.RestoreRequest_LocalConfig{
@@ -574,6 +577,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 			// Empty unless this is a Golden data resume.
 			GoldenSnapshotUriPrefix: state.GoldenSnapshotLocation,
 			ActorUid:                state.Actor.GetMetadata().Uid,
+			EgressGateway:           egressGateway,
 		}
 		_, err = client.Restore(ctx, req)
 		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while restoring durable snapshot")
@@ -597,6 +601,7 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 			SandboxAssets:          sandboxAssets,
 			Spec:                   workloadSpec,
 			ActorUid:               state.Actor.GetMetadata().Uid,
+			EgressGateway:          egressGateway,
 		}
 		_, err = client.Run(ctx, req)
 		return maybeCrashActor(ctx, s.store, input.ActorRef, err, "while creating workload from spec")
@@ -606,6 +611,13 @@ func (s *CallAteletRestoreStep) Execute(ctx context.Context, input *ResumeInput,
 
 func (s *CallAteletRestoreStep) RetryBackoff() *wait.Backoff { return nil }
 
+func (s *CallAteletRestoreStep) egressGateway() *ateletpb.EgressGateway {
+	if s.egressGatewayAddress == "" {
+		return nil
+	}
+	return &ateletpb.EgressGateway{Address: s.egressGatewayAddress}
+}
+
 type FinalizeRunningStep struct {
 	store store.Interface
 }

cmd/ateapi/internal/controlapi/workflow_testutil_test.go

@@ -42,7 +42,7 @@ func newTestActorWorkflow(t *testing.T, st store.Interface, tmplNamespace, tmplN
 	}); err != nil {
 		t.Fatalf("add template to indexer: %v", err)
 	}
-	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil)
+	return NewActorWorkflow(st, nil, nil, listersv1alpha1.NewActorTemplateLister(indexer), nil, nil, nil, "")
 }
 
 // seedWorkflowActor stores an actor with the given status, bound to the given