Repository: hyperledger/firefly
Pull request #1222 (2023-03-16 21:28:43)

Propagate API context to announce in sync token pool creation cases
When a token pool is created, the token connector can either confirm the token pool:
1. Synchronously if the token pool exists already on-chain - with a `200` return code
2. Asynchronously when using a factory contract - with a `202` return code, followed at some point with a websocket event

At this confirm stage, in a multi-party enabled namespace, the token pool is currently automatically published via .broadcast to all members of the network.

This automatic publishing occurs on the organization signing key, which might be different to the original key used to create the token pool on-chain (in the asynchronous case). This automated publishing is a little problematic for authentication checking based on the REST API context of the original token pool creation in the async case, because it is an asynchronous side effect of another REST API call.

In the synchronous case it's not problematic as it happens in-line with the REST API call. However, the code does not propagate the right context through the callbacks correctly to allow address resolution to occur with the right security context - meaning it might fail.

So this PR makes the small change to propagate that REST API context down into the callback in the synchronous case, such that the 2nd address resolution can occur correctly.

> This PR does **not** attempt to address the wider concern that publishing automatically as an asynchronous side-effect of creating a token pool, is not necessarily the right answer for all business networks and security models. That is a wider consideration that the community should consider outside the scope of this PR.


Patch:
diff --git a/internal/events/event_manager.go b/internal/events/event_manager.go
index c0322f8cf..7f98b9195 100644
--- a/internal/events/event_manager.go
+++ b/internal/events/event_manager.go
@@ -78,7 +78,7 @@ type EventManager interface {
 	SharedStorageBlobDownloaded(ss sharedstorage.Plugin, hash fftypes.Bytes32, size int64, payloadRef string, dataID *fftypes.UUID)
 
 	// Bound token callbacks
-	TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) error
+	TokenPoolCreated(ctx context.Context /* allows security context to be propagated when called in-line with the send TX */, ti tokens.Plugin, pool *tokens.TokenPool) error
 	TokensTransferred(ti tokens.Plugin, transfer *tokens.TokenTransfer) error
 	TokensApproved(ti tokens.Plugin, approval *tokens.TokenApproval) error
 
