ship: add table delta whitelist filtering

cmd/thalos/server.go

@@ -393,6 +393,7 @@ func serverCmd(cmd *cobra.Command, args []string) {
 	)
 
 	processor.SetBlacklist(conf.Ship.Blacklist)
+	processor.SetTableDeltaWhitelist(conf.Ship.TableDeltaWhitelist)
 	processor.FetchDeltas(conf.Ship.EnableTableDeltas)
 
 	// Run the application

config.example.yml

@@ -46,6 +46,18 @@ ship:
   # If true, Thalos will process table deltas
   # table_deltas: false
 
+  # Optional: whitelist table deltas by contract and table.
+  # When set, thalos will only process `contract_row` deltas where both
+  # `code` and `table` match this whitelist.
+  #
+  # NOTE: values must be arrays of table names.
+  # Use "*" to allow all tables for a contract.
+  # table_delta_whitelist:
+  #   eosio.token:
+  #     - accounts
+  #   mycontract:
+  #     - "*"
+
   # Blacklist contract/actions
   blacklist:
     # this is a "useless" action that results in alot of warning messages.

internal/config/builder.go

@@ -60,6 +60,7 @@ func NewBuilder() *Builder {
 			"ship.blacklist":              "blacklist",
 			"ship.blacklist_is_whitelist": "blacklist-is-whitelist",
 			"ship.table_deltas":           "table-deltas",
+			"ship.table_delta_whitelist":  "table-delta-whitelist",
 		},
 	}
 }
