[Utility] Add server-side validation on trustless relays (#789)

## Description

- Partial implementation of server-side validations for trustless
relays, along with placeholders for next validations to implement.
- Update Relay structure with the required fields.

## Issue

Part of work on #754 


## Type of change

Please mark the relevant option(s):

- [x] New feature, functionality or library
- [ ] Bug fix
- [ ] Code health or cleanup
- [ ] Major breaking change
- [] Documentation
- [ ] Other <!-- add details here if it a different type of change -->

## List of changes

- Add implementation for Servicer's validations, with placeholders for
the logic not implemented yet.
- Changed the Relay struct to include required fields, e.g.
`ApplicationAddress` in `RelayMeta`
- Remove AAT from Relay type and remove the referencing code that was
breaking the tests (AAT not fully removed yet, will be on the next PR)


## Testing

- [x] `make develop_test`; if any code changes were made
- [ ] `make test_e2e` on [k8s
LocalNet](https://github.com/pokt-network/pocket/blob/main/build/localnet/README.md);
if any code changes were made
- [ ] `e2e-devnet-test` passes tests on
[DevNet](https://pocketnetwork.notion.site/How-to-DevNet-ff1598f27efe44c09f34e2aa0051f0dd);
if any code was changed
- [ ] [Docker Compose
LocalNet](https://github.com/pokt-network/pocket/blob/main/docs/development/README.md);
if any major functionality was changed or introduced
- [ ] [k8s
LocalNet](https://github.com/pokt-network/pocket/blob/main/build/localnet/README.md);
if any infrastructure or configuration changes were made

## Required Checklist

- [x] I have performed a self-review of my own code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have added, or updated, [`godoc` format
comments](https://go.dev/blog/godoc) on touched members (see:
[tip.golang.org/doc/comment](https://tip.golang.org/doc/comment))
- [x] I have tested my changes using the available tooling
- [x] I have updated the corresponding CHANGELOG

### If Applicable Checklist

- [ ] I have updated the corresponding README(s); local and/or global
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] I have added, or updated,
[mermaid.js](https://mermaid-js.github.io) diagrams in the corresponding
README(s)
- [ ] I have added, or updated, documentation and
[mermaid.js](https://mermaid-js.github.io) diagrams in `shared/docs/*`
if I updated `shared/*`README(s)

---------

Signed-off-by: Arash Deshmeh <[email protected]>

rpc/doc/CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.20] - 2023-06-01
+
+- Removed AAT type from rpc handlers
+
 ## [0.0.0.19] - 2023-05-24
 
 - Updates rpc handlers to use updated BlockStore interface

rpc/handlers.go

@@ -97,18 +97,11 @@ func (s *rpcServer) PostV1ClientRelay(ctx echo.Context) error {
 		Id:   body.Meta.Geozone.Id,
 		Name: body.Meta.Geozone.Name,
 	}
-	aat := &coreTypes.AAT{
-		Version:              body.Meta.Token.Version,
-		ApplicationPublicKey: body.Meta.Token.AppPubKey,
-		ClientPublicKey:      body.Meta.Token.ClientPubKey,
-		ApplicationSignature: body.Meta.Token.AppSignature,
-	}
 	relayMeta := &coreTypes.RelayMeta{
 		BlockHeight:       body.Meta.BlockHeight,
 		ServicerPublicKey: body.Meta.ServicerPubKey,
 		RelayChain:        chain,
 		GeoZone:           geozone,
-		Token:             aat,
 		Signature:         body.Meta.Signature,
 	}
 

runtime/configs/config.go

@@ -120,6 +120,9 @@ func NewDefaultConfig(options ...func(*Config)) *Config {
 			},
 		},
 		Utility: &UtilityConfig{
+			ServicerConfig: &ServicerConfig{
+				Chains: []string{"0001"},
+			},
 			MaxMempoolTransactionBytes: defaults.DefaultUtilityMaxMempoolTransactionBytes,
 			MaxMempoolTransactions:     defaults.DefaultUtilityMaxMempoolTransactions,
 		},

runtime/configs/proto/utility_config.proto

@@ -7,4 +7,12 @@ option go_package = "github.com/pokt-network/pocket/runtime/configs";
 message UtilityConfig {
   uint64 max_mempool_transaction_bytes = 1;
   uint32 max_mempool_transactions = 2;
+  ServicerConfig servicer_config = 3;
+}
+
+// TODO: Reevalute whether each utility actor should contain address/pubKey configs or if it should be shared
+message ServicerConfig {
+  string public_key = 1;
+  string address = 2;
+  repeated string chains = 3;
 }

runtime/docs/CHANGELOG.md

@@ -7,6 +7,15 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.40] - 2023-06-01
+
+- Add an Address field to Servicer configuration
+- Add a default servicer configuration
+
+## [0.0.0.39] - 2023-05-25
+
+- Added a new ServicerConfig type
+
 ## [0.0.0.38] - 2023-05-08
 
 - Renamed `P2PConfig#MaxMempoolCount` to `P2PConfig#MaxNonces`

runtime/manager_test.go

@@ -4192,6 +4192,9 @@ func TestNewManagerFromReaders(t *testing.T) {
 						ServerModeEnabled: true,
 					},
 					Utility: &configs.UtilityConfig{
+						ServicerConfig: &configs.ServicerConfig{
+							Chains: []string{"0001"},
+						},
 						MaxMempoolTransactionBytes: 1073741824,
 						MaxMempoolTransactions:     9000,
 					},

shared/CHANGELOG.md

@@ -7,6 +7,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.59] - 2023-06-01
+
+- Use ApplicationAddress in RelayMeta
+- Fix SessionHeight field meaning comment
+
 ## [0.0.0.58] - 2023-06-01
 
 - Moved nonce field from RainTreeMessage to PocketEnvelope protobuf types

shared/core/types/proto/relay.proto

@@ -22,8 +22,8 @@ message RelayMeta {
     string servicer_public_key = 2;
     Identifiable relay_chain = 3;
     Identifiable geo_zone = 4;
-    AAT token = 5;
-    string signature = 6;
+    string signature = 5;
+    string application_address = 6;
 }
 
 message RelayResponse {

shared/core/types/proto/session.proto

@@ -11,7 +11,7 @@ import "actor.proto";
 message Session {
     string id = 1; // a universally unique ID for the session
     int64 session_number = 2; // a monotonically increasing number representing the # on the chain
-    int64 session_height = 3; // the block height at which this session started
+    int64 session_height = 3; // the number of blocks (out of numBlocksPerSession) in this session
     int64 num_session_blocks = 4; // the number of blocks the session is valid from
      // CONSIDERATION: Should we add a `RelayChain` enum and use it across the board?
      // CONSIDERATION: Should a single session support multiple relay chains?

shared/modules/utility_module.go

@@ -91,3 +91,9 @@ type LeaderUtilityUnitOfWork interface {
 type ReplicaUtilityUnitOfWork interface {
 	UtilityUnitOfWork
 }
+
+type Servicer interface {
+	Module
+
+	HandleRelay(*coreTypes.Relay) (*coreTypes.RelayResponse, error)
+}

utility/doc/CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.0.43] - 2023-06-01
+
+- Add a place-holder Servicer implementation
+- Use mockgen and Bus for DI
+- Cross-check the chains handled by a servicer using the Persistence module
+
 ## [0.0.0.42] - 2023-06-01
 
 - Added a utility feature list with must-have and nice-to-haves for MainNet and TestNet

utility/module.go

@@ -22,6 +22,9 @@ type utilityModule struct {
 
 	logger  *modules.Logger
 	mempool mempool.TXMempool
+
+	// TODO: initialize
+	servicer modules.Servicer
 }
 
 func Create(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {

utility/service/module.go

@@ -0,0 +1,9 @@
+package service
+
+import (
+	"github.com/pokt-network/pocket/shared/modules"
+)
+
+const servicerModuleName = "servicer"
+
+var _ modules.Module = &servicer{}

utility/service/service.go

@@ -1,11 +1,214 @@
 package service
 
 import (
+	"errors"
+	"fmt"
+
+	"golang.org/x/exp/slices"
+
+	"github.com/pokt-network/pocket/logger"
+	"github.com/pokt-network/pocket/runtime/configs"
 	coreTypes "github.com/pokt-network/pocket/shared/core/types"
 	"github.com/pokt-network/pocket/shared/crypto"
+	"github.com/pokt-network/pocket/shared/modules"
+	"github.com/pokt-network/pocket/shared/modules/base_modules"
+)
+
+var (
+	errValidateBlockHeight = errors.New("relay failed block height validation")
+	errValidateRelayMeta   = errors.New("relay failed metadata validation")
+
+	_ modules.Servicer = &servicer{}
 )
 
+type servicer struct {
+	base_modules.IntegratableModule
+	base_modules.InterruptableModule
+
+	logger *modules.Logger
+	config *configs.ServicerConfig
+}
+
+func CreateServicer(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {
+	return new(servicer).Create(bus, options...)
+}
+
+func (*servicer) Create(bus modules.Bus, options ...modules.ModuleOption) (modules.Module, error) {
+	s := &servicer{
+		logger: logger.Global.CreateLoggerForModule(servicerModuleName),
+	}
+
+	for _, option := range options {
+		option(s)
+	}
+
+	bus.RegisterModule(s)
+
+	cfg := bus.GetRuntimeMgr().GetConfig()
+	s.config = cfg.Utility.ServicerConfig
+
+	return s, nil
+}
+
+func (s *servicer) Start() error {
+	s.logger = logger.Global.CreateLoggerForModule(s.GetModuleName())
+	return nil
+}
+
+func (*servicer) GetModuleName() string {
+	return servicerModuleName
+}
+
+// HandleRelay processes a relay after performing validation.
+// It also updates the servicer's internal state to keep track of served relays.
+func (s *servicer) HandleRelay(relay *coreTypes.Relay) (*coreTypes.RelayResponse, error) {
+	if relay == nil {
+		return nil, fmt.Errorf("cannot serve nil relay")
+	}
+
+	if err := s.admitRelay(relay); err != nil {
+		return nil, fmt.Errorf("Error admitting relay: %w", err)
+	}
+
+	// TODO: implement Persist Relay
+	// TODO: implement execution
+	// TODO: implement state maintenance
+	// TODO: validate the response from the node?
+	// TODO: (QUESTION) Should we persist SignedRPC?
+	return nil, nil
+}
+
+// validateRelayMeta ensures the relay metadata is valid for being handled by the servicer
+// REFACTOR: move the meta-specific validation to a Validator method on RelayMeta struct
+func (s servicer) validateRelayMeta(meta *coreTypes.RelayMeta, currentHeight int64) error {
+	if meta == nil {
+		return fmt.Errorf("empty relay metadata")
+	}
+
+	if meta.RelayChain == nil {
+		return fmt.Errorf("relay chain unspecified")
+	}
+
+	// TODO: supported chains: needs to be crossed-checked with the world state from the persistence layer
+	if err := s.validateRelayChainSupport(meta.RelayChain, currentHeight); err != nil {
+		return fmt.Errorf("validation of support for relay chain %s failed: %w", meta.RelayChain.Id, err)
+	}
+
+	return nil
+}
+
+func (s servicer) validateRelayChainSupport(relayChain *coreTypes.Identifiable, currentHeight int64) error {
+	if !slices.Contains(s.config.Chains, relayChain.Id) {
+		return fmt.Errorf("chain %s not supported by servicer %s configuration", relayChain.Id, s.config.Address)
+	}
+
+	// DISCUSS: either update NewReadContext to take a uint64, or the GetCurrentHeight to return an int64.
+	readCtx, err := s.GetBus().GetPersistenceModule().NewReadContext(currentHeight)
+	if err != nil {
+		return fmt.Errorf("error getting persistence context at height %d: %w", currentHeight, err)
+	}
+	defer readCtx.Release() //nolint:errcheck // We only need to make sure the readCtx is released
+
+	// DISCUSS: should we update the GetServicer signature to take a string instead?
+	servicer, err := readCtx.GetServicer([]byte(s.config.Address), currentHeight)
+	if err != nil {
+		return fmt.Errorf("error reading servicer from persistence: %w", err)
+	}
+
+	if !slices.Contains(servicer.Chains, relayChain.Id) {
+		return fmt.Errorf("chain %s not supported by servicer %s configuration fetched from persistence", relayChain.Id, s.config.Address)
+	}
+
+	return nil
+}
+
+// TODO: implement
+// validateApplication makes sure the application has not received more relays than allocated in the current session.
+func (s servicer) validateApplication(meta *coreTypes.RelayMeta, session *coreTypes.Session) error {
+	/*
+		// if maxRelaysPerSession, overServiced := calculateAppSessionTokens(); overServiced {
+			return fmt.Errorf("application %s has exceeded its allocated relays %d for the session %d", meta.ApplicationPublicKey, maxRelaysPerSession)
+		}
+	*/
+	return nil
+}
+
+// validateServicer makes sure the servicer is A) active in the current session, and B) has not served more than its allocated relays for the session
+func (s servicer) validateServicer(meta *coreTypes.RelayMeta, session *coreTypes.Session) error {
+	if meta.ServicerPublicKey != s.config.PublicKey {
+		return fmt.Errorf("relay servicer key %s does not match this servicer instance %s", meta.ServicerPublicKey, s.config.PublicKey)
+	}
+
+	var found bool
+	for _, servicer := range session.Servicers {
+		if servicer != nil && servicer.PublicKey == meta.ServicerPublicKey {
+			found = true
+			break
+		}
+	}
+
+	if !found {
+		return fmt.Errorf("relay servicer key %s not found in session %d with %d servicers", meta.ServicerPublicKey, session.SessionNumber, len(session.Servicers))
+	}
+
+	// TODO: implement isServicerMaxedOut
+	return nil
+}
+
+// admitRelay decides whether the relay should be served
+func (s servicer) admitRelay(relay *coreTypes.Relay) error {
+	// TODO: utility module should initialize the servicer (if this module instance is a servicer)
+	const errPrefix = "Error admitting relay"
+
+	if relay == nil {
+		return fmt.Errorf("%s: relay is nil", errPrefix)
+	}
+
+	height := s.GetBus().GetConsensusModule().CurrentHeight()
+	if err := s.validateRelayMeta(relay.Meta, int64(height)); err != nil {
+		return fmt.Errorf("%w: %s", errValidateRelayMeta, err.Error())
+	}
+
+	// TODO: update the CLI to include ApplicationAddress(or Application Public Key) in the RelayMeta
+	session, err := s.GetBus().GetUtilityModule().GetSession(relay.Meta.ApplicationAddress, int64(height), relay.Meta.RelayChain.Id, relay.Meta.GeoZone.Id)
+	if err != nil {
+		return fmt.Errorf("%s: failed to get a session for height %d for relay meta %s: %w", errPrefix, height, relay.Meta, err)
+	}
+
+	// TODO: (REFACTOR) use a loop to run all validators: would also remove the need for passing the session around
+	if err := validateRelayBlockHeight(relay.Meta, session); err != nil {
+		return fmt.Errorf("%w: %s", errValidateBlockHeight, err.Error())
+	}
+
+	if err := s.validateApplication(relay.Meta, session); err != nil {
+		return fmt.Errorf("%s: relay failed application validation: %w", errPrefix, err)
+	}
+
+	if err := s.validateServicer(relay.Meta, session); err != nil {
+		return fmt.Errorf("%s: relay failed servicer instance validation: %w", errPrefix, err)
+	}
+
+	return nil
+}
+
+// IMPROVE: Add session height tolerance to account for session rollovers
+func validateRelayBlockHeight(relayMeta *coreTypes.RelayMeta, session *coreTypes.Session) error {
+	sessionStartingBlock := session.SessionNumber * session.NumSessionBlocks
+	sessionLastBlock := sessionStartingBlock + session.SessionHeight
+
+	if relayMeta.BlockHeight >= sessionStartingBlock && relayMeta.BlockHeight <= sessionLastBlock {
+		return nil
+	}
+
+	return fmt.Errorf("relay block height %d not within session ID %s starting block %d and last block %d",
+		relayMeta.BlockHeight,
+		session.Id,
+		sessionStartingBlock,
+		sessionLastBlock)
+}
+
 // TECHDEBT: These structures were copied as placeholders from v0 and need to be updated to reflect changes in v1
+// TODO: remove: use coreTypes.Relay instead
 type Relay interface {
 	RelayPayload
 	RelayMeta