From aeb733623e79d9b8f03036ec1a4c0717185b527d Mon Sep 17 00:00:00 2001
From: Elad <theman@elad.im>
Date: Fri, 19 Oct 2018 10:50:25 +0200
Subject: [PATCH] swarm/network: disallow historical retrieval requests
(#17936)
---
swarm/network/stream/delivery_test.go | 2 +-
swarm/network/stream/messages.go | 1 +
swarm/network/stream/snapshot_retrieval_test.go | 4 ++--
swarm/network/stream/snapshot_sync_test.go | 4 ++--
swarm/network/stream/stream.go | 7 +++++--
5 files changed, 11 insertions(+), 7 deletions(-)
diff --git a/swarm/network/stream/delivery_test.go b/swarm/network/stream/delivery_test.go
index ec6148a0b..949645558 100644
--- a/swarm/network/stream/delivery_test.go
+++ b/swarm/network/stream/delivery_test.go
@@ -416,7 +416,7 @@ func testDeliveryFromNodes(t *testing.T, nodes, conns, chunkCount int, skipCheck
return fmt.Errorf("No registry")
}
registry := item.(*Registry)
- err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), NewRange(0, 0), Top)
+ err = registry.Subscribe(sid, NewStream(swarmChunkServerStreamName, "", true), nil, Top)
if err != nil {
return err
}
diff --git a/swarm/network/stream/messages.go b/swarm/network/stream/messages.go
index 68503fe1f..0fe3e5eb4 100644
--- a/swarm/network/stream/messages.go
+++ b/swarm/network/stream/messages.go
@@ -158,6 +158,7 @@ type SubscribeErrorMsg struct {
}
func (p *Peer) handleSubscribeErrorMsg(req *SubscribeErrorMsg) (err error) {
+ //TODO the error should be channeled to whoever calls the subscribe
return fmt.Errorf("subscribe to peer %s: %v", p.ID(), req.Error)
}
diff --git a/swarm/network/stream/snapshot_retrieval_test.go b/swarm/network/stream/snapshot_retrieval_test.go
index 7625e02d7..b81cfc0ca 100644
--- a/swarm/network/stream/snapshot_retrieval_test.go
+++ b/swarm/network/stream/snapshot_retrieval_test.go
@@ -218,7 +218,7 @@ func runFileRetrievalTest(nodeCount int) error {
reader, _ := fileStore.Retrieve(context.TODO(), hash)
//check that we can read the file size and that it corresponds to the generated file size
if s, err := reader.Size(ctx, nil); err != nil || s != int64(len(randomFiles[i])) {
- log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id)
+ log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id)
time.Sleep(500 * time.Millisecond)
continue REPEAT
}
@@ -309,7 +309,7 @@ func runRetrievalTest(chunkCount int, nodeCount int) error {
reader, _ := fileStore.Retrieve(context.TODO(), hash)
//check that we can read the chunk size and that it corresponds to the generated chunk size
if s, err := reader.Size(ctx, nil); err != nil || s != int64(chunkSize) {
- log.Warn("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s)
+ log.Debug("Retrieve error", "err", err, "hash", hash, "nodeId", id, "size", s)
time.Sleep(500 * time.Millisecond)
continue REPEAT
}
diff --git a/swarm/network/stream/snapshot_sync_test.go b/swarm/network/stream/snapshot_sync_test.go
index 96e92c5cf..8d89f28cb 100644
--- a/swarm/network/stream/snapshot_sync_test.go
+++ b/swarm/network/stream/snapshot_sync_test.go
@@ -310,7 +310,7 @@ func runSim(conf *synctestConfig, ctx context.Context, sim *simulation.Simulatio
_, err = lstore.Get(ctx, chunk)
}
if err != nil {
- log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+ log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
// Do not get crazy with logging the warn message
time.Sleep(500 * time.Millisecond)
continue REPEAT
@@ -514,7 +514,7 @@ func testSyncingViaDirectSubscribe(t *testing.T, chunkCount int, nodeCount int)
_, err = lstore.Get(ctx, chunk)
}
if err != nil {
- log.Warn(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
+ log.Debug(fmt.Sprintf("Chunk %s NOT found for id %s", chunk, id))
// Do not get crazy with logging the warn message
time.Sleep(500 * time.Millisecond)
continue REPEAT
diff --git a/swarm/network/stream/stream.go b/swarm/network/stream/stream.go
index 1dc2a8cba..9d0e6c68b 100644
--- a/swarm/network/stream/stream.go
+++ b/swarm/network/stream/stream.go
@@ -18,6 +18,7 @@ package stream
import (
"context"
+ "errors"
"fmt"
"math"
"sync"
@@ -96,7 +97,10 @@ func NewRegistry(localID enode.ID, delivery *Delivery, syncChunkStore storage.Sy
delivery.getPeer = streamer.getPeer
if options.DoServeRetrieve {
- streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, _ bool) (Server, error) {
+ streamer.RegisterServerFunc(swarmChunkServerStreamName, func(_ *Peer, _ string, live bool) (Server, error) {
+ if !live {
+ return nil, errors.New("only live retrieval requests supported")
+ }
return NewSwarmChunkServer(delivery.chunkStore), nil
})
}
@@ -279,7 +283,6 @@ func (r *Registry) Subscribe(peerId enode.ID, s Stream, h *Range, priority uint8
if err != nil {
return err
}
-
if s.Live && h != nil {
if err := peer.setClientParams(
getHistoryStream(s),
--
GitLab