From 90a8de15feeccb9b1ec0b81bc40413d46bbff502 Mon Sep 17 00:00:00 2001 From: Henrik Hautakoski Date: Thu, 27 Aug 2026 20:01:19 +0200 Subject: [PATCH] ship: add table delta whitelist filtering --- cmd/thalos/server.go | 1 + config.example.yml | 12 ++++ internal/config/builder.go | 73 ++++++++++++++++++++---- internal/config/builder_test.go | 14 +++++ internal/config/cli.go | 1 + internal/config/config.go | 1 + internal/server/ship_processor.go | 79 +++++++++++++++++++++++++- internal/server/ship_processor_test.go | 79 ++++++++++++++++++++++++++ 8 files changed, 248 insertions(+), 12 deletions(-) create mode 100644 internal/server/ship_processor_test.go diff --git a/cmd/thalos/server.go b/cmd/thalos/server.go index 1e9b561..e62f369 100644 --- a/cmd/thalos/server.go +++ b/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 diff --git a/config.example.yml b/config.example.yml index 2dc438e..fac1da8 100644 --- a/config.example.yml +++ b/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. diff --git a/internal/config/builder.go b/internal/config/builder.go index e22bf45..2470deb 100644 --- a/internal/config/builder.go +++ b/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") + } +} diff --git a/internal/config/builder_test.go b/internal/config/builder_test.go index b133353..f874e9b 100644 --- a/internal/config/builder_test.go +++ b/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", diff --git a/internal/config/cli.go b/internal/config/cli.go index d4f26ac..7055a1c 100644 --- a/internal/config/cli.go +++ b/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") diff --git a/internal/config/config.go b/internal/config/config.go index feabb87..f7f1bef 100644 --- a/internal/config/config.go +++ b/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 { diff --git a/internal/server/ship_processor.go b/internal/server/ship_processor.go index 1c1ee39..34484db 100644 --- a/internal/server/ship_processor.go +++ b/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 { diff --git a/internal/server/ship_processor_test.go b/internal/server/ship_processor_test.go new file mode 100644 index 0000000..98dfa1f --- /dev/null +++ b/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) +}