ws: allow filtering notification by parameters

`Any` type with nil/null value is treated as a parameter filter that allows
any notification value. Not more than 16 filter parameters are allowed.
Closes #3624.

Signed-off-by: Pavel Karpy <carpawell@nspcc.ru>

docs/notifications.md

@@ -16,7 +16,8 @@ Currently supported events:
    Contents: transaction. Filters: sender and signer.
  * notification generated during execution
 
-   Contents: container hash, contract hash, notification name, stack item. Filters: contract hash, notification name.
+   Contents: container hash, contract hash, notification name, stack item.
+   Filters: contract hash, notification name, notification parameters.
  * transaction/persisting script executed
 
    Contents: application execution result. Filters: VM state, script container hash.
@@ -84,9 +85,13 @@ Recognized stream names:
    format for one of transaction's `Signers`.
  * `notification_from_execution`
    Filter: `contract` field containing a string with hex-encoded Uint160 (LE
-   representation) and/or `name` field containing a string with execution 
+   representation), `name` field containing a string with execution 
    notification name which should be a valid UTF-8 string not longer than 
-   32 bytes.
+   32 bytes and/or `parameters` field containing an ordered array of structs
+   with `type` and `value` fields. Parameter's `type` must be a simple builtin
+   type only: `Any`, `Boolean`, `Integer`, `ByteArray`, `String`, `Hash160`,
+   `Hash256`, `PublicKey` or `Signature`. No-op filter must be omitted or
+   explicitly be `Any` typed with zero value.
  * `transaction_executed`
    Filter: `state` field containing `HALT` or `FAULT` string for successful
    and failed executions respectively and/or `container` field containing

pkg/neorpc/filters.go

@@ -3,10 +3,13 @@ package neorpc
 import (
 	"errors"
 	"fmt"
+	"slices"
 
 	"github.com/nspcc-dev/neo-go/pkg/core/interop/runtime"
 	"github.com/nspcc-dev/neo-go/pkg/core/mempoolevent"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/vm/stackitem"
 	"github.com/nspcc-dev/neo-go/pkg/vm/vmstate"
 )
 
@@ -29,11 +32,26 @@ type (
 	}
 	// NotificationFilter is a wrapper structure representing a filter used for
 	// notifications generated during transaction execution. Notifications can
-	// be filtered by contract hash and/or by name. nil value treated as missing
+	// be filtered by contract hash, by event name and/or by notification
-	// filter.
+	// parameters. Notification parameter filters will be applied in the order
+	// corresponding to a produced notification's parameters. `Any`-typed
+	// parameter with zero value allows any notification parameter. Supported
+	// parameter types:
+	// - [smartcontract.AnyType]
+	// - [smartcontract.BoolType]
+	// - [smartcontract.IntegerType]
+	// - [smartcontract.ByteArrayType]
+	// - [smartcontract.StringType]
+	// - [smartcontract.Hash160Type]
+	// - [smartcontract.Hash256Type]
+	// - [smartcontract.PublicKeyType]
+	// - [smartcontract.SignatureType]
+	// nil value treated as missing filter.
 	NotificationFilter struct {
 		Contract   *util.Uint160             `json:"contract,omitempty"`
 		Name       *string                   `json:"name,omitempty"`
+		Parameters []smartcontract.Parameter `json:"parameters,omitempty"`
+		cachedSI   []stackitem.Item
 	}
 	// ExecutionFilter is a wrapper structure used for transaction and persisting
 	// scripts execution events. It allows to choose failing or successful
@@ -112,7 +130,9 @@ func (f TxFilter) IsValid() error {
 	return nil
 }
 
