Add FetchJWTSVIDs and SubscribeToJWTBundles

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

@@ -5,16 +5,21 @@ import (
 	"crypto/x509"
 	"fmt"
 	"sort"
+	"time"
 
 	"github.com/sirupsen/logrus"
+	"github.com/spiffe/go-spiffe/v2/spiffeid"
 	delegatedidentityv1 "github.com/spiffe/spire-api-sdk/proto/spire/api/agent/delegatedidentity/v1"
 	"github.com/spiffe/spire-api-sdk/proto/spire/api/types"
 	"github.com/spiffe/spire/pkg/agent/api/rpccontext"
 	workload_attestor "github.com/spiffe/spire/pkg/agent/attestor/workload"
+	"github.com/spiffe/spire/pkg/agent/client"
 	"github.com/spiffe/spire/pkg/agent/endpoints"
 	"github.com/spiffe/spire/pkg/agent/manager"
 	"github.com/spiffe/spire/pkg/agent/manager/cache"
+	"github.com/spiffe/spire/pkg/common/bundleutil"
 	"github.com/spiffe/spire/pkg/common/idutil"
+	"github.com/spiffe/spire/pkg/common/telemetry"
 	"github.com/spiffe/spire/pkg/common/x509util"
 	"github.com/spiffe/spire/pkg/server/api"
 	"github.com/spiffe/spire/proto/spire/common"
@@ -245,6 +250,124 @@ func (s *Service) SubscribeToX509Bundles(req *delegatedidentityv1.SubscribeToX50
 	}
 }
 
