extract IPNS over pubsub as a ValueStore
And: * Update for DHT changes. * Switch to the new record validation system. License: MIT Signed-off-by: Steven Allen <steven@stebalien.com>
Steven Allen committed
May 3, 2018 at 21:55 UTC
5dc0b7326e91b461a544a2c8890072117cf4bbf0
13 files changed
+220
-926
core/commands/ipnsps.go
+26
-19
@@ -2,15 +2,15 @@ package commands
2
3
import (
4
"errors"
5
- "fmt"
5
"io"
6
"strings"
7
8
cmds "github.com/ipfs/go-ipfs/commands"
9
e "github.com/ipfs/go-ipfs/core/commands/e"
11
- ns "github.com/ipfs/go-ipfs/namesys"
10
13
- "gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit"
11
+ record "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record"
12
+ peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
13
+ cmdkit "gx/ipfs/QmceUdzxkimdYsgtX733uNgzf1DLHyBKN6ehGSp85ayppM/go-ipfs-cmdkit"
14
)
15
16
type ipnsPubsubState struct {
@@ -49,8 +49,7 @@ var ipnspsStateCmd = &cmds.Command{
49
return
50
}
51
52
- _, ok := n.Namesys.GetResolver("pubsub")
53
- res.SetOutput(&ipnsPubsubState{ok})
52
+ res.SetOutput(&ipnsPubsubState{n.PSRouter != nil})
53
},
54
Type: ipnsPubsubState{},
55
Marshalers: cmds.MarshalerMap{
@@ -88,19 +87,26 @@ var ipnspsSubsCmd = &cmds.Command{
87
return
88
}
89
91
- r, ok := n.Namesys.GetResolver("pubsub")
92
- if !ok {
90
+ if n.PSRouter == nil {
91
res.SetError(errors.New("IPNS pubsub subsystem is not enabled"), cmdkit.ErrClient)
92
return
93
}
96
-
97
- psr, ok := r.(*ns.PubsubResolver)
98
- if !ok {
99
- res.SetError(fmt.Errorf("unexpected resolver type: %v", r), cmdkit.ErrNormal)
100
- return
94
+ var paths []string
95
+ for _, key := range n.PSRouter.GetSubscriptions() {
96
+ ns, k, err := record.SplitKey(key)
97
+ if err != nil || ns != "ipns" {
98
+ // Not necessarily an error.
99
+ continue
100
+ }
101
+ pid, err := peer.IDFromString(k)
102
+ if err != nil {
103
+ log.Errorf("ipns key not a valid peer ID: %s", err)
104
+ continue
105
+ }
106
+ paths = append(paths, "/ipns/"+peer.IDB58Encode(pid))
107
}
108
103
- res.SetOutput(&stringList{psr.GetSubscriptions()})
109
+ res.SetOutput(&stringList{paths})
110
},
111
Type: stringList{},
112
Marshalers: cmds.MarshalerMap{
@@ -119,19 +125,20 @@ var ipnspsCancelCmd = &cmds.Command{
125
return
126
}
127
122
- r, ok := n.Namesys.GetResolver("pubsub")
123
- if !ok {
128
+ if n.PSRouter == nil {
129
res.SetError(errors.New("IPNS pubsub subsystem is not enabled"), cmdkit.ErrClient)
130
return
131
}
132
128
- psr, ok := r.(*ns.PubsubResolver)
129
- if !ok {
130
- res.SetError(fmt.Errorf("unexpected resolver type: %v", r), cmdkit.ErrNormal)
133
+ name := req.Arguments()[0]
134
+ name = strings.TrimPrefix(name, "/ipns/")
135
+ pid, err := peer.IDB58Decode(name)
136
+ if err != nil {
137
+ res.SetError(err, cmdkit.ErrClient)
138
return
139
}
140
134
- ok = psr.Cancel(req.Arguments()[0])
141
+ ok := n.PSRouter.Cancel("/ipns/" + string(pid))
142
res.SetOutput(&ipnsPubsubCancel{ok})
143
},
144
Arguments: []cmdkit.Argument{
core/core.go
+55
-29
@@ -48,6 +48,7 @@ import (
48
mamask "gx/ipfs/QmSMZwvs3n4GBikZ7hKzT17c3bk65FmyZo2JqtJ16swqCv/multiaddr-filter"
49
logging "gx/ipfs/QmTG23dvpBCBjqQwyDxV8CQT6jmS4PSftNr1VqHhE3MLy7/go-log"
50
addrutil "gx/ipfs/QmTGSre9j1otFgsr1opCUQDXTPSM6BTZnMWwPeA5nYJM7w/go-addr-util"
51
+ record "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record"
52
routing "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing"
53
floodsub "gx/ipfs/QmVKrsEgixRtMWcMd6WQzuwqCUC3jfLf7Q7xcjnKoMMikS/go-libp2p-floodsub"
54
mssmux "gx/ipfs/QmVniQJkdzLZaZwzwMdd3dJTvWiJ1DQEkreVy6hs6h7Vk5/go-smux-multistream"
@@ -65,13 +66,16 @@ import (
66
peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
67
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
68
dht "gx/ipfs/Qmd3jqhBQFvhfBNTSJMQL15GgyVMpdxKTta69Napvx6Myd/go-libp2p-kad-dht"
69
+ dhtopts "gx/ipfs/Qmd3jqhBQFvhfBNTSJMQL15GgyVMpdxKTta69Napvx6Myd/go-libp2p-kad-dht/opts"
70
ipnet "gx/ipfs/Qmd3oYWVLCVWryDV6Pobv6whZcvDXAHqS3chemZ658y4a8/go-libp2p-interface-pnet"
71
+ psrouter "gx/ipfs/QmdSX2uedxXdsfNSqbkfSxcYi7pXBBdvp1Km2ZsjPWfydt/go-libp2p-pubsub-router"
72
exchange "gx/ipfs/QmdcAXgEHUueP4A7b5hjabKn2EooeHgMreMvFC249dGCgc/go-ipfs-exchange-interface"
73
pstore "gx/ipfs/QmdeiKhUy1TVGBaKxt7y1QmBDLBdisSrLJ1x58Eoj4PXUh/go-libp2p-peerstore"
74
ic "gx/ipfs/Qme1knMqwt1hKZbc1BmQFmnm9f36nyQGwXxPGVpVJ9rMK5/go-libp2p-crypto"
75
ipld "gx/ipfs/Qme5bWv7wtjUNGsK2BNGVUFPKiuxWrsqrtvYwCLRw8YFES/go-ipld-format"
76
ds "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore"
77
mplex "gx/ipfs/QmenmFuirGzv8S1R3DyvbZ6tFmQapkGeDCebgYzni1Ntn3/go-smux-multiplex"
78
+ rhelpers "gx/ipfs/QmeoG1seQ8a9b2vS3XJ8HPz9tXr6dzpS5NizjuDDSntQEk/go-libp2p-routing-helpers"
79
mafilter "gx/ipfs/Qmf2UAmRwDG4TvnkQpHZWPAzw7rpCYVhxmRXmYxXr5LD1g/go-maddr-filter"
80
ifconnmgr "gx/ipfs/QmfQNieWBPwmnUjXWPZbjJPzhNwFFabTb5RQ79dyVWGujQ/go-libp2p-interface-connmgr"
81
p2phost "gx/ipfs/QmfZTdmunzKzAGJrSvXXQbQ5kLLUiEMX5vdwux7iXkdk7D/go-libp2p-host"
@@ -135,6 +139,7 @@ type IpfsNode struct {
139
IpnsRepub *ipnsrp.Republisher
140
141
Floodsub *floodsub.PubSub
142
+ PSRouter *psrouter.PubsubValueStore
143
P2P *p2p.P2P
144
145
proc goprocess.Process
@@ -240,7 +245,7 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
245
return err
246
}
247
243
- if err := n.startOnlineServicesWithHost(ctx, peerhost, routingOption); err != nil {
248
+ if err := n.startOnlineServicesWithHost(ctx, peerhost, routingOption, pubsub, ipnsps); err != nil {
249
return err
250
}
251
@@ -249,21 +254,6 @@ func (n *IpfsNode) startOnlineServices(ctx context.Context, routingOption Routin
254
return err
255
}
256
252
- if pubsub || ipnsps {
253
- service, err := floodsub.NewFloodSub(ctx, peerhost)
254
- if err != nil {
255
- return err
256
- }
257
- n.Floodsub = service
258
- }
259
-
260
- if ipnsps {
261
- err = namesys.AddPubsubNameSystem(ctx, n.Namesys, n.PeerHost, n.Routing, n.Repo.Datastore(), n.Floodsub)
262
- if err != nil {
263
- return err
264
- }
265
- }
266
-
257
n.P2P = p2p.NewP2P(n.Identity, n.PeerHost, n.Peerstore)
258
259
// setup local discovery
@@ -437,17 +427,50 @@ func (n *IpfsNode) HandlePeerFound(p pstore.PeerInfo) {
427
428
// startOnlineServicesWithHost is the set of services which need to be
429
// initialized with the host and _before_ we start listening.
440
-func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost.Host, routingOption RoutingOption) error {
430
+func (n *IpfsNode) startOnlineServicesWithHost(ctx context.Context, host p2phost.Host, routingOption RoutingOption, pubsub bool, ipnsps bool) error {
431
// setup diagnostics service
432
n.Ping = ping.NewPingService(host)
433
434
+ if pubsub || ipnsps {
435
+ service, err := floodsub.NewFloodSub(ctx, host)
436
+ if err != nil {
437
+ return err
438
+ }
439
+ n.Floodsub = service
440
+ }
441
+
442
+ validator := record.NamespacedValidator{
443
+ "pk": record.PublicKeyValidator{},
444
+ "ipns": namesys.IpnsValidator{KeyBook: host.Peerstore()},
445
+ }
446
+
447
// setup routing service
445
- r, err := routingOption(ctx, host, n.Repo.Datastore())
448
+ r, err := routingOption(ctx, host, n.Repo.Datastore(), validator)
449
if err != nil {
450
return err
451
}
452
n.Routing = r
453
454
+ if ipnsps {
455
+ n.PSRouter = psrouter.NewPubsubValueStore(
456
+ ctx,
457
+ host,
458
+ n.Routing,
459
+ n.Floodsub,
460
+ validator,
461
+ )
462
+ n.Routing = rhelpers.Tiered{
463
+ // Always check pubsub first.
464
+ &rhelpers.Compose{
465
+ ValueStore: &rhelpers.LimitedValueStore{
466
+ ValueStore: n.PSRouter,
467
+ Namespaces: []string{"ipns"},
468
+ },
469
+ },
470
+ n.Routing,
471
+ }
472
+ }
473
+
474
// Wrap standard peer host with routing system to allow unknown peer lookups
475
n.PeerHost = rhost.Wrap(host, n.Routing)
476
@@ -935,21 +958,24 @@ func startListening(host p2phost.Host, cfg *config.Config) error {
958
return nil
959
}
960
938
-func constructDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching) (routing.IpfsRouting, error) {
939
- dhtRouting := dht.NewDHT(ctx, host, dstore)
940
- dhtRouting.Validator[IpnsValidatorTag] = namesys.NewIpnsRecordValidator(host.Peerstore())
941
- dhtRouting.Selector[IpnsValidatorTag] = namesys.IpnsSelectorFunc
942
- return dhtRouting, nil
961
+func constructDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
962
+ return dht.New(
963
+ ctx, host,
964
+ dhtopts.Datastore(dstore),
965
+ dhtopts.Validator(validator),
966
+ )
967
}
968
945
-func constructClientDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching) (routing.IpfsRouting, error) {
946
- dhtRouting := dht.NewDHTClient(ctx, host, dstore)
947
- dhtRouting.Validator[IpnsValidatorTag] = namesys.NewIpnsRecordValidator(host.Peerstore())
948
- dhtRouting.Selector[IpnsValidatorTag] = namesys.IpnsSelectorFunc
949
- return dhtRouting, nil
969
+func constructClientDHTRouting(ctx context.Context, host p2phost.Host, dstore ds.Batching, validator record.Validator) (routing.IpfsRouting, error) {
970
+ return dht.New(
971
+ ctx, host,
972
+ dhtopts.Client(true),
973
+ dhtopts.Datastore(dstore),
974
+ dhtopts.Validator(validator),
975
+ )
976
}
977
952
-type RoutingOption func(context.Context, p2phost.Host, ds.Batching) (routing.IpfsRouting, error)
978
+type RoutingOption func(context.Context, p2phost.Host, ds.Batching, record.Validator) (routing.IpfsRouting, error)
979
980
type DiscoveryOption func(context.Context, p2phost.Host) (discovery.Service, error)
981
namesys/interface.go
-8
@@ -61,7 +61,6 @@ var ErrPublishFailed = errors.New("could not publish name")
61
type NameSystem interface {
62
Resolver
63
Publisher
64
- ResolverLookup
64
}
65
66
// Resolver is an object capable of resolving names.
@@ -95,10 +94,3 @@ type Publisher interface {
94
// call once the records spec is implemented
95
PublishWithEOL(ctx context.Context, name ci.PrivKey, value path.Path, eol time.Time) error
96
}
98
-
99
-// ResolverLookup is an object capable of finding resolvers for a subsystem
100
-type ResolverLookup interface {
101
-
102
- // GetResolver retrieves a resolver associated with a subsystem
103
- GetResolver(subs string) (Resolver, bool)
104
-}
namesys/ipns_validate_test.go
+25
-53
@@ -12,8 +12,8 @@ import (
12
u "gx/ipfs/QmNiJuT8Ja3hMVpBHXv3Q6dwmperaQ6JjLtpMQgMCD7xvx/go-ipfs-util"
13
mockrouting "gx/ipfs/QmPuPdzoG4b5uyYSQCjLEHB8NM593m3BW19UHX2jZ6Wzfm/go-ipfs-routing/mock"
14
record "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record"
15
- recordpb "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record/pb"
15
routing "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing"
16
+ ropts "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing/options"
17
testutil "gx/ipfs/QmUJzxQQ2kzwQubsMqBTr1NGDpLfh7pGA2E1oaJULcKDPq/go-testutil"
18
proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
19
peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
@@ -23,8 +23,10 @@ import (
23
dssync "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore/sync"
24
)
25
26
-func testValidatorCase(t *testing.T, priv ci.PrivKey, kbook pstore.KeyBook, ns string, key string, val []byte, eol time.Time, exp error) {
27
- validChecker := NewIpnsRecordValidator(kbook)
26
+func testValidatorCase(t *testing.T, priv ci.PrivKey, kbook pstore.KeyBook, key string, val []byte, eol time.Time, exp error) {
27
+ t.Helper()
28
+
29
+ validator := IpnsValidator{kbook}
30
31
p := path.Path("/ipfs/QmfM2r8seH2GiRaC4esTjeraXEachRt8ZsSeGaWTPLyMoG")
32
entry, err := CreateRoutingEntryData(priv, p, 1, eol)
@@ -39,15 +41,9 @@ func testValidatorCase(t *testing.T, priv ci.PrivKey, kbook pstore.KeyBook, ns s
41
t.Fatal(err)
42
}
43
}
42
- rec := &record.ValidationRecord{
43
- Namespace: ns,
44
- Key: key,
45
- Value: data,
46
- }
47
-
48
- err = validChecker.Func(rec)
44
+ err = validator.Validate(key, data)
45
if err != exp {
50
- params := fmt.Sprintf("namespace: %s\nkey: %s\neol: %s\n", ns, key, eol)
46
+ params := fmt.Sprintf("key: %s\neol: %s\n", key, eol)
47
if exp == nil {
48
t.Fatalf("Unexpected error %s for params %s", err, params)
49
} else if err == nil {
@@ -67,15 +63,15 @@ func TestValidator(t *testing.T) {
63
kbook.AddPubKey(id, priv.GetPublic())
64
emptyKbook := pstore.NewPeerstore()
65
70
- testValidatorCase(t, priv, kbook, "ipns", string(id), nil, ts.Add(time.Hour), nil)
71
- testValidatorCase(t, priv, kbook, "ipns", string(id), nil, ts.Add(time.Hour*-1), ErrExpiredRecord)
72
- testValidatorCase(t, priv, kbook, "ipns", string(id), []byte("bad data"), ts.Add(time.Hour), ErrBadRecord)
73
- testValidatorCase(t, priv, kbook, "ipns", "bad key", nil, ts.Add(time.Hour), ErrKeyFormat)
74
- testValidatorCase(t, priv, emptyKbook, "ipns", string(id), nil, ts.Add(time.Hour), ErrPublicKeyNotFound)
75
- testValidatorCase(t, priv2, kbook, "ipns", string(id2), nil, ts.Add(time.Hour), ErrPublicKeyNotFound)
76
- testValidatorCase(t, priv2, kbook, "ipns", string(id), nil, ts.Add(time.Hour), ErrSignature)
77
- testValidatorCase(t, priv, kbook, "", string(id), nil, ts.Add(time.Hour), ErrInvalidPath)
78
- testValidatorCase(t, priv, kbook, "wrong", string(id), nil, ts.Add(time.Hour), ErrInvalidPath)
66
+ testValidatorCase(t, priv, kbook, "/ipns/"+string(id), nil, ts.Add(time.Hour), nil)
67
+ testValidatorCase(t, priv, kbook, "/ipns/"+string(id), nil, ts.Add(time.Hour*-1), ErrExpiredRecord)
68
+ testValidatorCase(t, priv, kbook, "/ipns/"+string(id), []byte("bad data"), ts.Add(time.Hour), ErrBadRecord)
69
+ testValidatorCase(t, priv, kbook, "/ipns/"+"bad key", nil, ts.Add(time.Hour), ErrKeyFormat)
70
+ testValidatorCase(t, priv, emptyKbook, "/ipns/"+string(id), nil, ts.Add(time.Hour), ErrPublicKeyNotFound)
71
+ testValidatorCase(t, priv2, kbook, "/ipns/"+string(id2), nil, ts.Add(time.Hour), ErrPublicKeyNotFound)
72
+ testValidatorCase(t, priv2, kbook, "/ipns/"+string(id), nil, ts.Add(time.Hour), ErrSignature)
73
+ testValidatorCase(t, priv, kbook, "//"+string(id), nil, ts.Add(time.Hour), ErrInvalidPath)
74
+ testValidatorCase(t, priv, kbook, "/wrong/"+string(id), nil, ts.Add(time.Hour), ErrInvalidPath)
75
}
76
77
func TestResolverValidation(t *testing.T) {
@@ -85,13 +81,6 @@ func TestResolverValidation(t *testing.T) {
81
peerstore := pstore.NewPeerstore()
82
83
vstore := newMockValueStore(rid, dstore, peerstore)
88
- vstore.Validator["ipns"] = NewIpnsRecordValidator(peerstore)
89
- vstore.Validator["pk"] = &record.ValidChecker{
90
- Func: func(r *record.ValidationRecord) error {
91
- return nil
92
- },
93
- Sign: false,
94
- }
84
resolver := NewRoutingResolver(vstore, 0)
85
86
// Create entry with expiry in one hour
@@ -224,19 +213,19 @@ type mockValueStore struct {
213
func newMockValueStore(id testutil.Identity, dstore ds.Datastore, kbook pstore.KeyBook) *mockValueStore {
214
serv := mockrouting.NewServer()
215
r := serv.ClientWithDatastore(context.Background(), id, dstore)
227
- return &mockValueStore{r, kbook, make(record.Validator)}
216
+ return &mockValueStore{r, kbook, record.NamespacedValidator{
217
+ "ipns": IpnsValidator{kbook},
218
+ "pk": record.PublicKeyValidator{},
219
+ }}
220
}
221
230
-func (m *mockValueStore) GetValue(ctx context.Context, k string) ([]byte, error) {
231
- data, err := m.r.GetValue(ctx, k)
222
+func (m *mockValueStore) GetValue(ctx context.Context, k string, opts ...ropts.Option) ([]byte, error) {
223
+ data, err := m.r.GetValue(ctx, k, opts...)
224
if err != nil {
225
return data, err
226
}
227
236
- rec := new(recordpb.Record)
237
- rec.Key = proto.String(k)
238
- rec.Value = data
239
- if err = m.Validator.VerifyRecord(rec); err != nil {
228
+ if err = m.Validator.Validate(k, data); err != nil {
229
return nil, err
230
}
231
@@ -263,23 +252,6 @@ func (m *mockValueStore) GetPublicKey(ctx context.Context, p peer.ID) (ci.PubKey
252
return pk, m.kbook.AddPubKey(p, pk)
253
}
254
266
-func (m *mockValueStore) GetValues(ctx context.Context, k string, count int) ([]routing.RecvdVal, error) {
267
- vals, err := m.r.GetValues(ctx, k, count)
268
- if err != nil {
269
- return nil, err
270
- }
271
- valid := make([]routing.RecvdVal, 0, len(vals))
272
- for _, v := range vals {
273
- rec := new(recordpb.Record)
274
- rec.Key = proto.String(k)
275
- rec.Value = v.Val
276
- if err = m.Validator.VerifyRecord(rec); err == nil {
277
- valid = append(valid, v)
278
- }
279
- }
280
- return valid, nil
281
-}
282
-
283
-func (m *mockValueStore) PutValue(ctx context.Context, k string, d []byte) error {
284
- return m.r.PutValue(ctx, k, d)
255
+func (m *mockValueStore) PutValue(ctx context.Context, k string, d []byte, opts ...ropts.Option) error {
256
+ return m.r.PutValue(ctx, k, d, opts...)
257
}
namesys/namesys.go
-33
@@ -2,7 +2,6 @@ package namesys
2
3
import (
4
"context"
5
- "errors"
5
"strings"
6
"sync"
7
"time"
@@ -11,13 +10,11 @@ import (
10
path "github.com/ipfs/go-ipfs/path"
11
12
routing "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing"
14
- floodsub "gx/ipfs/QmVKrsEgixRtMWcMd6WQzuwqCUC3jfLf7Q7xcjnKoMMikS/go-libp2p-floodsub"
13
isd "gx/ipfs/QmZmmuAXgX73UQmX1jRKjTGmjzq24Jinqkq8vzkBtno4uX/go-is-domain"
14
mh "gx/ipfs/QmZyZDi491cCNTLfAhwcaDii2Kg4pwKRkhqQzURGDvY6ua/go-multihash"
15
peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
16
ci "gx/ipfs/Qme1knMqwt1hKZbc1BmQFmnm9f36nyQGwXxPGVpVJ9rMK5/go-libp2p-crypto"
17
ds "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore"
20
- p2phost "gx/ipfs/QmfZTdmunzKzAGJrSvXXQbQ5kLLUiEMX5vdwux7iXkdk7D/go-libp2p-host"
18
)
19
20
// mpns (a multi-protocol NameSystem) implements generic IPFS naming.
@@ -48,23 +45,6 @@ func NewNameSystem(r routing.ValueStore, ds ds.Datastore, cachesize int) NameSys
45
}
46
}
47
51
-// AddPubsubNameSystem adds the pubsub publisher and resolver to the namesystem
52
-func AddPubsubNameSystem(ctx context.Context, ns NameSystem, host p2phost.Host, r routing.IpfsRouting, ds ds.Datastore, ps *floodsub.PubSub) error {
53
- mpns, ok := ns.(*mpns)
54
- if !ok {
55
- return errors.New("unexpected NameSystem; not an mpns instance")
56
- }
57
-
58
- pkf, ok := r.(routing.PubKeyFetcher)
59
- if !ok {
60
- return errors.New("unexpected IpfsRouting; not a PubKeyFetcher instance")
61
- }
62
-
63
- mpns.resolvers["pubsub"] = NewPubsubResolver(ctx, host, r, pkf, ps)
64
- mpns.publishers["pubsub"] = NewPubsubPublisher(ctx, host, ds, r, ps)
65
- return nil
66
-}
67
-
48
const DefaultResolverCacheTTL = time.Minute
49
50
// Resolve implements Resolver.
@@ -219,16 +199,3 @@ func (ns *mpns) addToDHTCache(key ci.PrivKey, value path.Path, eol time.Time) {
199
eol: eol,
200
})
201
}
222
-
223
-// GetResolver implements ResolverLookup
224
-func (ns *mpns) GetResolver(subs string) (Resolver, bool) {
225
- res, ok := ns.resolvers[subs]
226
- if ok {
227
- ires, ok := res.(Resolver)
228
- if ok {
229
- return ires, true
230
- }
231
- }
232
-
233
- return nil, false
234
-}
namesys/publisher.go
+1
-1
@@ -142,7 +142,7 @@ func PutRecordToRouting(ctx context.Context, k ci.PrivKey, value path.Path, seqn
142
errs := make(chan error, 2) // At most two errors (IPNS, and public key)
143
144
// Attempt to extract the public key from the ID
145
- extractedPublicKey := id.ExtractPublicKey()
145
+ extractedPublicKey, _ := id.ExtractPublicKey()
146
147
go func() {
148
errs <- PublishEntry(ctx, r, ipnskey, entry)
namesys/publisher_test.go
+1
-8
@@ -49,14 +49,7 @@ func testNamekeyPublisher(t *testing.T, keyType int, expectedErr error, expected
49
}
50
51
// ID
52
- var id peer.ID
53
- switch keyType {
54
- case ci.Ed25519:
55
- id, err = peer.IDFromEd25519PublicKey(pubKey)
56
- default:
57
- id, err = peer.IDFromPublicKey(pubKey)
58
- }
59
-
52
+ id, err := peer.IDFromPublicKey(pubKey)
53
if err != nil {
54
t.Fatal(err)
55
}
namesys/pubsub.go
deleted
-426
@@ -1,426 +0,0 @@
1
-package namesys
2
-
3
-import (
4
- "context"
5
- "errors"
6
- "fmt"
7
- "strings"
8
- "sync"
9
- "time"
10
-
11
- opts "github.com/ipfs/go-ipfs/namesys/opts"
12
- pb "github.com/ipfs/go-ipfs/namesys/pb"
13
- path "github.com/ipfs/go-ipfs/path"
14
-
15
- u "gx/ipfs/QmNiJuT8Ja3hMVpBHXv3Q6dwmperaQ6JjLtpMQgMCD7xvx/go-ipfs-util"
16
- record "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record"
17
- dhtpb "gx/ipfs/QmTUyK82BVPA6LmSzEJpfEunk9uBaQzWtMsNP917tVj4sT/go-libp2p-record/pb"
18
- routing "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing"
19
- floodsub "gx/ipfs/QmVKrsEgixRtMWcMd6WQzuwqCUC3jfLf7Q7xcjnKoMMikS/go-libp2p-floodsub"
20
- dshelp "gx/ipfs/QmYJgz1Z5PbBGP7n2XA8uv5sF1EKLfYUjL7kFemVAjMNqC/go-ipfs-ds-help"
21
- proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
22
- mh "gx/ipfs/QmZyZDi491cCNTLfAhwcaDii2Kg4pwKRkhqQzURGDvY6ua/go-multihash"
23
- peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
24
- cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
25
- pstore "gx/ipfs/QmdeiKhUy1TVGBaKxt7y1QmBDLBdisSrLJ1x58Eoj4PXUh/go-libp2p-peerstore"
26
- ci "gx/ipfs/Qme1knMqwt1hKZbc1BmQFmnm9f36nyQGwXxPGVpVJ9rMK5/go-libp2p-crypto"
27
- ds "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore"
28
- dssync "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore/sync"
29
- p2phost "gx/ipfs/QmfZTdmunzKzAGJrSvXXQbQ5kLLUiEMX5vdwux7iXkdk7D/go-libp2p-host"
30
-)
31
-
32
-// PubsubPublisher is a publisher that distributes IPNS records through pubsub
33
-type PubsubPublisher struct {
34
- ctx context.Context
35
- ds ds.Datastore
36
- host p2phost.Host
37
- cr routing.ContentRouting
38
- ps *floodsub.PubSub
39
-
40
- mx sync.Mutex
41
- subs map[string]struct{}
42
-}
43
-
44
-// PubsubResolver is a resolver that receives IPNS records through pubsub
45
-type PubsubResolver struct {
46
- ctx context.Context
47
- ds ds.Datastore
48
- host p2phost.Host
49
- cr routing.ContentRouting
50
- pkf routing.PubKeyFetcher
51
- ps *floodsub.PubSub
52
-
53
- mx sync.Mutex
54
- subs map[string]*floodsub.Subscription
55
-}
56
-
57
-// NewPubsubPublisher constructs a new Publisher that publishes IPNS records through pubsub.
58
-// The constructor interface is complicated by the need to bootstrap the pubsub topic.
59
-// This could be greatly simplified if the pubsub implementation handled bootstrap itself
60
-func NewPubsubPublisher(ctx context.Context, host p2phost.Host, ds ds.Datastore, cr routing.ContentRouting, ps *floodsub.PubSub) *PubsubPublisher {
61
- return &PubsubPublisher{
62
- ctx: ctx,
63
- ds: ds,
64
- host: host, // needed for pubsub bootstrap
65
- cr: cr, // needed for pubsub bootstrap
66
- ps: ps,
67
- subs: make(map[string]struct{}),
68
- }
69
-}
70
-
71
-// NewPubsubResolver constructs a new Resolver that resolves IPNS records through pubsub.
72
-// same as above for pubsub bootstrap dependencies
73
-func NewPubsubResolver(ctx context.Context, host p2phost.Host, cr routing.ContentRouting, pkf routing.PubKeyFetcher, ps *floodsub.PubSub) *PubsubResolver {
74
- return &PubsubResolver{
75
- ctx: ctx,
76
- ds: dssync.MutexWrap(ds.NewMapDatastore()),
77
- host: host, // needed for pubsub bootstrap
78
- cr: cr, // needed for pubsub bootstrap
79
- pkf: pkf,
80
- ps: ps,
81
- subs: make(map[string]*floodsub.Subscription),
82
- }
83
-}
84
-
85
-// Publish publishes an IPNS record through pubsub with default TTL
86
-func (p *PubsubPublisher) Publish(ctx context.Context, k ci.PrivKey, value path.Path) error {
87
- return p.PublishWithEOL(ctx, k, value, time.Now().Add(DefaultRecordTTL))
88
-}
89
-
90
-// PublishWithEOL publishes an IPNS record through pubsub
91
-func (p *PubsubPublisher) PublishWithEOL(ctx context.Context, k ci.PrivKey, value path.Path, eol time.Time) error {
92
- id, err := peer.IDFromPrivateKey(k)
93
- if err != nil {
94
- return err
95
- }
96
-
97
- _, ipnskey := IpnsKeysForID(id)
98
-
99
- seqno, err := p.getPreviousSeqNo(ctx, ipnskey)
100
- if err != nil {
101
- return err
102
- }
103
-
104
- seqno++
105
-
106
- return p.publishRecord(ctx, k, value, seqno, eol, ipnskey, id)
107
-}
108
-
109
-func (p *PubsubPublisher) getPreviousSeqNo(ctx context.Context, ipnskey string) (uint64, error) {
110
- // the datastore is shared with the routing publisher to properly increment and persist
111
- // ipns record sequence numbers.
112
- prevrec, err := p.ds.Get(dshelp.NewKeyFromBinary([]byte(ipnskey)))
113
- if err != nil {
114
- if err == ds.ErrNotFound {
115
- // None found, lets start at zero!
116
- return 0, nil
117
- }
118
- return 0, err
119
- }
120
-
121
- prbytes, ok := prevrec.([]byte)
122
- if !ok {
123
- return 0, fmt.Errorf("unexpected type returned from datastore: %#v", prevrec)
124
- }
125
-
126
- var dsrec dhtpb.Record
127
- err = proto.Unmarshal(prbytes, &dsrec)
128
- if err != nil {
129
- return 0, err
130
- }
131
-
132
- var entry pb.IpnsEntry
133
- err = proto.Unmarshal(dsrec.GetValue(), &entry)
134
- if err != nil {
135
- return 0, err
136
- }
137
-
138
- return entry.GetSequence(), nil
139
-}
140
-
141
-func (p *PubsubPublisher) publishRecord(ctx context.Context, k ci.PrivKey, value path.Path, seqno uint64, eol time.Time, ipnskey string, ID peer.ID) error {
142
- entry, err := CreateRoutingEntryData(k, value, seqno, eol)
143
- if err != nil {
144
- return err
145
- }
146
-
147
- data, err := proto.Marshal(entry)
148
- if err != nil {
149
- return err
150
- }
151
-
152
- // the datastore is shared with the routing publisher to properly increment and persist
153
- // ipns record sequence numbers; so we need to Record our new entry in the datastore
154
- dsrec, err := record.MakePutRecord(k, ipnskey, data, true)
155
- if err != nil {
156
- return err
157
- }
158
-
159
- dsdata, err := proto.Marshal(dsrec)
160
- if err != nil {
161
- return err
162
- }
163
-
164
- err = p.ds.Put(dshelp.NewKeyFromBinary([]byte(ipnskey)), dsdata)
165
- if err != nil {
166
- return err
167
- }
168
-
169
- // now we publish, but we also need to bootstrap pubsub for our messages to propagate
170
- topic := "/ipns/" + ID.Pretty()
171
-
172
- p.mx.Lock()
173
- _, ok := p.subs[topic]
174
-
175
- if !ok {
176
- p.subs[topic] = struct{}{}
177
- p.mx.Unlock()
178
-
179
- bootstrapPubsub(p.ctx, p.cr, p.host, topic)
180
- } else {
181
- p.mx.Unlock()
182
- }
183
-
184
- log.Debugf("PubsubPublish: publish IPNS record for %s (%d)", topic, seqno)
185
- return p.ps.Publish(topic, data)
186
-}
187
-
188
-// Resolve resolves a name through pubsub and default depth limit
189
-func (r *PubsubResolver) Resolve(ctx context.Context, name string, options ...opts.ResolveOpt) (path.Path, error) {
190
- return resolve(ctx, r, name, opts.ProcessOpts(options), "/ipns/")
191
-}
192
-
193
-func (r *PubsubResolver) resolveOnce(ctx context.Context, name string, options *opts.ResolveOpts) (path.Path, error) {
194
- log.Debugf("PubsubResolve: resolve '%s'", name)
195
-
196
- // retrieve the public key once (for verifying messages)
197
- xname := strings.TrimPrefix(name, "/ipns/")
198
- hash, err := mh.FromB58String(xname)
199
- if err != nil {
200
- log.Warningf("PubsubResolve: bad input hash: [%s]", xname)
201
- return "", err
202
- }
203
-
204
- id := peer.ID(hash)
205
- if r.host.Peerstore().PrivKey(id) != nil {
206
- return "", errors.New("cannot resolve own name through pubsub")
207
- }
208
-
209
- pubk := id.ExtractPublicKey()
210
- if pubk == nil {
211
- pubk, err = r.pkf.GetPublicKey(ctx, id)
212
- if err != nil {
213
- log.Warningf("PubsubResolve: error fetching public key: %s [%s]", err.Error(), xname)
214
- return "", err
215
- }
216
- }
217
-
218
- // the topic is /ipns/Qmhash
219
- if !strings.HasPrefix(name, "/ipns/") {
220
- name = "/ipns/" + name
221
- }
222
-
223
- r.mx.Lock()
224
- // see if we already have a pubsub subscription; if not, subscribe
225
- sub, ok := r.subs[name]
226
- if !ok {
227
- sub, err = r.ps.Subscribe(name)
228
- if err != nil {
229
- r.mx.Unlock()
230
- return "", err
231
- }
232
-
233
- log.Debugf("PubsubResolve: subscribed to %s", name)
234
-
235
- r.subs[name] = sub
236
-
237
- ctx, cancel := context.WithCancel(r.ctx)
238
- go r.handleSubscription(sub, name, pubk, cancel)
239
- go bootstrapPubsub(ctx, r.cr, r.host, name)
240
- }
241
- r.mx.Unlock()
242
-
243
- // resolve to what we may already have in the datastore
244
- dsval, err := r.ds.Get(dshelp.NewKeyFromBinary([]byte(name)))
245
- if err != nil {
246
- if err == ds.ErrNotFound {
247
- return "", ErrResolveFailed
248
- }
249
- return "", err
250
- }
251
-
252
- data := dsval.([]byte)
253
- entry := new(pb.IpnsEntry)
254
-
255
- err = proto.Unmarshal(data, entry)
256
- if err != nil {
257
- return "", err
258
- }
259
-
260
- // check EOL; if the entry has expired, delete from datastore and return ds.ErrNotFound
261
- eol, ok := checkEOL(entry)
262
- if ok && eol.Before(time.Now()) {
263
- err = r.ds.Delete(dshelp.NewKeyFromBinary([]byte(name)))
264
- if err != nil {
265
- log.Warningf("PubsubResolve: error deleting stale value for %s: %s", name, err.Error())
266
- }
267
-
268
- return "", ErrResolveFailed
269
- }
270
-
271
- value, err := path.ParsePath(string(entry.GetValue()))
272
- return value, err
273
-}
274
-
275
-// GetSubscriptions retrieves a list of active topic subscriptions
276
-func (r *PubsubResolver) GetSubscriptions() []string {
277
- r.mx.Lock()
278
- defer r.mx.Unlock()
279
-
280
- var res []string
281
- for sub := range r.subs {
282
- res = append(res, sub)
283
- }
284
-
285
- return res
286
-}
287
-
288
-// Cancel cancels a topic subscription; returns true if an active
289
-// subscription was canceled
290
-func (r *PubsubResolver) Cancel(name string) bool {
291
- r.mx.Lock()
292
- defer r.mx.Unlock()
293
-
294
- sub, ok := r.subs[name]
295
- if ok {
296
- sub.Cancel()
297
- delete(r.subs, name)
298
- }
299
-
300
- return ok
301
-}
302
-
303
-func (r *PubsubResolver) handleSubscription(sub *floodsub.Subscription, name string, pubk ci.PubKey, cancel func()) {
304
- defer sub.Cancel()
305
- defer cancel()
306
-
307
- for {
308
- msg, err := sub.Next(r.ctx)
309
- if err != nil {
310
- if err != context.Canceled {
311
- log.Warningf("PubsubResolve: subscription error in %s: %s", name, err.Error())
312
- }
313
- return
314
- }
315
-
316
- err = r.receive(msg, name, pubk)
317
- if err != nil {
318
- log.Warningf("PubsubResolve: error processing update for %s: %s", name, err.Error())
319
- }
320
- }
321
-}
322
-
323
-func (r *PubsubResolver) receive(msg *floodsub.Message, name string, pubk ci.PubKey) error {
324
- data := msg.GetData()
325
- if data == nil {
326
- return errors.New("empty message")
327
- }
328
-
329
- entry := new(pb.IpnsEntry)
330
- err := proto.Unmarshal(data, entry)
331
- if err != nil {
332
- return err
333
- }
334
-
335
- ok, err := pubk.Verify(ipnsEntryDataForSig(entry), entry.GetSignature())
336
- if err != nil || !ok {
337
- return errors.New("signature verification failed")
338
- }
339
-
340
- _, err = path.ParsePath(string(entry.GetValue()))
341
- if err != nil {
342
- return err
343
- }
344
-
345
- eol, ok := checkEOL(entry)
346
- if ok && eol.Before(time.Now()) {
347
- return errors.New("stale update; EOL exceeded")
348
- }
349
-
350
- // check the sequence number against what we may already have in our datastore
351
- oval, err := r.ds.Get(dshelp.NewKeyFromBinary([]byte(name)))
352
- if err == nil {
353
- odata := oval.([]byte)
354
- oentry := new(pb.IpnsEntry)
355
-
356
- err = proto.Unmarshal(odata, oentry)
357
- if err != nil {
358
- return err
359
- }
360
-
361
- if entry.GetSequence() <= oentry.GetSequence() {
362
- return errors.New("stale update; sequence number too small")
363
- }
364
- }
365
-
366
- log.Debugf("PubsubResolve: receive IPNS record for %s", name)
367
-
368
- return r.ds.Put(dshelp.NewKeyFromBinary([]byte(name)), data)
369
-}
370
-
371
-// rendezvous with peers in the name topic through provider records
372
-// Note: rendezvous/boostrap should really be handled by the pubsub implementation itself!
373
-func bootstrapPubsub(ctx context.Context, cr routing.ContentRouting, host p2phost.Host, name string) {
374
- topic := "floodsub:" + name
375
- hash := u.Hash([]byte(topic))
376
- rz := cid.NewCidV1(cid.Raw, hash)
377
-
378
- err := cr.Provide(ctx, rz, true)
379
- if err != nil {
380
- log.Warningf("bootstrapPubsub: error providing rendezvous for %s: %s", topic, err.Error())
381
- }
382
-
383
- go func() {
384
- for {
385
- select {
386
- case <-time.After(8 * time.Hour):
387
- err := cr.Provide(ctx, rz, true)
388
- if err != nil {
389
- log.Warningf("bootstrapPubsub: error providing rendezvous for %s: %s", topic, err.Error())
390
- }
391
- case <-ctx.Done():
392
- return
393
- }
394
- }
395
- }()
396
-
397
- rzctx, cancel := context.WithTimeout(ctx, time.Second*10)
398
- defer cancel()
399
-
400
- wg := &sync.WaitGroup{}
401
- for pi := range cr.FindProvidersAsync(rzctx, rz, 10) {
402
- if pi.ID == host.ID() {
403
- continue
404
- }
405
- wg.Add(1)
406
- go func(pi pstore.PeerInfo) {
407
- defer wg.Done()
408
-
409
- ctx, cancel := context.WithTimeout(ctx, time.Second*10)
410
- defer cancel()
411
-
412
- err := host.Connect(ctx, pi)
413
- if err != nil {
414
- log.Debugf("Error connecting to pubsub peer %s: %s", pi.ID, err.Error())
415
- return
416
- }
417
-
418
- // delay to let pubsub perform its handshake
419
- time.Sleep(time.Millisecond * 250)
420
-
421
- log.Debugf("Connected to pubsub peer %s", pi.ID)
422
- }(pi)
423
- }
424
-
425
- wg.Wait()
426
-}
namesys/pubsub_test.go
deleted
-197
@@ -1,197 +0,0 @@
1
-package namesys
2
-
3
-import (
4
- "context"
5
- "sync"
6
- "testing"
7
- "time"
8
-
9
- path "github.com/ipfs/go-ipfs/path"
10
-
11
- mockrouting "gx/ipfs/QmPuPdzoG4b5uyYSQCjLEHB8NM593m3BW19UHX2jZ6Wzfm/go-ipfs-routing/mock"
12
- routing "gx/ipfs/QmUHRKTeaoASDvDj7cTAXsmjAY7KQ13ErtzkQHZQq6uFUz/go-libp2p-routing"
13
- testutil "gx/ipfs/QmUJzxQQ2kzwQubsMqBTr1NGDpLfh7pGA2E1oaJULcKDPq/go-testutil"
14
- floodsub "gx/ipfs/QmVKrsEgixRtMWcMd6WQzuwqCUC3jfLf7Q7xcjnKoMMikS/go-libp2p-floodsub"
15
- netutil "gx/ipfs/Qmb6BsZf6Y3kxffXMNTubGPF1w1bkHtpvhfYbmnwP3NQyw/go-libp2p-netutil"
16
- bhost "gx/ipfs/Qmc64U41EEB4nPG7wxjEqFwKJajS2f8kk5q2TvUrQf78Xu/go-libp2p-blankhost"
17
- peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
18
- pstore "gx/ipfs/QmdeiKhUy1TVGBaKxt7y1QmBDLBdisSrLJ1x58Eoj4PXUh/go-libp2p-peerstore"
19
- ci "gx/ipfs/Qme1knMqwt1hKZbc1BmQFmnm9f36nyQGwXxPGVpVJ9rMK5/go-libp2p-crypto"
20
- ds "gx/ipfs/QmeiCcJfDW1GJnWUArudsv5rQsihpi4oyddPhdqo3CfX6i/go-datastore"
21
- p2phost "gx/ipfs/QmfZTdmunzKzAGJrSvXXQbQ5kLLUiEMX5vdwux7iXkdk7D/go-libp2p-host"
22
-)
23
-
24
-func newNetHost(ctx context.Context, t *testing.T) p2phost.Host {
25
- netw := netutil.GenSwarmNetwork(t, ctx)
26
- return bhost.NewBlankHost(netw)
27
-}
28
-
29
-func newNetHosts(ctx context.Context, t *testing.T, n int) []p2phost.Host {
30
- var out []p2phost.Host
31
-
32
- for i := 0; i < n; i++ {
33
- h := newNetHost(ctx, t)
34
- out = append(out, h)
35
- }
36
-
37
- return out
38
-}
39
-
40
-// PubKeyFetcher implementation with a global key store
41
-type mockKeyStore struct {
42
- keys map[peer.ID]ci.PubKey
43
- mx sync.Mutex
44
-}
45
-
46
-func (m *mockKeyStore) addPubKey(id peer.ID, pkey ci.PubKey) {
47
- m.mx.Lock()
48
- defer m.mx.Unlock()
49
- m.keys[id] = pkey
50
-}
51
-
52
-func (m *mockKeyStore) getPubKey(id peer.ID) (ci.PubKey, error) {
53
- m.mx.Lock()
54
- defer m.mx.Unlock()
55
- pkey, ok := m.keys[id]
56
- if ok {
57
- return pkey, nil
58
- }
59
-
60
- return nil, routing.ErrNotFound
61
-}
62
-
63
-func (m *mockKeyStore) GetPublicKey(ctx context.Context, id peer.ID) (ci.PubKey, error) {
64
- return m.getPubKey(id)
65
-}
66
-
67
-func newMockKeyStore() *mockKeyStore {
68
- return &mockKeyStore{
69
- keys: make(map[peer.ID]ci.PubKey),
70
- }
71
-}
72
-
73
-// ConentRouting mock
74
-func newMockRouting(ms mockrouting.Server, ks *mockKeyStore, host p2phost.Host) routing.ContentRouting {
75
- id := host.ID()
76
-
77
- privk := host.Peerstore().PrivKey(id)
78
- pubk := host.Peerstore().PubKey(id)
79
- pi := host.Peerstore().PeerInfo(id)
80
-
81
- ks.addPubKey(id, pubk)
82
- return ms.Client(testutil.NewIdentity(id, pi.Addrs[0], privk, pubk))
83
-}
84
-
85
-func newMockRoutingForHosts(ms mockrouting.Server, ks *mockKeyStore, hosts []p2phost.Host) []routing.ContentRouting {
86
- rs := make([]routing.ContentRouting, len(hosts))
87
- for i := 0; i < len(hosts); i++ {
88
- rs[i] = newMockRouting(ms, ks, hosts[i])
89
- }
90
- return rs
91
-}
92
-
93
-// tests
94
-func TestPubsubPublishSubscribe(t *testing.T) {
95
- ctx, cancel := context.WithCancel(context.Background())
96
- defer cancel()
97
-
98
- ms := mockrouting.NewServer()
99
- ks := newMockKeyStore()
100
-
101
- pubhost := newNetHost(ctx, t)
102
- pubmr := newMockRouting(ms, ks, pubhost)
103
- fs, err := floodsub.NewFloodSub(ctx, pubhost)
104
- if err != nil {
105
- t.Fatal(err)
106
- }
107
- pub := NewPubsubPublisher(ctx, pubhost, ds.NewMapDatastore(), pubmr, fs)
108
- privk := pubhost.Peerstore().PrivKey(pubhost.ID())
109
- pubpinfo := pstore.PeerInfo{ID: pubhost.ID(), Addrs: pubhost.Addrs()}
110
-
111
- name := "/ipns/" + pubhost.ID().Pretty()
112
-
113
- reshosts := newNetHosts(ctx, t, 5)
114
- resmrs := newMockRoutingForHosts(ms, ks, reshosts)
115
- res := make([]*PubsubResolver, len(reshosts))
116
- for i := 0; i < len(res); i++ {
117
-
118
- fs, err := floodsub.NewFloodSub(ctx, reshosts[i])
119
- if err != nil {
120
- t.Fatal(err)
121
- }
122
-
123
- res[i] = NewPubsubResolver(ctx, reshosts[i], resmrs[i], ks, fs)
124
- if err := reshosts[i].Connect(ctx, pubpinfo); err != nil {
125
- t.Fatal(err)
126
- }
127
- }
128
-
129
- time.Sleep(time.Millisecond * 100)
130
- for i := 0; i < len(res); i++ {
131
- checkResolveNotFound(ctx, t, i, res[i], name)
132
- // delay to avoid connection storms
133
- time.Sleep(time.Millisecond * 100)
134
- }
135
-
136
- // let the bootstrap finish
137
- time.Sleep(time.Second * 1)
138
-
139
- val := path.Path("/ipfs/QmP1DfoUjiWH2ZBo1PBH6FupdBucbDepx3HpWmEY6JMUpY")
140
- err = pub.Publish(ctx, privk, val)
141
- if err != nil {
142
- t.Fatal(err)
143
- }
144
-
145
- // let the flood propagate
146
- time.Sleep(time.Second * 1)
147
- for i := 0; i < len(res); i++ {
148
- checkResolve(ctx, t, i, res[i], name, val)
149
- }
150
-
151
- val = path.Path("/ipfs/QmP1wMAqk6aZYRZirbaAwmrNeqFRgQrwBt3orUtvSa1UYD")
152
- err = pub.Publish(ctx, privk, val)
153
- if err != nil {
154
- t.Fatal(err)
155
- }
156
-
157
- // let the flood propagate
158
- time.Sleep(time.Second * 1)
159
- for i := 0; i < len(res); i++ {
160
- checkResolve(ctx, t, i, res[i], name, val)
161
- }
162
-
163
- // cancel subscriptions
164
- for i := 0; i < len(res); i++ {
165
- res[i].Cancel(name)
166
- }
167
- time.Sleep(time.Millisecond * 100)
168
-
169
- nval := path.Path("/ipfs/QmPgDWmTmuzvP7QE5zwo1TmjbJme9pmZHNujB2453jkCTr")
170
- err = pub.Publish(ctx, privk, nval)
171
- if err != nil {
172
- t.Fatal(err)
173
- }
174
-
175
- // check we still have the old value in the resolver
176
- time.Sleep(time.Second * 1)
177
- for i := 0; i < len(res); i++ {
178
- checkResolve(ctx, t, i, res[i], name, val)
179
- }
180
-}
181
-
182
-func checkResolveNotFound(ctx context.Context, t *testing.T, i int, resolver Resolver, name string) {
183
- _, err := resolver.Resolve(ctx, name)
184
- if err != ErrResolveFailed {
185
- t.Fatalf("[resolver %d] unexpected error: %s", i, err.Error())
186
- }
187
-}
188
-
189
-func checkResolve(ctx context.Context, t *testing.T, i int, resolver Resolver, name string, val path.Path) {
190
- xval, err := resolver.Resolve(ctx, name)
191
- if err != nil {
192
- t.Fatalf("[resolver %d] resolve failed: %s", i, err.Error())
193
- }
194
- if xval != val {
195
- t.Fatalf("[resolver %d] unexpected value: %s %s", i, val, xval)
196
- }
197
-}
namesys/routing.go
+9
-41
@@ -17,6 +17,7 @@ import (
17
mh "gx/ipfs/QmZyZDi491cCNTLfAhwcaDii2Kg4pwKRkhqQzURGDvY6ua/go-multihash"
18
peer "gx/ipfs/QmcJukH2sAFjY3HdBKq35WDzWoL3UUu2gt9wdfqZTUyM74/go-libp2p-peer"
19
cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
20
+ dht "gx/ipfs/Qmd3jqhBQFvhfBNTSJMQL15GgyVMpdxKTta69Napvx6Myd/go-libp2p-kad-dht"
21
)
22
23
var log = logging.Logger("namesys")
@@ -133,28 +134,28 @@ func (r *routingResolver) resolveOnce(ctx context.Context, name string, options
134
return "", err
135
}
136
137
+ pid, err := peer.IDFromBytes(hash)
138
+ if err != nil {
139
+ log.Debugf("RoutingResolver: could not convert public key hash %s to peer ID: %s\n", name, err)
140
+ return "", err
141
+ }
142
+
143
// Name should be the hash of a public key retrievable from ipfs.
144
// We retrieve the public key here to make certain that it's in the peer
145
// store before calling GetValue() on the DHT - the DHT will call the
146
// ipns validator, which in turn will get the public key from the peer
147
// store to verify the record signature
141
- _, err = routing.GetPublicKey(r.routing, ctx, hash)
148
+ _, err = routing.GetPublicKey(r.routing, ctx, pid)
149
if err != nil {
150
log.Debugf("RoutingResolver: could not retrieve public key %s: %s\n", name, err)
151
return "", err
152
}
153
147
- pid, err := peer.IDFromBytes(hash)
148
- if err != nil {
149
- log.Debugf("RoutingResolver: could not convert public key hash %s to peer ID: %s\n", name, err)
150
- return "", err
151
- }
152
-
154
// Use the routing system to get the name.
155
// Note that the DHT will call the ipns validator when retrieving
156
// the value, which in turn verifies the ipns record signature
157
_, ipnsKey := IpnsKeysForID(pid)
157
- val, err := r.getValue(ctx, ipnsKey, options)
158
+ val, err := r.routing.GetValue(ctx, ipnsKey, dht.Quorum(int(options.DhtRecordCount)))
159
if err != nil {
160
log.Debugf("RoutingResolver: dht get for name %s failed: %s", name, err)
161
return "", err
@@ -187,39 +188,6 @@ func (r *routingResolver) resolveOnce(ctx context.Context, name string, options
188
}
189
}
190
190
-func (r *routingResolver) getValue(ctx context.Context, ipnsKey string, options *opts.ResolveOpts) ([]byte, error) {
191
- // Get specified number of values from the DHT
192
- vals, err := r.routing.GetValues(ctx, ipnsKey, int(options.DhtRecordCount))
193
- if err != nil {
194
- return nil, err
195
- }
196
-
197
- // Select the best value
198
- recs := make([][]byte, 0, len(vals))
199
- for _, v := range vals {
200
- if v.Val != nil {
201
- recs = append(recs, v.Val)
202
- }
203
- }
204
-
205
- if len(recs) == 0 {
206
- return nil, routing.ErrNotFound
207
- }
208
-
209
- i, err := IpnsSelectorFunc(ipnsKey, recs)
210
- if err != nil {
211
- return nil, err
212
- }
213
-
214
- best := recs[i]
215
- if best == nil {
216
- log.Errorf("GetValues %s yielded record with nil value", ipnsKey)
217
- return nil, routing.ErrNotFound
218
- }
219
-
220
- return best, nil
221
-}
222
-
191
func checkEOL(e *pb.IpnsEntry) (time.Time, bool) {
192
if e.GetValidityType() == pb.IpnsEntry_EOL {
193
eol, err := u.ParseRFC3339(string(e.GetValidity()))
namesys/selector.go
deleted
-63
@@ -1,63 +0,0 @@
1
-package namesys
2
-
3
-import (
4
- "bytes"
5
- "errors"
6
-
7
- pb "github.com/ipfs/go-ipfs/namesys/pb"
8
-
9
- u "gx/ipfs/QmNiJuT8Ja3hMVpBHXv3Q6dwmperaQ6JjLtpMQgMCD7xvx/go-ipfs-util"
10
- proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
11
-)
12
-
13
-// IpnsSelectorFunc selects the best record by checking which has the highest
14
-// sequence number and latest EOL
15
-func IpnsSelectorFunc(k string, vals [][]byte) (int, error) {
16
- var recs []*pb.IpnsEntry
17
- for _, v := range vals {
18
- e := new(pb.IpnsEntry)
19
- err := proto.Unmarshal(v, e)
20
- if err == nil {
21
- recs = append(recs, e)
22
- } else {
23
- recs = append(recs, nil)
24
- }
25
- }
26
-
27
- return selectRecord(recs, vals)
28
-}
29
-
30
-func selectRecord(recs []*pb.IpnsEntry, vals [][]byte) (int, error) {
31
- var bestSeq uint64
32
- besti := -1
33
-
34
- for i, r := range recs {
35
- if r == nil || r.GetSequence() < bestSeq {
36
- continue
37
- }
38
- rt, err := u.ParseRFC3339(string(r.GetValidity()))
39
- if err != nil {
40
- log.Errorf("failed to parse ipns record EOL %s", r.GetValidity())
41
- continue
42
- }
43
-
44
- if besti == -1 || r.GetSequence() > bestSeq {
45
- bestSeq = r.GetSequence()
46
- besti = i
47
- } else if r.GetSequence() == bestSeq {
48
- bestt, _ := u.ParseRFC3339(string(recs[besti].GetValidity()))
49
- if rt.After(bestt) {
50
- besti = i
51
- } else if rt == bestt {
52
- if bytes.Compare(vals[i], vals[besti]) > 0 {
53
- besti = i
54
- }
55
- }
56
- }
57
- }
58
- if besti == -1 {
59
- return 0, errors.New("no usable records in given set")
60
- }
61
-
62
- return besti, nil
63
-}
namesys/validator.go
+91
-48
@@ -1,6 +1,7 @@
1
package namesys
2
3
import (
4
+ "bytes"
5
"errors"
6
"time"
7
@@ -41,64 +42,106 @@ var ErrKeyFormat = errors.New("record key could not be parsed into peer ID")
42
// from the peer store
43
var ErrPublicKeyNotFound = errors.New("public key not found in peer store")
44
44
-// NewIpnsRecordValidator returns a ValidChecker for IPNS records.
45
-// The validator function will get a public key from the KeyBook
46
-// to verify the record's signature. Note that the public key must
47
-// already have been fetched from the network and put into the KeyBook
48
-// by the caller.
49
-func NewIpnsRecordValidator(kbook pstore.KeyBook) *record.ValidChecker {
50
- // ValidateIpnsRecord implements ValidatorFunc and verifies that the
51
- // given record's value is an IpnsEntry, that the entry has been correctly
52
- // signed, and that the entry has not expired
53
- ValidateIpnsRecord := func(r *record.ValidationRecord) error {
54
- if r.Namespace != "ipns" {
55
- return ErrInvalidPath
56
- }
45
+type IpnsValidator struct {
46
+ KeyBook pstore.KeyBook
47
+}
48
58
- // Parse the value into an IpnsEntry
59
- entry := new(pb.IpnsEntry)
60
- err := proto.Unmarshal(r.Value, entry)
61
- if err != nil {
62
- return ErrBadRecord
63
- }
49
+func (v IpnsValidator) Validate(key string, value []byte) error {
50
+ ns, pidString, err := record.SplitKey(key)
51
+ if err != nil || ns != "ipns" {
52
+ return ErrInvalidPath
53
+ }
54
+
55
+ // Parse the value into an IpnsEntry
56
+ entry := new(pb.IpnsEntry)
57
+ err = proto.Unmarshal(value, entry)
58
+ if err != nil {
59
+ return ErrBadRecord
60
+ }
61
65
- // Get the public key defined by the ipns path
66
- pid, err := peer.IDFromString(r.Key)
62
+ // Get the public key defined by the ipns path
63
+ pid, err := peer.IDFromString(pidString)
64
+ if err != nil {
65
+ log.Debugf("failed to parse ipns record key %s into peer ID", pidString)
66
+ return ErrKeyFormat
67
+ }
68
+ pubk := v.KeyBook.PubKey(pid)
69
+ if pubk == nil {
70
+ log.Debugf("public key with hash %s not found in peer store", pid)
71
+ return ErrPublicKeyNotFound
72
+ }
73
+
74
+ // Check the ipns record signature with the public key
75
+ if ok, err := pubk.Verify(ipnsEntryDataForSig(entry), entry.GetSignature()); err != nil || !ok {
76
+ log.Debugf("failed to verify signature for ipns record %s", pidString)
77
+ return ErrSignature
78
+ }
79
+
80
+ // Check that record has not expired
81
+ switch entry.GetValidityType() {
82
+ case pb.IpnsEntry_EOL:
83
+ t, err := u.ParseRFC3339(string(entry.GetValidity()))
84
if err != nil {
68
- log.Debugf("failed to parse ipns record key %s into peer ID", r.Key)
69
- return ErrKeyFormat
85
+ log.Debugf("failed parsing time for ipns record EOL in record %s", pidString)
86
+ return err
87
}
71
- pubk := kbook.PubKey(pid)
72
- if pubk == nil {
73
- log.Debugf("public key with hash %s not found in peer store", pid)
74
- return ErrPublicKeyNotFound
88
+ if time.Now().After(t) {
89
+ return ErrExpiredRecord
90
}
91
+ default:
92
+ return ErrUnrecognizedValidity
93
+ }
94
+ return nil
95
+}
96
77
- // Check the ipns record signature with the public key
78
- if ok, err := pubk.Verify(ipnsEntryDataForSig(entry), entry.GetSignature()); err != nil || !ok {
79
- log.Debugf("failed to verify signature for ipns record %s", r.Key)
80
- return ErrSignature
97
+// IpnsSelectorFunc selects the best record by checking which has the highest
98
+// sequence number and latest EOL
99
+func (v IpnsValidator) Select(k string, vals [][]byte) (int, error) {
100
+ var recs []*pb.IpnsEntry
101
+ for _, v := range vals {
102
+ e := new(pb.IpnsEntry)
103
+ err := proto.Unmarshal(v, e)
104
+ if err == nil {
105
+ recs = append(recs, e)
106
+ } else {
107
+ recs = append(recs, nil)
108
}
109
+ }
110
83
- // Check that record has not expired
84
- switch entry.GetValidityType() {
85
- case pb.IpnsEntry_EOL:
86
- t, err := u.ParseRFC3339(string(entry.GetValidity()))
87
- if err != nil {
88
- log.Debugf("failed parsing time for ipns record EOL in record %s", r.Key)
89
- return err
90
- }
91
- if time.Now().After(t) {
92
- return ErrExpiredRecord
111
+ return selectRecord(recs, vals)
112
+}
113
+
114
+func selectRecord(recs []*pb.IpnsEntry, vals [][]byte) (int, error) {
115
+ var bestSeq uint64
116
+ besti := -1
117
+
118
+ for i, r := range recs {
119
+ if r == nil || r.GetSequence() < bestSeq {
120
+ continue
121
+ }
122
+ rt, err := u.ParseRFC3339(string(r.GetValidity()))
123
+ if err != nil {
124
+ log.Errorf("failed to parse ipns record EOL %s", r.GetValidity())
125
+ continue
126
+ }
127
+
128
+ if besti == -1 || r.GetSequence() > bestSeq {
129
+ bestSeq = r.GetSequence()
130
+ besti = i
131
+ } else if r.GetSequence() == bestSeq {
132
+ bestt, _ := u.ParseRFC3339(string(recs[besti].GetValidity()))
133
+ if rt.After(bestt) {
134
+ besti = i
135
+ } else if rt == bestt {
136
+ if bytes.Compare(vals[i], vals[besti]) > 0 {
137
+ besti = i
138
+ }
139
}
94
- default:
95
- return ErrUnrecognizedValidity
140
}
97
- return nil
141
}
99
-
100
- return &record.ValidChecker{
101
- Func: ValidateIpnsRecord,
102
- Sign: false,
142
+ if besti == -1 {
143
+ return 0, errors.New("no usable records in given set")
144
}
145
+
146
+ return besti, nil
147
}
package.json
+12
@@ -599,6 +599,18 @@
599
"hash": "QmTbBs3Y3u5F69XNJzdnnc6SP5GKgcXxCDzx6w8m6piVRT",
600
"name": "go-bitfield",
601
"version": "0.1.1"
602
+ },
603
+ {
604
+ "author": "stebalien",
605
+ "hash": "QmdSX2uedxXdsfNSqbkfSxcYi7pXBBdvp1Km2ZsjPWfydt",
606
+ "name": "go-libp2p-pubsub-router",
607
+ "version": "0.2.1"
608
+ },
609
+ {
610
+ "author": "Stebalien",
611
+ "hash": "QmeoG1seQ8a9b2vS3XJ8HPz9tXr6dzpS5NizjuDDSntQEk",
612
+ "name": "go-libp2p-routing-helpers",
613
+ "version": "0.1.0"
614
}
615
],
616
"gxVersion": "0.10.0",