Merge pull request #583 from azdagron/node-workload-federated-bundle-…

…support Node and Workload API support for federated bundles
spiffe · Sep 7, 2018 · 8927a86 · 8927a86
2 parents 7ef6699 + f4ef97a
commit 8927a86
Show file tree

Hide file tree

Showing 28 changed files with 688 additions and 495 deletions.
diff --git a/pkg/agent/attestor/node/node.go b/pkg/agent/attestor/node/node.go
@@ -307,6 +307,9 @@ func (a *attestor) serverCredFunc(bundle []*x509.Certificate) func() (credential
 }
 
 func (a *attestor) parseAttestationResponse(id string, r *node.AttestResponse) (*x509.Certificate, []*x509.Certificate, error) {
+	if r.SvidUpdate == nil {
+		return nil, nil, errors.New("response missing svid update")
+	}
 	if len(r.SvidUpdate.Svids) < 1 {
 		return nil, nil, errors.New("no svid received")
 	}
@@ -321,12 +324,21 @@ func (a *attestor) parseAttestationResponse(id string, r *node.AttestResponse) (
 		return nil, nil, fmt.Errorf("invalid svid: %v", err)
 	}
 
-	bundle, err := x509.ParseCertificates(r.SvidUpdate.Bundle)
+	if r.SvidUpdate.Bundles == nil {
+		return nil, nil, errors.New("missing bundles")
+	}
+
+	bundle := r.SvidUpdate.Bundles[a.c.TrustDomain.String()]
+	if bundle == nil {
+		return nil, nil, errors.New("missing bundle")
+	}
+
+	bundleCerts, err := x509.ParseCertificates(bundle.CaCerts)
 	if err != nil {
 		return nil, nil, fmt.Errorf("invalid bundle: %v", bundle)
 	}
 
-	return svid, bundle, nil
+	return svid, bundleCerts, nil
 }
 
 func (a *attestor) serverID() *url.URL {

diff --git a/pkg/agent/attestor/node/node_test.go b/pkg/agent/attestor/node/node_test.go
@@ -237,6 +237,9 @@ func (s *NodeAttestorTestSuite) setAttestResponse(challenges []challengeResponse
 	svid, _, err := util.LoadSVIDFixture()
 	s.Require().NoError(err)
 
+	bundle, err := util.LoadBundleFixture()
+	s.Require().NoError(err)
+
 	stream := mock_node.NewMockNode_AttestClient(s.ctrl)
 	stream.EXPECT().Send(gomock.Any())
 	for _, challenge := range challenges {
@@ -251,7 +254,14 @@ func (s *NodeAttestorTestSuite) setAttestResponse(challenges []challengeResponse
 				"spiffe://example.com/spire/agent/join_token/foobar": &node.X509SVID{
 					Cert:      svid.Raw,
 					ExpiresAt: svid.NotAfter.Unix(),
-				}},
+				},
+			},
+			Bundles: map[string]*node.Bundle{
+				"spiffe://example.com": &node.Bundle{
+					Id:      "spiffe://example.com",
+					CaCerts: bundle[0].Raw,
+				},
+			},
 		}}, nil)
 	stream.EXPECT().CloseSend()
 	stream.EXPECT().Recv().Return(nil, io.EOF)