-// Copy creates a deep copy of the NotificationFilter. It handles nil NotificationFilter correctly.
+// Copy creates a deep copy of the NotificationFilter.
+// If [NotificationFilter.ParametersSI] has been called before, cached values
+// are cleared. It handles nil NotificationFilter correctly.
 func (f *NotificationFilter) Copy() *NotificationFilter {
 	if f == nil {
 		return nil
@@ -126,14 +146,65 @@ func (f *NotificationFilter) Copy() *NotificationFilter {
 		res.Name = new(string)
 		*res.Name = *f.Name
 	}
+	if len(f.Parameters) != 0 {
+		res.Parameters = slices.Clone(f.Parameters)
+	}
+	f.cachedSI = nil
 	return res
 }
 
+// ParametersSI returns [stackitem.Item] version of [NotificationFilter.Parameters]
+// according to [smartcontract.Parameter.ToStackItem]; result is cached.
+// It mainly should be used by server code. Must not be used concurrently.
+func (f *NotificationFilter) ParametersSI() ([]stackitem.Item, error) {
+	if f.cachedSI != nil {
+		return f.cachedSI, nil
+	}
+	f.cachedSI = make([]stackitem.Item, 0, len(f.Parameters))
+	for i, p := range f.Parameters {
+		si, err := p.ToStackItem()
+		if err != nil {
+			f.cachedSI = nil
+			return nil, fmt.Errorf("converting %d parameter it stack item: %w", i, err)
+		}
+		f.cachedSI = append(f.cachedSI, si)
+	}
+	return f.cachedSI, nil
+}
+
+// MaxNotificationFilterParameters is a reasonable filter's parameter limit
+// that does not allow attackers to increase node resources usage but that
+// also should be enough for real applications.
+const MaxNotificationFilterParameters = 16
+
 // IsValid implements SubscriptionFilter interface.
 func (f NotificationFilter) IsValid() error {
 	if f.Name != nil && len(*f.Name) > runtime.MaxEventNameLen {
 		return fmt.Errorf("%w: NotificationFilter name parameter must be less than %d", ErrInvalidSubscriptionFilter, runtime.MaxEventNameLen)
 	}
+	if l := len(f.Parameters); l != 0 {
+		if l > MaxNotificationFilterParameters {
+			return fmt.Errorf("%w: NotificationFilter's parameters number exceeded: %d > %d", ErrInvalidSubscriptionFilter, l, MaxNotificationFilterParameters)
+		}
+		for i, parameter := range f.Parameters {
+			switch parameter.Type {
+			case smartcontract.AnyType,
+				smartcontract.BoolType,
+				smartcontract.IntegerType,
+				smartcontract.ByteArrayType,
+				smartcontract.StringType,
+				smartcontract.Hash160Type,
+				smartcontract.Hash256Type,
+				smartcontract.PublicKeyType,
+				smartcontract.SignatureType:
+			default:
+				return fmt.Errorf("%w: NotificationFilter type parameter %d is unsupported: %s", ErrInvalidSubscriptionFilter, i, parameter.Type)
+			}
+			if _, err := parameter.ToStackItem(); err != nil {
+				return fmt.Errorf("%w: NotificationFilter filter parameter does not correspond to any stack item: %w", ErrInvalidSubscriptionFilter, err)
+			}
+		}
+	}
 	return nil
 }
 

pkg/neorpc/filters_test.go

@@ -3,6 +3,7 @@ package neorpc
 import (
 	"testing"
 
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/stretchr/testify/require"
 )
@@ -88,6 +89,15 @@ func TestNotificationFilterCopy(t *testing.T) {
 	require.Equal(t, bf, tf)
 	*bf.Name = "azaza"
 	require.NotEqual(t, bf, tf)
+
+	var err error
+	bf.Parameters, err = smartcontract.NewParametersFromValues(1, "2", []byte{3})
+	require.NoError(t, err)
+
+	tf = bf.Copy()
+	require.Equal(t, bf, tf)
+	bf.Parameters[0], bf.Parameters[1] = bf.Parameters[1], bf.Parameters[0]
+	require.NotEqual(t, bf, tf)
 }
 
 func TestExecutionFilterCopy(t *testing.T) {

pkg/neorpc/rpcevent/filter.go

@@ -6,6 +6,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core/transaction"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc/result"
+	"github.com/nspcc-dev/neo-go/pkg/vm/stackitem"
 )
 
 type (
@@ -66,7 +67,28 @@ func Matches(f Comparator, r Container) bool {
 		notification := r.EventPayload().(*state.ContainedNotificationEvent)
 		hashOk := filt.Contract == nil || notification.ScriptHash.Equals(*filt.Contract)
 		nameOk := filt.Name == nil || notification.Name == *filt.Name
-		return hashOk && nameOk
+		parametersOk := true
+		if len(filt.Parameters) > 0 {
+			stackItems := notification.Item.Value().([]stackitem.Item)
+			parameters, err := filt.ParametersSI()
+			if err != nil {
+				return false
+			}
+			for i, p := range parameters {
+				if p.Type() == stackitem.AnyT && p.Value() == nil {
+					continue
+				}
+				if i >= len(stackItems) {
+					parametersOk = false
+					break
+				}
+				if !p.Equals(stackItems[i]) {
+					parametersOk = false
+					break
+				}
+			}
+		}
+		return hashOk && nameOk && parametersOk
 	case neorpc.ExecutionEventID:
 		filt := filter.(neorpc.ExecutionFilter)
 		applog := r.EventPayload().(*state.AppExecResult)

pkg/neorpc/rpcevent/filter_test.go

@@ -10,7 +10,9 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/neorpc"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc/result"
 	"github.com/nspcc-dev/neo-go/pkg/network/payload"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
+	"github.com/nspcc-dev/neo-go/pkg/vm/stackitem"
 	"github.com/nspcc-dev/neo-go/pkg/vm/vmstate"
 	"github.com/stretchr/testify/require"
 )
@@ -55,6 +57,10 @@ func TestMatches(t *testing.T) {
 	name := "ntf name"
 	badName := "bad name"
 	badType := mempoolevent.TransactionRemoved
+	parameters, err := smartcontract.NewParametersFromValues(1, "2", []byte{3})
+	require.NoError(t, err)
+	badParameters, err := smartcontract.NewParametersFromValues([]byte{3}, "2", []byte{1})
+	require.NoError(t, err)
 	bContainer := testContainer{
 		id: neorpc.BlockEventID,
 		pld: &block.Block{
@@ -76,6 +82,16 @@ func TestMatches(t *testing.T) {
 		id:  neorpc.NotificationEventID,
 		pld: &state.ContainedNotificationEvent{NotificationEvent: state.NotificationEvent{ScriptHash: contract, Name: name}},
 	}
+	ntfContainerParameters := testContainer{
+		id: neorpc.NotificationEventID,
+		pld: &state.ContainedNotificationEvent{
+			NotificationEvent: state.NotificationEvent{
+				ScriptHash: contract,
+				Name:       name,
+				Item:       stackitem.NewArray(prmsToStack(t, parameters)),
+			},
+		},
+	}
 	exContainer := testContainer{
 		id:  neorpc.ExecutionEventID,
 		pld: &state.AppExecResult{Container: cnt, Execution: state.Execution{VMState: st}},
@@ -261,6 +277,24 @@ func TestMatches(t *testing.T) {
 			container: ntfContainer,
 			expected:  true,
 		},
+		{
+			name: "notification, parameters match",
+			comparator: testComparator{
+				id:     neorpc.NotificationEventID,
+				filter: neorpc.NotificationFilter{Name: &name, Contract: &contract, Parameters: parameters},
+			},
+			container: ntfContainerParameters,
+			expected:  true,
+		},
+		{
+			name: "notification, parameters mismatch",
+			comparator: testComparator{
+				id:     neorpc.NotificationEventID,
+				filter: neorpc.NotificationFilter{Name: &name, Contract: &contract, Parameters: badParameters},
+			},
+			container: ntfContainerParameters,
+			expected:  false,
+		},
 		{
 			name:       "execution, no filter",
 			comparator: testComparator{id: neorpc.ExecutionEventID},
@@ -343,3 +377,13 @@ func TestMatches(t *testing.T) {
 		})
 	}
 }
+
+func prmsToStack(t *testing.T, pp []smartcontract.Parameter) []stackitem.Item {
+	res := make([]stackitem.Item, 0, len(pp))
+	for _, p := range pp {
+		s, err := p.ToStackItem()
+		require.NoError(t, err)
+		res = append(res, s)
+	}
+	return res
+}

pkg/rpcclient/wsclient_test.go

@@ -24,6 +24,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/neorpc/result"
 	"github.com/nspcc-dev/neo-go/pkg/network/payload"
 	"github.com/nspcc-dev/neo-go/pkg/services/rpcsrv/params"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/nspcc-dev/neo-go/pkg/vm/vmstate"
 	"github.com/stretchr/testify/assert"
@@ -591,6 +592,7 @@ func TestWSFilteredSubscriptions(t *testing.T) {
 				require.NoError(t, json.Unmarshal(param.RawMessage, filt))
 				require.Equal(t, util.Uint160{1, 2, 3, 4, 5}, *filt.Contract)
 				require.Nil(t, filt.Name)
+				require.Empty(t, filt.Parameters)
 			},
 		},
 		{"notifications name",
@@ -605,6 +607,7 @@ func TestWSFilteredSubscriptions(t *testing.T) {
 				require.NoError(t, json.Unmarshal(param.RawMessage, filt))
 				require.Equal(t, "my_pretty_notification", *filt.Name)
 				require.Nil(t, filt.Contract)
+				require.Empty(t, filt.Parameters)
 			},
 		},
 		{"notifications contract hash and name",
@@ -620,6 +623,27 @@ func TestWSFilteredSubscriptions(t *testing.T) {
 				require.NoError(t, json.Unmarshal(param.RawMessage, filt))
 				require.Equal(t, util.Uint160{1, 2, 3, 4, 5}, *filt.Contract)
 				require.Equal(t, "my_pretty_notification", *filt.Name)
+				require.Empty(t, filt.Parameters)
+			},
+		},
+		{"notifications parameters",
+			func(t *testing.T, wsc *WSClient) {
+				contract := util.Uint160{1, 2, 3, 4, 5}
+				name := "my_pretty_notification"
+				prms, err := smartcontract.NewParametersFromValues(1, "2", []byte{3})
+				require.NoError(t, err)
+				_, err = wsc.ReceiveExecutionNotifications(&neorpc.NotificationFilter{Contract: &contract, Name: &name, Parameters: prms}, make(chan *state.ContainedNotificationEvent))
+				require.NoError(t, err)
+			},
+			func(t *testing.T, p *params.Params) {
+				param := p.Value(1)
+				filt := new(neorpc.NotificationFilter)
+				prms, err := smartcontract.NewParametersFromValues(1, "2", []byte{3})
+				require.NoError(t, err)
+				require.NoError(t, json.Unmarshal(param.RawMessage, filt))
+				require.Equal(t, util.Uint160{1, 2, 3, 4, 5}, *filt.Contract)
+				require.Equal(t, "my_pretty_notification", *filt.Name)
+				require.Equal(t, prms, filt.Parameters)
 			},
 		},
 		{"executions state",

pkg/services/rpcsrv/subscription_test.go

@@ -1,8 +1,11 @@
 package rpcsrv
 
 import (
+	"encoding/base64"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
+	"slices"
 	"strings"
 	"sync"
 	"testing"
@@ -14,6 +17,7 @@ import (
 	"github.com/nspcc-dev/neo-go/pkg/core"
 	"github.com/nspcc-dev/neo-go/pkg/encoding/address"
 	"github.com/nspcc-dev/neo-go/pkg/neorpc"
+	"github.com/nspcc-dev/neo-go/pkg/smartcontract"
 	"github.com/nspcc-dev/neo-go/pkg/util"
 	"github.com/stretchr/testify/require"
 )
@@ -238,6 +242,9 @@ func TestFilteredSubscriptions(t *testing.T) {
 				rmap := resp.Payload[0].(map[string]any)
 				require.Equal(t, neorpc.NotificationEventID, resp.Event)
 				c := rmap["contract"].(string)
+				for s, a := range rmap {
+					fmt.Println(s, a)
+				}
 				require.Equal(t, "0x"+testContractHash, c)
 			},
 		},
@@ -261,6 +268,39 @@ func TestFilteredSubscriptions(t *testing.T) {
 				require.Equal(t, "my_pretty_notification", n)
 			},
 		},
+		"notification matching contract hash and parameter": {
+			params: `["notification_from_execution", {"contract":"` + testContractHash + `", "parameters":[{"type":"Any","value":null},{"type":"Hash160","value":"449fe8fbd4523072f5e3a4dfa17a494c119d4c08"}]}]`,
+			check: func(t *testing.T, resp *neorpc.Notification) {
+				rmap := resp.Payload[0].(map[string]any)
+				require.Equal(t, neorpc.NotificationEventID, resp.Event)
+				c := rmap["contract"].(string)
+				require.Equal(t, "0x"+testContractHash, c)
+				// it should be exact unique "Init" call sending all the tokens to the contract itself
+				parameters := rmap["state"].(map[string]any)["value"].([]any)
+				require.Len(t, parameters, 3)
+				// sender
+				transferReceiverType := parameters[1].(map[string]any)["type"].(string)
+				require.Equal(t, smartcontract.Hash160Type.ConvertToStackitemType().String(), transferReceiverType)
+				transferReceiver := parameters[1].(map[string]any)["value"].(string)
+				hashExp, err := hex.DecodeString(testContractHash)
+				require.NoError(t, err)
+				slices.Reverse(hashExp)
+				hashGot, err := base64.StdEncoding.DecodeString(transferReceiver)
+				require.NoError(t, err)
+				require.Equal(t, hashExp, hashGot)
+				// this amount happens only for initial token distribution
+				amountType := parameters[2].(map[string]any)["type"].(string)
+				require.Equal(t, smartcontract.IntegerType.ConvertToStackitemType().String(), amountType)
+				amount := parameters[2].(map[string]any)["value"].(string)
+				require.Equal(t, amount, "1000000")
+			},
+		},
+		"notification matching contract hash but unknown parameter": {
+			params: `["notification_from_execution", {"contract":"` + testContractHash + `", "parameters":[{"type":"Any","value":null},{"type":"Hash160","value":"ffffffffffffffffffffffffffffffffffffffff"}]}]`,
+			check: func(t *testing.T, resp *neorpc.Notification) {
+				t.Fatal("this filter should not return any notification from test contract")
+			},
+		},
 		"execution matching state": {
 			params: `["transaction_executed", {"state":"HALT"}]`,
 			check: func(t *testing.T, resp *neorpc.Notification) {