From d3f056bd68fb6a8e9ffa3938d5404c6a209e0313 Mon Sep 17 00:00:00 2001 From: Ferenc Szabo Date: Fri, 21 Sep 2018 12:56:43 +0200 Subject: swarm/network/stream: fix DoS invalid hash length (#927) --- swarm/network/stream/messages.go | 14 +++++-- swarm/network/stream/streamer_test.go | 77 ++++++++++++++++++++++++++++++++--- 2 files changed, 82 insertions(+), 9 deletions(-) (limited to 'swarm/network/stream') diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go index 2e1a81e82..62c46b120 100644 --- a/swarm/network/stream/messages.go +++ b/swarm/network/stream/messages.go @@ -26,7 +26,7 @@ import ( bv "github.com/ethereum/go-ethereum/swarm/network/bitvector" "github.com/ethereum/go-ethereum/swarm/spancontext" "github.com/ethereum/go-ethereum/swarm/storage" - opentracing "github.com/opentracing/opentracing-go" + "github.com/opentracing/opentracing-go" ) var syncBatchTimeout = 30 * time.Second @@ -195,10 +195,16 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg if err != nil { return err } + hashes := req.Hashes - want, err := bv.New(len(hashes) / HashSize) + lenHashes := len(hashes) + if lenHashes%HashSize != 0 { + return fmt.Errorf("error invalid hashes length (len: %v)", lenHashes) + } + + want, err := bv.New(lenHashes / HashSize) if err != nil { - return fmt.Errorf("error initiaising bitvector of length %v: %v", len(hashes)/HashSize, err) + return fmt.Errorf("error initiaising bitvector of length %v: %v", lenHashes/HashSize, err) } ctr := 0 @@ -206,7 +212,7 @@ func (p *Peer) handleOfferedHashesMsg(ctx context.Context, req *OfferedHashesMsg ctx, cancel := context.WithTimeout(ctx, syncBatchTimeout) ctx = context.WithValue(ctx, "source", p.ID().String()) - for i := 0; i < len(hashes); i += HashSize { + for i := 0; i < lenHashes; i += HashSize { hash := hashes[i : i+HashSize] if wait := c.NeedData(ctx, hash); wait != nil { diff --git a/swarm/network/stream/streamer_test.go b/swarm/network/stream/streamer_test.go index 06e96b9a9..dba4e7ea3 100644 --- a/swarm/network/stream/streamer_test.go +++ b/swarm/network/stream/streamer_test.go @@ -19,6 +19,7 @@ package stream import ( "bytes" "context" + "errors" "testing" "time" @@ -55,11 +56,12 @@ func TestStreamerRequestSubscription(t *testing.T) { } var ( - hash0 = sha3.Sum256([]byte{0}) - hash1 = sha3.Sum256([]byte{1}) - hash2 = sha3.Sum256([]byte{2}) - hashesTmp = append(hash0[:], hash1[:]...) - hashes = append(hashesTmp, hash2[:]...) + hash0 = sha3.Sum256([]byte{0}) + hash1 = sha3.Sum256([]byte{1}) + hash2 = sha3.Sum256([]byte{2}) + hashesTmp = append(hash0[:], hash1[:]...) + hashes = append(hashesTmp, hash2[:]...) + corruptHashes = append(hashes[:40]) ) type testClient struct { @@ -459,6 +461,71 @@ func TestStreamerUpstreamSubscribeLiveAndHistory(t *testing.T) { } } +func TestStreamerDownstreamCorruptHashesMsgExchange(t *testing.T) { + tester, streamer, _, teardown, err := newStreamerTester(t) + defer teardown() + if err != nil { + t.Fatal(err) + } + + stream := NewStream("foo", "", true) + + var tc *testClient + + streamer.RegisterClientFunc("foo", func(p *Peer, t string, live bool) (Client, error) { + tc = newTestClient(t) + return tc, nil + }) + + peerID := tester.IDs[0] + + err = streamer.Subscribe(peerID, stream, NewRange(5, 8), Top) + if err != nil { + t.Fatalf("Expected no error, got %v", err) + } + + err = tester.TestExchanges(p2ptest.Exchange{ + Label: "Subscribe message", + Expects: []p2ptest.Expect{ + { + Code: 4, + Msg: &SubscribeMsg{ + Stream: stream, + History: NewRange(5, 8), + Priority: Top, + }, + Peer: peerID, + }, + }, + }, + p2ptest.Exchange{ + Label: "Corrupt offered hash message", + Triggers: []p2ptest.Trigger{ + { + Code: 1, + Msg: &OfferedHashesMsg{ + HandoverProof: &HandoverProof{ + Handover: &Handover{}, + }, + Hashes: corruptHashes, + From: 5, + To: 8, + Stream: stream, + }, + Peer: peerID, + }, + }, + }) + if err != nil { + t.Fatal(err) + } + + expectedError := errors.New("Message handler error: (msg code 1): error invalid hashes length (len: 40)") + if err := tester.TestDisconnected(&p2ptest.Disconnect{Peer: tester.IDs[0], Error: expectedError}); err != nil { + t.Fatal(err) + } +} + func TestStreamerDownstreamOfferedHashesMsgExchange(t *testing.T) { tester, streamer, _, teardown, err := newStreamerTester(t) defer teardown() -- cgit From 3f7acbbeb929bc3a2a3073bae15977ec69761bab Mon Sep 17 00:00:00 2001 From: Balint Gabor Date: Tue, 25 Sep 2018 17:35:54 +0200 Subject: swarm: prevent forever running retrieve request loops --- swarm/network/stream/delivery.go | 6 +++++- swarm/network/stream/stream.go | 2 +- 2 files changed, 6 insertions(+), 2 deletions(-) (limited to 'swarm/network/stream') diff --git a/swarm/network/stream/delivery.go b/swarm/network/stream/delivery.go index 431136ab1..c2adb1009 100644 --- a/swarm/network/stream/delivery.go +++ b/swarm/network/stream/delivery.go @@ -128,6 +128,7 @@ func (s *SwarmChunkServer) GetData(ctx context.Context, key []byte) ([]byte, err type RetrieveRequestMsg struct { Addr storage.Address SkipCheck bool + HopCount uint8 } func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req *RetrieveRequestMsg) error { @@ -148,7 +149,9 @@ func (d *Delivery) handleRetrieveRequestMsg(ctx context.Context, sp *Peer, req * var cancel func() // TODO: do something with this hardcoded timeout, maybe use TTL in the future - ctx, cancel = context.WithTimeout(context.WithValue(ctx, "peer", sp.ID().String()), network.RequestTimeout) + ctx = context.WithValue(ctx, "peer", sp.ID().String()) + ctx = context.WithValue(ctx, "hopcount", req.HopCount) + ctx, cancel = context.WithTimeout(ctx, network.RequestTimeout) go func() { select { @@ -247,6 +250,7 @@ func (d *Delivery) RequestFromPeers(ctx context.Context, req *network.Request) ( err := sp.SendPriority(ctx, &RetrieveRequestMsg{ Addr: req.Addr, SkipCheck: req.SkipCheck, + HopCount: req.HopCount, }, Top) if err != nil { return nil, nil, err diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go index ea7cce8cb..65b8dff5a 100644 --- a/swarm/network/stream/stream.go +++ b/swarm/network/stream/stream.go @@ -639,7 +639,7 @@ func (c *clientParams) clientCreated() { // Spec is the spec of the streamer protocol var Spec = &protocols.Spec{ Name: "stream", - Version: 6, + Version: 7, MaxMsgSize: 10 * 1024 * 1024, Messages: []interface{}{ UnsubscribeMsg{}, -- cgit