forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-07-27 23:26:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

swarm/pss: WIP whisper keys handling

Browse source
This commit is contained in:
lash 2017-07-13 17:44:16 +02:00
parent d9a883910e
commit 78d0a9d1d5
4 changed files with 445 additions and 414 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

4
pot/address.go
View file

@ -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[:])
}

72
swarm/pss/pss.go
View file

@ -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,

779
swarm/pss/pss_test.go
View file

@ -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
//}

4
swarm/pss/pssapi.go
View file

@ -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
}