@@ -152,9 +153,9 @@ func decodeIntoBlacklist(in any) (*types.Blacklist, error) {
 	// Sometimes we have a slice of interfaces.
 	// Need to convert it to a slice of strings.
 	case []any:
-		sv := make([]string, len(v))
-		for i, j := range v {
-			sv[i] = j.(string)
+		sv, err := toStringSlice(v)
+		if err != nil {
+			return nil, err
 		}
 		return blacklistParseSlice(sv)
 	}
@@ -165,19 +166,48 @@ func decodeIntoBlacklist(in any) (*types.Blacklist, error) {
 // Blacklist map parser
 func blacklistParseMap(in map[string]any) (*types.Blacklist, error) {
 	list := &types.Blacklist{}
-	for k, v := range in {
-		switch v := v.(type) {
-		case []any:
-			for _, v := range v {
-				list.Add(k, v.(string))
-			}
-		case any:
-			list.Add(k, v.(string))
+	for contract, value := range in {
+		if err := blacklistParseValue(list, contract, value); err != nil {
+			return nil, err
 		}
 	}
 	return list, nil
 }
 
+func blacklistParseValue(list *types.Blacklist, contract string, in any) error {
+	switch value := in.(type) {
+	case map[string]any:
+		for nested, nestedValue := range value {
+			if err := blacklistParseValue(list, contract+"."+nested, nestedValue); err != nil {
+				return err
+			}
+		}
+		return nil
+	case map[any]any:
+		for nested, nestedValue := range value {
+			nestedKey, ok := nested.(string)
+			if !ok {
+				return fmt.Errorf("Must be a string slice")
+			}
+			if err := blacklistParseValue(list, contract+"."+nestedKey, nestedValue); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+
+	values, err := toStringSlice(in)
+	if err != nil {
+		return err
+	}
+
+	for _, entry := range values {
+		list.Add(contract, entry)
+	}
+
+	return nil
+}
+
 // Blacklist slice parser
 func blacklistParseSlice(in []string) (*types.Blacklist, error) {
 	list := &types.Blacklist{}
@@ -195,3 +225,24 @@ func blacklistParseSlice(in []string) (*types.Blacklist, error) {
 	}
 	return list, nil
 }
+
+func toStringSlice(in any) ([]string, error) {
+	switch value := in.(type) {
+	case string:
+		return []string{value}, nil
+	case []string:
+		return value, nil
+	case []any:
+		out := make([]string, 0, len(value))
+		for _, item := range value {
+			str, ok := item.(string)
+			if !ok {
+				return nil, fmt.Errorf("Must be a string slice")
+			}
+			out = append(out, str)
+		}
+		return out, nil
+	default:
+		return nil, fmt.Errorf("Must be a string slice")
+	}
+}

internal/config/builder_test.go

@@ -46,6 +46,10 @@ func TestBuilder(t *testing.T) {
 				"contract": {"skip1", "skip2"},
 			}),
 			BlacklistIsWhitelist: true,
+			TableDeltaWhitelist: *types.NewBlacklist(map[string][]string{
+				"eosio.token": {"accounts"},
+				"mycontract":  {"*"},
+			}),
 		},
 		Telegram: TelegramConfig{
 			Id:      "110201543:AAHdqTcvCH1vGWJxfSeofSAs0K5PALDsaw",
@@ -85,6 +89,11 @@ ship:
   max_messages_in_flight: 1337
   start_block_num: 23671836
   end_block_num: 23872222
+  table_delta_whitelist:
+    eosio.token:
+      - accounts
+    mycontract:
+      - "*"
   blacklist:
     eosio: noop
     contract:
@@ -176,6 +185,7 @@ func TestBuilder_Flags(t *testing.T) {
 	require.NoError(t, flags.Set("blacklist", "contract:action1,contract:action2,contract2:action1"))
 	require.NoError(t, flags.Set("blacklist-is-whitelist", "true"))
 	require.NoError(t, flags.Set("table-deltas", "false"))
+	require.NoError(t, flags.Set("table-delta-whitelist", "eosio.token:accounts,mycontract:*"))
 
 	cfg, err := NewBuilder().
 		SetSource(bytes.NewReader([]byte(``))).
@@ -208,6 +218,10 @@ func TestBuilder_Flags(t *testing.T) {
 				"contract2": {"action1"},
 			}),
 			BlacklistIsWhitelist: true,
+			TableDeltaWhitelist: *types.NewBlacklist(map[string][]string{
+				"eosio.token": {"accounts"},
+				"mycontract":  {"*"},
+			}),
 		},
 		Telegram: TelegramConfig{
 			Id:      "72983126312982618",

internal/config/cli.go

@@ -47,6 +47,7 @@ func GetFlags() *pflag.FlagSet {
 	flags.Lookup("end-block").DefValue = "none"
 
 	flags.Bool("table-deltas", true, "True if thalos should receive and process table deltas from ship.")
+	flags.StringSlice("table-delta-whitelist", []string{}, "Define a list of 'contract:table' pairs to whitelist table deltas. Use '*' for wildcard table names, e.g. 'mycontract:*'.")
 
 	flags.Bool("irreversible-only", false, "Only stream irreversible blocks from ship")
 	flags.Int("max-msg-in-flight", 10, "Maximum messages that can be sent from SHIP without acknowledgement")

internal/config/config.go

@@ -40,6 +40,7 @@ type ShipConfig struct {
 	Blacklist            types.Blacklist `yaml:"blacklist" mapstructure:"blacklist"`
 	BlacklistIsWhitelist bool            `yaml:"blacklist_is_whitelist" mapstructure:"blacklist_is_whitelist"`
 	EnableTableDeltas    bool            `yaml:"table_deltas" mapstructure:"table_deltas"`
+	TableDeltaWhitelist  types.Blacklist `yaml:"table_delta_whitelist" mapstructure:"table_delta_whitelist"`
 }
 
 type Config struct {

internal/server/ship_processor.go

@@ -42,6 +42,9 @@ type ShipProcessor struct {
 
 	// Action blacklist
 	blacklist types.Blacklist
+
+	// Optional whitelist for contract_row table delta contract+table names.
+	tableDeltaWhitelist types.Blacklist
 }
 
 // SpawnProcessor creates a new ShipProccessor that consumes the shipclient.Stream passed to it.
@@ -80,6 +83,69 @@ func (processor *ShipProcessor) SetBlacklist(list types.Blacklist) {
 	processor.blacklist = list
 }
 
+func (processor *ShipProcessor) SetTableDeltaWhitelist(list types.Blacklist) {
+	list.SetWhitelist(true)
+	processor.tableDeltaWhitelist = list
+}
+
+func (processor *ShipProcessor) hasTableDeltaWhitelist() bool {
+	return !processor.tableDeltaWhitelist.Empty()
+}
+
+func mapValueToString(data map[string]any, key string) (string, bool) {
+	value, ok := data[key]
+	if !ok {
+		return "", false
+	}
+
+	if name, ok := value.(string); ok {
+		return name, true
+	}
+
+	if name, ok := value.(interface{ String() string }); ok {
+		return name.String(), true
+	}
+
+	return "", false
+}
+
+func (processor *ShipProcessor) shouldProcessTableDeltaName(name string) bool {
+	if !processor.hasTableDeltaWhitelist() {
+		return true
+	}
+
+	return name == "contract_row"
+}
+
+func (processor *ShipProcessor) filterWhitelistedContractRows(rows []message.TableDeltaRow) []message.TableDeltaRow {
+	if !processor.hasTableDeltaWhitelist() {
+		return rows
+	}
+
+	out := make([]message.TableDeltaRow, 0, len(rows))
+	for _, row := range rows {
+		if row.Data == nil {
+			continue
+		}
+
+		contract, ok := mapValueToString(row.Data, "code")
+		if !ok {
+			continue
+		}
+
+		table, ok := mapValueToString(row.Data, "table")
+		if !ok {
+			continue
+		}
+
+		if processor.tableDeltaWhitelist.IsAllowed(contract, table) {
+			out = append(out, row)
+		}
+	}
+
+	return out
+}
+
 func (processor *ShipProcessor) initHandler(abi *chain.Abi) {
 	processor.shipABI = abi
 }
@@ -333,12 +399,23 @@ func (processor *ShipProcessor) processBlock(blockResult *ship.GetBlocksResultV0
 		} else {
 			logger := mainLogger.WithField("type", "table_delta").Dup()
 			for _, delta := range deltas {
+				if !processor.shouldProcessTableDeltaName(delta.V0.Name) {
+					continue
+				}
+
+				rows := processor.proccessDeltaRows(logger, delta.V0.Name, delta.V0.Rows)
+				if processor.hasTableDeltaWhitelist() {
+					rows = processor.filterWhitelistedContractRows(rows)
+					if len(rows) < 1 {
+						continue
+					}
+				}
 
 				msg := message.TableDelta{
 					BlockNum:  blockNumber,
 					Timestamp: timestamp,
 					Name:      delta.V0.Name,
-					Rows:      processor.proccessDeltaRows(logger, delta.V0.Name, delta.V0.Rows),
+					Rows:      rows,
 				}
 
 				if err := processor.queue.PostTableDelta(msg); err != nil {

internal/server/ship_processor_test.go

@@ -0,0 +1,79 @@
+package server
+
+import (
+	"testing"
+
+	"github.com/eosswedenorg/thalos/api/message"
+	"github.com/eosswedenorg/thalos/internal/types"
+	"github.com/shufflingpixels/antelope-go/chain"
+	"github.com/stretchr/testify/require"
+)
+
+func TestShipProcessor_ShouldProcessTableDeltaName(t *testing.T) {
+	processor := &ShipProcessor{}
+
+	require.True(t, processor.shouldProcessTableDeltaName("contract_row"))
+	require.True(t, processor.shouldProcessTableDeltaName("resource_limits"))
+
+	processor.SetTableDeltaWhitelist(*types.NewBlacklist(map[string][]string{
+		"eosio.token": {"accounts"},
+	}))
+
+	require.True(t, processor.shouldProcessTableDeltaName("contract_row"))
+	require.False(t, processor.shouldProcessTableDeltaName("resource_limits"))
+}
+
+func TestShipProcessor_FilterWhitelistedContractRows(t *testing.T) {
+	processor := &ShipProcessor{}
+	processor.SetTableDeltaWhitelist(*types.NewBlacklist(map[string][]string{
+		"eosio.token": {"accounts"},
+		"eosio":       {"*"},
+	}))
+
+	rows := []message.TableDeltaRow{
+		{Data: map[string]any{"code": "eosio.token", "table": "accounts"}},
+		{Data: map[string]any{"code": "eosio.token", "table": "stat"}},
+		{Data: map[string]any{"code": "eosio", "table": "users"}},
+		{Data: map[string]any{"scope": "missing-code"}},
+		{Data: map[string]any{"code": 42, "table": "accounts"}},
+		{Data: map[string]any{"code": "eosio.token", "table": 1}},
+		{Data: nil},
+	}
+
+	actual := processor.filterWhitelistedContractRows(rows)
+
+	require.Len(t, actual, 2)
+	require.Equal(t, "eosio.token", actual[0].Data["code"])
+	require.Equal(t, "accounts", actual[0].Data["table"])
+	require.Equal(t, "eosio", actual[1].Data["code"])
+}
+
+func TestShipProcessor_FilterWhitelistedContractRows_StringerCode(t *testing.T) {
+	processor := &ShipProcessor{}
+	processor.SetTableDeltaWhitelist(*types.NewBlacklist(map[string][]string{
+		"eosio.token": {"accounts"},
+	}))
+
+	rows := []message.TableDeltaRow{
+		{Data: map[string]any{"code": chain.N("eosio.token"), "table": chain.N("accounts")}},
+		{Data: map[string]any{"code": chain.N("eosio.token"), "table": chain.N("stat")}},
+	}
+
+	actual := processor.filterWhitelistedContractRows(rows)
+
+	require.Len(t, actual, 1)
+	require.Equal(t, chain.N("eosio.token"), actual[0].Data["code"])
+}
+
+func TestShipProcessor_FilterWhitelistedContractRows_NoWhitelist(t *testing.T) {
+	processor := &ShipProcessor{}
+
+	rows := []message.TableDeltaRow{
+		{Data: map[string]any{"code": "eosio.token", "table": "accounts"}},
+		{Data: map[string]any{"scope": "missing-code"}},
+	}
+
+	actual := processor.filterWhitelistedContractRows(rows)
+
+	require.Equal(t, rows, actual)
+}