forks/go-ethereum
1
0
Fork 1
mirror of https://github.com/ethereum/go-ethereum.git synced 2026-08-17 17:33:47 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 1

swarm/pss: fix WithTimeout cancel leaks; fix fmt.Errorf formats

Browse source
This commit is contained in:
Anton Evangelatov 2018-02-12 13:28:02 +01:00
parent 5d30973288
commit b263690654
7 changed files with 48 additions and 29 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
swarm/network/protocol.go
View file

@ -207,10 +207,11 @@ func (b *Bzz) RunProtocol(spec *protocols.Spec, run func(*BzzPeer) error) func(*
// performHandshake implements the negotiation of the bzz handshake
// shared among swarm subprotocols
func performHandshake(p *protocols.Peer, handshake *HandshakeMsg) error {
ctx, _ := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
// defer cancel()
// ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
defer close(handshake.done)
ctx, cancel := context.WithTimeout(context.Background(), bzzHandshakeTimeout)
defer func() {
close(handshake.done)
cancel()
}()
rsh, err := p.Handshake(ctx, handshake, checkHandshake)
if err != nil {
handshake.err = err

10
swarm/pss/client/client_test.go
View file

@ -104,7 +104,8 @@ func TestClientHandshake(t *testing.T) {
lproto := pss.NewPingProtocol(lpssping)
rproto := pss.NewPingProtocol(rpssping)
ctx, _ := context.WithTimeout(context.Background(), time.Second*10)
ctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
err = lpsc.RunProtocol(ctx, lproto)
if err != nil {
t.Fatal(err)
@ -231,13 +232,14 @@ func newServices() adapters.Services {
"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)
return nil, fmt.Errorf("create pss cache tmpdir failed: %s", err)
}
dpa, err := storage.NewLocalDPA(cachedir, make([]byte, 32))
if err != nil {
return nil, fmt.Errorf("local dpa creation failed", "error", err)
return nil, fmt.Errorf("local dpa creation failed: %s", err)
}
ctxlocal, _ := context.WithTimeout(context.Background(), time.Second)
ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctxlocal)
privkey, err := w.GetPrivateKey(keys)
psparams := pss.NewPssParams(privkey)

5
swarm/pss/handshake.go
View file

@ -268,7 +268,7 @@ func (self *HandshakeController) handler(msg []byte, p *p2p.Peer, asymmetric boo
if !asymmetric {
if self.symKeyIndex[symkeyid] != nil {
if self.symKeyIndex[symkeyid].count >= self.symKeyIndex[symkeyid].limit {
return fmt.Errorf("discarding message using expired key", "symkeyid", symkeyid)
return fmt.Errorf("discarding message using expired key: %s", symkeyid)
}
self.symKeyIndex[symkeyid].count++
log.Trace("increment symkey recv use", "symsymkeyid", symkeyid, "count", self.symKeyIndex[symkeyid].count, "limit", self.symKeyIndex[symkeyid].limit, "receiver", common.ToHex(crypto.FromECDSAPub(self.pss.PublicKey())))
@ -457,7 +457,8 @@ func (self *HandshakeAPI) Handshake(pubkeyid string, topic Topic, sync bool, flu
return keys, err
}
if sync {
ctx, _ := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout)
ctx, cancel := context.WithTimeout(context.Background(), self.ctrl.symKeyRequestTimeout)
defer cancel()
select {
case keys = <-hsc:
log.Trace("sync handshake response receive", "key", keys)

2
swarm/pss/protocol.go
View file

@ -227,7 +227,7 @@ func (self *Protocol) AddPeer(p *p2p.Peer, run func(*p2p.Peer, p2p.MsgReadWriter
}
go func() {
err := run(p, rw)
log.Warn(fmt.Sprintf("pss vprotocol quit on addr %v topic %v: %v", topic, err))
log.Warn(fmt.Sprintf("pss vprotocol quit topic %v: %v", topic, err))
}()
return rw, nil
}

6
swarm/pss/protocol_test.go
View file

@ -73,11 +73,13 @@ func testProtocol(t *testing.T) {
time.Sleep(time.Millisecond * 1000) // replace with hive healthy code
lmsgC := make(chan APIMsg)
lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
lctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
defer lsub.Unsubscribe()
rmsgC := make(chan APIMsg)
rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
rctx, cancel := context.WithTimeout(context.Background(), time.Second*10)
defer cancel()
rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
defer rsub.Unsubscribe()

2
swarm/pss/pss.go
View file

@ -501,7 +501,7 @@ func (self *Pss) processSym(envelope *whisper.Envelope) (*whisper.ReceivedMessag
func (self *Pss) processAsym(envelope *whisper.Envelope) (*whisper.ReceivedMessage, string, *PssAddress, error) {
recvmsg, err := envelope.OpenAsymmetric(self.privateKey)
if err != nil {
return nil, "", nil, fmt.Errorf("could not decrypt message: %v", "err", err)
return nil, "", nil, fmt.Errorf("could not decrypt message: %s", err)
}
// check signature (if signed), strip padding
if !recvmsg.Validate() {

43
swarm/pss/pss_test.go
View file

@ -137,7 +137,8 @@ func TestTopic(t *testing.T) {
func TestCache(t *testing.T) {
var err error
to, _ := hex.DecodeString("08090a0b0c0d0e0f1011121314150001020304050607161718191a1b1c1d1e1f")
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
privkey, err := w.GetPrivateKey(keys)
if err != nil {
@ -211,7 +212,8 @@ func TestAddressMatch(t *testing.T) {
remoteaddr := []byte("feedbeef")
kadparams := network.NewKadParams()
kad := network.NewKademlia(localaddr, kadparams)
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
if err != nil {
t.Fatalf("Could not generate private key: %v", err)
@ -255,12 +257,14 @@ func TestAddressMatch(t *testing.T) {
// set and generate pubkeys and symkeys
func TestKeys(t *testing.T) {
// make our key and init pss with it
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
ourkeys, err := wapi.NewKeyPair(ctx)
if err != nil {
t.Fatalf("create 'our' key fail")
}
ctx, _ = context.WithTimeout(context.Background(), time.Second)
ctx, cancel2 := context.WithTimeout(context.Background(), time.Second)
defer cancel2()
theirkeys, err := wapi.NewKeyPair(ctx)
if err != nil {
t.Fatalf("create 'their' key fail")
@ -449,12 +453,14 @@ func testSymSend(t *testing.T) {
// at this point we've verified that symkeys are saved and match on each peer
// now try sending symmetrically encrypted message, both directions
lmsgC := make(chan APIMsg)
lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
lctx, lcancel := context.WithTimeout(context.Background(), time.Second*10)
defer lcancel()
lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
log.Trace("lsub", "id", lsub)
defer lsub.Unsubscribe()
rmsgC := make(chan APIMsg)
rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
rctx, rcancel := context.WithTimeout(context.Background(), time.Second*10)
defer rcancel()
rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
log.Trace("rsub", "id", rsub)
defer rsub.Unsubscribe()
@ -562,12 +568,14 @@ func testAsymSend(t *testing.T) {
time.Sleep(time.Millisecond * 500) // replace with hive healthy code
lmsgC := make(chan APIMsg)
lctx, _ := context.WithTimeout(context.Background(), time.Second*10)
lctx, lcancel := context.WithTimeout(context.Background(), time.Second*10)
defer lcancel()
lsub, err := clients[0].Subscribe(lctx, "pss", lmsgC, "receive", topic)
log.Trace("lsub", "id", lsub)
defer lsub.Unsubscribe()
rmsgC := make(chan APIMsg)
rctx, _ := context.WithTimeout(context.Background(), time.Second*10)
rctx, rcancel := context.WithTimeout(context.Background(), time.Second*10)
defer rcancel()
rsub, err := clients[1].Subscribe(rctx, "pss", rmsgC, "receive", topic)
log.Trace("rsub", "id", rsub)
defer rsub.Unsubscribe()
@ -834,7 +842,8 @@ func benchmarkSymKeySend(b *testing.B) {
if err != nil {
b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err)
}
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
privkey, err := w.GetPrivateKey(keys)
ps := newTestPss(privkey, nil, nil)
@ -877,7 +886,8 @@ func benchmarkAsymKeySend(b *testing.B) {
if err != nil {
b.Fatalf("benchmark called with invalid msgsize param '%s': %v", msgsizestring[1], err)
}
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
privkey, err := w.GetPrivateKey(keys)
ps := newTestPss(privkey, nil, nil)
@ -922,7 +932,8 @@ func benchmarkSymkeyBruteforceChangeaddr(b *testing.B) {
}
pssmsgs := make([]*PssMsg, 0, keycount)
var keyid string
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
privkey, err := w.GetPrivateKey(keys)
if cachesize > 0 {
@ -1004,7 +1015,8 @@ func benchmarkSymkeyBruteforceSameaddr(b *testing.B) {
}
}
addr := make([]PssAddress, keycount)
ctx, _ := context.WithTimeout(context.Background(), time.Second)
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctx)
privkey, err := w.GetPrivateKey(keys)
if cachesize > 0 {
@ -1121,17 +1133,18 @@ func newServices() adapters.Services {
pssProtocolName: 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)
return nil, fmt.Errorf("create pss cache tmpdir failed: %s", err)
}
dpa, err := storage.NewLocalDPA(cachedir, network.NewAddrFromNodeID(ctx.Config.ID).Over())
if err != nil {
return nil, fmt.Errorf("local dpa creation failed", "error", err)
return nil, fmt.Errorf("local dpa creation failed: %s", err)
}
// execadapter does not exec init()
initTest()
ctxlocal, _ := context.WithTimeout(context.Background(), time.Second)
ctxlocal, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
keys, err := wapi.NewKeyPair(ctxlocal)
privkey, err := w.GetPrivateKey(keys)
pssp := NewPssParams(privkey)