diff --git a/pkg/agent/client/client.go b/pkg/agent/client/client.go
@@ -120,7 +120,7 @@ func (c *client) FetchUpdates(ctx context.Context, req *node.FetchX509SVIDReques
 
 	regEntries := map[string]*common.RegistrationEntry{}
 	svids := map[string]*node.X509SVID{}
-	var lastBundle []byte
+	bundles := map[string]*node.Bundle{}
 	// Read all the server responses from the stream.
 	for {
 		resp, err := stream.Recv()
@@ -129,7 +129,13 @@ func (c *client) FetchUpdates(ctx context.Context, req *node.FetchX509SVIDReques
 		}
 		if err != nil {
 			// There was an error receiving a response, exit loop to return what we have.
-			return &Update{regEntries, svids, lastBundle}, err
+			logrus.Errorf("failed to consume entire SVID update stream: %v", err)
+			return nil, err
+		}
+
+		if resp.SvidUpdate == nil {
+			logrus.Warn("empty update in SVID update stream")
+			continue
 		}
 
 		for _, re := range resp.SvidUpdate.RegistrationEntries {
@@ -138,12 +144,14 @@ func (c *client) FetchUpdates(ctx context.Context, req *node.FetchX509SVIDReques
 		for spiffeid, svid := range resp.SvidUpdate.Svids {
 			svids[spiffeid] = svid
 		}
-		lastBundle = resp.SvidUpdate.Bundle
+		for spiffeid, bundle := range resp.SvidUpdate.Bundles {
+			bundles[spiffeid] = bundle
+		}
 	}
 	return &Update{
 		Entries: regEntries,
 		SVIDs:   svids,
-		Bundle:  lastBundle,
+		Bundles: bundles,
 	}, nil
 }
 

diff --git a/pkg/agent/client/client_test.go b/pkg/agent/client/client_test.go
@@ -36,7 +36,6 @@ func TestFetchUpdates(t *testing.T) {
 	}
 	res := &node.FetchX509SVIDResponse{
 		SvidUpdate: &node.X509SVIDUpdate{
-			Bundle: []byte{10, 20, 30, 40},
 			RegistrationEntries: []*common.RegistrationEntry{{
 				EntryId: "1",
 			}},
@@ -45,6 +44,12 @@ func TestFetchUpdates(t *testing.T) {
 					Cert: []byte{11, 22, 33},
 				},
 			},
+			Bundles: map[string]*node.Bundle{
+				"spiffe://example.org": {
+					Id:      "spiffe://example.org",
+					CaCerts: []byte{10, 20, 30, 40},
+				},
+			},
 		},
 	}
 
@@ -57,7 +62,7 @@ func TestFetchUpdates(t *testing.T) {
 	update, err := client.FetchUpdates(context.Background(), req)
 	require.Nil(t, err)
 
-	assert.Equal(t, res.SvidUpdate.Bundle, update.Bundle)
+	assert.Equal(t, res.SvidUpdate.Bundles, update.Bundles)
 	assert.Equal(t, res.SvidUpdate.Svids, update.SVIDs)
 	for _, entry := range res.SvidUpdate.RegistrationEntries {
 		assert.Equal(t, entry, update.Entries[entry.EntryId])

diff --git a/pkg/agent/client/update.go b/pkg/agent/client/update.go
@@ -11,7 +11,7 @@ import (
 type Update struct {
 	Entries map[string]*common.RegistrationEntry
 	SVIDs   map[string]*node.X509SVID
-	Bundle  []byte
+	Bundles map[string]*node.Bundle
 }
 
 func (u *Update) String() string {
@@ -38,12 +38,11 @@ func (u *Update) String() string {
 		}
 		buffer.WriteString(" ")
 	}
-	buffer.WriteString("], Bundle: ")
-	if u.Bundle != nil && len(u.Bundle) > 0 {
-		buffer.WriteString("bytes")
-	} else {
-		buffer.WriteString("none")
+	buffer.WriteString("], Bundles: [")
+	for spiffeid := range u.Bundles {
+		buffer.WriteString(spiffeid)
+		buffer.WriteString(" ")
 	}
-	buffer.WriteString("}")
+	buffer.WriteString("]}")
 	return buffer.String()
 }
diff --git a/pkg/agent/client/update_test.go b/pkg/agent/client/update_test.go
@@ -15,17 +15,22 @@ var (
 func TestString(t *testing.T) {
 	entries := regEntriesMap["resp1"]
 	u := &Update{
-		Bundle:  []byte{1, 2, 3},
 		Entries: map[string]*common.RegistrationEntry{entries[0].EntryId: entries[0]},
 		SVIDs: map[string]*node.X509SVID{
 			"spiffe://example.org": {
 				Cert:      []byte{4, 5},
 				ExpiresAt: 5,
 			},
 		},
+		Bundles: map[string]*node.Bundle{
+			"spiffe://example.org": {
+				Id:      "spiffe://example.org",
+				CaCerts: []byte{1, 2, 3},
+			},
+		},
 	}
 
-	expected := "{ Entries: [{ spiffeID: spiffe://example.org/spire/agent, parentID: spiffe://example.org/spire/agent/join_token/abcd, selectors: [type:\"spiffe_id\" value:\"spiffe://example.org/spire/agent/join_token/abcd\" ]}], SVIDs: [spiffe://example.org: cert:\"\\004\\005\" expires_at:5  ], Bundle: bytes}"
+	expected := "{ Entries: [{ spiffeID: spiffe://example.org/spire/agent, parentID: spiffe://example.org/spire/agent/join_token/abcd, selectors: [type:\"spiffe_id\" value:\"spiffe://example.org/spire/agent/join_token/abcd\" ]}], SVIDs: [spiffe://example.org: cert:\"\\004\\005\" expires_at:5  ], Bundles: [spiffe://example.org ]}"
 	if u.String() != expected {
 		t.Errorf("expected: %s, got: %s", expected, u.String())
 	}

diff --git a/pkg/agent/endpoints/workload/handler.go b/pkg/agent/endpoints/workload/handler.go
@@ -100,10 +100,12 @@ func (h *Handler) sendResponse(update *cache.WorkloadUpdate, stream workload.Spi
 func (h *Handler) composeResponse(update *cache.WorkloadUpdate) (*workload.X509SVIDResponse, error) {
 	resp := new(workload.X509SVIDResponse)
 	resp.Svids = []*workload.X509SVID{}
+	resp.FederatedBundles = make(map[string][]byte)
 
-	bundle := []byte{}
-	for _, c := range update.Bundle {
-		bundle = append(bundle, c.Raw...)
+	bundle := marshalBundle(update.Bundle)
+
+	for id, federatedBundle := range update.FederatedBundles {
+		resp.FederatedBundles[id] = marshalBundle(federatedBundle)
 	}
 
 	for _, e := range update.Entries {
@@ -115,10 +117,11 @@ func (h *Handler) composeResponse(update *cache.WorkloadUpdate) (*workload.X509S
 		}
 
 		svid := &workload.X509SVID{
-			SpiffeId:    id,
-			X509Svid:    e.SVID.Raw,
-			X509SvidKey: keyData,
-			Bundle:      bundle,
+			SpiffeId:      id,
+			X509Svid:      e.SVID.Raw,
+			X509SvidKey:   keyData,
+			Bundle:        bundle,
+			FederatesWith: e.RegistrationEntry.FederatesWith,
 		}
 
 		resp.Svids = append(resp.Svids, svid)
@@ -148,3 +151,11 @@ func (h *Handler) callerPID(ctx context.Context) (pid int32, err error) {
 
 	return info.PID, nil
 }
+
+func marshalBundle(certs []*x509.Certificate) []byte {
+	bundle := []byte{}
+	for _, c := range certs {
+		bundle = append(bundle, c.Raw...)
+	}
+	return bundle
+}
diff --git a/pkg/agent/endpoints/workload/handler_test.go b/pkg/agent/endpoints/workload/handler_test.go
@@ -146,13 +146,17 @@ func (s *HandlerTestSuite) TestComposeResponse() {
 	s.Require().NoError(err)
 
 	svidMsg := &workload.X509SVID{
-		SpiffeId:    "spiffe://example.org/foo",
-		X509Svid:    update.Entries[0].SVID.Raw,
-		X509SvidKey: keyData,
-		Bundle:      update.Bundle[0].Raw,
+		SpiffeId:      "spiffe://example.org/foo",
+		X509Svid:      update.Entries[0].SVID.Raw,
+		X509SvidKey:   keyData,
+		Bundle:        update.Bundle[0].Raw,
+		FederatesWith: []string{"spiffe://otherdomain.test"},
 	}
 	apiMsg := &workload.X509SVIDResponse{
 		Svids: []*workload.X509SVID{svidMsg},
+		FederatedBundles: map[string][]byte{
+			"spiffe://otherdomain.test": update.Bundle[0].Raw,
+		},
 	}
 
 	resp, err := s.h.composeResponse(s.workloadUpdate())
@@ -200,12 +204,16 @@ func (s *HandlerTestSuite) workloadUpdate() *cache.WorkloadUpdate {
 		SVID:       svid,
 		PrivateKey: key,
 		RegistrationEntry: &common.RegistrationEntry{
-			SpiffeId: "spiffe://example.org/foo",
+			SpiffeId:      "spiffe://example.org/foo",
+			FederatesWith: []string{"spiffe://otherdomain.test"},
 		},
 	}
 	update := &cache.WorkloadUpdate{
 		Entries: []*cache.Entry{&entry},
 		Bundle:  []*x509.Certificate{ca},
+		FederatedBundles: map[string][]*x509.Certificate{
+			"spiffe://otherdomain.test": {ca},
+		},
 	}
 
 	return update