adding verifiers to ProofService

fraudserv/service.go

@@ -44,11 +44,13 @@ type ProofService struct {
 	storesLk sync.RWMutex
 	stores   map[fraud.ProofType]datastore.Datastore
 
-	pubsub *pubsub.PubSub
-	host   host.Host
-	getter fraud.HeaderFetcher
-	ds     datastore.Datastore
+	verifiersLk sync.RWMutex
+	verifiers   map[fraud.ProofType]fraud.Verifier
 
+	pubsub        *pubsub.PubSub
+	host          host.Host
+	getter        fraud.HeaderFetcher
+	ds            datastore.Datastore
 	syncerEnabled bool
 }
 
@@ -64,6 +66,7 @@ func NewProofService(
 		pubsub:        p,
 		host:          host,
 		getter:        getter,
+		verifiers:     make(map[fraud.ProofType]fraud.Verifier),
 		topics:        make(map[fraud.ProofType]*pubsub.Topic),
 		stores:        make(map[fraud.ProofType]datastore.Datastore),
 		ds:            ds,
@@ -138,6 +141,16 @@ func (f *ProofService) Broadcast(ctx context.Context, p fraud.Proof) error {
 	return t.Publish(ctx, bin)
 }
 
+func (f *ProofService) AddVerifier(proofType fraud.ProofType, verifier fraud.Verifier) error {
+	f.verifiersLk.Lock()
+	defer f.verifiersLk.Unlock()
+	if _, ok := f.verifiers[proofType]; ok {
+		return fmt.Errorf("verifier for proof type %s already exist", proofType)
+	}
+	f.verifiers[proofType] = verifier
+	return nil
+}
+
 // processIncoming encompasses the logic for validating fraud proofs.
 func (f *ProofService) processIncoming(
 	ctx context.Context,
@@ -181,6 +194,23 @@ func (f *ProofService) processIncoming(
 			"err", err, "proofType", proof.Type(), "height", proof.Height())
 		return pubsub.ValidationIgnore
 	}
+
+	// execute the verifier for proof type if exists
+	f.verifiersLk.RLock()
+	verifier, ok := f.verifiers[proofType]
+	f.verifiersLk.RUnlock()
+	if ok {
+		status, err := verifier(proof)
+		if err != nil {
+			log.Errorw("failed to run the verifier", "err", err, "proofType", proof.Type())
+			return pubsub.ValidationReject
+		}
+		if !status {
+			log.Errorw("invalid fraud proof", "proofType", proof.Type())
+			return pubsub.ValidationReject
+		}
+	}
+
 	// validate the fraud proof.
 	// Peer will be added to black list if the validation fails.
 	err = proof.Validate(extHeader)

fraudserv/service_test.go

@@ -2,6 +2,7 @@ package fraudserv
 
 import (
 	"context"
+	"errors"
 	"testing"
 	"time"
 
@@ -15,6 +16,7 @@ import (
 	"github.com/celestiaorg/go-header"
 	"github.com/celestiaorg/go-header/headertest"
 
+	gofraud "github.com/celestiaorg/go-fraud"
 	"github.com/celestiaorg/go-fraud/fraudtest"
 )
 
@@ -35,6 +37,47 @@ func TestService_SubscribeBroadcastValid(t *testing.T) {
 	require.NoError(t, err)
 }
 
+func TestService_SubscribeBroadcastWithVerifiers(t *testing.T) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	t.Cleanup(cancel)
+
+	serv := newTestService(ctx, t, false)
+	require.NoError(t, serv.Start(ctx))
+
+	fraud := fraudtest.NewValidProof()
+	require.NoError(t, serv.AddVerifier(fraud.Type(), func(fraudProof gofraud.Proof) (bool, error) {
+		return true, nil
+	}))
+
+	// test for error while adding the verifier for the second time
+	require.Error(t, serv.AddVerifier(fraud.Type(), func(fraudProof gofraud.Proof) (bool, error) {
+		return true, nil
+	}))
+	sub, err := serv.Subscribe(fraud.Type())
+	require.NoError(t, err)
+	defer sub.Cancel()
+
+	require.NoError(t, serv.Broadcast(ctx, fraud))
+	_, err = sub.Proof(ctx)
+	require.NoError(t, err)
+
+	// test for invalid fraud proof verifier
+	serv = newTestService(ctx, t, false)
+	require.NoError(t, serv.Start(ctx))
+	require.NoError(t, serv.AddVerifier(fraud.Type(), func(fraudProof gofraud.Proof) (bool, error) {
+		return false, nil
+	}))
+	require.Error(t, serv.Broadcast(ctx, fraud))
+
+	// test for error case of fraud proof verifier
+	serv = newTestService(ctx, t, false)
+	require.NoError(t, serv.Start(ctx))
+	require.NoError(t, serv.AddVerifier(fraud.Type(), func(fraudProof gofraud.Proof) (bool, error) {
+		return true, errors.New("throws error")
+	}))
+	require.Error(t, serv.Broadcast(ctx, fraud))
+}
+
 func TestService_SubscribeBroadcastInvalid(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
 	t.Cleanup(cancel)

interface.go

@@ -9,6 +9,9 @@ import (
 // HeaderFetcher aliases a function that is used to fetch an ExtendedHeader from store by height.
 type HeaderFetcher func(context.Context, uint64) (header.Header, error)
 
+// Verifier is a function that is executed as part of processing the incoming fraud proof
+type Verifier func(fraud Proof) (bool, error)
+
 // ProofUnmarshaler aliases a function that parses data to `Proof`.
 type ProofUnmarshaler func([]byte) (Proof, error)
 
@@ -35,6 +38,10 @@ type Broadcaster interface {
 type Subscriber interface {
 	// Subscribe allows to subscribe on a Proof pub sub topic by its type.
 	Subscribe(ProofType) (Subscription, error)
+	// AddVerifier allows for supplying additional verification logic which
+	// will be run as part of processing the incoming fraud proof.
+	// This only supplements the main validation done by Proof.Validate
+	AddVerifier(ProofType, Verifier) error
 }
 
 // Getter encompasses the behavior to fetch stored fraud proofs.