Allow DelegatedIdentity API clients to subscribe by PID

doc/spire_agent.md

@@ -369,7 +369,31 @@ plugins {
 
 ## Delegated Identity API
 
-The Delegated Identity API allows an authorized (i.e. delegated) workload to obtain SVIDs and bundles on behalf of workloads that cannot be attested by SPIRE Agent directly. The authorized workload does so by providing SPIRE Agent the selectors that would normally be obtained during workload attestation. The Delegated Identity API is served over the admin API endpoint.
+The Delegated Identity API allows an authorized (i.e. delegated) workload to obtain SVIDs and bundles on behalf of workloads that cannot be attested by SPIRE Agent directly. 
+
+The Delegated Identity API is served over the SPIRE Agent's admin API endpoint.
+
+Note that this explicitly and by-design grants the authorized delegate workload the ability to impersonate any of the other workloads it can obtain SVIDs for. Any workload authorized to use the
+Delegated Identity API becomes a "trusted delegate" of the SPIRE Agent, and may impersonate and act on behalf of all workload SVIDs it obtains from the SPIRE Agent.
+
+The trusted delegate workload itself is attested by the SPIRE Agent first, and the delegate's SPIFFE ID is checked against an allowlist of authorized delegates.
+
+Once these requirements are met, the trusted delegate workload can obtain SVIDS for any workloads in the scope of the SPIRE Agent instance it is interacting with.
+
+There are two ways the trusted delegate workload can request SVIDs for other workloads from the SPIRE Agent:
+
+1. By attesting the other workload itself, building a set of selectors, and then providing SPIRE Agent those selectors over the Delegated Identity API.
+ In this approach, the trusted delegate workload is entirely responsible for attesting the other workload and building the attested selectors.
+ When those selectors are presented to the SPIRE Agent, the SPIRE Agent will simply return SVIDs for any workload registration entries that match the provided selectors.
+ No other checks or attestations will be performed by the SPIRE Agent.
+
+1. By obtaining a PID for the other workload, and providing that PID to the SPIRE Agent over the Delegated Identity API.
+ In this approach, the SPIRE Agent will do attestation for the provided PID, build the attested selectors, 
+ and return SVIDs for any workload registration entries that match the selectors the SPIRE Agent attested from that PID.
+ This differs from the previous approach in that the SPIRE Agent itself (not the trusted delegate) handles the attestation of the other workload.
+ As on most platforms PIDs are not stable identifiers, the trusted delegate workload **must** ensure that the PID it provides to the SPIRE Agent 
+ via the Delegated Identity API for attestation is not recycled between the time a Delegate Identity API request is made, and a response is received (e.g. by using pidfds).
+ SVIDs and attestation results obtained via the Delegated Identity API for a PID are valid until the process referred to by the PID terminates. 
 
 To enable the Delegated Identity API, configure the admin API endpoint address and the list of SPIFFE IDs for authorized delegates. For example:
 

go.mod

@@ -2,6 +2,8 @@ module github.com/spiffe/spire
 
 go 1.22.3
 
+replace github.com/spiffe/spire-api-sdk => ../spire-api-sdk
+
 require (
  cloud.google.com/go/iam v1.1.8
  cloud.google.com/go/kms v1.18.0

pkg/agent/api/delegatedidentity/v1/service.go

@@ -54,20 +54,22 @@ func New(config Config) *Service {
  }
 
  return &Service{
- manager: config.Manager,
- attestor: endpoints.PeerTrackerAttestor{Attestor: config.Attestor},
- metrics: config.Metrics,
- authorizedDelegates: AuthorizedDelegates,
+ manager: config.Manager,
+ peerAttestor: endpoints.PeerTrackerAttestor{Attestor: config.Attestor},
+ delegateWorkloadAttestor: config.Attestor,
+ metrics: config.Metrics,
+ authorizedDelegates: AuthorizedDelegates,
  }
 }
 
 // Service implements the delegated identity server
 type Service struct {
  delegatedidentityv1.UnsafeDelegatedIdentityServer
 
- manager manager.Manager
- attestor attestor
- metrics telemetry.Metrics
+ manager manager.Manager
+ peerAttestor attestor
+ delegateWorkloadAttestor workloadattestor.Attestor
+ metrics telemetry.Metrics
 
  // SPIFFE IDs of delegates that are authorized to use this API
  authorizedDelegates map[string]bool
@@ -79,7 +81,7 @@ func (s *Service) isCallerAuthorized(ctx context.Context, log logrus.FieldLogger
  callerSelectors := cachedSelectors
 
  if callerSelectors == nil {
- callerSelectors, err = s.attestor.Attest(ctx)
+ callerSelectors, err = s.peerAttestor.Attest(ctx)
  if err != nil {
  log.WithError(err).Error("Workload attestation failed")
  return nil, status.Error(codes.Internal, "workload attestation failed")
@@ -111,6 +113,63 @@ func (s *Service) isCallerAuthorized(ctx context.Context, log logrus.FieldLogger
  return nil, status.Error(codes.PermissionDenied, "caller not configured as an authorized delegate")
 }
 
+// Attempt to attest and authorize the delegate, and then take the PID the delegate gave us
+// and attempt to attest that into a set of selectors + a subscription to changes of those selectors.
+//
+// Note that the trusted delegate is responsible for ensuring the PID is valid and not recycled,
+// from initiation of this call until the termination of the response stream, and if it is,
+// must discard any stream contents provided by this call as invalid.
+func (s *Service) SubscribeToX509SVIDsByPID(req *delegatedidentityv1.SubscribeToX509SVIDsByPIDRequest, stream delegatedidentityv1.DelegatedIdentity_SubscribeToX509SVIDsByPIDServer) error {
+ latency := adminapi.StartFirstX509SVIDUpdateLatency(s.metrics)
+ ctx := stream.Context()
+ log := rpccontext.Logger(ctx)
+ var receivedFirstUpdate bool
+
+ cachedSelectors, err := s.isCallerAuthorized(ctx, log, nil)
+ if err != nil {
+ return err
+ }
+
+ // Delegate authorized, use PID the delegate gave us to try and attest OBO
+ selectors, err := s.delegateWorkloadAttestor.Attest(ctx, int(req.Pid))
+ if err != nil {
+ return err
+ }
+
+ log.WithFields(logrus.Fields{
+ "delegate_selectors": cachedSelectors,
+ "pid_selectors": selectors,
+ }).Info("Subscribing to cache changes")
+
+ subscriber, err := s.manager.SubscribeToCacheChanges(ctx, selectors)
+ if err != nil {
+ log.WithError(err).Error("Subscribe to cache changes failed")
+ return err
+ }
+ defer subscriber.Finish()
+
+ for {
+ select {
+ case update := <-subscriber.Updates():
+ if len(update.Identities) > 0 && !receivedFirstUpdate {
+ // emit latency metric for first update containing an SVID.
+ latency.Measure()
+ receivedFirstUpdate = true
+ }
+
+ if _, err := s.isCallerAuthorized(ctx, log, cachedSelectors); err != nil {
+ return err
+ }
+
+ if err := sendX509SVIDResponse(update, stream, log); err != nil {
+ return err
+ }
+ case <-ctx.Done():
+ return nil
+ }
+ }
+}
+
 func (s *Service) SubscribeToX509SVIDs(req *delegatedidentityv1.SubscribeToX509SVIDsRequest, stream delegatedidentityv1.DelegatedIdentity_SubscribeToX509SVIDsServer) error {
  latency := adminapi.StartFirstX509SVIDUpdateLatency(s.metrics)
  ctx := stream.Context()
@@ -290,6 +349,70 @@ func (s *Service) SubscribeToX509Bundles(_ *delegatedidentityv1.SubscribeToX509B
  }
 }
 
+// Attempt to attest and authorize the delegate, and then take the PID the delegate gave us
+// and attempt to attest that into a set of selectors.
+//
+// Note that the trusted delegate is responsible for ensuring the PID is valid and not recycled,
+// from initiation of this call until the return of this call, and if it is must discard any response
+// provided by this call as invalid.
+func (s *Service) FetchJWTSVIDsByPID(ctx context.Context, req *delegatedidentityv1.FetchJWTSVIDsByPIDRequest) (resp *delegatedidentityv1.FetchJWTSVIDsResponse, err error) {
+ log := rpccontext.Logger(ctx)
+ if len(req.Audience) == 0 {
+ log.Error("Missing required audience parameter")
+ return nil, status.Error(codes.InvalidArgument, "audience must be specified")
+ }
+
+ if _, err = s.isCallerAuthorized(ctx, log, nil); err != nil {
+ return nil, err
+ }
+
+ // Delegate authorized, use PID the delegate gave us to try and attest OBO
+ selectors, err := s.delegateWorkloadAttestor.Attest(ctx, int(req.Pid))
+ if err != nil {
+ return nil, err
+ }
+
+ resp = new(delegatedidentityv1.FetchJWTSVIDsResponse)
+
+ entries := s.manager.MatchingRegistrationEntries(selectors)
+ for _, entry := range entries {
+ spiffeID, err := spiffeid.FromString(entry.SpiffeId)
+ if err != nil {
+ log.WithField(telemetry.SPIFFEID, entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
+ return nil, status.Errorf(codes.InvalidArgument, "invalid requested SPIFFE ID: %v", err)
+ }
+
+ loopLog := log.WithField(telemetry.SPIFFEID, spiffeID.String())
+
+ var svid *client.JWTSVID
+ svid, err = s.manager.FetchJWTSVID(ctx, entry, req.Audience)
+ if err != nil {
+ loopLog.WithError(err).Error("Could not fetch JWT-SVID")
+ return nil, status.Errorf(codes.Unavailable, "could not fetch JWT-SVID: %v", err)
+ }
+ resp.Svids = append(resp.Svids, &types.JWTSVID{
+ Token: svid.Token,
+ Id: &types.SPIFFEID{
+ TrustDomain: spiffeID.TrustDomain().Name(),
+ Path: spiffeID.Path(),
+ },
+ ExpiresAt: svid.ExpiresAt.Unix(),
+ IssuedAt: svid.IssuedAt.Unix(),
+ Hint: entry.Hint,
+ })
+
+ ttl := time.Until(svid.ExpiresAt)
+ loopLog.WithField(telemetry.TTL, ttl.Seconds()).Debug("Fetched JWT SVID")
+ }
+
+ if len(resp.Svids) == 0 {
+ log.Error("No identity issued")
+ return nil, status.Error(codes.PermissionDenied, "no identity issued")
+ }
+
+ return resp, nil
+}
+
 func (s *Service) FetchJWTSVIDs(ctx context.Context, req *delegatedidentityv1.FetchJWTSVIDsRequest) (resp *delegatedidentityv1.FetchJWTSVIDsResponse, err error) {
  log := rpccontext.Logger(ctx)
  if len(req.Audience) == 0 {