diff --git a/pkg/agent/client/client.go b/pkg/agent/client/client.go index 14af4fd7d2..9431e51b38 100644 --- a/pkg/agent/client/client.go +++ b/pkg/agent/client/client.go @@ -72,6 +72,12 @@ type JWTSVID struct { ExpiresAt time.Time } +type WITSVID struct { + Token string + IssuedAt time.Time + ExpiresAt time.Time +} + type SyncStats struct { Entries SyncEntriesStats Bundles SyncBundlesStats @@ -94,6 +100,7 @@ type Client interface { RenewSVID(ctx context.Context, csr []byte) (*X509SVID, error) NewX509SVIDs(ctx context.Context, csrs map[string][]byte) (map[string]*X509SVID, error) NewJWTSVID(ctx context.Context, entryID string, audience []string, hasCacheHit bool) (*JWTSVID, spiffeid.ID, error) + NewWITSVIDs(ctx context.Context, publicKeys map[string]crypto.PublicKey) (map[string]*WITSVID, error) PostStatus(ctx context.Context, agentVersion string) error // Release releases any resources that were held by this Client, if any. @@ -402,6 +409,49 @@ func (c *client) NewJWTSVID(ctx context.Context, entryID string, audience []stri }, spiffeId, nil } +func (c *client) NewWITSVIDs(ctx context.Context, publicKeys map[string]crypto.PublicKey) (map[string]*WITSVID, error) { + c.c.RotMtx.RLock() + defer c.c.RotMtx.RUnlock() + + ctx, cancel := context.WithTimeout(ctx, rpcTimeout) + defer cancel() + + svids := make(map[string]*WITSVID) + var params []*svidv1.NewWITSVIDParams + for entryID, publicKey := range publicKeys { + pkixPublicKey, err := x509.MarshalPKIXPublicKey(publicKey) + if err != nil { + return nil, fmt.Errorf("failed to marshal PKIX public key for entry %q: %w", entryID, err) + } + + params = append(params, &svidv1.NewWITSVIDParams{ + EntryId: entryID, + PublicKey: pkixPublicKey, + }) + } + + protoSVIDs, err := c.fetchWITSVIDs(ctx, params) + if err != nil { + return nil, err + } + + for i, s := range protoSVIDs { + entryID := params[i].EntryId + if s == nil { + c.c.Log.WithField(telemetry.RegistrationID, entryID).Debug("Entry not found") + continue + } + + svids[entryID] = &WITSVID{ + Token: s.Token, + IssuedAt: time.Unix(s.IssuedAt, 0).UTC(), + ExpiresAt: time.Unix(s.ExpiresAt, 0).UTC(), + } + } + + return svids, nil +} + // Release the underlying connection. func (c *client) Release() { c.release(nil) @@ -768,6 +818,39 @@ func (c *client) fetchSVIDs(ctx context.Context, params []*svidv1.NewX509SVIDPar return svids, nil } +func (c *client) fetchWITSVIDs(ctx context.Context, params []*svidv1.NewWITSVIDParams) ([]*types.WITSVID, error) { + svidClient, connection, err := c.newSVIDClient() + if err != nil { + return nil, err + } + defer connection.Release() + + resp, err := svidClient.BatchNewWITSVID(ctx, &svidv1.BatchNewWITSVIDRequest{ + Params: params, + }) + if err != nil { + c.release(connection) + c.withErrorFields(err).Error("Failed to batch new WIT-SVID(s)") + return nil, fmt.Errorf("failed to batch new WIT-SVID(s): %w", err) + } + + okStatus := int32(codes.OK) + var svids []*types.WITSVID + for i, r := range resp.Results { + if r.Status.Code != okStatus { + c.c.Log.WithFields(logrus.Fields{ + telemetry.RegistrationID: params[i].EntryId, + telemetry.Status: r.Status.Code, + telemetry.Error: r.Status.Message, + }).Warn("Failed to mint WIT-SVID") + } + + svids = append(svids, r.Svid) + } + + return svids, nil +} + func (c *client) newEntryClient() (entryv1.EntryClient, *nodeConn, error) { conn, err := c.getOrOpenConn() if err != nil { diff --git a/pkg/agent/client/client_test.go b/pkg/agent/client/client_test.go index b53da479b3..6e3f1e9eca 100644 --- a/pkg/agent/client/client_test.go +++ b/pkg/agent/client/client_test.go @@ -3,6 +3,9 @@ package client import ( "context" "crypto" + "crypto/ecdsa" + "crypto/elliptic" + "crypto/rand" "crypto/x509" "errors" "fmt" @@ -80,12 +83,20 @@ var ( }, } - testSvids = map[string]*X509SVID{ + testX509SVIDs = map[string]*X509SVID{ "entry-id": { CertChain: []byte{11, 22, 33}, }, } + testWITSVIDs = map[string]*WITSVID{ + "entry-id": { + Token: "SOME TOKEN", + IssuedAt: time.Unix(12345, 0).UTC(), + ExpiresAt: time.Unix(54321, 0).UTC(), + }, + } + testBundles = map[string]*common.Bundle{ "spiffe://example.org": { TrustDomainId: "spiffe://example.org", @@ -505,7 +516,7 @@ func TestNewX509SVIDs(t *testing.T) { batchSVIDErr: nil, wantError: assert.NoError, assertFuncConn: assertConnectionIsNotNil, - testSvids: testSvids, + testSvids: testX509SVIDs, }, { name: "failed", @@ -570,6 +581,15 @@ func newTestCSRs() map[string][]byte { } } +func newTestPublicKeys(t *testing.T) map[string]crypto.PublicKey { + signer, err := ecdsa.GenerateKey(elliptic.P256(), rand.Reader) + require.NoError(t, err) + + return map[string]crypto.PublicKey{ + "entry-id": signer.Public(), + } +} + func TestFetchReleaseWaitsForFetchUpdatesToFinish(t *testing.T) { client, tc := createClient(t) @@ -990,6 +1010,139 @@ func TestFetchJWTSVID(t *testing.T) { } } +func TestNewWITSVIDs(t *testing.T) { + logHook.Reset() + + sClient, tc := createClient(t) + entries := []*types.Entry{ + { + Id: "ENTRYID1", + ParentId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/host"}, + SpiffeId: &types.SPIFFEID{ + TrustDomain: "example.org", + Path: "/id1", + }, + Selectors: []*types.Selector{ + {Type: "S", Value: "1"}, + }, + FederatesWith: []string{"domain1.test"}, + RevisionNumber: 1234, + }, + // This entry should be ignored since it is missing an entry ID + { + ParentId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/host"}, + SpiffeId: &types.SPIFFEID{ + TrustDomain: "example.org", + Path: "/id2", + }, + Selectors: []*types.Selector{ + {Type: "S", Value: "2"}, + }, + FederatesWith: []string{"domain2.test"}, + }, + // This entry should be ignored since it is missing a SPIFFE ID + { + Id: "ENTRYID3", + ParentId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/host"}, + Selectors: []*types.Selector{ + {Type: "S", Value: "3"}, + }, + }, + // This entry should be ignored since it is missing selectors + { + Id: "ENTRYID4", + ParentId: &types.SPIFFEID{TrustDomain: "example.org", Path: "/host"}, + SpiffeId: &types.SPIFFEID{ + TrustDomain: "example.org", + Path: "/id4", + }, + }, + } + witSVIDs := map[string]*types.WITSVID{ + "entry-id": { + Id: &types.SPIFFEID{TrustDomain: "example.org", Path: "/path"}, + Token: "SOME TOKEN", + IssuedAt: 12345, + ExpiresAt: 54321, + }, + } + + tests := []struct { + name string + entries []*types.Entry + witSVIDs map[string]*types.WITSVID + batchSVIDErr error + wantError assert.ErrorAssertionFunc + assertFuncConn func(t *testing.T, client *client) + testSvids map[string]*WITSVID + expectedLogs []spiretest.LogEntry + }{ + { + name: "success", + entries: entries, + witSVIDs: witSVIDs, + batchSVIDErr: nil, + wantError: assert.NoError, + assertFuncConn: assertConnectionIsNotNil, + testSvids: testWITSVIDs, + }, + { + name: "failed", + entries: entries, + witSVIDs: witSVIDs, + batchSVIDErr: status.Error(codes.NotFound, "not found when executing BatchNewWITSVID"), + wantError: assert.Error, + assertFuncConn: assertConnectionIsNil, + testSvids: nil, + expectedLogs: []spiretest.LogEntry{ + { + Level: logrus.ErrorLevel, + Message: "Failed to batch new WIT-SVID(s)", + Data: logrus.Fields{ + telemetry.StatusCode: "NotFound", + telemetry.StatusMessage: "not found when executing BatchNewWITSVID", + logrus.ErrorKey: "rpc error: code = NotFound desc = not found when executing BatchNewWITSVID", + }, + }, + }, + }, + } + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + tc.entryServer.entries = tt.entries + tc.svidServer.witSVIDs = tt.witSVIDs + tc.svidServer.batchSVIDErr = tt.batchSVIDErr + + // Simulate an ongoing SVID rotation (request should not be made in the middle of a rotation) + sClient.c.RotMtx.Lock() + + // Do the request in a different go routine + var wg sync.WaitGroup + var svids map[string]*WITSVID + err := errors.New("a not nil error") + wg.Go(func() { + svids, err = sClient.NewWITSVIDs(ctx, newTestPublicKeys(t)) + }) + + // The request should wait until the SVID rotation finishes + require.Contains(t, "a not nil error", err.Error()) + require.Nil(t, svids) + + // Simulate the end of the SVID rotation + sClient.c.RotMtx.Unlock() + wg.Wait() + + // Assert results + spiretest.AssertLogsContainEntries(t, logHook.AllEntries(), tt.expectedLogs) + tt.assertFuncConn(t, sClient) + if !tt.wantError(t, err, fmt.Sprintf("error was not expected for test case %s", tt.name)) { + return + } + assert.Equal(t, tt.testSvids, svids) + }) + } +} + // createClient creates a sample client with mocked components for testing purposes func createClient(t *testing.T) (*client, *testServer) { tc := &testServer{ @@ -1118,6 +1271,7 @@ type fakeSVIDServer struct { newJWTSVID error x509SVIDs map[string]*types.X509SVID jwtSVID *types.JWTSVID + witSVIDs map[string]*types.WITSVID simulateRelease func() } @@ -1166,6 +1320,42 @@ func (c *fakeSVIDServer) NewJWTSVID(context.Context, *svidv1.NewJWTSVIDRequest) }, nil } +func (c *fakeSVIDServer) BatchNewWITSVID(_ context.Context, in *svidv1.BatchNewWITSVIDRequest) (*svidv1.BatchNewWITSVIDResponse, error) { + if c.batchSVIDErr != nil { + return nil, c.batchSVIDErr + } + + // Simulate async calls + if c.simulateRelease != nil { + go c.simulateRelease() + } + + var results []*svidv1.BatchNewWITSVIDResponse_Result + for _, param := range in.Params { + svid, ok := c.witSVIDs[param.EntryId] + switch { + case ok: + results = append(results, &svidv1.BatchNewWITSVIDResponse_Result{ + Status: &types.Status{ + Code: int32(codes.OK), + }, + Svid: svid, + }) + default: + results = append(results, &svidv1.BatchNewWITSVIDResponse_Result{ + Status: &types.Status{ + Code: int32(codes.NotFound), + Message: "svid not found", + }, + }) + } + } + + return &svidv1.BatchNewWITSVIDResponse{ + Results: results, + }, nil +} + type fakeAgentServer struct { agentv1.UnimplementedAgentServer err error