fraud: implement fraud service

fraud/interface.go

@@ -2,8 +2,17 @@ package fraud
 
 import (
 	"context"
+
+	logging "github.com/ipfs/go-log/v2"
+
+	"github.com/celestiaorg/celestia-node/header"
 )
 
+var log = logging.Logger("fraud")
+
+// headerFetcher aliases a function that is used to fetch an ExtendedHeader from store by height.
+type headerFetcher func(context.Context, uint64) (*header.ExtendedHeader, error)
+
 // ProofUnmarshaler aliases a function that parses data to `Proof`.
 type ProofUnmarshaler func([]byte) (Proof, error)
 

fraud/service.go

@@ -0,0 +1,130 @@
+package fraud
+
+import (
+	"context"
+	"encoding/hex"
+	"errors"
+	"sync"
+	"time"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+)
+
+// service is responsible for validating and propagating Fraud Proofs.
+// It implements the Service interface.
+type service struct {
+	pubsub *pubsub.PubSub
+	getter headerFetcher
+
+	mu     sync.RWMutex
+	topics map[ProofType]*topic
+}
+
+func NewService(p *pubsub.PubSub, getter headerFetcher) Service {
+	return &service{
+		pubsub: p,
+		getter: getter,
+		topics: make(map[ProofType]*topic),
+	}
+}
+
+func (f *service) Subscribe(proofType ProofType) (Subscription, error) {
+	// TODO: @vgonkivs check if fraud proof is in fraud store, then return with error
+	f.mu.Lock()
+	t, ok := f.topics[proofType]
+	f.mu.Unlock()
+	if !ok {
+		return nil, errors.New("fraud: topic was not created")
+	}
+	return newSubscription(t)
+}
+
+func (f *service) RegisterUnmarshaler(proofType ProofType, u ProofUnmarshaler) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+
+	t, err := createTopic(f.pubsub, proofType, u, f.processIncoming)
+	if err != nil {
+		return err
+	}
+	f.topics[proofType] = t
+	return nil
+}
+
+func (f *service) UnregisterUnmarshaler(proofType ProofType) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	t, ok := f.topics[proofType]
+	if !ok {
+		return errors.New("fraud: topic was not created")
+	}
+	delete(f.topics, proofType)
+	return t.close()
+
+}
+
+func (f *service) Broadcast(ctx context.Context, p Proof) error {
+	bin, err := p.MarshalBinary()
+	if err != nil {
+		return err
+	}
+	f.mu.RLock()
+	t, ok := f.topics[p.Type()]
+	f.mu.RUnlock()
+	if !ok {
+		return errors.New("fraud: topic was not created")
+	}
+	return t.publish(ctx, bin)
+}
+
+func (f *service) processIncoming(
+	ctx context.Context,
+	proofType ProofType,
+	msg *pubsub.Message,
+) pubsub.ValidationResult {
+	f.mu.RLock()
+	t, ok := f.topics[proofType]
+	f.mu.RUnlock()
+	if !ok {
+		log.Error("topic was not created")
+		return pubsub.ValidationReject
+	}
+	proof, err := t.unmarshal(msg.Data)
+	if err != nil {
+		log.Errorw("unmarshaling header error", err)
+		return pubsub.ValidationReject
+	}
+
+	// create a timeout for block fetching since our validator will be called synchronously
+	// and getter is a blocking function.
+	newCtx, cancel := context.WithTimeout(ctx, time.Minute*2)
+	extHeader, err := f.getter(newCtx, proof.Height())
+	if err != nil {
+		cancel()
+		// Timeout means there is a problem with the network.
+		// As we cannot prove or discard Fraud Proof, user must restart the node.
+		if errors.Is(err, context.DeadlineExceeded) {
+			log.Fatal("could not get block. Timeout reached. Please restart your node.")
+		}
+		log.Errorw("failed to fetch block: ",
+			"err", err)
+		return pubsub.ValidationReject
+	}
+	defer cancel()
+	err = proof.Validate(extHeader)
+	if err != nil {
+		log.Errorw("validation err: ",
+			"err", err)
+		f.pubsub.BlacklistPeer(msg.ReceivedFrom)
+		return pubsub.ValidationReject
+	}
+	log.Warnw("received an inbound proof", "kind", proof.Type(),
+		"hash", hex.EncodeToString(extHeader.DAH.Hash()),
+		"from", msg.ReceivedFrom.String(),
+	)
+	return pubsub.ValidationAccept
+}
+
+func getSubTopic(p ProofType) string {
+	return p.String() + "-sub"
+}

fraud/subscription.go

@@ -0,0 +1,38 @@
+package fraud
+
+import (
+	"context"
+
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+)
+
+// subscription handles Fraud Proof from the pubsub topic.
+type subscription struct {
+	subscription *pubsub.Subscription
+	unmarshaler  ProofUnmarshaler
+}
+
+func newSubscription(t *topic) (*subscription, error) {
+	sub, err := t.topic.Subscribe()
+	if err != nil {
+		return nil, err
+	}
+
+	return &subscription{sub, t.unmarshal}, nil
+}
+
+func (s *subscription) Proof(ctx context.Context) (Proof, error) {
+	data, err := s.subscription.Next(ctx)
+	if err != nil {
+		return nil, err
+	}
+	proof, err := s.unmarshaler(data.Data)
+	if err != nil {
+		return nil, err
+	}
+	return proof, nil
+}
+
+func (s *subscription) Cancel() {
+	s.subscription.Cancel()
+}

fraud/topic.go

@@ -0,0 +1,57 @@
+package fraud
+
+import (
+	"context"
+
+	"github.com/libp2p/go-libp2p-core/peer"
+	pubsub "github.com/libp2p/go-libp2p-pubsub"
+)
+
+// topic holds pubsub topic and unmarshaler of corresponded Fraud Proof
+type topic struct {
+	topic     *pubsub.Topic
+	unmarshal ProofUnmarshaler
+}
+
+// publish allows to publish Fraud Proofs to the network
+func (t *topic) publish(ctx context.Context, data []byte) error {
+	return t.topic.Publish(ctx, data)
+}
+
+// close removes unmarshaler and closes pubsub topic
+func (t *topic) close() error {
+	t.unmarshal = nil
+	return t.topic.Close()
+}
+
+func createTopic(
+	p *pubsub.PubSub,
+	proofType ProofType,
+	unmarshal ProofUnmarshaler,
+	val func(context.Context, ProofType, *pubsub.Message) pubsub.ValidationResult) (*topic, error) {
+	t, err := p.Join(getSubTopic(proofType))
+	if err != nil {
+		return nil, err
+	}
+	if err = registerValidator(p, proofType, val); err != nil {
+		return nil, err
+	}
+	log.Debugf("successfully subscribed to topic: %s", getSubTopic(proofType))
+	return &topic{topic: t, unmarshal: unmarshal}, nil
+}
+
+// registerValidator adds an internal validation to topic inside libp2p for provided ProofType.
+func registerValidator(
+	p *pubsub.PubSub,
+	proofType ProofType,
+	val func(context.Context, ProofType, *pubsub.Message) pubsub.ValidationResult,
+) error {
+	return p.RegisterTopicValidator(
+		getSubTopic(proofType),
+		func(ctx context.Context, _ peer.ID, msg *pubsub.Message) pubsub.ValidationResult {
+			return val(ctx, proofType, msg)
+		},
+		// make validation synchronous.
+		pubsub.WithValidatorInline(true),
+	)
+}