Repository: cosmos/relayer
Pull request #812 (2022-06-27 17:44:51)

PathProcessor - client, connection, and channel filtering
Adds `Rule` and `FilterList` to `PathEnd` packet message channel filtering (i.e. which channels should packet messages be relayed for.) Connection and Channel handshake messages are handled differently to allow automatic handshake completion.

Adds logic to merging messages for a `pathEndRuntime` so that only applicable messages are cached.

Desired behavior:
- `Rule` can be either `allowlist` or `denylist` to filter the channels for relaying packets. This incorporates into the existing config definition. No filter will be applied if the `Rule` is anything else other than those 2 strings.
- `PathProcessor` will cache and relay packet messages that abide by the `Rule`/`FilterList`.
- For connection handshake messages, `PathProcessor` will cache and relay messages where the message's connection uses the same client as the `pathEnd`. This will allow the relayer to automatically complete connection handshakes on top of clients that are in the configured paths.
- For channel handshake messages, `PathProcessor` will cache and relay messages where the message's channel uses connections that use the same client as the `pathEnd`. This will allow the relayer to automatically complete channel handshakes where the underlying client is in the relayer's configured paths.


Patch:
diff --git a/relayer/processor/path_end.go b/relayer/processor/path_end.go
index fcb16cb..c226231 100644
--- a/relayer/processor/path_end.go
+++ b/relayer/processor/path_end.go
@@ -1,14 +1,85 @@
 package processor
 
+// PathEnd references one chain involved in a path.
+// A path is composed of two PathEnds.
 type PathEnd struct {
-	ChainID      string
-	ClientID     string
-	ConnectionID string
+	ChainID  string
+	ClientID string
+	// ConnectionIDs are tracked by pathEndRuntime in PathProcessor for known connections on this client
 
-	// TODO channel filter stuff (allowlist, denylist)
+	// Can be either "allowlist" or "denylist"
+	Rule       string
+	FilterList []ChannelKey // which channels to allow or deny
 }
 
+// NewPathEnd constructs a PathEnd, validating initial parameters.
+func NewPathEnd(chainID string, clientID string, rule string, filterList []ChannelKey) PathEnd {
+	return PathEnd{
+		ChainID:    chainID,
+		ClientID:   clientID,
+		Rule:       rule,
+		FilterList: filterList,
+	}
+}
+
+const (
+	RuleAllowList = "allowlist"
+	RuleDenyList  = "denylist"
+)
+
+func (pe PathEnd) checkChannelMatch(listChannelID, listPortID string, channelKey ChannelKey) bool {
+	if listChannelID == "" {
+		return false
+	}
+	if listChannelID == channelKey.ChannelID {
+		if listPortID == "" {
+			return true
+		}
+		if listPortID == channelKey.PortID {
+			return true
+		}
+	}
+	if listChannelID == channelKey.CounterpartyChannelID {
+		if listPortID == "" {
+			return true
+		}
+		if listPortID == channelKey.CounterpartyPortID {
+			return true
+		}
+	}
+	return false
+}
+
+func (pe PathEnd) shouldRelayChannelSingle(channelKey ChannelKey, listChannel ChannelKey, allowList bool) bool {
+	if pe.checkChannelMatch(listChannel.ChannelID, listChannel.PortID, channelKey) {
+		return allowList
+	}
+	if pe.checkChannelMatch(listChannel.CounterpartyChannelID, listChannel.CounterpartyPortID, channelKey) {
+		return allowList
+	}
+	return !allowList
+}
+
+// if port ID is empty on allowlist channel, allow all ports
+// if port ID is non-empty on allowlist channel, allow only that specific port
+// if port ID is empty on blocklist channel, block all ports
+// if port ID is non-empty on blocklist channel, block only that specific port
 func (pe PathEnd) ShouldRelayChannel(channelKey ChannelKey) bool {
-	// TODO return based on channel filter
+	if pe.Rule == RuleAllowList {
+		for _, allowedChannel := range pe.FilterList {
+			if pe.shouldRelayChannelSingle(channelKey, allowedChannel, true) {
+				return true
+			}
+		}
+		return false
+	} else if pe.Rule == RuleDenyList {
+		for _, blockedChannel := range pe.FilterList {
+			if !pe.shouldRelayChannelSingle(channelKey, blockedChannel, false) {
+				return false
+			}
+		}
+		return true
+	}
+	// if neither allow list or block list are provided, all channels are okay
 	return true
 }
