Validate message topics/tag before sending

internal/assets/token_pool.go

@@ -29,7 +29,7 @@ func (am *assetManager) CreateTokenPool(ctx context.Context, ns string, pool *ff
 	if err := am.data.VerifyNamespaceExists(ctx, ns); err != nil {
 		return nil, err
 	}
-	if err := fftypes.ValidateFFNameField(ctx, pool.Name, "name"); err != nil {
+	if err := fftypes.ValidateFFNameFieldNoUUID(ctx, pool.Name, "name"); err != nil {
 		return nil, err
 	}
 	pool.ID = fftypes.NewUUID()
@@ -154,7 +154,7 @@ func (am *assetManager) GetTokenPool(ctx context.Context, ns, connector, poolNam
 	if err := fftypes.ValidateFFNameField(ctx, ns, "namespace"); err != nil {
 		return nil, err
 	}
-	if err := fftypes.ValidateFFNameField(ctx, poolName, "name"); err != nil {
+	if err := fftypes.ValidateFFNameFieldNoUUID(ctx, poolName, "name"); err != nil {
 		return nil, err
 	}
 	pool, err := am.database.GetTokenPool(ctx, ns, poolName)

internal/events/batch_pin_complete_test.go

@@ -375,7 +375,7 @@ func TestPersistBatchSwallowBadData(t *testing.T) {
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, false).Return(nil)
 
 	valid, err := em.persistBatch(context.Background(), batch)
-	assert.True(t, valid)
+	assert.False(t, valid)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
@@ -519,7 +519,7 @@ func TestPersistBatchGoodMessageAuthorMismatch(t *testing.T) {
 	mdi.On("UpsertBatch", mock.Anything, mock.Anything, false).Return(nil)
 
 	valid, err := em.persistBatch(context.Background(), batch)
-	assert.True(t, valid)
+	assert.False(t, valid)
 	assert.NoError(t, err)
 }
 
@@ -615,7 +615,8 @@ func TestPersistBatchMessageNilData(t *testing.T) {
 			ID: fftypes.NewUUID(),
 		},
 	}
-	err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	valid, err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	assert.False(t, valid)
 	assert.NoError(t, err)
 }
 
@@ -637,7 +638,8 @@ func TestPersistBatchMessageUpsertHashMismatch(t *testing.T) {
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationSkip).Return(database.HashMismatch)
 
-	err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	valid, err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	assert.False(t, valid)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }
@@ -660,7 +662,8 @@ func TestPersistBatchMessageUpsertMessageFail(t *testing.T) {
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationSkip).Return(fmt.Errorf("pop"))
 
-	err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	valid, err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	assert.False(t, valid)
 	assert.EqualError(t, err, "pop")
 }
 
@@ -682,7 +685,8 @@ func TestPersistBatchMessageOK(t *testing.T) {
 	mdi := em.database.(*databasemocks.Plugin)
 	mdi.On("UpsertMessage", mock.Anything, mock.Anything, database.UpsertOptimizationSkip).Return(nil)
 
-	err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	valid, err := em.persistBatchMessage(context.Background(), batch, 0, msg, database.UpsertOptimizationSkip)
+	assert.True(t, valid)
 	assert.NoError(t, err)
 	mdi.AssertExpectations(t)
 }

internal/events/persist_batch.go

