From b90598fb57395f0fa2594920569e4f007454f30c Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Wed, 25 May 2022 19:05:06 +0300 Subject: [PATCH 1/6] - added PublishWithSk functionality on the Topic implementation --- topic.go | 55 +++++++++++++++++++++++++++++ topic_test.go | 95 ++++++++++++++++++++++++++++++++++++++++++++++++++- 2 files changed, 149 insertions(+), 1 deletion(-) diff --git a/topic.go b/topic.go index 6460782c..9fb4951d 100644 --- a/topic.go +++ b/topic.go @@ -6,6 +6,7 @@ import ( "fmt" "sync" + "github.com/libp2p/go-libp2p-core/crypto" pb "github.com/libp2p/go-libp2p-pubsub/pb" "github.com/libp2p/go-libp2p-core/peer" @@ -14,6 +15,12 @@ import ( // ErrTopicClosed is returned if a Topic is utilized after it has been closed var ErrTopicClosed = errors.New("this Topic is closed, try opening a new one") +// ErrNilSignKey is returned if a nil private key was provided +var ErrNilSignKey = errors.New("nil sign key") + +// ErrEmptyPeerID is returned if an empty peer ID was provided +var ErrEmptyPeerID = errors.New("empty peer ID") + // Topic is the handle for a pubsub topic type Topic struct { p *PubSub @@ -248,6 +255,54 @@ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error return t.p.val.PushLocal(&Message{m, t.p.host.ID(), nil}) } +// PublishWithSk publishes data to topic using the provided secret key and generated peer ID. Providing also the peer ID +// (which can be cached) will improve the performance as it does not require the generation of the public key +// and the resulting peer ID. +// This method can be used as to be able to publish messages on the network without having a "real", connectable host. +func (t *Topic) PublishWithSk(ctx context.Context, data []byte, signKey crypto.PrivKey, pid peer.ID, opts ...PubOpt) error { + if signKey == nil { + return ErrNilSignKey + } + if len(pid) == 0 { + return ErrEmptyPeerID + } + + t.mux.RLock() + defer t.mux.RUnlock() + if t.closed { + return ErrTopicClosed + } + + m := &pb.Message{ + Data: data, + Topic: &t.topic, + From: nil, + Seqno: nil, + } + if t.p.signID != "" { + m.Seqno = t.p.nextSeqno() + } + m.From = []byte(pid) + err := signMessage(pid, signKey, m) + if err != nil { + return err + } + + pub := &PublishOptions{} + for _, opt := range opts { + errOpts := opt(pub) + if errOpts != nil { + return errOpts + } + } + + if pub.ready != nil { + t.p.disc.Bootstrap(ctx, t.topic, pub.ready) + } + + return t.p.val.PushLocal(&Message{m, t.p.host.ID(), nil}) +} + // WithReadiness returns a publishing option for only publishing when the router is ready. // This option is not useful unless PubSub is also using WithDiscovery func WithReadiness(ready RouterReady) PubOpt { diff --git a/topic_test.go b/topic_test.go index 2169d358..267813ca 100644 --- a/topic_test.go +++ b/topic_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/libp2p/go-libp2p-core/peer" + tnet "github.com/libp2p/go-libp2p-testing/net" ) func getTopics(psubs []*PubSub, topicID string, opts ...TopicOpt) []*Topic { @@ -377,7 +378,7 @@ func TestSubscriptionLeaveNotification(t *testing.T) { // Test removing peers and verifying that they cause events subs[1].Cancel() - hosts[2].Close() + _ = hosts[2].Close() psubs[0].BlacklistPeer(hosts[3].ID()) leavingPeers := make(map[peer.ID]struct{}) @@ -780,3 +781,95 @@ func readAllQueuedEvents(ctx context.Context, t *testing.T, evt *TopicEventHandl } return peerState } + +func TestTopic_PublishWithSk(t *testing.T) { + t.Parallel() + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + const topic = "foobar" + const numHosts = 5 + + virtualPeer := tnet.RandPeerNetParamsOrFatal(t) + hosts := getNetHosts(t, ctx, numHosts) + topics := getTopics(getPubsubs(ctx, hosts), topic) + + t.Run("nil sign private key should error", func(t *testing.T) { + err := topics[0].PublishWithSk(ctx, []byte("buff"), nil, virtualPeer.ID) + if err != ErrNilSignKey { + t.Fatal("error should have been of type errNilSignKey") + } + }) + t.Run("empty peer ID should error", func(t *testing.T) { + err := topics[0].PublishWithSk(ctx, []byte("buff"), virtualPeer.PrivKey, "") + if err != ErrEmptyPeerID { + t.Fatal("error should have been of type errEmptyPeerID") + } + }) +} + +func TestTopicRelay_PublishWithSk(t *testing.T) { + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + + const topic = "foobar" + const numHosts = 5 + + virtualPeer := tnet.RandPeerNetParamsOrFatal(t) + hosts := getNetHosts(t, ctx, numHosts) + topics := getTopics(getPubsubs(ctx, hosts), topic) + + // [0.Rel] - [1.Rel] - [2.Sub] + // | + // [3.Rel] - [4.Sub] + + connect(t, hosts[0], hosts[1]) + connect(t, hosts[1], hosts[2]) + connect(t, hosts[1], hosts[3]) + connect(t, hosts[3], hosts[4]) + + time.Sleep(time.Millisecond * 100) + + var subs []*Subscription + + for i, topicValue := range topics { + if i == 2 || i == 4 { + sub, err := topicValue.Subscribe() + if err != nil { + t.Fatal(err) + } + + subs = append(subs, sub) + } else { + _, err := topicValue.Relay() + if err != nil { + t.Fatal(err) + } + } + } + + time.Sleep(time.Millisecond * 100) + + for i := 0; i < 100; i++ { + msg := []byte("message") + + owner := rand.Intn(len(topics)) + + err := topics[owner].PublishWithSk(ctx, msg, virtualPeer.PrivKey, virtualPeer.ID) + if err != nil { + t.Fatal(err) + } + + for _, sub := range subs { + received, errSub := sub.Next(ctx) + if errSub != nil { + t.Fatal(errSub) + } + + if !bytes.Equal(msg, received.Data) { + t.Fatal("received message is other than expected") + } + } + } +} From 3907872ef991cbf9df764407b8c47f63c8325ed1 Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Wed, 25 May 2022 19:10:12 +0300 Subject: [PATCH 2/6] - fixes after merge --- topic.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/topic.go b/topic.go index 669e94d6..bc02f59b 100644 --- a/topic.go +++ b/topic.go @@ -338,7 +338,7 @@ func (t *Topic) PublishWithSk(ctx context.Context, data []byte, signKey crypto.P t.p.disc.Bootstrap(ctx, t.topic, pub.ready) } - return t.p.val.PushLocal(&Message{m, t.p.host.ID(), nil}) + return t.p.val.PushLocal(&Message{m, "", t.p.host.ID(), nil}) } // WithReadiness returns a publishing option for only publishing when the router is ready. From 6cab88ef4704130b9267f341e8f320875ae444ce Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Wed, 25 May 2022 19:48:28 +0300 Subject: [PATCH 3/6] - complete refactor: used publishing options instead of a new function --- topic.go | 101 ++++++++++++++++++++------------------------------ topic_test.go | 13 ++++--- 2 files changed, 49 insertions(+), 65 deletions(-) diff --git a/topic.go b/topic.go index bc02f59b..ecebf2ca 100644 --- a/topic.go +++ b/topic.go @@ -209,8 +209,12 @@ func (t *Topic) Relay() (RelayCancelFunc, error) { // RouterReady is a function that decides if a router is ready to publish type RouterReady func(rt PubSubRouter, topic string) (bool, error) +// ProvideKey is a function that provides a private key and its associated peer ID when publishing a new message +type ProvideKey func() (crypto.PrivKey, peer.ID) + type PublishOptions struct { - ready RouterReady + ready RouterReady + customKey ProvideKey } type PubOpt func(pub *PublishOptions) error @@ -223,6 +227,27 @@ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error return ErrTopicClosed } + pid := t.p.signID + key := t.p.signKey + + pub := &PublishOptions{} + for _, opt := range opts { + err := opt(pub) + if err != nil { + return err + } + } + + if pub.customKey != nil { + key, pid = pub.customKey() + if key == nil { + return ErrNilSignKey + } + if len(pid) == 0 { + return ErrEmptyPeerID + } + } + m := &pb.Message{ Data: data, Topic: &t.topic, @@ -230,20 +255,12 @@ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error Seqno: nil, } if t.p.signID != "" { - m.From = []byte(t.p.signID) + m.From = []byte(pid) m.Seqno = t.p.nextSeqno() } if t.p.signKey != nil { - m.From = []byte(t.p.signID) - err := signMessage(t.p.signID, t.p.signKey, m) - if err != nil { - return err - } - } - - pub := &PublishOptions{} - for _, opt := range opts { - err := opt(pub) + m.From = []byte(pid) + err := signMessage(pid, key, m) if err != nil { return err } @@ -293,54 +310,6 @@ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error return t.p.val.PushLocal(&Message{m, "", t.p.host.ID(), nil}) } -// PublishWithSk publishes data to topic using the provided secret key and generated peer ID. Providing also the peer ID -// (which can be cached) will improve the performance as it does not require the generation of the public key -// and the resulting peer ID. -// This method can be used as to be able to publish messages on the network without having a "real", connectable host. -func (t *Topic) PublishWithSk(ctx context.Context, data []byte, signKey crypto.PrivKey, pid peer.ID, opts ...PubOpt) error { - if signKey == nil { - return ErrNilSignKey - } - if len(pid) == 0 { - return ErrEmptyPeerID - } - - t.mux.RLock() - defer t.mux.RUnlock() - if t.closed { - return ErrTopicClosed - } - - m := &pb.Message{ - Data: data, - Topic: &t.topic, - From: nil, - Seqno: nil, - } - if t.p.signID != "" { - m.Seqno = t.p.nextSeqno() - } - m.From = []byte(pid) - err := signMessage(pid, signKey, m) - if err != nil { - return err - } - - pub := &PublishOptions{} - for _, opt := range opts { - errOpts := opt(pub) - if errOpts != nil { - return errOpts - } - } - - if pub.ready != nil { - t.p.disc.Bootstrap(ctx, t.topic, pub.ready) - } - - return t.p.val.PushLocal(&Message{m, "", t.p.host.ID(), nil}) -} - // WithReadiness returns a publishing option for only publishing when the router is ready. // This option is not useful unless PubSub is also using WithDiscovery func WithReadiness(ready RouterReady) PubOpt { @@ -350,6 +319,18 @@ func WithReadiness(ready RouterReady) PubOpt { } } +// WithSecretKeyAndPeerId returns a publishing option for providing a custom private key and its corresponding peer ID +// This option is useful when we want to send messages from "virtual", never-connectable peers in the network +func WithSecretKeyAndPeerId(key crypto.PrivKey, pid peer.ID) PubOpt { + return func(pub *PublishOptions) error { + pub.customKey = func() (crypto.PrivKey, peer.ID) { + return key, pid + } + + return nil + } +} + // Close closes down the topic. Will return an error unless there are no active event handlers or subscriptions. // Does not error if the topic is already closed. func (t *Topic) Close() error { diff --git a/topic_test.go b/topic_test.go index 84c90218..b8102007 100644 --- a/topic_test.go +++ b/topic_test.go @@ -879,7 +879,7 @@ func TestWithTopicMsgIdFunction(t *testing.T) { })) connectAll(t, hosts) - topicsA := getTopics(pubsubs, topicA) // uses global msgIdFn + topicsA := getTopics(pubsubs, topicA) // uses global msgIdFn topicsB := getTopics(pubsubs, topicB, WithTopicMessageIdFn(func(pmsg *pb.Message) string { // uses custom hash := sha1.Sum(pmsg.Data) return string(hash[:]) @@ -936,20 +936,22 @@ func TestTopic_PublishWithSk(t *testing.T) { topics := getTopics(getPubsubs(ctx, hosts), topic) t.Run("nil sign private key should error", func(t *testing.T) { - err := topics[0].PublishWithSk(ctx, []byte("buff"), nil, virtualPeer.ID) + withVirtualKey := WithSecretKeyAndPeerId(nil, virtualPeer.ID) + err := topics[0].Publish(ctx, []byte("buff"), withVirtualKey) if err != ErrNilSignKey { t.Fatal("error should have been of type errNilSignKey") } }) t.Run("empty peer ID should error", func(t *testing.T) { - err := topics[0].PublishWithSk(ctx, []byte("buff"), virtualPeer.PrivKey, "") + withVirtualKey := WithSecretKeyAndPeerId(virtualPeer.PrivKey, "") + err := topics[0].Publish(ctx, []byte("buff"), withVirtualKey) if err != ErrEmptyPeerID { t.Fatal("error should have been of type errEmptyPeerID") } }) } -func TestTopicRelay_PublishWithSk(t *testing.T) { +func TestTopicRelay_PublishWithKey(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel() @@ -996,7 +998,8 @@ func TestTopicRelay_PublishWithSk(t *testing.T) { owner := rand.Intn(len(topics)) - err := topics[owner].PublishWithSk(ctx, msg, virtualPeer.PrivKey, virtualPeer.ID) + withVirtualKey := WithSecretKeyAndPeerId(virtualPeer.PrivKey, virtualPeer.ID) + err := topics[owner].Publish(ctx, msg, withVirtualKey) if err != nil { t.Fatal(err) } From 433864f69a1a5c7b57bb56b6d1cd0fafcfaecccc Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Wed, 25 May 2022 20:08:25 +0300 Subject: [PATCH 4/6] - added one more check in unit test --- topic_test.go | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/topic_test.go b/topic_test.go index b8102007..2943b327 100644 --- a/topic_test.go +++ b/topic_test.go @@ -922,7 +922,7 @@ func TestWithTopicMsgIdFunction(t *testing.T) { } } -func TestTopic_PublishWithSk(t *testing.T) { +func TestTopic_PublishWithKeyInvalidParameters(t *testing.T) { t.Parallel() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -1013,6 +1013,9 @@ func TestTopicRelay_PublishWithKey(t *testing.T) { if !bytes.Equal(msg, received.Data) { t.Fatal("received message is other than expected") } + if string(received.From) != string(virtualPeer.ID) { + t.Fatal("received message is not from the virtual peer") + } } } } From d31b062f4dce3b89a0cff4b4812e131bafa405fe Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Thu, 26 May 2022 10:52:44 +0300 Subject: [PATCH 5/6] - bugfix: use local variables instead of fields --- topic.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/topic.go b/topic.go index ecebf2ca..901138bf 100644 --- a/topic.go +++ b/topic.go @@ -254,11 +254,11 @@ func (t *Topic) Publish(ctx context.Context, data []byte, opts ...PubOpt) error From: nil, Seqno: nil, } - if t.p.signID != "" { + if pid != "" { m.From = []byte(pid) m.Seqno = t.p.nextSeqno() } - if t.p.signKey != nil { + if key != nil { m.From = []byte(pid) err := signMessage(pid, key, m) if err != nil { From 7a9e99c6db8cedc5b6d53afb4b6aa13f473fc60c Mon Sep 17 00:00:00 2001 From: Iulian Pascalau Date: Thu, 26 May 2022 11:06:07 +0300 Subject: [PATCH 6/6] - renamed tests --- topic_test.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/topic_test.go b/topic_test.go index 2943b327..b87a9b6f 100644 --- a/topic_test.go +++ b/topic_test.go @@ -922,7 +922,7 @@ func TestWithTopicMsgIdFunction(t *testing.T) { } } -func TestTopic_PublishWithKeyInvalidParameters(t *testing.T) { +func TestTopicPublishWithKeyInvalidParameters(t *testing.T) { t.Parallel() ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) @@ -951,7 +951,7 @@ func TestTopic_PublishWithKeyInvalidParameters(t *testing.T) { }) } -func TestTopicRelay_PublishWithKey(t *testing.T) { +func TestTopicRelayPublishWithKey(t *testing.T) { ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) defer cancel()