+func (s *Service) FetchJWTSVIDs(ctx context.Context, req *delegatedidentityv1.FetchJWTSVIDsRequest) (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
+	}
+
+	selectors, err := api.SelectorsFromProto(req.Selectors)
+	if err != nil {
+		log.WithError(err).Error("Invalid argument; could not parse provided selectors")
+		return nil, status.Error(codes.InvalidArgument, "could not parse provided selectors")
+	}
+	var spiffeIDs []spiffeid.ID
+
+	log = log.WithField(telemetry.Registered, true)
+
+	identities := s.manager.MatchingIdentities(selectors)
+	for _, identity := range identities {
+		spiffeID, err := spiffeid.FromString(identity.Entry.SpiffeId)
+		if err != nil {
+			log.WithField(telemetry.SPIFFEID, identity.Entry.SpiffeId).WithError(err).Error("Invalid requested SPIFFE ID")
+			return nil, status.Errorf(codes.InvalidArgument, "invalid requested SPIFFE ID: %v", err)
+		}
+
+		spiffeIDs = append(spiffeIDs, spiffeID)
+	}
+
+	if len(spiffeIDs) == 0 {
+		log.WithField(telemetry.Registered, false).Error("No identity issued")
+		return nil, status.Error(codes.PermissionDenied, "no identity issued")
+	}
+
+	resp = new(delegatedidentityv1.FetchJWTSVIDsResponse)
+	for _, id := range spiffeIDs {
+		loopLog := log.WithField(telemetry.SPIFFEID, id.String())
+
+		var svid *client.JWTSVID
+		svid, err = s.manager.FetchJWTSVID(ctx, id, 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: id.TrustDomain().String(),
+				Path:        id.Path(),
+			},
+			ExpiresAt: svid.ExpiresAt.Unix(),
+			IssuedAt:  svid.IssuedAt.Unix(),
+		})
+
+		ttl := time.Until(svid.ExpiresAt)
+		loopLog.WithField(telemetry.TTL, ttl.Seconds()).Debug("Fetched JWT SVID")
+	}
+
+	return resp, nil
+}
+
+func (s *Service) SubscribeToJWTBundles(req *delegatedidentityv1.SubscribeToJWTBundlesRequest, stream delegatedidentityv1.DelegatedIdentity_SubscribeToJWTBundlesServer) error {
+	ctx := stream.Context()
+	log := rpccontext.Logger(ctx)
+	cachedSelectors, err := s.isCallerAuthorized(ctx, log, nil)
+
+	if err != nil {
+		return err
+	}
+
+	subscriber := s.manager.SubscribeToBundleChanges()
+
+	// send initial update....
+	jwtbundles := make(map[string][]byte)
+	for td, bundle := range subscriber.Value() {
+		jwksBytes, err := bundleutil.Marshal(bundle, bundleutil.NoX509SVIDKeys(), bundleutil.StandardJWKS())
+		if err != nil {
+			return err
+		}
+		jwtbundles[td.IDString()] = jwksBytes
+	}
+
+	resp := &delegatedidentityv1.SubscribeToJWTBundlesResponse{
+		Bundles: jwtbundles,
+	}
+
+	if err := stream.Send(resp); err != nil {
+		return err
+	}
+	for {
+		select {
+		case <-subscriber.Changes():
+			if _, err := s.isCallerAuthorized(ctx, log, cachedSelectors); err != nil {
+				return err
+			}
+			for td, bundle := range subscriber.Next() {
+				jwksBytes, err := bundleutil.Marshal(bundle, bundleutil.NoX509SVIDKeys(), bundleutil.StandardJWKS())
+				if err != nil {
+					return err
+				}
+				jwtbundles[td.IDString()] = jwksBytes
+			}
+
+			resp := &delegatedidentityv1.SubscribeToJWTBundlesResponse{
+				Bundles: jwtbundles,
+			}
+
+			if err := stream.Send(resp); err != nil {
+				return err
+			}
+		case <-ctx.Done():
+			return nil
+		}
+	}
+}
+
 func marshalBundle(certs []*x509.Certificate) []byte {
 	bundle := []byte{}
 	for _, c := range certs {

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

@@ -13,11 +13,14 @@ import (
 	"github.com/sirupsen/logrus/hooks/test"
 	"github.com/spiffe/go-spiffe/v2/bundle/spiffebundle"
 	"github.com/spiffe/go-spiffe/v2/spiffeid"
+	"github.com/spiffe/go-spiffe/v2/svid/jwtsvid"
 	"github.com/spiffe/go-spiffe/v2/svid/x509svid"
 	delegatedidentityv1 "github.com/spiffe/spire-api-sdk/proto/spire/api/agent/delegatedidentity/v1"
 	"github.com/spiffe/spire-api-sdk/proto/spire/api/types"
+	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
 
+	"github.com/spiffe/spire/pkg/agent/client"
 	"github.com/spiffe/spire/pkg/agent/manager"
 	"github.com/spiffe/spire/pkg/agent/manager/cache"
 	"github.com/spiffe/spire/pkg/common/api/middleware"
@@ -43,6 +46,9 @@ var (
 
 	bundle1 = bundleutil.BundleFromRootCA(trustDomain1, &x509.Certificate{Raw: []byte("AAA")})
 	bundle2 = bundleutil.BundleFromRootCA(trustDomain2, &x509.Certificate{Raw: []byte("BBB")})
+
+	jwksBundle1, _ = bundleutil.Marshal(bundle1, bundleutil.NoX509SVIDKeys(), bundleutil.StandardJWKS())
+	jwksBundle2, _ = bundleutil.Marshal(bundle2, bundleutil.NoX509SVIDKeys(), bundleutil.StandardJWKS())
 )
 
 func TestSubscribeToX509SVIDs(t *testing.T) {
@@ -351,6 +357,205 @@ func TestSubscribeToX509Bundles(t *testing.T) {
 	}
 }
 
+func TestFetchJWTSVIDs(t *testing.T) {
+	ca := testca.New(t, trustDomain1)
+
+	x509SVID1 := ca.CreateX509SVID(id1)
+	x509SVID2 := ca.CreateX509SVID(id2)
+
+	for _, tt := range []struct {
+		testName       string
+		identities     []cache.Identity
+		authSpiffeID   []string
+		audience       []string
+		expectCode     codes.Code
+		expectMsg      string
+		attestErr      error
+		managerErr     error
+		expectTokenIDs []spiffeid.ID
+	}{
+		{
+			testName:   "missing required audience",
+			expectCode: codes.InvalidArgument,
+			expectMsg:  "audience must be specified",
+		},
+		{
+			testName:   "Attest error",
+			attestErr:  errors.New("ohno"),
+			audience:   []string{"AUDIENCE"},
+			expectCode: codes.Internal,
+			expectMsg:  "workload attestation failed",
+		},
+		{
+			testName:     "Access to \"privileged\" admin API denied",
+			authSpiffeID: []string{"spiffe://example.org/one/wrong"},
+			audience:     []string{"AUDIENCE"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			expectCode: codes.PermissionDenied,
+			expectMsg:  "caller not configured as an authorized delegate",
+		},
+		{
+			testName:     "fetch error",
+			authSpiffeID: []string{"spiffe://example.org/one"},
+			audience:     []string{"AUDIENCE"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			managerErr: errors.New("ohno"),
+			expectCode: codes.Unavailable,
+			expectMsg:  "could not fetch JWT-SVID: ohno",
+		},
+		{
+			testName:     "success with one identities",
+			authSpiffeID: []string{"spiffe://example.org/one"},
+			audience:     []string{"AUDIENCE"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			expectTokenIDs: []spiffeid.ID{x509SVID1.ID},
+		},
+		{
+			testName:     "success with two identities",
+			authSpiffeID: []string{"spiffe://example.org/one"},
+			audience:     []string{"AUDIENCE"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+				identityFromX509SVID(x509SVID2),
+			},
+			expectTokenIDs: []spiffeid.ID{x509SVID1.ID, x509SVID2.ID},
+		},
+	} {
+		tt := tt
+		t.Run(tt.testName, func(t *testing.T) {
+			params := testParams{
+				CA:           ca,
+				Identities:   tt.identities,
+				AuthSpiffeID: tt.authSpiffeID,
+				AttestErr:    tt.attestErr,
+				ManagerErr:   tt.managerErr,
+			}
+			runTest(t, params,
+				func(ctx context.Context, client delegatedidentityv1.DelegatedIdentityClient) {
+					selectors := []*types.Selector{{Type: "sa", Value: "foo"}}
+					resp, err := client.FetchJWTSVIDs(ctx, &delegatedidentityv1.FetchJWTSVIDsRequest{
+						Audience:  tt.audience,
+						Selectors: selectors,
+					})
+
+					spiretest.RequireGRPCStatus(t, err, tt.expectCode, tt.expectMsg)
+					if tt.expectCode != codes.OK {
+						assert.Nil(t, resp)
+						return
+					}
+					var tokenIDs []spiffeid.ID
+					for _, svid := range resp.Svids {
+						parsedSVID, err := jwtsvid.ParseInsecure(svid.Token, tt.audience)
+						require.NoError(t, err, "JWT-SVID token is malformed")
+						tokenIDs = append(tokenIDs, parsedSVID.ID)
+					}
+					assert.Equal(t, tt.expectTokenIDs, tokenIDs)
+				})
+		})
+	}
+}
+func TestSubscribeToJWTBundles(t *testing.T) {
+	ca := testca.New(t, trustDomain1)
+
+	x509SVID1 := ca.CreateX509SVID(id1)
+
+	for _, tt := range []struct {
+		testName     string
+		identities   []cache.Identity
+		authSpiffeID []string
+		expectCode   codes.Code
+		expectMsg    string
+		attestErr    error
+		expectResp   []*delegatedidentityv1.SubscribeToJWTBundlesResponse
+		cacheUpdates map[spiffeid.TrustDomain]*cache.Bundle
+	}{
+
+		{
+			testName:   "Attest error",
+			attestErr:  errors.New("ohno"),
+			expectCode: codes.Internal,
+			expectMsg:  "workload attestation failed",
+		},
+		{
+			testName:     "Access to \"privileged\" admin API denied",
+			authSpiffeID: []string{"spiffe://example.org/one/wrong"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			expectCode: codes.PermissionDenied,
+			expectMsg:  "caller not configured as an authorized delegate",
+		},
+		{
+			testName:     "cache bundle update - one bundle",
+			authSpiffeID: []string{"spiffe://example.org/one"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			cacheUpdates: map[spiffeid.TrustDomain]*cache.Bundle{
+				spiffeid.RequireTrustDomainFromString(bundle1.TrustDomainID()): bundle1,
+			},
+			expectResp: []*delegatedidentityv1.SubscribeToJWTBundlesResponse{
+				{
+					Bundles: map[string][]byte{
+						bundle1.TrustDomainID(): jwksBundle1,
+					},
+				},
+			},
+		},
+		{
+			testName:     "cache bundle update - two bundles",
+			authSpiffeID: []string{"spiffe://example.org/one"},
+			identities: []cache.Identity{
+				identityFromX509SVID(x509SVID1),
+			},
+			cacheUpdates: map[spiffeid.TrustDomain]*cache.Bundle{
+				spiffeid.RequireTrustDomainFromString(bundle1.TrustDomainID()): bundle1,
+				spiffeid.RequireTrustDomainFromString(bundle2.TrustDomainID()): bundle2,
+			},
+			expectResp: []*delegatedidentityv1.SubscribeToJWTBundlesResponse{
+				{
+					Bundles: map[string][]byte{
+						bundle1.TrustDomainID(): jwksBundle1,
+						bundle2.TrustDomainID(): jwksBundle2,
+					},
+				},
+			},
+		},
+	} {
+		tt := tt
+		t.Run(tt.testName, func(t *testing.T) {
+			params := testParams{
+				CA:           ca,
+				Identities:   tt.identities,
+				AuthSpiffeID: tt.authSpiffeID,
+				AttestErr:    tt.attestErr,
+				CacheUpdates: tt.cacheUpdates,
+			}
+			runTest(t, params,
+				func(ctx context.Context, client delegatedidentityv1.DelegatedIdentityClient) {
+					req := &delegatedidentityv1.SubscribeToJWTBundlesRequest{}
+
+					stream, err := client.SubscribeToJWTBundles(ctx, req)
+
+					require.NoError(t, err)
+
+					for _, multiResp := range tt.expectResp {
+						resp, err := stream.Recv()
+
+						spiretest.RequireGRPCStatus(t, err, tt.expectCode, tt.expectMsg)
+						spiretest.RequireProtoEqual(t, multiResp, resp)
+					}
+				})
+		})
+	}
+}
+
 type testParams struct {
 	CA           *testca.CA
 	Identities   []cache.Identity
@@ -441,6 +646,16 @@ func (m *FakeManager) SubscribeToCacheChanges(selectors cache.Selectors) cache.S
 	return newFakeSubscriber(m, m.updates)
 }
 
+func (m *FakeManager) FetchJWTSVID(ctx context.Context, spiffeID spiffeid.ID, audience []string) (*client.JWTSVID, error) {
+	svid := m.ca.CreateJWTSVID(spiffeID, audience)
+	if m.err != nil {
+		return nil, m.err
+	}
+	return &client.JWTSVID{
+		Token: svid.Marshal(),
+	}, nil
+}
+
 type fakeSubscriber struct {
 	m      *FakeManager
 	ch     chan *cache.WorkloadUpdate