PR feedback

Signed-off-by: hfuss <[email protected]>

Author: onelapahead

go.work.sum

@@ -306,10 +306,7 @@ func (h *FFDX) beforeConnect(ctx context.Context, w wsclient.WSClient) error {
 	}
 
 	for _, cb := range h.callbacks.handlers {
-		err := cb.DXConnect(h)
-		if err != nil {
-			log.L(ctx).Errorf("error handling DX connect event: %v", err)
-		}
+		cb.DXConnect(h)
 	}
 
 	h.initialized = true
@@ -479,51 +476,56 @@ func (h *FFDX) CheckNodeIdentityStatus(ctx context.Context, node *core.Identity)
 		if h.metrics != nil && h.metrics.IsMetricsEnabled() {
 			h.metrics.NodeIdentityDXCertMismatch(node.Namespace, mismatchState)
 		}
+		log.L(ctx).Debugf("Identity status checked against DX node='%s' mismatch_state='%s'", node.Name, mismatchState)
 	}()
 
 	dxPeer, err := h.GetEndpointInfo(ctx, node.Name) // should be the same as the local node
 	if err != nil {
 		return err
 	}
 
+	dxPeerCert := dxPeer.GetString("cert")
 	// if this occurs, it is either a misconfigured / broken DX or likely a DX that is compatible from an API perspective
 	// but does not have the same peer info as the HTTPS mTLS DX
