Identity: Move actor JWT/cert minting into the main API

benchmarking/locust/common/ateapi_pb2_grpc.py

@@ -104,6 +104,16 @@ def __init__(self, channel):
                 request_serializer=ateapi__pb2.DeleteActorEgressPolicyRequest.SerializeToString,
                 response_deserializer=ateapi__pb2.EgressPolicy.FromString,
                 _registered_method=True)
+        self.MintActorJWT = channel.unary_unary(
+                '/ateapi.Control/MintActorJWT',
+                request_serializer=ateapi__pb2.MintActorJWTRequest.SerializeToString,
+                response_deserializer=ateapi__pb2.MintActorJWTResponse.FromString,
+                _registered_method=True)
+        self.MintActorCertificate = channel.unary_unary(
+                '/ateapi.Control/MintActorCertificate',
+                request_serializer=ateapi__pb2.MintActorCertificateRequest.SerializeToString,
+                response_deserializer=ateapi__pb2.MintActorCertificateResponse.FromString,
+                _registered_method=True)
         self.GetActorSnapshot = channel.unary_unary(
                 '/ateapi.Control/GetActorSnapshot',
                 request_serializer=ateapi__pb2.GetActorSnapshotRequest.SerializeToString,
@@ -294,6 +304,28 @@ def DeleteActorEgressPolicy(self, request, context):
         context.set_details('Method not implemented!')
         raise NotImplementedError('Method not implemented!')
 
+    def MintActorJWT(self, request, context):
+        """Create a Substrate-issued JWT asserting the actor identity.
+
+        * Called by the egress gateway when actor JWT injection is configured for outbound requests.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
+    def MintActorCertificate(self, request, context):
+        """Create a Substrate-issued SPIFFE certificate asserting the actor identity.
+
+        * Called by atelet to provision an atunnel with a certificate for
+        communication with the egress gateway.  TODO(ahmedtd): Migrate this use
+        case to a distinct certificate to prevent actor/atunnel confusion.
+        * Called by the egress gateway when actor client certificate injection is
+        configured for outbound requests.
+        """
+        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
+        context.set_details('Method not implemented!')
+        raise NotImplementedError('Method not implemented!')
+
     def GetActorSnapshot(self, request, context):
         """Get an ActorSnapshot.
         """
@@ -502,6 +534,16 @@ def add_ControlServicer_to_server(servicer, server):
                     request_deserializer=ateapi__pb2.DeleteActorEgressPolicyRequest.FromString,
                     response_serializer=ateapi__pb2.EgressPolicy.SerializeToString,
             ),
+            'MintActorJWT': grpc.unary_unary_rpc_method_handler(
+                    servicer.MintActorJWT,
+                    request_deserializer=ateapi__pb2.MintActorJWTRequest.FromString,
+                    response_serializer=ateapi__pb2.MintActorJWTResponse.SerializeToString,
+            ),
+            'MintActorCertificate': grpc.unary_unary_rpc_method_handler(
+                    servicer.MintActorCertificate,
+                    request_deserializer=ateapi__pb2.MintActorCertificateRequest.FromString,
+                    response_serializer=ateapi__pb2.MintActorCertificateResponse.SerializeToString,
+            ),
             'GetActorSnapshot': grpc.unary_unary_rpc_method_handler(
                     servicer.GetActorSnapshot,
                     request_deserializer=ateapi__pb2.GetActorSnapshotRequest.FromString,
@@ -916,6 +958,60 @@ def DeleteActorEgressPolicy(request,
             metadata,
             _registered_method=True)
 
+    @staticmethod
+    def MintActorJWT(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/ateapi.Control/MintActorJWT',
+            ateapi__pb2.MintActorJWTRequest.SerializeToString,
+            ateapi__pb2.MintActorJWTResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
+    @staticmethod
+    def MintActorCertificate(request,
+            target,
+            options=(),
+            channel_credentials=None,
+            call_credentials=None,
+            insecure=False,
+            compression=None,
+            wait_for_ready=None,
+            timeout=None,
+            metadata=None):
+        return grpc.experimental.unary_unary(
+            request,
+            target,
+            '/ateapi.Control/MintActorCertificate',
+            ateapi__pb2.MintActorCertificateRequest.SerializeToString,
+            ateapi__pb2.MintActorCertificateResponse.FromString,
+            options,
+            channel_credentials,
+            insecure,
+            call_credentials,
+            compression,
+            wait_for_ready,
+            timeout,
+            metadata,
+            _registered_method=True)
+
     @staticmethod
     def GetActorSnapshot(request,
             target,
@@ -1482,151 +1578,3 @@ def DeleteActorTemplate(request,
             timeout,
             metadata,
             _registered_method=True)
-
-
-class ActorIdentityStub:
-    """ActorIdentity allows substrate workloads to exchange their
-    infrastructure-level credentials (k8s service account token, etc.) for a
-    substrate actor-level credential. A given substrate actor might migrate
-    between many different physical workers over the course of its lifecycle,
-    whereas the actor credential's identity will be stable for the life of the
-    actor.
-    """
-
-    def __init__(self, channel):
-        """Constructor.
-
-        Args:
-            channel: A grpc.Channel.
-        """
-        self.MintJWT = channel.unary_unary(
-                '/ateapi.ActorIdentity/MintJWT',
-                request_serializer=ateapi__pb2.MintJWTRequest.SerializeToString,
-                response_deserializer=ateapi__pb2.MintJWTResponse.FromString,
-                _registered_method=True)
-        self.MintCert = channel.unary_unary(
-                '/ateapi.ActorIdentity/MintCert',
-                request_serializer=ateapi__pb2.MintCertRequest.SerializeToString,
-                response_deserializer=ateapi__pb2.MintCertResponse.FromString,
-                _registered_method=True)
-
-
-class ActorIdentityServicer:
-    """ActorIdentity allows substrate workloads to exchange their
-    infrastructure-level credentials (k8s service account token, etc.) for a
-    substrate actor-level credential. A given substrate actor might migrate
-    between many different physical workers over the course of its lifecycle,
-    whereas the actor credential's identity will be stable for the life of the
-    actor.
-    """
-
-    def MintJWT(self, request, context):
-        """Request an Actor Identity JWT.
-
-        To call this RPC, you must be authenticated as the Kubernetes Pod that is
-        currently running the requested actor.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-    def MintCert(self, request, context):
-        """Request an Actor Identity Certificate for an actor.
-
-        Actors do not call this RPC themselves. The atelet hosting the actor calls
-        it on the actor's behalf, authenticating with its own client certificate
-        rather than a bearer token.
-
-        Authorization is decided on that client certificate and the worker
-        identity attested by atelet. Ateapi verifies that the worker is assigned to
-        the actor and that the actor points back to that exact worker before signing.
-
-        The certificate in the response is the actor's identity, not the atelet's.
-        """
-        context.set_code(grpc.StatusCode.UNIMPLEMENTED)
-        context.set_details('Method not implemented!')
-        raise NotImplementedError('Method not implemented!')
-
-
-def add_ActorIdentityServicer_to_server(servicer, server):
-    rpc_method_handlers = {
-            'MintJWT': grpc.unary_unary_rpc_method_handler(
-                    servicer.MintJWT,
-                    request_deserializer=ateapi__pb2.MintJWTRequest.FromString,
-                    response_serializer=ateapi__pb2.MintJWTResponse.SerializeToString,
-            ),
-            'MintCert': grpc.unary_unary_rpc_method_handler(
-                    servicer.MintCert,
-                    request_deserializer=ateapi__pb2.MintCertRequest.FromString,
-                    response_serializer=ateapi__pb2.MintCertResponse.SerializeToString,
-            ),
-    }
-    generic_handler = grpc.method_handlers_generic_handler(
-            'ateapi.ActorIdentity', rpc_method_handlers)
-    server.add_generic_rpc_handlers((generic_handler,))
-    server.add_registered_method_handlers('ateapi.ActorIdentity', rpc_method_handlers)
-
-
- # This class is part of an EXPERIMENTAL API.
-class ActorIdentity:
-    """ActorIdentity allows substrate workloads to exchange their
-    infrastructure-level credentials (k8s service account token, etc.) for a
-    substrate actor-level credential. A given substrate actor might migrate
-    between many different physical workers over the course of its lifecycle,
-    whereas the actor credential's identity will be stable for the life of the
-    actor.
-    """
-
-    @staticmethod
-    def MintJWT(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(
-            request,
-            target,
-            '/ateapi.ActorIdentity/MintJWT',
-            ateapi__pb2.MintJWTRequest.SerializeToString,
-            ateapi__pb2.MintJWTResponse.FromString,
-            options,
-            channel_credentials,
-            insecure,
-            call_credentials,
-            compression,
-            wait_for_ready,
-            timeout,
-            metadata,
-            _registered_method=True)
-
-    @staticmethod
-    def MintCert(request,
-            target,
-            options=(),
-            channel_credentials=None,
-            call_credentials=None,
-            insecure=False,
-            compression=None,
-            wait_for_ready=None,
-            timeout=None,
-            metadata=None):
-        return grpc.experimental.unary_unary(
-            request,
-            target,
-            '/ateapi.ActorIdentity/MintCert',
-            ateapi__pb2.MintCertRequest.SerializeToString,
-            ateapi__pb2.MintCertResponse.FromString,
-            options,
-            channel_credentials,
-            insecure,
-            call_credentials,
-            compression,
-            wait_for_ready,
-            timeout,
-            metadata,
-            _registered_method=True)

cmd/ate-setup/internal/steps/create.go

@@ -255,8 +255,9 @@ func (e *Env) createJWTPool(ctx context.Context, namespace, name string) error {
 	if err != nil {
 		return fmt.Errorf("while generating the JWT authority for %s/%s: %w", namespace, name, err)
 	}
-	poolBytes, err := localjwtauthority.Marshal(&localjwtauthority.Pool{
-		Authorities: []*localjwtauthority.Authority{authority},
+	poolBytes, err := localjwtauthority.Marshal(&localjwtauthority.ConcretePool{
+		Authorities:      []*localjwtauthority.Authority{authority},
+		ActiveForSigning: poolKeyID,
 	})
 	if err != nil {
 		return fmt.Errorf("while marshaling the JWT pool for %s/%s: %w", namespace, name, err)

cmd/ateapi/internal/actoridentity/actoridentity.go

@@ -23,14 +23,13 @@
 	"fmt"
 	"log/slog"
 	"net/url"
-	"os"
 	"path"
 	"time"
 
-	"github.com/agent-substrate/substrate/cmd/ateapi/internal/actoridjwt"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/controlapi"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/store"
 	"github.com/agent-substrate/substrate/cmd/ateapi/internal/workercache"
+	"github.com/agent-substrate/substrate/internal/actoridjwt"
 	"github.com/agent-substrate/substrate/internal/localca"
 	"github.com/agent-substrate/substrate/internal/localjwtauthority"
 	"github.com/agent-substrate/substrate/internal/principal"
@@ -47,13 +46,14 @@
 
 // Server implements ateapipb.ActorIdentityServer
 type Server struct {
 	ateapipb.UnimplementedActorIdentityServer
 
+	// TODO(identity): Issuer is probably logically a property of the JWT
+	// signing pool.
 	actorIdentityJWTIssuer string
 
-	// TODO: Cache the signing keys in memory, so we don't read from a file every time.
-	actorIDJWTPoolFile string
-	actorIDCAPool      localca.Pool
+	actorIDJWTPool localjwtauthority.Pool
+	actorIDCAPool  localca.Pool
 
 	// store is the actor database. MintCert consults it to confirm the caller
 	// is entitled to the actor it is asking for a credential for.
@@ -61,12 +61,12 @@
 	workers *workercache.Cache
 }
 
 var _ ateapipb.ActorIdentityServer = (*Server)(nil)
 
-func New(actorIdentityJWTIssuer, actorIDJWTPoolFile string, actorIDCAPool localca.Pool, store store.Interface, workers *workercache.Cache) *Server {
+func New(actorIdentityJWTIssuer string, actorIDJWTPool localjwtauthority.Pool, actorIDCAPool localca.Pool, store store.Interface, workers *workercache.Cache) *Server {
 	return &Server{
 		actorIdentityJWTIssuer: actorIdentityJWTIssuer,
-		actorIDJWTPoolFile:     actorIDJWTPoolFile,
+		actorIDJWTPool:         actorIDJWTPool,
 		actorIDCAPool:          actorIDCAPool,
 		store:                  store,
 		workers:                workers,
@@ -87,7 +87,7 @@
 	actorCertificateLifetime = time.Hour
 )
 
 func (s *Server) MintJWT(ctx context.Context, req *ateapipb.MintJWTRequest) (*ateapipb.MintJWTResponse, error) {
 	caller, ok := principal.FromContext(ctx)
 	if !ok || caller.Kind != principal.KindJWT {
 		return nil, status.Errorf(codes.Unauthenticated, "JWT authentication is required")
@@ -102,15 +102,9 @@
 
 	// TODO: Cross-check the verified caller and requested actor against the actor database.
 
-	// TODO: Cache signing keys in memory, so we don't read from disk every time.
-	signingPoolBytes, err := os.ReadFile(s.actorIDJWTPoolFile)
-	if err != nil {
-		return nil, fmt.Errorf("while reading signing pool bytes: %w", err)
-	}
-
-	signingPool, err := localjwtauthority.Unmarshal(signingPoolBytes)
-	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")
 	}
 
 	actorClaims := &actoridjwt.Claims{
@@ -131,13 +125,7 @@
 		},
 	}
 
-	actorWireClaims, err := actoridjwt.ClaimsToWire(actorClaims)
-	if err != nil {
-		return nil, fmt.Errorf("while making actor JWT claims: %w", err)
-	}
-
-	// Assume the first authority is the one to use for signing.
-	actorJWT, err := actoridjwt.Sign(actorWireClaims, signingPool.Authorities[0].SigningKey, signingPool.Authorities[0].Algorithm, signingPool.Authorities[0].ID)
+	actorJWT, err := s.actorIDJWTPool.SignJWT(actorClaims)
 	if err != nil {
 		return nil, fmt.Errorf("while signing actor JWT: %w", err)
 	}
@@ -147,7 +135,7 @@
 	}, nil
 }
 
 func (s *Server) MintCert(ctx context.Context, req *ateapipb.MintCertRequest) (*ateapipb.MintCertResponse, error) {
 	caller, err := authenticateAtelet(ctx)
 	if err != nil {
 		return nil, err
@@ -286,13 +274,13 @@
 	return &ateletCaller{podName: identity.PodName, nodeName: identity.NodeName}, nil
 }
 
 func validateMintJWTRequest(ctx context.Context, req *ateapipb.MintJWTRequest) field.ErrorList {
 	// Call the generated validation.
 	op := operation.Operation{Type: operation.Create}
 	return controlapi.Validate_MintJWTRequest(ctx, op, nil, req, nil)
 }
 
 func validateMintCertRequest(ctx context.Context, req *ateapipb.MintCertRequest) field.ErrorList {
 	// Call the generated validation.
 	op := operation.Operation{Type: operation.Create}
 	return controlapi.Validate_MintCertRequest(ctx, op, nil, req, nil)
@@ -304,7 +292,7 @@
 // The worker is resolved from cache first (hot path), but cache misses and
 // denials fall back to the authoritative store to handle watch-delivery lag
 // right after ResumeActor.
 func (s *Server) authorizeActor(ctx context.Context, caller *ateletCaller, req *ateapipb.MintCertRequest) (*ateapipb.Actor, resources.ActorRef, error) {
 	reason := "worker not found"
 	worker, err := s.workers.Worker(req.GetWorker().GetName())
 	if err != nil && !errors.Is(err, store.ErrNotFound) {
@@ -349,7 +337,7 @@
 // denyMint logs the internal reason and returns a uniform PermissionDenied.
 // Denials are deliberately indistinguishable from each other: a caller that
 // is not entitled to a worker should not learn its assignment.
 func (s *Server) denyMint(ctx context.Context, caller *ateletCaller, req *ateapipb.MintCertRequest, reason string, args ...any) error {
 	slog.WarnContext(ctx, "ActorIdentity denied: "+reason,
 		append([]any{slog.String("worker", req.GetWorker().GetName()), slog.String("callerPod", caller.podName), slog.String("callerNode", caller.nodeName)}, args...)...)
 	return status.Error(codes.PermissionDenied, "caller is not permitted to mint credentials for this actor")