@@ -132,8 +132,8 @@ func (em *eventManager) persistBatch(ctx context.Context /* db TX context*/, bat
 
 	// Insert the message entries
 	for i, msg := range batch.Payload.Messages {
-		if err = em.persistBatchMessage(ctx, batch, i, msg, optimization); err != nil {
-			return false, err
+		if valid, err = em.persistBatchMessage(ctx, batch, i, msg, optimization); !valid || err != nil {
+			return valid, err
 		}
 	}
 
@@ -191,14 +191,13 @@ func (em *eventManager) persistReceivedData(ctx context.Context /* db TX context
 	return true, nil
 }
 
-func (em *eventManager) persistBatchMessage(ctx context.Context /* db TX context*/, batch *fftypes.Batch, i int, msg *fftypes.Message, optimization database.UpsertOptimization) error {
+func (em *eventManager) persistBatchMessage(ctx context.Context /* db TX context*/, batch *fftypes.Batch, i int, msg *fftypes.Message, optimization database.UpsertOptimization) (bool, error) {
 	if msg != nil && (msg.Header.Author != batch.Author || msg.Header.Key != batch.Key) {
 		log.L(ctx).Errorf("Mismatched key/author '%s'/'%s' on message entry %d in batch '%s'", msg.Header.Key, msg.Header.Author, i, batch.ID)
-		return nil // skip entry
+		return false, nil // skip entry
 	}
 
-	_, err := em.persistReceivedMessage(ctx, i, msg, "batch", batch.ID, optimization)
-	return err
+	return em.persistReceivedMessage(ctx, i, msg, "batch", batch.ID, optimization)
 }
 
 func (em *eventManager) persistReceivedMessage(ctx context.Context /* db TX context*/, i int, msg *fftypes.Message, mType string, mID *fftypes.UUID, optimization database.UpsertOptimization) (bool, error) {

internal/orchestrator/data_query.go

@@ -128,7 +128,7 @@ func (or *orchestrator) GetDatatypeByName(ctx context.Context, ns, name, version
 	if err := fftypes.ValidateFFNameField(ctx, ns, "namespace"); err != nil {
 		return nil, err
 	}
-	if err := fftypes.ValidateFFNameField(ctx, name, "name"); err != nil {
+	if err := fftypes.ValidateFFNameFieldNoUUID(ctx, name, "name"); err != nil {
 		return nil, err
 	}
 	return or.database.GetDatatypeByName(ctx, ns, name, version)

internal/orchestrator/subscriptions.go

@@ -41,7 +41,7 @@ func (or *orchestrator) createUpdateSubscription(ctx context.Context, ns string,
 	if err := or.data.VerifyNamespaceExists(ctx, subDef.Namespace); err != nil {
 		return nil, err
 	}
-	if err := fftypes.ValidateFFNameField(ctx, subDef.Name, "name"); err != nil {
+	if err := fftypes.ValidateFFNameFieldNoUUID(ctx, subDef.Name, "name"); err != nil {
 		return nil, err
 	}
 	if subDef.Transport == system.SystemEventsTransport {

pkg/fftypes/datatype.go

@@ -53,7 +53,7 @@ func (dt *Datatype) Validate(ctx context.Context, existing bool) (err error) {
 	if err = ValidateFFNameField(ctx, dt.Namespace, "namespace"); err != nil {
 		return err
 	}
-	if err = ValidateFFNameField(ctx, dt.Name, "name"); err != nil {
+	if err = ValidateFFNameFieldNoUUID(ctx, dt.Name, "name"); err != nil {
 		return err
 	}
 	if err = ValidateFFNameField(ctx, dt.Version, "version"); err != nil {

pkg/fftypes/group.go

@@ -68,7 +68,7 @@ func (group *Group) Validate(ctx context.Context, existing bool) (err error) {
 	}
 	// We allow a blank name for a group (for auto creation)
 	if group.Name != "" {
-		if err = ValidateFFNameField(ctx, group.Name, "name"); err != nil {
+		if err = ValidateFFNameFieldNoUUID(ctx, group.Name, "name"); err != nil {
 			return err
 		}
 	}

pkg/fftypes/message.go

@@ -154,6 +154,14 @@ func (m *Message) Seal(ctx context.Context) (err error) {
 	if len(m.Header.Topics) == 0 {
 		m.Header.Topics = []string{DefaultTopic}
 	}
+	if err := m.Header.Topics.Validate(ctx, "header.topics"); err != nil {
+		return err
+	}
+	if m.Header.Tag != "" {
+		if err := ValidateFFNameField(ctx, m.Header.Tag, "header.tag"); err != nil {
+			return err
+		}
+	}
 	if m.Header.ID == nil {
 		m.Header.ID = NewUUID()
 	}

pkg/fftypes/message_test.go

@@ -34,6 +34,26 @@ func TestSealBareMessage(t *testing.T) {
 	assert.NotNil(t, msg.Hash)
 }
 
+func TestSealEmptyTopicString(t *testing.T) {
+	msg := Message{
+		Header: MessageHeader{
+			Topics: []string{""},
+		},
+	}
+	err := msg.Seal(context.Background())
+	assert.Regexp(t, `FF10131.*header.topics\[0\]`, err)
+}
+
+func TestSealBadTagString(t *testing.T) {
+	msg := Message{
+		Header: MessageHeader{
+			Tag: "!wrong",
+		},
+	}
+	err := msg.Seal(context.Background())
+	assert.Regexp(t, `FF10131.*header.tag`, err)
+}
+
 func TestVerifyEmptyTopicString(t *testing.T) {
 	msg := Message{
 		Header: MessageHeader{

pkg/fftypes/namearray.go

@@ -25,10 +25,13 @@ import (
 	"github.com/hyperledger/firefly/internal/i18n"
 )
 
-// FFNameArray is an array of strings, each conforming to the requirements
-// of a FireFly name, with a combined length (when joined with commas) of 1024
+// FFNameArray is an array of strings, each conforming to the requirements of a FireFly name
 type FFNameArray []string
 
+// Because each FFName has a max length of 64, 15 names (plus comma delimeters) is a safe max
+// to pack into a string column of length 1024
+const FFNameArrayMax = 15
+
 func (na FFNameArray) Value() (driver.Value, error) {
 	if na == nil {
 		return "", nil
@@ -80,8 +83,8 @@ func (na FFNameArray) Validate(ctx context.Context, fieldName string) error {
 			return err
 		}
 	}
-	if len(na) > 15 {
-		return i18n.NewError(ctx, i18n.MsgTooManyItems, fieldName, 15, len(na))
+	if len(na) > FFNameArrayMax {
+		return i18n.NewError(ctx, i18n.MsgTooManyItems, fieldName, FFNameArrayMax, len(na))
 	}
 	return nil
 }

pkg/fftypes/node.go

@@ -40,7 +40,7 @@ type DXInfo struct {
 }
 
 func (n *Node) Validate(ctx context.Context, existing bool) (err error) {
-	if err = ValidateFFNameField(ctx, n.Name, "name"); err != nil {
+	if err = ValidateFFNameFieldNoUUID(ctx, n.Name, "name"); err != nil {
 		return err
 	}
 	if err = ValidateLength(ctx, n.Description, "description", 4096); err != nil {
@@ -58,7 +58,7 @@ func (n *Node) Validate(ctx context.Context, existing bool) (err error) {
 }
 
 func (n *Node) Topic() string {
-	return orgTopic(n.Owner)
+	return OrgTopic
 }
 
 func (n *Node) SetBroadcastMessage(msgID *UUID) {

pkg/fftypes/node_test.go

@@ -48,8 +48,7 @@ func TestNodeValidation(t *testing.T) {
 	assert.Regexp(t, "FF10203", n.Validate(context.Background(), true))
 
 	var def Definition = n
-	n.Owner = "owner"
-	assert.Equal(t, "ff_org_owner", def.Topic())
+	assert.Equal(t, "ff_organizations", def.Topic())
 	def.SetBroadcastMessage(NewUUID())
 	assert.NotNil(t, n.Message)
 }

pkg/fftypes/organization.go

@@ -19,14 +19,13 @@ package fftypes
 import (
 	"context"
 	"fmt"
-	"strings"
-	"unicode"
 
 	"github.com/hyperledger/firefly/internal/i18n"
 )
 
 const (
 	FireflyOrgDIDPrefix = "did:firefly:org/"
+	OrgTopic            = "ff_organizations"
 )
 
 // Organization is a top-level identity in the network
@@ -42,7 +41,7 @@ type Organization struct {
 }
 
 func (org *Organization) Validate(ctx context.Context, existing bool) (err error) {
-	if err = ValidateFFNameField(ctx, org.Name, "name"); err != nil {
+	if err = ValidateFFNameFieldNoUUID(ctx, org.Name, "name"); err != nil {
 		return err
 	}
 	if err = ValidateLength(ctx, org.Description, "description", 4096); err != nil {
@@ -56,23 +55,8 @@ func (org *Organization) Validate(ctx context.Context, existing bool) (err error
 	return nil
 }
 
-func orgTopic(orgIdentity string) string {
-	buf := strings.Builder{}
-	for _, r := range orgIdentity {
-		if buf.Len() > 64 {
-			break
-		}
-		if unicode.IsLetter(r) || unicode.IsNumber(r) {
-			buf.WriteRune(r)
-		} else {
-			buf.WriteRune('_')
-		}
-	}
-	return fmt.Sprintf("ff_org_%s", buf.String())
-}
-
 func (org *Organization) Topic() string {
-	return orgTopic(org.Identity)
+	return OrgTopic
 }
 
 func (org *Organization) SetBroadcastMessage(msgID *UUID) {

pkg/fftypes/organization_test.go

@@ -47,8 +47,7 @@ func TestOrganizationValidation(t *testing.T) {
 	assert.Regexp(t, "FF10203", org.Validate(context.Background(), true))
 
 	var def Definition = org
-	org.Identity = `A B C D E F G H I J K L M N O P Q R S T U V W X Y Z $ ( ) + ! 0 1 2 3 4 5 6 7 8 9`
-	assert.Equal(t, "ff_org_A_B_C_D_E_F_G_H_I_J_K_L_M_N_O_P_Q_R_S_T_U_V_W_X_Y_Z___________0_1", def.Topic())
+	assert.Equal(t, "ff_organizations", def.Topic())
 	def.SetBroadcastMessage(NewUUID())
 	assert.NotNil(t, org.Message)
 }

pkg/fftypes/tokenpool.go

@@ -66,7 +66,7 @@ func (t *TokenPool) Validate(ctx context.Context) (err error) {
 	if err = ValidateFFNameField(ctx, t.Namespace, "namespace"); err != nil {
 		return err
 	}
-	if err = ValidateFFNameField(ctx, t.Name, "name"); err != nil {
+	if err = ValidateFFNameFieldNoUUID(ctx, t.Name, "name"); err != nil {
 		return err
 	}
 	return nil

pkg/fftypes/validations.go

@@ -31,11 +31,15 @@ func ValidateFFNameField(ctx context.Context, str string, fieldName string) erro
 	if !ffNameValidator.MatchString(str) {
 		return i18n.NewError(ctx, i18n.MsgInvalidName, fieldName)
 	}
+	return nil
+}
+
+func ValidateFFNameFieldNoUUID(ctx context.Context, str string, fieldName string) error {
 	if _, err := ParseUUID(ctx, str); err == nil {
 		// Name must not be a UUID
 		return i18n.NewError(ctx, i18n.MsgNoUUID, fieldName)
 	}
-	return nil
+	return ValidateFFNameField(ctx, str, fieldName)
 }
 
 func ValidateLength(ctx context.Context, str string, fieldName string, max int) error {

pkg/fftypes/validations_test.go

@@ -38,6 +38,9 @@ func TestValidateFFNameField(t *testing.T) {
 	assert.Regexp(t, "FF10131.*badField", err)
 
 	err = ValidateFFNameField(context.Background(), "af34658e-a728-4b21-b9cf-8451f07be065", "badField")
+	assert.NoError(t, err)
+
+	err = ValidateFFNameFieldNoUUID(context.Background(), "af34658e-a728-4b21-b9cf-8451f07be065", "badField")
 	assert.Regexp(t, "FF10288.*badField", err)
 
 }