diff --git a/internal/events/token_pool_created.go b/internal/events/token_pool_created.go
index 7d06b835c..559598fa4 100644
--- a/internal/events/token_pool_created.go
+++ b/internal/events/token_pool_created.go
@@ -51,7 +51,7 @@ func addPoolDetailsFromPlugin(ffPool *core.TokenPool, pluginPool *tokens.TokenPo
 }
 
 func (em *eventManager) confirmPool(ctx context.Context, pool *core.TokenPool, ev *blockchain.Event) error {
-	log.L(em.ctx).Debugf("Confirming pool ID=%s Locator='%s'", pool.ID, pool.Locator)
+	log.L(ctx).Debugf("Confirming pool ID=%s Locator='%s'", pool.ID, pool.Locator)
 	var blockchainID string
 	if ev != nil {
 		// Some pools will not include a blockchain event for creation (such as when indexing a pre-existing pool)
@@ -132,12 +132,16 @@ func (em *eventManager) loadFromOperation(ctx context.Context, pool *tokens.Toke
 // It will be at least invoked on the submitter when the pool is first created, to trigger the submitter to announce it.
 // It will be invoked on every node (including the submitter) after the pool is announced+activated, to trigger confirmation of the pool.
 // When received in any other scenario, it should be ignored.
-func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) (err error) {
+//
+// The context passed to this callback is dependent on what phase it is called in.
+// In the case that it is called synchronously on the submitter, in order to trigger the announcement, the original context
+// of the REST API will be propagated (so it can be used for the resolution of the org signing key).
+func (em *eventManager) TokenPoolCreated(ctx context.Context, ti tokens.Plugin, pool *tokens.TokenPool) (err error) {
 	var msgIDforRewind *fftypes.UUID
 	var announcePool *core.TokenPool
 
-	err = em.retry.Do(em.ctx, "persist token pool transaction", func(attempt int) (bool, error) {
-		err := em.database.RunAsGroup(em.ctx, func(ctx context.Context) error {
+	err = em.retry.Do(ctx, "persist token pool transaction", func(attempt int) (bool, error) {
+		err := em.database.RunAsGroup(ctx, func(ctx context.Context) error {
 			// See if this is a confirmation of an unconfirmed pool
 			existingPool, err := em.loadExisting(ctx, pool)
 			if err != nil {
@@ -145,7 +149,7 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 			}
 			if existingPool != nil {
 				if existingPool.State == core.TokenPoolStateConfirmed {
-					log.L(em.ctx).Debugf("Token pool ID=%s Locator='%s' already confirmed", existingPool.ID, pool.PoolLocator)
+					log.L(ctx).Debugf("Token pool ID=%s Locator='%s' already confirmed", existingPool.ID, pool.PoolLocator)
 					return nil // already confirmed
 				}
 				msgIDforRewind = existingPool.Message
@@ -153,7 +157,7 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 			} else if pool.TX.ID == nil {
 				// TransactionID is required if the pool doesn't exist yet
 				// (but it may be omitted when activating a pool that was received via definition broadcast)
-				log.L(em.ctx).Errorf("Invalid token pool transaction - ID is nil")
+				log.L(ctx).Errorf("Invalid token pool transaction - ID is nil")
 				return nil // move on
 			}
 
@@ -184,16 +188,16 @@ func (em *eventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPoo
 		if announcePool != nil {
 			// If the pool is tied to a contract interface, resolve the methods to be used for later operations
 			if announcePool.Interface != nil && announcePool.Interface.ID != nil && announcePool.InterfaceFormat != "" {
-				log.L(em.ctx).Infof("Querying token connector interface, id=%s", announcePool.ID)
-				if err := em.assets.ResolvePoolMethods(em.ctx, announcePool); err != nil {
+				log.L(ctx).Infof("Querying token connector interface, id=%s", announcePool.ID)
+				if err := em.assets.ResolvePoolMethods(ctx, announcePool); err != nil {
 					return err
 				}
 			}
 
 			// Announce the details of the new token pool
 			// Other nodes will pass these details to their own token connector for validation/activation of the pool
-			log.L(em.ctx).Infof("Announcing token pool, id=%s", announcePool.ID)
-			err = em.defsender.DefineTokenPool(em.ctx, &core.TokenPoolAnnouncement{
+			log.L(ctx).Infof("Announcing token pool, id=%s", announcePool.ID)
+			err = em.defsender.DefineTokenPool(ctx, &core.TokenPoolAnnouncement{
 				Pool: announcePool,
 			}, false)
 		}
diff --git a/internal/tokens/fftokens/fftokens.go b/internal/tokens/fftokens/fftokens.go
index adc1ebefb..b60e4e017 100644
--- a/internal/tokens/fftokens/fftokens.go
+++ b/internal/tokens/fftokens/fftokens.go
@@ -70,7 +70,7 @@ func (cb *callbacks) OperationUpdate(ctx context.Context, nsOpID string, status
 func (cb *callbacks) TokenPoolCreated(ctx context.Context, pool *tokens.TokenPool) error {
 	// Deliver token pool creation events to every handler
 	for _, handler := range cb.handlers {
-		if err := handler.TokenPoolCreated(cb.plugin, pool); err != nil {
+		if err := handler.TokenPoolCreated(ctx, cb.plugin, pool); err != nil {
 			return err
 		}
 	}
diff --git a/mocks/eventmocks/event_manager.go b/mocks/eventmocks/event_manager.go
index f61da6151..b67e116cd 100644
--- a/mocks/eventmocks/event_manager.go
+++ b/mocks/eventmocks/event_manager.go
@@ -288,13 +288,13 @@ func (_m *EventManager) SubscriptionUpdates() chan<- *fftypes.UUID {
 	return r0
 }
 
-// TokenPoolCreated provides a mock function with given fields: ti, pool
-func (_m *EventManager) TokenPoolCreated(ti tokens.Plugin, pool *tokens.TokenPool) error {
-	ret := _m.Called(ti, pool)
+// TokenPoolCreated provides a mock function with given fields: ctx, ti, pool
+func (_m *EventManager) TokenPoolCreated(ctx context.Context, ti tokens.Plugin, pool *tokens.TokenPool) error {
+	ret := _m.Called(ctx, ti, pool)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(tokens.Plugin, *tokens.TokenPool) error); ok {
-		r0 = rf(ti, pool)
+	if rf, ok := ret.Get(0).(func(context.Context, tokens.Plugin, *tokens.TokenPool) error); ok {
+		r0 = rf(ctx, ti, pool)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/mocks/tokenmocks/callbacks.go b/mocks/tokenmocks/callbacks.go
index e138da9fb..3c19a8ecd 100644
--- a/mocks/tokenmocks/callbacks.go
+++ b/mocks/tokenmocks/callbacks.go
@@ -3,6 +3,8 @@
 package tokenmocks
 
 import (
+	context "context"
+
 	tokens "github.com/hyperledger/firefly/pkg/tokens"
 	mock "github.com/stretchr/testify/mock"
 )
@@ -12,13 +14,13 @@ type Callbacks struct {
 	mock.Mock
 }
 
-// TokenPoolCreated provides a mock function with given fields: plugin, pool
-func (_m *Callbacks) TokenPoolCreated(plugin tokens.Plugin, pool *tokens.TokenPool) error {
-	ret := _m.Called(plugin, pool)
+// TokenPoolCreated provides a mock function with given fields: ctx, plugin, pool
+func (_m *Callbacks) TokenPoolCreated(ctx context.Context, plugin tokens.Plugin, pool *tokens.TokenPool) error {
+	ret := _m.Called(ctx, plugin, pool)
 
 	var r0 error
-	if rf, ok := ret.Get(0).(func(tokens.Plugin, *tokens.TokenPool) error); ok {
-		r0 = rf(plugin, pool)
+	if rf, ok := ret.Get(0).(func(context.Context, tokens.Plugin, *tokens.TokenPool) error); ok {
+		r0 = rf(ctx, plugin, pool)
 	} else {
 		r0 = ret.Error(0)
 	}
diff --git a/pkg/tokens/plugin.go b/pkg/tokens/plugin.go
index 941a56fcd..8f581c8a9 100644
--- a/pkg/tokens/plugin.go
+++ b/pkg/tokens/plugin.go
@@ -81,7 +81,11 @@ type Callbacks interface {
 	// submitted by us, or by any other authorized party in the network.
 	//
 	// Error should only be returned in shutdown scenarios
-	TokenPoolCreated(plugin Plugin, pool *TokenPool) error
+	//
+	// Note: The context is passed on this callback (unlike most callbacks), as it might be
+	//       involved in-line with the original REST API call in the special case of the
+	//       submitter.
+	TokenPoolCreated(ctx context.Context, plugin Plugin, pool *TokenPool) error
 
 	// TokensTransferred notifies on a transfer between token accounts.
 	//


Test patch:
diff --git a/internal/events/token_pool_created_test.go b/internal/events/token_pool_created_test.go
index 5723d60a2..e0afa9e21 100644
--- a/internal/events/token_pool_created_test.go
+++ b/internal/events/token_pool_created_test.go
@@ -55,7 +55,7 @@ func TestTokenPoolCreatedIgnore(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -79,7 +79,7 @@ func TestTokenPoolCreatedIgnoreNoTX(t *testing.T) {
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 }
 
@@ -134,7 +134,7 @@ func TestTokenPoolCreatedConfirm(t *testing.T) {
 		return e.Type == core.EventTypePoolConfirmed && *e.Reference == *storedPool.ID
 	})).Return(nil).Once()
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 	assert.Equal(t, "ERC1155", storedPool.Standard)
@@ -176,7 +176,7 @@ func TestTokenPoolCreatedAlreadyConfirmed(t *testing.T) {
 
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(storedPool, nil)
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 }
@@ -220,7 +220,7 @@ func TestTokenPoolCreatedConfirmFailBadSymbol(t *testing.T) {
 		ID: opID,
 	}}, nil, nil)
 
-	err := em.TokenPoolCreated(mti, chainPool)
+	err := em.TokenPoolCreated(em.ctx, mti, chainPool)
 	assert.NoError(t, err)
 
 }
@@ -364,7 +364,7 @@ func TestTokenPoolCreatedAnnounce(t *testing.T) {
 		return pool.Pool.Namespace == "ns1" && pool.Pool.Name == "my-pool" && *pool.Pool.ID == *poolID
 	}), false).Return(nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 	mti.AssertExpectations(t)
@@ -415,7 +415,7 @@ func TestTokenPoolCreatedAnnounceBadInterface(t *testing.T) {
 		return pool.Locator == "123" && pool.Name == "my-pool"
 	})).Return(fmt.Errorf("pop"))
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.EqualError(t, err, "pop")
 
 	mti.AssertExpectations(t)
@@ -453,7 +453,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputID(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -492,7 +492,7 @@ func TestTokenPoolCreatedAnnounceBadOpInputNS(t *testing.T) {
 	em.mdi.On("GetTokenPoolByLocator", em.ctx, "ns1", "erc1155", "123").Return(nil, nil)
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil)
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 }
@@ -536,7 +536,7 @@ func TestTokenPoolCreatedAnnounceBadSymbol(t *testing.T) {
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(nil, nil, fmt.Errorf("pop")).Once()
 	em.mdi.On("GetOperations", em.ctx, "ns1", mock.Anything).Return(operations, nil, nil).Once()
 
-	err := em.TokenPoolCreated(mti, pool)
+	err := em.TokenPoolCreated(em.ctx, mti, pool)
 	assert.NoError(t, err)
 
 	mti.AssertExpectations(t)
diff --git a/internal/tokens/fftokens/fftokens_test.go b/internal/tokens/fftokens/fftokens_test.go
index 9736cfe55..3f1f41ac8 100644
--- a/internal/tokens/fftokens/fftokens_test.go
+++ b/internal/tokens/fftokens/fftokens_test.go
@@ -267,11 +267,13 @@ func TestCreateTokenPoolSynchronous(t *testing.T) {
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
 
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	ctx, cancelCtx := context.WithCancel(context.Background())
+	defer cancelCtx()
+	mcb.On("TokenPoolCreated", ctx /* important this gets through */, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && *p.TX.ID == *pool.TX.ID
 	})).Return(nil)
 
-	complete, err := h.CreateTokenPool(context.Background(), nsOpID, pool)
+	complete, err := h.CreateTokenPool(ctx, nsOpID, pool)
 	assert.True(t, complete)
 	assert.NoError(t, err)
 }
@@ -425,7 +427,7 @@ func TestActivateTokenPoolSynchronous(t *testing.T) {
 
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil && p.Event == nil
 	})).Return(nil)
 
@@ -473,7 +475,7 @@ func TestActivateTokenPoolSynchronousBadResponse(t *testing.T) {
 
 	mcb := &tokenmocks.Callbacks{}
 	h.SetHandler("ns1", mcb)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil
 	})).Return(nil)
 
@@ -993,7 +995,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"6"},"event":"ack"}`, string(msg))
 
 	// token-pool: invalid uuid (success)
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && p.TX.ID == nil && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
@@ -1017,7 +1019,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"7"},"event":"ack"}`, string(msg))
 
 	// token-pool: success
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && txID.Equals(p.TX.ID) && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(nil).Once()
 	fromServer <- fftypes.JSONObject{
@@ -1042,7 +1044,7 @@ func TestPoolEvents(t *testing.T) {
 	assert.Equal(t, `{"data":{"id":"8"},"event":"ack"}`, string(msg))
 
 	// token-pool: batch + callback fail
-	mcb.On("TokenPoolCreated", h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
+	mcb.On("TokenPoolCreated", mock.Anything, h, mock.MatchedBy(func(p *tokens.TokenPool) bool {
 		return p.PoolLocator == "F1" && p.Type == core.TokenTypeFungible && txID.Equals(p.TX.ID) && p.Event.ProtocolID == "000000000010/000020/000030"
 	})).Return(fmt.Errorf("pop")).Once()
 	fromServer <- fftypes.JSONObject{