-	if dxPeer.GetString("cert") == "" {
-		log.L(ctx).Warnf("DX peer does not have a 'cert', DX plugin may be unsupported")
+	if dxPeerCert == "" {
+		log.L(ctx).Debugf("DX peer does not have a 'cert', DX plugin may be unsupported")
 		return nil
 	}
 
-	if h.metrics != nil && h.metrics.IsMetricsEnabled() {
-		expiry, err := extractSoonestExpiryFromCertBundle(strings.ReplaceAll(dxPeer.GetString("cert"), `\n`, "\n"))
-		if err == nil {
-			if expiry.Before(time.Now()) {
-				log.L(ctx).Warnf("Certificate for node '%s' has expired", node.Name)
-			}
+	expiry, err := extractSoonestExpiryFromCertBundle(strings.ReplaceAll(dxPeerCert, `\n`, "\n"))
+	if err == nil {
+		if expiry.Before(time.Now()) {
+			log.L(ctx).Warnf("DX certificate for node '%s' has expired", node.Name)
+		}
 
+		if h.metrics != nil && h.metrics.IsMetricsEnabled() {
 			h.metrics.NodeIdentityDXCertExpiry(node.Namespace, expiry)
-		} else {
-			log.L(ctx).Errorf("Failed to find x509 cert within DX cert bundle node='%s' namespace='%s'", node.Name, node.Namespace)
 		}
+	} else {
+		log.L(ctx).Errorf("Failed to find x509 cert within DX cert bundle node='%s' namespace='%s'", node.Name, node.Namespace)
 	}
 
 	if node.Profile == nil {
 		return i18n.NewError(ctx, coremsgs.MsgNodeNotProvidedForCheck)
 	}
 
-	if node.Profile.GetString("cert") != "" {
+	nodeCert := node.Profile.GetString("cert")
+	if nodeCert != "" {
 		mismatchState = metrics.NodeIdentityDXCertMismatchStatusHealthy
-		if dxPeer.GetString("cert") != node.Profile.GetString("cert") {
+		if dxPeerCert != nodeCert {
+			log.L(ctx).Warnf("DX certificate for node '%s' is out-of-sync with on-chain identity", node.Name)
 			mismatchState = metrics.NodeIdentityDXCertMismatchStatusMismatched
 		}
 	}
 
 	return nil
 }
 
-// we assume the cert with the soonest expiry is the leaf cert, but even if its the CA,
-// thats what will invalidate the leaf anyways, so really we only care about the soonest expiry
+// We assume the cert with the soonest expiry is the leaf cert, but even if its the CA,
+// that's what will invalidate the leaf anyways, so really we only care about the soonest expiry.
+// So we loop through the bundle finding the soonest expiry, not necessarily the leaf.
 func extractSoonestExpiryFromCertBundle(certBundle string) (time.Time, error) {
-	var leafCert *x509.Certificate
+	var expiringCert *x509.Certificate
 	var block *pem.Block
 	var rest = []byte(certBundle)
 
@@ -537,16 +539,16 @@ func extractSoonestExpiryFromCertBundle(certBundle string) (time.Time, error) {
 		if err != nil {
 			return time.Time{}, fmt.Errorf("failed to parse non-certificate within bundle: %v", err)
 		}
-		if leafCert == nil || cert.NotAfter.Before(leafCert.NotAfter) {
-			leafCert = cert
+		if expiringCert == nil || cert.NotAfter.Before(expiringCert.NotAfter) {
+			expiringCert = cert
 		}
 	}
 
-	if leafCert == nil {
+	if expiringCert == nil {
 		return time.Time{}, errors.New("no valid certificate found")
 	}
 
-	return leafCert.NotAfter.UTC(), nil
+	return expiringCert.NotAfter.UTC(), nil
 }
 
 func (h *FFDX) ackLoop() {

internal/dataexchange/ffdx/ffdx.go

@@ -52,8 +52,8 @@ import (
 var utConfig = config.RootSection("ffdx_unit_tests")
 
 const (
-	// NOTE: the CA cert expires on Monday, February 28, 2035 7:30:57 PM,
-	//       and the leaf cert expires on Monday, February 28, 2028 7:30:57 PM
+	// NOTE: the CA cert expires on Monday, February 28, 2035 7:30:57 PM UTC,
+	//       and the leaf cert expires on Monday, February 28, 2028 7:30:57 PM UTC
 	testCertBundle = `
 -----BEGIN CERTIFICATE-----
 MIIDqTCCApGgAwIBAgIUbZT+Ds4f2oDmGpgVi+SaQq9gxvcwDQYJKoZIhvcNAQEL
@@ -100,6 +100,9 @@ R0gnXX/IZFnLhLh6UpLOBB0KIGENh75EEU7755jMKDKFj16D0uA1Lzrh5YxicTMy
 ydFYQLpLycsWl2oV3JB4pO5TIzjY9awkRE0MeMMc
 -----END CERTIFICATE-----
 `
+	// The CA is at the bottom of the bundle and expires same as above,
+	// Monday, February 28, 2035 7:30:57 PM. The leaf cert at the top of
+	// the bundle is already expired.
 	testExpiredCert = `
 -----BEGIN CERTIFICATE-----
 MIIDqjCCApKgAwIBAgIUWnobAQ4vq8gWBAXZBf7XZG3oSicwDQYJKoZIhvcNAQEL
@@ -123,6 +126,28 @@ p/TUQ6q+Mg3W9pSXqLm8jmWNBfDViQF1v9Z3ASFYHUF/yak8jMdBEUpAqDadd/ay
 BHm9m8IvFevQjpUw6kyyg77ehEBBn7H/ISTL3HTCpUbkR3qUnFjOyBJ0G02XoozB
 I/hI0mpd6y+/JwyvG0smbD2lioiO/JQaUEZGU8pU
 -----END CERTIFICATE-----
+-----BEGIN CERTIFICATE-----
+MIIDqTCCApGgAwIBAgIUbZT+Ds4f2oDmGpgVi+SaQq9gxvcwDQYJKoZIhvcNAQEL
+BQAwZDELMAkGA1UEBhMCVVMxEzARBgNVBAgMCkNhbGlmb3JuaWExFjAUBgNVBAcM
+DVNhbiBGcmFuY2lzY28xEzARBgNVBAoMCkV4YW1wbGUgQ0ExEzARBgNVBAMMCmV4
+YW1wbGUtY2EwHhcNMjUwMjI4MTkzMDM4WhcNMzUwMjI2MTkzMDM4WjBkMQswCQYD
+VQQGEwJVUzETMBEGA1UECAwKQ2FsaWZvcm5pYTEWMBQGA1UEBwwNU2FuIEZyYW5j
+aXNjbzETMBEGA1UECgwKRXhhbXBsZSBDQTETMBEGA1UEAwwKZXhhbXBsZS1jYTCC
+ASIwDQYJKoZIhvcNAQEBBQADggEPADCCAQoCggEBAOWBryFqk0YqQ6pzGJvBDjbV
+4BnkMzsv+Fq869Xks09OP4eW44oqfFUmpCFyS3fEmRCz+389t4mKvxcCRIJMW0f5
+K9jffG1QKUKL4UuNfEPFpM0MXTwhI+dCdvofdelzc+KBGA6CDYlnWYcCKFSuWeSu
+xrb/qCEvhcCaSYt3e2WcRHRuK+OLzM3REeJctC4G/pq858OUV5CZU2B6aGV/9uFL
+ZW3TCrOaj+Khzzt5FNvjVdLiUw0FS8VESxFA4kH8p+XUshs9S0e7LfIBSID2NU8+
++5D6HliqNqikbsny1Ps6GhLa+nI37LOVj7nFcG7uk+gb6HUN1+0YvjOJ0/zvnLEC
+AwEAAaNTMFEwHQYDVR0OBBYEFJfNoXmIn5S6W7Lcj5G/huW5q1YQMB8GA1UdIwQY
+MBaAFJfNoXmIn5S6W7Lcj5G/huW5q1YQMA8GA1UdEwEB/wQFMAMBAf8wDQYJKoZI
+hvcNAQELBQADggEBALsdRYJHQMkhjLcrO4Yha1KXh2d+irmi8AqQqQgbLIsSzuqG
+bKFiYnJ8PKHaISHlev2xRM9kEjDZ/9q8T4aUELg4eBjj7VK+gs+gSBO6peJ+AcEg
+TepsE5GHmhoIIiE/3dIP6XnaM6NBb8q0ewsIg1c5vLlrt8W96LY6Og7f+742VvoV
+H31srpGjy7c5nYjBTn/Bu84eb5Lxfvy10sJjnenkXDJvzkUcnfbRzDQ9k5ZuPa05
+x+BsxonN0iaeZH91F+Y3kgJidLnU5EhIB/1KXYjuEbl9qUxD6GFHRststPRPeOmj
+7C+BtJCIjjavysSqVMvQWLQ6rXms3SpRPAimWqM=
+-----END CERTIFICATE-----
 `
 )
 
@@ -999,16 +1024,15 @@ type mockDXCallbacks struct {
 	connectCalls int
 }
 
-func (m *mockDXCallbacks) DXConnect(plugin dataexchange.Plugin) error {
+func (m *mockDXCallbacks) DXConnect(plugin dataexchange.Plugin) {
 	m.connectCalls++
-	return errors.New("connect callback failed")
 }
 
 func (m *mockDXCallbacks) DXEvent(plugin dataexchange.Plugin, event dataexchange.DXEvent) error {
 	panic("implement me")
 }
 
-func TestWebsocketDXConnectFails(t *testing.T) {
+func TestWebsocketDXConnect(t *testing.T) {
 	mockedClient := &http.Client{}
 	httpmock.ActivateNonDefault(mockedClient)
 	defer httpmock.DeactivateAndReset()

internal/dataexchange/ffdx/ffdx_test.go

@@ -247,7 +247,7 @@ func (dh *definitionHandler) handleIdentityClaim(ctx context.Context, state *cor
 		return dh.database.InsertEvent(ctx, event)
 	})
 
-	if identity.Type == core.IdentityTypeNode {
+	if dh.multiparty && identity.Type == core.IdentityTypeNode {
 		nodeDID, err := dh.identity.GetLocalNodeDID(ctx)
 		if err != nil {
 			return HandlerResult{Action: core.ActionRetry}, err

internal/definitions/handler_identity_claim.go

@@ -631,7 +631,6 @@ func TestHandleDefinitionIdentityClaimLocalNodeWithFailingStatusCheck(t *testing
 	dh.mdx.On("CheckNodeIdentityStatus", ctx, node1).Return(errors.New("failed to check status but no worries"))
 
 	dh.multiparty = true
-
 	action, err := dh.HandleDefinitionBroadcast(ctx, &bs.BatchState, claimMsg, core.DataArray{claimData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: core.ActionConfirm}, action)
 	assert.NoError(t, err)
@@ -669,7 +668,6 @@ func TestHandleDefinitionIdentityClaimLocaNodeMisconfigured(t *testing.T) {
 	dh.mim.On("GetLocalNodeDID", ctx).Return(node1.DID, errors.New("somehow local node isnt configured but we  got this far"))
 
 	dh.multiparty = true
-
 	action, err := dh.HandleDefinitionBroadcast(ctx, &bs.BatchState, claimMsg, core.DataArray{claimData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: core.ActionRetry}, action)
 	assert.Error(t, err)

internal/definitions/handler_identity_claim_test.go

@@ -88,7 +88,7 @@ func (dh *definitionHandler) handleIdentityUpdate(ctx context.Context, state *co
 		return dh.database.InsertEvent(ctx, event)
 	})
 
-	if identity.Type == core.IdentityTypeNode {
+	if dh.multiparty && identity.Type == core.IdentityTypeNode {
 		nodeDID, err := dh.identity.GetLocalNodeDID(ctx)
 		if err != nil {
 			return HandlerResult{Action: core.ActionRetry}, err

internal/definitions/handler_identity_update.go

@@ -293,8 +293,9 @@ func TestHandleDefinitionIdentityUpdateLocalNodeOk(t *testing.T) {
 	dh, bs := newTestDefinitionHandler(t)
 	ctx := context.Background()
 
-	_, node1, updateMsg, updateData, iu := testNodeIdentityUpdate(t)
+	org1, node1, updateMsg, updateData, iu := testNodeIdentityUpdate(t)
 
+	dh.mim.On("VerifyIdentityChain", ctx, node1).Return(org1, false, nil)
 	dh.mim.On("CachedIdentityLookupByID", ctx, node1.ID).Return(node1, nil)
 	dh.mdi.On("UpsertIdentity", ctx, mock.MatchedBy(func(identity *core.Identity) bool {
 		assert.Equal(t, *updateMsg.Header.ID, *identity.Messages.Update)
@@ -308,6 +309,7 @@ func TestHandleDefinitionIdentityUpdateLocalNodeOk(t *testing.T) {
 	dh.mim.On("GetLocalNodeDID", ctx).Return(node1.DID, nil)
 	dh.mdx.On("CheckNodeIdentityStatus", ctx, node1).Return(errors.New("failed to check status but no worries"))
 
+	dh.multiparty = true
 	action, err := dh.HandleDefinitionBroadcast(ctx, &bs.BatchState, updateMsg, core.DataArray{updateData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: core.ActionConfirm}, action)
 	assert.NoError(t, err)
@@ -320,8 +322,9 @@ func TestHandleDefinitionIdentityUpdateLocalNodeMisconfigured(t *testing.T) {
 	dh, bs := newTestDefinitionHandler(t)
 	ctx := context.Background()
 
-	_, node1, updateMsg, updateData, iu := testNodeIdentityUpdate(t)
+	org1, node1, updateMsg, updateData, iu := testNodeIdentityUpdate(t)
 
+	dh.mim.On("VerifyIdentityChain", ctx, node1).Return(org1, false, nil)
 	dh.mim.On("CachedIdentityLookupByID", ctx, node1.ID).Return(node1, nil)
 	dh.mdi.On("UpsertIdentity", ctx, mock.MatchedBy(func(identity *core.Identity) bool {
 		assert.Equal(t, *updateMsg.Header.ID, *identity.Messages.Update)
@@ -334,6 +337,7 @@ func TestHandleDefinitionIdentityUpdateLocalNodeMisconfigured(t *testing.T) {
 	})).Return(nil)
 	dh.mim.On("GetLocalNodeDID", ctx).Return(node1.DID, errors.New("no local node but somehow we got this far"))
 
+	dh.multiparty = true
 	action, err := dh.HandleDefinitionBroadcast(ctx, &bs.BatchState, updateMsg, core.DataArray{updateData}, fftypes.NewUUID())
 	assert.Equal(t, HandlerResult{Action: core.ActionRetry}, action)
 	assert.Error(t, err)

internal/definitions/handler_identity_update_test.go

@@ -104,6 +104,8 @@ func ParseKeyNormalizationConfig(strConfigVal string) int {
 	}
 }
 
+// GetLocalNodeDID returns the expected local node DID based on the multiparty configuration, the node
+// identity does not need to be registered yet in order for this to succeed.
 func (im *identityManager) GetLocalNodeDID(ctx context.Context) (string, error) {
 	nodeName := im.multiparty.LocalNode().Name
 	if nodeName == "" {
@@ -113,6 +115,7 @@ func (im *identityManager) GetLocalNodeDID(ctx context.Context) (string, error)
 	return fmt.Sprintf("%s%s", core.FireFlyNodeDIDPrefix, nodeName), nil
 }
 
+// GetLocalNode returns the local node identity, if it has been registered and is in the DB/cache.
 func (im *identityManager) GetLocalNode(ctx context.Context) (node *core.Identity, err error) {
 	nodeDID, err := im.GetLocalNodeDID(ctx)
 	if err != nil {

internal/identity/identitymanager.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"sync"
 
+	"github.com/hyperledger/firefly-common/pkg/log"
+
 	"github.com/hyperledger/firefly-common/pkg/fftypes"
 	"github.com/hyperledger/firefly-common/pkg/i18n"
 	"github.com/hyperledger/firefly/internal/coremsgs"
@@ -98,9 +100,12 @@ func (bc *boundCallbacks) TokensApproved(plugin tokens.Plugin, approval *tokens.
 	return bc.o.events.TokensApproved(plugin, approval)
 }
 
-func (bc *boundCallbacks) DXConnect(plugin dataexchange.Plugin) error {
-	if err := bc.checkStopped(); err != nil {
-		return err
+func (bc *boundCallbacks) DXConnect(plugin dataexchange.Plugin) {
+	err := bc.checkStopped()
+	if err == nil {
+		err = bc.o.networkmap.CheckNodeIdentityStatus(bc.o.ctx)
+	}
+	if err != nil {
+		log.L(bc.o.ctx).Errorf("Error handling DX connect callback: %s", err)
 	}
-	return bc.o.networkmap.CheckNodeIdentityStatus(bc.o.ctx)
 }

internal/orchestrator/bound_callbacks.go

@@ -107,7 +107,7 @@ func TestBoundCallbacks(t *testing.T) {
 	assert.NoError(t, err)
 
 	mnm.On("CheckNodeIdentityStatus", mock.Anything, mock.Anything).Return(nil)
-	err = bc.DXConnect(mdx)
+	bc.DXConnect(mdx)
 	assert.NoError(t, err)
 
 	mei.On("TokenPoolCreated", mock.Anything, mti, &tokens.TokenPool{}).Return(nil)
@@ -145,8 +145,8 @@ func TestBoundCallbacksStopped(t *testing.T) {
 	err = bc.DXEvent(nil, &dataexchangemocks.DXEvent{})
 	assert.Regexp(t, "FF10446", err)
 
-	err = bc.DXConnect(nil)
-	assert.Regexp(t, "FF10446", err)
+	bc.DXConnect(nil)
+	// no-op
 
 	err = bc.TokenPoolCreated(context.Background(), nil, &tokens.TokenPool{})
 	assert.Regexp(t, "FF10446", err)