diff --git a/pot/address.go b/pot/address.go index a8e65fa93f..48acb29e58 100644 --- a/pot/address.go +++ b/pot/address.go @@ -41,6 +41,10 @@ func NewAddressFromBytes(b []byte) Address { return Address(h) } +func (a Address) IsZero() bool { + return a.Bin() == zerosBin +} + func (a Address) String() string { return fmt.Sprintf("%x", a[:]) } diff --git a/swarm/pss/pss.go b/swarm/pss/pss.go index 420763cae8..e3f0d1c1c7 100644 --- a/swarm/pss/pss.go +++ b/swarm/pss/pss.go @@ -9,7 +9,6 @@ import ( "time" "github.com/ethereum/go-ethereum/common" - "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" "github.com/ethereum/go-ethereum/p2p" "github.com/ethereum/go-ethereum/p2p/discover" @@ -65,11 +64,11 @@ type Pss struct { network.Overlay // we can get the overlayaddress from this peerPool map[pot.Address]map[whisper.TopicType]p2p.MsgReadWriter // keep track of all virtual p2p.Peers we are currently speaking to fwdPool map[discover.NodeID]*protocols.Peer // keep track of all peers sitting on the pssmsg routing layer - keyPool map[pot.Address]ecdsa.PublicKey // keep track of all public keys so we can encrypt for our peers - reverseKeyPool map[string]pot.Address // as above but reverse lookup - handlers map[whisper.TopicType]map[*Handler]bool // topic and version based pss payload handlers - fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg - cachettl time.Duration // how long to keep messages in fwdcache + keyPool map[pot.Address]map[whisper.TopicType]ecdsa.PublicKey // keep track of all public keys so we can encrypt for our peers + reverseKeyPool map[ecdsa.PublicKey]map[whisper.TopicType]pot.Address + handlers map[whisper.TopicType]map[*Handler]bool // topic and version based pss payload handlers + fwdcache map[pssDigest]pssCacheEntry // checksum of unique fields from pssmsg mapped to expiry, cache to determine whether to drop msg + cachettl time.Duration // how long to keep messages in fwdcache lock sync.Mutex dpa *storage.DPA privatekey *ecdsa.PrivateKey @@ -98,8 +97,8 @@ func NewPss(k network.Overlay, dpa *storage.DPA, params *PssParams) *Pss { Overlay: k, peerPool: make(map[pot.Address]map[whisper.TopicType]p2p.MsgReadWriter, PssPeerCapacity), fwdPool: make(map[discover.NodeID]*protocols.Peer), - keyPool: make(map[pot.Address]ecdsa.PublicKey), - reverseKeyPool: make(map[string]pot.Address), + keyPool: make(map[pot.Address]map[whisper.TopicType]ecdsa.PublicKey), + reverseKeyPool: make(map[ecdsa.PublicKey]map[whisper.TopicType]pot.Address), handlers: make(map[whisper.TopicType]map[*Handler]bool), fwdcache: make(map[pssDigest]pssCacheEntry), cachettl: params.Cachettl, @@ -176,25 +175,45 @@ func (self *Pss) Register(topic *whisper.TopicType, handler Handler) func() { return func() { self.deregister(topic, &handler) } } -func (self *Pss) AddAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) { +// Add a Public key address mapping +// returns false if identical mapping already exists +func (self *Pss) AddPublicKey(addr pot.Address, topic whisper.TopicType, pubkey ecdsa.PublicKey) bool { self.lock.Lock() defer self.lock.Unlock() - self.keyPool[addr] = pubkey - self.reverseKeyPool[common.ToHex(crypto.FromECDSAPub(&pubkey))] = addr + if len(self.keyPool[addr]) == 0 { + self.keyPool[addr] = make(map[whisper.TopicType]ecdsa.PublicKey) + self.reverseKeyPool[pubkey] = make(map[whisper.TopicType]pot.Address) + } + self.keyPool[addr][topic] = pubkey + self.reverseKeyPool[pubkey][topic] = addr + return true } -// may need these later, please let them be -////func (self *Pss) RemoveAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) { -// delete(self.reverseKeyPool, &self.keyPool[addr]) -// delete(self.keyPool, addr) -//} -// -//func (self *Pss) hasAddressKeyPair(addr pot.Address, pubkey ecdsa.PublicKey) bool { -// if self.keyPool[addr] != nil { -// return *self.keyPool[addr] == pubkey -// } -// return false -//} +func (self *Pss) RemovePublicKey(addr pot.Address, topic whisper.TopicType, pubkey ecdsa.PublicKey) bool { + if len(self.keyPool[addr]) == 0 { + return false + } + zeroKey := ecdsa.PublicKey{} + if self.keyPool[addr][topic] == zeroKey { + return false + } + delete(self.reverseKeyPool, pubkey) + self.keyPool[addr][topic] = zeroKey + return true +} + +func (self *Pss) GetKeys(addr pot.Address) (keys []ecdsa.PublicKey) { +outer: + for _, key := range self.keyPool[addr] { + for _, havekey := range keys { + if havekey == key { + continue outer + } + } + keys = append(keys, key) + } + return +} func (self *Pss) deregister(topic *whisper.TopicType, h *Handler) { self.lock.Lock() @@ -300,9 +319,10 @@ func (self *Pss) Process(pssmsg *PssMsg) error { nid, _ := discover.HexID("0x00") p := p2p.NewPeer(nid, fmt.Sprintf("%x", recvmsg.Src), []p2p.Cap{}) - addr := self.reverseKeyPool[common.ToHex(crypto.FromECDSAPub(recvmsg.Src))] + //addr := self.reverseKeyPool[common.ToHex(crypto.FromECDSAPub(recvmsg.Src))] + addr := self.reverseKeyPool[*recvmsg.Src][recvmsg.Topic] log.Warn("recvkey", "key", *recvmsg.Src, "addr", addr) - if bytes.Equal([]byte{}, addr[:]) { + if addr.IsZero() { return fmt.Errorf("unknown key", "addr", addr) } @@ -323,7 +343,7 @@ func (self *Pss) Process(pssmsg *PssMsg) error { func (self *Pss) SendAsym(to []byte, topic whisper.TopicType, msg []byte) error { var potaddr pot.Address copy(potaddr[:], to) - topubkey := self.keyPool[potaddr] + topubkey := self.keyPool[potaddr][topic] log.Debug("using pubkey", "pubkey", topubkey) wparams := &whisper.MessageParams{ TTL: DefaultTTL, diff --git a/swarm/pss/pss_test.go b/swarm/pss/pss_test.go index 2cfdedb512..9234ca9a4a 100644 --- a/swarm/pss/pss_test.go +++ b/swarm/pss/pss_test.go @@ -2,29 +2,31 @@ package pss import ( "bytes" - "context" + // "context" + // "crypto/ecdsa" "encoding/hex" - "encoding/json" + // "encoding/json" + "flag" "fmt" - "io/ioutil" + // "io/ioutil" "math/rand" "os" - "sync" + // "sync" "testing" "time" - "flag" -// "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/log" - "github.com/ethereum/go-ethereum/node" + // "github.com/ethereum/go-ethereum/node" "github.com/ethereum/go-ethereum/p2p" - "github.com/ethereum/go-ethereum/p2p/discover" - "github.com/ethereum/go-ethereum/p2p/protocols" - "github.com/ethereum/go-ethereum/p2p/simulations" - "github.com/ethereum/go-ethereum/p2p/simulations/adapters" - p2ptest "github.com/ethereum/go-ethereum/p2p/testing" + // "github.com/ethereum/go-ethereum/p2p/discover" + // "github.com/ethereum/go-ethereum/p2p/protocols" + // "github.com/ethereum/go-ethereum/p2p/simulations" + // "github.com/ethereum/go-ethereum/p2p/simulations/adapters" + // p2ptest "github.com/ethereum/go-ethereum/p2p/testing" "github.com/ethereum/go-ethereum/swarm/network" - "github.com/ethereum/go-ethereum/swarm/storage" + // "github.com/ethereum/go-ethereum/swarm/storage" + whisper "github.com/ethereum/go-ethereum/whisper/whisperv5" ) const ( @@ -34,11 +36,10 @@ const ( var ( snapshotfile string - debugflag = flag.Bool("v", false, "verbose") + debugflag = flag.Bool("v", false, "verbose") // custom logging psslogmain log.Logger - ) var services = newServices() @@ -63,6 +64,12 @@ func init() { } +func TestKeys(t *testing.T) { + key, err := crypto.GenerateKey() + addr := network.RandomAddress.Over() + topic := whisper.NewTopic("foo", 42) +} + func TestCache(t *testing.T) { var err error to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f") @@ -199,374 +206,374 @@ func TestRegisterHandler(t *testing.T) { } } -func TestSimpleLinear(t *testing.T) { - var err error - nodeconfig := adapters.RandomNodeConfig() - addr := network.NewAddrFromNodeID(nodeconfig.ID) - _ = p2ptest.NewTestPeerPool() - ps := NewTestPss(addr.Over()) - - ping := &Ping{ - C: make(chan struct{}), - } - - ps.Register(&PingTopic, RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)).Handle) - - if err != nil { - t.Fatalf("Failed to register virtual protocol in pss: %v", err) - } - run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { - id := p.ID() - pp := protocols.NewPeer(p, rw, pssSpec) - bp := &testOverlayConn{ - Peer: pp, - addr: network.ToOverlayAddr(id[:]), - } - //a := pot.NewAddressFromBytes(bp.addr) - //ps.fwdPool[a] = pp - ps.fwdPool[id] = pp - ps.Overlay.On(bp) - defer ps.Overlay.Off(bp) - log.Debug(fmt.Sprintf("%v", ps.Overlay)) - return bp.Run(ps.handlePssMsg) - } - - pt := p2ptest.NewProtocolTester(t, nodeconfig.ID, 2, run) - - msg := NewPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PingProtocol, PingTopic, []byte{1, 2, 3}) - - exchange := p2ptest.Exchange{ - Expects: []p2ptest.Expect{ - p2ptest.Expect{ - Code: 0, - Msg: msg, - Peer: pt.IDs[0], - }, - }, - Triggers: []p2ptest.Trigger{ - p2ptest.Trigger{ - Code: 0, - Msg: msg, - Peer: pt.IDs[1], - }, - }, - } - - err = pt.TestExchanges(exchange) - if err != nil { - t.Fatalf("exchange failed %v", err) - } -} - -func TestSnapshot_50_5(t *testing.T) { - testSnapshot(t, "testdata/snapshot_50.json", 5, true) -} - -func TestSnapshot_5_50(t *testing.T) { - testSnapshot(t, "testdata/snapshot_5.json", 50, true) -} - -func TestSnapshot_5_5(t *testing.T) { - testSnapshot(t, "testdata/snapshot_5.json", 5, true) -} - -func testSnapshot(t *testing.T, snapshotfile string, msgcount int, sim bool) { - - - // choose the adapter to use - var adapter adapters.NodeAdapter - if sim { - adapter = adapters.NewSimAdapter(services) - } else { - baseDir, err := ioutil.TempDir("", "swarm-test") - if err != nil { - t.Fatal(err) - } - defer os.RemoveAll(baseDir) - adapter = adapters.NewExecAdapter(baseDir) - } - - // process shapshot - jsonsnapshot, err := ioutil.ReadFile(snapshotfile) - if err != nil { - t.Fatalf("cant read snapshot: %s", snapshotfile) - } - snapshot := &simulations.Snapshot{} - err = json.Unmarshal(jsonsnapshot, snapshot) - if err != nil { - t.Fatalf("snapshot file unreadable: %v", err) - } - for _, node := range snapshot.Nodes { - node.Config.Services = []string{"bzz", "pss"} - } - - // setup network with snapshot - net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ - ID: "0", - }) - defer net.Shutdown() - - err = net.Load(snapshot) - if err != nil { - t.Fatalf("invalid snapshot: %v", err) - } - - timeout := 15 * time.Second - ctx, cancelmain := context.WithTimeout(context.Background(), timeout) - defer cancelmain() - - // nodes expecting messages - recvids := make([]discover.NodeID, msgcount) - - // the overlay address map to recvids - recvaddrs := make(map[discover.NodeID][]byte) - - // messages actually received (registered through trigger and test check) - var msgreceived []discover.NodeID - - // trigger for expect in test - trigger := make(chan discover.NodeID) - - // one wait for every message - wg := sync.WaitGroup{} - wg.Add(msgcount) - - action := func(ctx context.Context) error { - var rpcerr error - var rpcbyte []byte - for _, simnode := range net.Nodes { - if simnode == nil { - return fmt.Errorf("unknown node: %s", simnode.ID()) - } - client, err := simnode.Client() - if err != nil { - return fmt.Errorf("error getting recp node client: %s", err) - } - - err = client.Call(&rpcbyte, "pss_baseAddr") - if err != nil { - t.Fatalf("cant get overlayaddr: %v", err) - } - - recvaddrs[simnode.ID()] = rpcbyte - err = client.Call(&rpcbyte, "pss_baseAddr") - if err != nil { - t.Fatalf("cant get overlayaddr: %v", err) - } - - err = triggerChecks(ctx, &wg, &trigger, net, simnode.ID()) - if err != nil { - t.Fatalf("trigger setup failed: %v", err) - } - } - for i := 0; i < msgcount; i++ { - - idx := rand.Intn(len(net.Nodes)) - sendernode := net.Nodes[idx] - toidx := rand.Intn(len(net.Nodes)-1) - if toidx >= idx { - toidx++ - } - recvnode := net.Nodes[toidx] - recvids[i] = recvnode.ID() - msg := PingMsg{Created: time.Now()} - code, _ := PingProtocol.GetCode(&PingMsg{}) - pmsg, _ := NewProtocolMsg(code, msg) - - client, err := sendernode.Client() - if err != nil { - return fmt.Errorf("error getting sendernode client: %s", err) - } - client.CallContext(ctx, &rpcerr, "pss_send", PingTopic, APIMsg{ - Addr: recvaddrs[recvnode.ID()], - Msg: pmsg, - }) - if rpcerr != nil { - return fmt.Errorf("error rpc send id %x: %v", sendernode.ID(), rpcerr) - } - } - return nil - } - check := func(ctx context.Context, id discover.NodeID) (bool, error) { - select { - case <-ctx.Done(): - wg.Done() - return false, ctx.Err() - default: - } - msgreceived = append(msgreceived, id) - psslogmain.Info("trigger received", "id", id, "len", len(msgreceived)) - wg.Done() - return true, nil - } - - result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ - Action: action, - Trigger: trigger, - Expect: &simulations.Expectation{ - Nodes: recvids, - Check: check, - }, - }) - if result.Error != nil { - psslogmain.Error("msg failed!", "err", result.Error) - cancelmain() - t.Fatalf("simulation failed: %s", result.Error) - } - - wg.Wait() - - if len(msgreceived) != msgcount { - t.Fatalf("Simulation Failed, got %d of %d msgs", len(msgreceived), msgcount) - } - - psslogmain.Info("done!") - t.Logf("Simulation Passed, got %d of %d msgs", len(msgreceived), msgcount) - //t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt)) -} - -// triggerChecks triggers a simulation step check whenever a peer is added or -// removed from the given node -// connections and connectionstarget are temporary kademlia check workarounds -func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discover.NodeID, net *simulations.Network, id discover.NodeID) error { - - quitC := make(chan struct{}) - - node := net.GetNode(id) - if node == nil { - return fmt.Errorf("unknown node: %s", id) - } - client, err := node.Client() - if err != nil { - return err - } - - peerevents := make(chan *p2p.PeerEvent) - peersub, err := client.Subscribe(context.Background(), "admin", peerevents, "peerEvents") - if err != nil { - return fmt.Errorf("error getting peer events for node %v: %s", id, err) - } - - msgevents := make(chan APIMsg) - msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receive", PingTopic) - if err != nil { - return fmt.Errorf("error getting msg events for node %v: %s", id, err) - } - - go func() { - defer msgsub.Unsubscribe() - defer peersub.Unsubscribe() - for { - select { - case <-msgevents: - psslogmain.Debug("incoming msg", "node", id) - *trigger <- id - case err := <-peersub.Err(): - if err != nil { - log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err) - } - return - - case err := <-msgsub.Err(): - if err != nil { - log.Error(fmt.Sprintf("error getting msg for node %v", id), "err", err) - } - return - case <-quitC: - return - } - } - }() - - go func() { - wg.Wait() - quitC <- struct{}{} - }() - - return nil -} - -func newServices() adapters.Services { - stateStore := adapters.NewSimStateStore() - kademlias := make(map[discover.NodeID]*network.Kademlia) - kademlia := func(id discover.NodeID) *network.Kademlia { - if k, ok := kademlias[id]; ok { - return k - } - addr := network.NewAddrFromNodeID(id) - params := network.NewKadParams() - params.MinProxBinSize = 2 - params.MaxBinSize = 3 - params.MinBinSize = 1 - params.MaxRetries = 1000 - params.RetryExponent = 2 - params.RetryInterval = 1000000 - kademlias[id] = network.NewKademlia(addr.Over(), params) - return kademlias[id] - } - return adapters.Services{ - //"pss": func(id discover.NodeID, snapshot []byte) node.Service { - "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { - cachedir, err := ioutil.TempDir("", "pss-cache") - if err != nil { - return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err) - } - dpa, err := storage.NewLocalDPA(cachedir) - if err != nil { - return nil, fmt.Errorf("local dpa creation failed", "error", err) - } - - pssp := NewPssParams(true) - ps := NewPss(kademlia(ctx.Config.ID), dpa, pssp) - - ping := &Ping{ - C: make(chan struct{}), - } - ps.Register(&PingTopic, RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)).Handle) - if err != nil { - log.Error("Couldnt register pss protocol", "err", err) - os.Exit(1) - } - - return ps, nil - }, - //"bzz": func(id discover.NodeID, snapshot []byte) node.Service { - "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { - addr := network.NewAddrFromNodeID(ctx.Config.ID) - hp := network.NewHiveParams() - hp.Discovery = false - config := &network.BzzConfig{ - OverlayAddr: addr.Over(), - UnderlayAddr: addr.Under(), - HiveParams: hp, - } - return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil - }, - } -} - -type connmap struct { - conns map[discover.NodeID][]discover.NodeID - healthy map[discover.NodeID]bool - lock sync.Mutex -} - -type testOverlayConn struct { - *protocols.Peer - addr []byte -} - -func (self *testOverlayConn) Address() []byte { - return self.addr -} - -func (self *testOverlayConn) Off() network.OverlayAddr { - return self -} - -func (self *testOverlayConn) Drop(err error) { -} - -func (self *testOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr { - return self -} +//func TestSimpleLinear(t *testing.T) { +// var err error +// nodeconfig := adapters.RandomNodeConfig() +// addr := network.NewAddrFromNodeID(nodeconfig.ID) +// _ = p2ptest.NewTestPeerPool() +// ps := NewTestPss(addr.Over()) +// +// ping := &Ping{ +// C: make(chan struct{}), +// } +// +// ps.Register(&PingTopic, RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)).Handle) +// +// if err != nil { +// t.Fatalf("Failed to register virtual protocol in pss: %v", err) +// } +// run := func(p *p2p.Peer, rw p2p.MsgReadWriter) error { +// id := p.ID() +// pp := protocols.NewPeer(p, rw, pssSpec) +// bp := &testOverlayConn{ +// Peer: pp, +// addr: network.ToOverlayAddr(id[:]), +// } +// //a := pot.NewAddressFromBytes(bp.addr) +// //ps.fwdPool[a] = pp +// ps.fwdPool[id] = pp +// ps.Overlay.On(bp) +// defer ps.Overlay.Off(bp) +// log.Debug(fmt.Sprintf("%v", ps.Overlay)) +// return bp.Run(ps.handlePssMsg) +// } +// +// pt := p2ptest.NewProtocolTester(t, nodeconfig.ID, 2, run) +// +// msg := NewPingMsg(network.ToOverlayAddr(pt.IDs[0].Bytes()), PingProtocol, PingTopic, []byte{1, 2, 3}) +// +// exchange := p2ptest.Exchange{ +// Expects: []p2ptest.Expect{ +// p2ptest.Expect{ +// Code: 0, +// Msg: msg, +// Peer: pt.IDs[0], +// }, +// }, +// Triggers: []p2ptest.Trigger{ +// p2ptest.Trigger{ +// Code: 0, +// Msg: msg, +// Peer: pt.IDs[1], +// }, +// }, +// } +// +// err = pt.TestExchanges(exchange) +// if err != nil { +// t.Fatalf("exchange failed %v", err) +// } +//} +// +//func TestSnapshot_50_5(t *testing.T) { +// testSnapshot(t, "testdata/snapshot_50.json", 5, true) +//} +// +//func TestSnapshot_5_50(t *testing.T) { +// testSnapshot(t, "testdata/snapshot_5.json", 50, true) +//} +// +//func TestSnapshot_5_5(t *testing.T) { +// testSnapshot(t, "testdata/snapshot_5.json", 5, true) +//} +// +//func testSnapshot(t *testing.T, snapshotfile string, msgcount int, sim bool) { +// +// +// // choose the adapter to use +// var adapter adapters.NodeAdapter +// if sim { +// adapter = adapters.NewSimAdapter(services) +// } else { +// baseDir, err := ioutil.TempDir("", "swarm-test") +// if err != nil { +// t.Fatal(err) +// } +// defer os.RemoveAll(baseDir) +// adapter = adapters.NewExecAdapter(baseDir) +// } +// +// // process shapshot +// jsonsnapshot, err := ioutil.ReadFile(snapshotfile) +// if err != nil { +// t.Fatalf("cant read snapshot: %s", snapshotfile) +// } +// snapshot := &simulations.Snapshot{} +// err = json.Unmarshal(jsonsnapshot, snapshot) +// if err != nil { +// t.Fatalf("snapshot file unreadable: %v", err) +// } +// for _, node := range snapshot.Nodes { +// node.Config.Services = []string{"bzz", "pss"} +// } +// +// // setup network with snapshot +// net := simulations.NewNetwork(adapter, &simulations.NetworkConfig{ +// ID: "0", +// }) +// defer net.Shutdown() +// +// err = net.Load(snapshot) +// if err != nil { +// t.Fatalf("invalid snapshot: %v", err) +// } +// +// timeout := 15 * time.Second +// ctx, cancelmain := context.WithTimeout(context.Background(), timeout) +// defer cancelmain() +// +// // nodes expecting messages +// recvids := make([]discover.NodeID, msgcount) +// +// // the overlay address map to recvids +// recvaddrs := make(map[discover.NodeID][]byte) +// +// // messages actually received (registered through trigger and test check) +// var msgreceived []discover.NodeID +// +// // trigger for expect in test +// trigger := make(chan discover.NodeID) +// +// // one wait for every message +// wg := sync.WaitGroup{} +// wg.Add(msgcount) +// +// action := func(ctx context.Context) error { +// var rpcerr error +// var rpcbyte []byte +// for _, simnode := range net.Nodes { +// if simnode == nil { +// return fmt.Errorf("unknown node: %s", simnode.ID()) +// } +// client, err := simnode.Client() +// if err != nil { +// return fmt.Errorf("error getting recp node client: %s", err) +// } +// +// err = client.Call(&rpcbyte, "pss_baseAddr") +// if err != nil { +// t.Fatalf("cant get overlayaddr: %v", err) +// } +// +// recvaddrs[simnode.ID()] = rpcbyte +// err = client.Call(&rpcbyte, "pss_baseAddr") +// if err != nil { +// t.Fatalf("cant get overlayaddr: %v", err) +// } +// +// err = triggerChecks(ctx, &wg, &trigger, net, simnode.ID()) +// if err != nil { +// t.Fatalf("trigger setup failed: %v", err) +// } +// } +// for i := 0; i < msgcount; i++ { +// +// idx := rand.Intn(len(net.Nodes)) +// sendernode := net.Nodes[idx] +// toidx := rand.Intn(len(net.Nodes)-1) +// if toidx >= idx { +// toidx++ +// } +// recvnode := net.Nodes[toidx] +// recvids[i] = recvnode.ID() +// msg := PingMsg{Created: time.Now()} +// code, _ := PingProtocol.GetCode(&PingMsg{}) +// pmsg, _ := NewProtocolMsg(code, msg) +// +// client, err := sendernode.Client() +// if err != nil { +// return fmt.Errorf("error getting sendernode client: %s", err) +// } +// client.CallContext(ctx, &rpcerr, "pss_send", PingTopic, APIMsg{ +// Addr: recvaddrs[recvnode.ID()], +// Msg: pmsg, +// }) +// if rpcerr != nil { +// return fmt.Errorf("error rpc send id %x: %v", sendernode.ID(), rpcerr) +// } +// } +// return nil +// } +// check := func(ctx context.Context, id discover.NodeID) (bool, error) { +// select { +// case <-ctx.Done(): +// wg.Done() +// return false, ctx.Err() +// default: +// } +// msgreceived = append(msgreceived, id) +// psslogmain.Info("trigger received", "id", id, "len", len(msgreceived)) +// wg.Done() +// return true, nil +// } +// +// result := simulations.NewSimulation(net).Run(ctx, &simulations.Step{ +// Action: action, +// Trigger: trigger, +// Expect: &simulations.Expectation{ +// Nodes: recvids, +// Check: check, +// }, +// }) +// if result.Error != nil { +// psslogmain.Error("msg failed!", "err", result.Error) +// cancelmain() +// t.Fatalf("simulation failed: %s", result.Error) +// } +// +// wg.Wait() +// +// if len(msgreceived) != msgcount { +// t.Fatalf("Simulation Failed, got %d of %d msgs", len(msgreceived), msgcount) +// } +// +// psslogmain.Info("done!") +// t.Logf("Simulation Passed, got %d of %d msgs", len(msgreceived), msgcount) +// //t.Logf("Duration: %s", result.FinishedAt.Sub(result.StartedAt)) +//} +// +//// triggerChecks triggers a simulation step check whenever a peer is added or +//// removed from the given node +//// connections and connectionstarget are temporary kademlia check workarounds +//func triggerChecks(ctx context.Context, wg *sync.WaitGroup, trigger *chan discover.NodeID, net *simulations.Network, id discover.NodeID) error { +// +// quitC := make(chan struct{}) +// +// node := net.GetNode(id) +// if node == nil { +// return fmt.Errorf("unknown node: %s", id) +// } +// client, err := node.Client() +// if err != nil { +// return err +// } +// +// peerevents := make(chan *p2p.PeerEvent) +// peersub, err := client.Subscribe(context.Background(), "admin", peerevents, "peerEvents") +// if err != nil { +// return fmt.Errorf("error getting peer events for node %v: %s", id, err) +// } +// +// msgevents := make(chan APIMsg) +// msgsub, err := client.Subscribe(context.Background(), "pss", msgevents, "receive", PingTopic) +// if err != nil { +// return fmt.Errorf("error getting msg events for node %v: %s", id, err) +// } +// +// go func() { +// defer msgsub.Unsubscribe() +// defer peersub.Unsubscribe() +// for { +// select { +// case <-msgevents: +// psslogmain.Debug("incoming msg", "node", id) +// *trigger <- id +// case err := <-peersub.Err(): +// if err != nil { +// log.Error(fmt.Sprintf("error getting peer events for node %v", id), "err", err) +// } +// return +// +// case err := <-msgsub.Err(): +// if err != nil { +// log.Error(fmt.Sprintf("error getting msg for node %v", id), "err", err) +// } +// return +// case <-quitC: +// return +// } +// } +// }() +// +// go func() { +// wg.Wait() +// quitC <- struct{}{} +// }() +// +// return nil +//} +// +//func newServices() adapters.Services { +// stateStore := adapters.NewSimStateStore() +// kademlias := make(map[discover.NodeID]*network.Kademlia) +// kademlia := func(id discover.NodeID) *network.Kademlia { +// if k, ok := kademlias[id]; ok { +// return k +// } +// addr := network.NewAddrFromNodeID(id) +// params := network.NewKadParams() +// params.MinProxBinSize = 2 +// params.MaxBinSize = 3 +// params.MinBinSize = 1 +// params.MaxRetries = 1000 +// params.RetryExponent = 2 +// params.RetryInterval = 1000000 +// kademlias[id] = network.NewKademlia(addr.Over(), params) +// return kademlias[id] +// } +// return adapters.Services{ +// //"pss": func(id discover.NodeID, snapshot []byte) node.Service { +// "pss": func(ctx *adapters.ServiceContext) (node.Service, error) { +// cachedir, err := ioutil.TempDir("", "pss-cache") +// if err != nil { +// return nil, fmt.Errorf("create pss cache tmpdir failed", "error", err) +// } +// dpa, err := storage.NewLocalDPA(cachedir) +// if err != nil { +// return nil, fmt.Errorf("local dpa creation failed", "error", err) +// } +// +// pssp := NewPssParams(true) +// ps := NewPss(kademlia(ctx.Config.ID), dpa, pssp) +// +// ping := &Ping{ +// C: make(chan struct{}), +// } +// ps.Register(&PingTopic, RegisterPssProtocol(ps, &PingTopic, PingProtocol, NewPingProtocol(ping.PingHandler)).Handle) +// if err != nil { +// log.Error("Couldnt register pss protocol", "err", err) +// os.Exit(1) +// } +// +// return ps, nil +// }, +// //"bzz": func(id discover.NodeID, snapshot []byte) node.Service { +// "bzz": func(ctx *adapters.ServiceContext) (node.Service, error) { +// addr := network.NewAddrFromNodeID(ctx.Config.ID) +// hp := network.NewHiveParams() +// hp.Discovery = false +// config := &network.BzzConfig{ +// OverlayAddr: addr.Over(), +// UnderlayAddr: addr.Under(), +// HiveParams: hp, +// } +// return network.NewBzz(config, kademlia(ctx.Config.ID), stateStore), nil +// }, +// } +//} +// +//type connmap struct { +// conns map[discover.NodeID][]discover.NodeID +// healthy map[discover.NodeID]bool +// lock sync.Mutex +//} +// +//type testOverlayConn struct { +// *protocols.Peer +// addr []byte +//} +// +//func (self *testOverlayConn) Address() []byte { +// return self.addr +//} +// +//func (self *testOverlayConn) Off() network.OverlayAddr { +// return self +//} +// +//func (self *testOverlayConn) Drop(err error) { +//} +// +//func (self *testOverlayConn) Update(o network.OverlayAddr) network.OverlayAddr { +// return self +//} diff --git a/swarm/pss/pssapi.go b/swarm/pss/pssapi.go index 98df4766ea..819c5b3e23 100644 --- a/swarm/pss/pssapi.go +++ b/swarm/pss/pssapi.go @@ -74,10 +74,10 @@ func (pssapi *API) BaseAddr() ([]byte, error) { return pssapi.Pss.BaseAddr(), nil } -func (pssapi *API) AddAddressKeyPair(addr []byte, pubkey ecdsa.PublicKey) error { +func (pssapi *API) AddPublicKey(addr []byte, topic whisper.TopicType, pubkey ecdsa.PublicKey) error { var potaddr pot.Address copy(potaddr[:], addr) - pssapi.Pss.AddAddressKeyPair(potaddr, pubkey) + pssapi.Pss.AddPublicKey(potaddr, topic, pubkey) return nil }