diff --git a/relayer/processor/path_end_runtime.go b/relayer/processor/path_end_runtime.go
index bf8b9d7..fa9f84b 100644
--- a/relayer/processor/path_end_runtime.go
+++ b/relayer/processor/path_end_runtime.go
@@ -8,6 +8,14 @@ import "github.com/cosmos/relayer/v2/relayer/provider"
 type pathEndRuntime struct {
 	info PathEnd
 
+	// TODO populate known connections and channels as they are discovered on this client.
+	// channel handshakes will not work until these are populated properly.
+
+	// includes all known and discovered connections that use this PathEnd's client.
+	knownConnections []string
+	// includes all known and discovered channels on connections that use this PathEnd's client.
+	knownChannels []string
+
 	chainProvider provider.ChainProvider
 
 	// cached data
@@ -44,20 +52,76 @@ func newPathEndRuntime(pathEnd PathEnd) *pathEndRuntime {
 	}
 }
 
+func (pathEnd *pathEndRuntime) isRelevantConnection(connectionID string) bool {
+	for _, conn := range pathEnd.knownConnections {
+		if conn == connectionID {
+			return true
+		}
+	}
+	return false
+}
+
+// mergeMessageCache merges relevant IBC messages for packet flows, connection handshakes, and channel handshakes.
+func (pathEnd *pathEndRuntime) mergeMessageCache(messageCache IBCMessagesCache) {
+	packetMessages := make(ChannelPacketMessagesCache)
+	connectionHandshakeMessages := make(ConnectionMessagesCache)
+	channelHandshakeMessages := make(ChannelMessagesCache)
+
+	for ch, pmc := range messageCache.PacketFlow {
+		if !pathEnd.info.ShouldRelayChannel(ch) {
+			packetMessages[ch] = pmc
+		}
+	}
+	pathEnd.messageCache.PacketFlow.Merge(packetMessages)
+
+	for action, cmc := range messageCache.ConnectionHandshake {
+		newCmc := make(ConnectionMessageCache)
+		for conn, msg := range cmc {
+			if pathEnd.info.ClientID == conn.ClientID {
+				// can complete connection handshakes on this client
+				// since PathProcessor holds reference to the counterparty chain pathEndRuntime.
+				newCmc[conn] = msg
+				break
+			}
+		}
+		if len(newCmc) == 0 {
+			continue
+		}
+		connectionHandshakeMessages[action] = newCmc
+	}
+	pathEnd.messageCache.ConnectionHandshake.Merge(connectionHandshakeMessages)
+
+	for action, cmc := range messageCache.ChannelHandshake {
+		newCmc := make(ChannelMessageCache)
+		for ch, msg := range cmc {
+			for _, chID := range pathEnd.knownChannels {
+				if chID == ch.ChannelID {
+					// if channel's connection is known, and connection is on top of this client,
+					// then we can complete channel handshake for this channel
+					// since PathProcessor holds reference to the counterparty chain pathEndRuntime.
+					newCmc[ch] = msg
+					break
+				}
+			}
+		}
+		if len(newCmc) == 0 {
+			continue
+		}
+		channelHandshakeMessages[action] = newCmc
+	}
+	pathEnd.messageCache.ChannelHandshake.Merge(channelHandshakeMessages)
+}
+
 func (pathEnd *pathEndRuntime) MergeCacheData(d ChainProcessorCacheData) {
 	pathEnd.inSync = d.InSync
 	pathEnd.latestBlock = d.LatestBlock
 	pathEnd.latestHeader = d.LatestHeader
 	pathEnd.clientState = d.ClientState
 
-	// TODO messageCache Merge will likely need to move into the PathProcessor, and make sure passes client, connection, or channel filter
-	pathEnd.messageCache.Merge(d.IBCMessagesCache) // Merge incoming packet IBC messages into the backlog
+	pathEnd.mergeMessageCache(d.IBCMessagesCache) // Merge incoming packet IBC messages into the backlog
 
-	// TODO connectionStateCache Merge will likely need to move into the PathProcessor, and make sure passes connection filter
 	pathEnd.connectionStateCache.Merge(d.ConnectionStateCache) // Update latest connection open state for chain
-
-	// TODO channelStateCache Merge will likely need to move into the PathProcessor, and make sure passes connection filter
-	pathEnd.channelStateCache.Merge(d.ChannelStateCache) // Update latest channel open state for chain
+	pathEnd.channelStateCache.Merge(d.ChannelStateCache)       // Update latest channel open state for chain
 
 	pathEnd.ibcHeaderCache.Merge(d.IBCHeaderCache)  // Update latest IBC header state
 	pathEnd.ibcHeaderCache.Prune(ibcHeadersToCache) // Only keep most recent IBC headers
diff --git a/relayer/processor/path_processor.go b/relayer/processor/path_processor.go
index 02acd37..0346142 100644
--- a/relayer/processor/path_processor.go
+++ b/relayer/processor/path_processor.go
@@ -132,19 +132,19 @@ func (pp *PathProcessor) IsRelayedChannel(chainID string, channelKey ChannelKey)
 }
 
 func (pp *PathProcessor) IsRelevantClient(chainID string, clientID string) bool {
-	if pp.pathEnd1.info.ChainID == chainID && pp.pathEnd1.info.ClientID == clientID {
-		return true
-	} else if pp.pathEnd2.info.ChainID == chainID && pp.pathEnd2.info.ClientID == clientID {
-		return true
+	if pp.pathEnd1.info.ChainID == chainID {
+		return pp.pathEnd1.info.ClientID == clientID
+	} else if pp.pathEnd2.info.ChainID == chainID {
+		return pp.pathEnd2.info.ClientID == clientID
 	}
 	return false
 }
 
 func (pp *PathProcessor) IsRelevantConnection(chainID string, connectionID string) bool {
-	if pp.pathEnd1.info.ChainID == chainID && pp.pathEnd1.info.ConnectionID == connectionID {
-		return true
-	} else if pp.pathEnd2.info.ChainID == chainID && pp.pathEnd2.info.ConnectionID == connectionID {
-		return true
+	if pp.pathEnd1.info.ChainID == chainID {
+		return pp.pathEnd1.isRelevantConnection(connectionID)
+	} else if pp.pathEnd2.info.ChainID == chainID {
+		return pp.pathEnd2.isRelevantConnection(connectionID)
 	}
 	return false
 }
diff --git a/relayer/processor/types.go b/relayer/processor/types.go
index 5444b64..7c70c87 100644
--- a/relayer/processor/types.go
+++ b/relayer/processor/types.go
@@ -188,12 +188,6 @@ func (c PacketMessagesCache) DeleteCachedMessages(toDelete ...map[string][]uint6
 	}
 }
 
-func (c IBCMessagesCache) Merge(other IBCMessagesCache) {
-	c.ConnectionHandshake.Merge(other.ConnectionHandshake)
-	c.ChannelHandshake.Merge(other.ChannelHandshake)
-	c.PacketFlow.Merge(other.PacketFlow)
-}
-
 // Merge merges another ChannelPacketMessagesCache into this one.
 func (c ChannelPacketMessagesCache) Merge(other ChannelPacketMessagesCache) {
 	for channelKey, messageCache := range other {


Test patch:
diff --git a/relayer/processor/path_end_test.go b/relayer/processor/path_end_test.go
new file mode 100644
index 0000000..7f6dac4
--- /dev/null
+++ b/relayer/processor/path_end_test.go
@@ -0,0 +1,174 @@
+package processor_test
+
+import (
+	"testing"
+
+	"github.com/cosmos/relayer/v2/relayer/processor"
+	"github.com/stretchr/testify/require"
+)
+
+const (
+	testChannel0 = "test-channel-0"
+	testPort0    = "test-port-0"
+	testChannel1 = "test-channel-1"
+	testPort1    = "test-port-1"
+)
+
+// empty allow list and block list should relay everything
+func TestAllowAllChannels(t *testing.T) {
+	mockPathEnd := processor.PathEnd{}
+
+	mockAllowedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort0,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel), "does not allow channel to be relayed, even though allow list and block list are empty")
+
+	// test counterparty
+	mockAllowedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel1,
+		CounterpartyPortID:    testPort1,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel2), "does not allow counterparty channel to be relayed, even though allow list and block list are empty")
+}
+
+//
+func TestAllowAllPortsForChannel(t *testing.T) {
+	mockAllowList := []processor.ChannelKey{
+		{ChannelID: testChannel0},
+	}
+	mockPathEnd := processor.PathEnd{
+		Rule:       processor.RuleAllowList,
+		FilterList: mockAllowList,
+	}
+
+	mockAllowedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort0,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel), "does not allow channel to be relayed, even though channelID is in allow list")
+
+	// test counterparty
+	mockAllowedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort0,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel2), "does not allow counterparty channel to be relayed, even though channelID is in allow list")
+
+	mockBlockedChannel := processor.ChannelKey{
+		ChannelID: testChannel1,
+		PortID:    testPort1,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel), "allows channel to be relayed, even though channelID is not in allow list")
+
+	mockBlockedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel1,
+		CounterpartyPortID:    testPort1,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel2), "allows channel to be relayed, even though channelID is not in allow list")
+}
+
+func TestAllowSpecificPortForChannel(t *testing.T) {
+	mockAllowList := []processor.ChannelKey{
+		{ChannelID: testChannel0, PortID: testPort0},
+	}
+	mockPathEnd := processor.PathEnd{
+		Rule:       processor.RuleAllowList,
+		FilterList: mockAllowList,
+	}
+
+	mockAllowedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort0,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel), "does not allow channel to be relayed, even though channelID is in allow list")
+
+	// test counterparty
+	mockAllowedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort0,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel2), "does not allow counterparty channel to be relayed, even though channelID is in allow list")
+
+	mockBlockedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort1,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel), "allows channel to be relayed, even though portID is not in allow list")
+
+	mockBlockedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort1,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel2), "allows channel to be relayed, even though portID is not in allow list")
+}
+
+func TestBlockAllPortsForChannel(t *testing.T) {
+	mockBlockList := []processor.ChannelKey{
+		{ChannelID: testChannel0},
+	}
+	mockPathEnd := processor.PathEnd{
+		Rule:       processor.RuleDenyList,
+		FilterList: mockBlockList,
+	}
+
+	mockBlockedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort0,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel), "allows channel to be relayed, even though channelID is in block list")
+
+	// test counterparty
+	mockBlockedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort0,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel2), "allows counterparty channel to be relayed, even though channelID is in block list")
+
+	mockAllowedChannel := processor.ChannelKey{
+		ChannelID: testChannel1,
+		PortID:    testPort1,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel), "does not allow channel to be relayed, even though channelID is not in block list")
+
+	mockAllowedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel1,
+		CounterpartyPortID:    testPort1,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel2), "does not allow counterparty channel to be relayed, even though channelID is not in block list")
+}
+
+func TestBlockSpecificPortForChannel(t *testing.T) {
+	mockBlockList := []processor.ChannelKey{
+		{ChannelID: testChannel0, PortID: testPort0},
+	}
+	mockPathEnd := processor.PathEnd{
+		Rule:       processor.RuleDenyList,
+		FilterList: mockBlockList,
+	}
+
+	mockBlockedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort0,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel), "allows channel to be relayed, even though channelID/portID is in block list")
+
+	// test counterparty
+	mockBlockedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort0,
+	}
+	require.False(t, mockPathEnd.ShouldRelayChannel(mockBlockedChannel2), "allows counterparty channel to be relayed, even though channelID/portID is in block list")
+
+	mockAllowedChannel := processor.ChannelKey{
+		ChannelID: testChannel0,
+		PortID:    testPort1,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel), "does not allow channel to be relayed, even though portID is not in block list")
+
+	mockAllowedChannel2 := processor.ChannelKey{
+		CounterpartyChannelID: testChannel0,
+		CounterpartyPortID:    testPort1,
+	}
+	require.True(t, mockPathEnd.ShouldRelayChannel(mockAllowedChannel2), "does not allow counterparty channel to be relayed, even though portID is not in block list")
+}