diff --git a/les/flowcontrol/control.go b/les/flowcontrol/control.go index d50eb809cc..e1751507e5 100644 --- a/les/flowcontrol/control.go +++ b/les/flowcontrol/control.go @@ -31,7 +31,7 @@ type ServerParams struct { } type ClientNode struct { - params *ServerParams + params ServerParams bufValue uint64 lastTime mclock.AbsTime lock sync.Mutex @@ -39,7 +39,7 @@ type ClientNode struct { cmNode *cmNode } -func NewClientNode(cm *ClientManager, params *ServerParams) *ClientNode { +func NewClientNode(cm *ClientManager, params ServerParams) *ClientNode { node := &ClientNode{ cm: cm, params: params, @@ -96,13 +96,13 @@ func (peer *ClientNode) RequestProcessed(cost uint64) (bv, realCost uint64) { type ServerNode struct { bufEstimate uint64 lastTime mclock.AbsTime - params *ServerParams + params ServerParams sumCost uint64 // sum of req costs sent to this server pending map[uint64]uint64 // value = sumCost after sending the given req lock sync.RWMutex } -func NewServerNode(params *ServerParams) *ServerNode { +func NewServerNode(params ServerParams) *ServerNode { return &ServerNode{ bufEstimate: params.BufLimit, lastTime: mclock.Now(), @@ -111,6 +111,22 @@ func NewServerNode(params *ServerParams) *ServerNode { } } +// UpdateParams updates flow control parameters +func (peer *ServerNode) UpdateParams(params ServerParams) { + peer.lock.Lock() + defer peer.lock.Unlock() + + peer.recalcBLE(mclock.Now()) + if params.BufLimit > peer.params.BufLimit { + peer.bufEstimate += params.BufLimit - peer.params.BufLimit + } else { + if peer.bufEstimate > params.BufLimit { + peer.bufEstimate = params.BufLimit + } + } + peer.params = params +} + func (peer *ServerNode) recalcBLE(time mclock.AbsTime) { dt := uint64(time - peer.lastTime) if time < peer.lastTime { diff --git a/les/handler.go b/les/handler.go index 19ccbcd2b6..1fcc0c4699 100644 --- a/les/handler.go +++ b/les/handler.go @@ -380,17 +380,25 @@ func (pm *ProtocolManager) handleMsg(p *peer) error { return errResp(ErrDecode, "%v: %v", msg, err) } - if p.requestAnnounceType == announceTypeSigned { - if err := req.checkSignature(p.ID()); err != nil { - p.Log().Trace("Invalid announcement signature", "err", err) - return err - } - p.Log().Trace("Valid announcement signature") + update, size := req.Update.decode() + if p.rejectUpdate(size) { + return errResp(ErrRequestRejected, "") } + p.updateFlowControl(update) - p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth) - if pm.fetcher != nil { - pm.fetcher.announce(p, &req) + if req.Hash != (common.Hash{}) { + if p.requestAnnounceType == announceTypeSigned { + if err := req.checkSignature(p.ID(), update); err != nil { + p.Log().Trace("Invalid announcement signature", "err", err) + return err + } + p.Log().Trace("Valid announcement signature") + } + + p.Log().Trace("Announce message content", "number", req.Number, "hash", req.Hash, "td", req.Td, "reorg", req.ReorgDepth) + if pm.fetcher != nil { + pm.fetcher.announce(p, &req) + } } case GetBlockHeadersMsg: diff --git a/les/helper_test.go b/les/helper_test.go index b46d41f174..8de187e9a0 100644 --- a/les/helper_test.go +++ b/les/helper_test.go @@ -184,7 +184,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor srv := &LesServer{lesCommons: lesCommons{protocolManager: pm}} pm.server = srv - srv.defParams = &flowcontrol.ServerParams{ + srv.defParams = flowcontrol.ServerParams{ BufLimit: testBufLimit, MinRecharge: 1, } @@ -313,7 +313,7 @@ func (p *testPeer) handshake(t *testing.T, td *big.Int, head common.Hash, headNu t.Fatalf("status send: %v", err) } - p.fcServerParams = &flowcontrol.ServerParams{ + p.fcServerParams = flowcontrol.ServerParams{ BufLimit: testBufLimit, MinRecharge: 1, } diff --git a/les/peer.go b/les/peer.go index 678384f0eb..4b69c70350 100644 --- a/les/peer.go +++ b/les/peer.go @@ -25,6 +25,7 @@ import ( "time" "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/common/mclock" "github.com/ethereum/go-ethereum/core/types" "github.com/ethereum/go-ethereum/eth" "github.com/ethereum/go-ethereum/les/flowcontrol" @@ -42,6 +43,12 @@ var ( const maxResponseErrors = 50 // number of invalid responses tolerated (makes the protocol less brittle but still avoids spam) +// bandwidth limitation for parameter updates +const ( + allowedUpdateBytes = 100000 // initial/maximum allowed update size + allowedUpdateRate = time.Millisecond * 10 // time constant for recharging one byte of allowance +) + const ( announceTypeNone = iota announceTypeSimple @@ -69,10 +76,12 @@ type peer struct { poolEntry *poolEntry hasBlock func(common.Hash, uint64, bool) bool responseErrors int + updateCounter uint64 + updateTime mclock.AbsTime fcClient *flowcontrol.ClientNode // nil if the peer is server only fcServer *flowcontrol.ServerNode // nil if the peer is client only - fcServerParams *flowcontrol.ServerParams + fcServerParams flowcontrol.ServerParams fcCosts requestCostTable } @@ -89,6 +98,27 @@ func newPeer(version int, network uint64, p *p2p.Peer, rw p2p.MsgReadWriter) *pe } } +// rejectUpdate returns true if a parameter update has to be rejected because +// the size and/or rate of updates exceed the bandwidth limitation +func (p *peer) rejectUpdate(size uint64) bool { + now := mclock.Now() + if p.updateCounter == 0 { + p.updateTime = now + } else { + dt := now - p.updateTime + r := uint64(dt / mclock.AbsTime(allowedUpdateRate)) + if p.updateCounter > r { + p.updateCounter -= r + p.updateTime += mclock.AbsTime(allowedUpdateRate * time.Duration(r)) + } else { + p.updateCounter = 0 + p.updateTime = now + } + } + p.updateCounter += size + return p.updateCounter > allowedUpdateBytes +} + func (p *peer) canQueue() bool { return p.sendQueue.canQueue() } @@ -340,12 +370,14 @@ func (l keyValueList) add(key string, val interface{}) keyValueList { return append(l, entry) } -func (l keyValueList) decode() keyValueMap { +func (l keyValueList) decode() (keyValueMap, uint64) { m := make(keyValueMap) + var size uint64 for _, entry := range l { m[entry.Key] = entry.Value + size += uint64(len(entry.Key)) + uint64(len(entry.Value)) + 8 } - return m + return m, size } func (m keyValueMap) get(key string, val interface{}) error { @@ -418,7 +450,10 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis if err != nil { return err } - recv := recvList.decode() + recv, size := recvList.decode() + if p.rejectUpdate(size) { + return errResp(ErrRequestRejected, "") + } var rGenesis, rHash common.Hash var rVersion, rNetwork, rNum uint64 @@ -471,7 +506,7 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis if recv.get("txRelay", nil) != nil { return errResp(ErrUselessPeer, "peer cannot relay transactions") } - params := &flowcontrol.ServerParams{} + var params flowcontrol.ServerParams if err := recv.get("flowControl/BL", ¶ms.BufLimit); err != nil { return err } @@ -491,6 +526,30 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis return nil } +// updateFlowControl updates the flow control parameters belonging to the server +// node if the announced key/value set contains relevant fields +func (p *peer) updateFlowControl(update keyValueMap) { + if p.fcServer == nil { + return + } + params := p.fcServerParams + updateParams := false + if update.get("flowControl/BL", ¶ms.BufLimit) == nil { + updateParams = true + } + if update.get("flowControl/MRR", ¶ms.MinRecharge) == nil { + updateParams = true + } + if updateParams { + p.fcServerParams = params + p.fcServer.UpdateParams(params) + } + var MRC RequestCostList + if update.get("flowControl/MRC", &MRC) == nil { + p.fcCosts = MRC.decode() + } +} + // String implements fmt.Stringer. func (p *peer) String() string { return fmt.Sprintf("Peer %s [%s]", p.id, diff --git a/les/protocol.go b/les/protocol.go index 0b24f5aedb..03e1ea8275 100644 --- a/les/protocol.go +++ b/les/protocol.go @@ -146,9 +146,9 @@ func (a *announceData) sign(privKey *ecdsa.PrivateKey) { } // checkSignature verifies if the block announcement has a valid signature by the given pubKey -func (a *announceData) checkSignature(id enode.ID) error { +func (a *announceData) checkSignature(id enode.ID, update keyValueMap) error { var sig []byte - if err := a.Update.decode().get("sign", &sig); err != nil { + if err := update.get("sign", &sig); err != nil { return err } rlp, _ := rlp.EncodeToBytes(announceBlock{a.Hash, a.Number, a.Td}) diff --git a/les/server.go b/les/server.go index 2fa0456d69..3a1978113d 100644 --- a/les/server.go +++ b/les/server.go @@ -43,7 +43,7 @@ type LesServer struct { fcManager *flowcontrol.ClientManager // nil if our node is client only fcCostStats *requestCostStats - defParams *flowcontrol.ServerParams + defParams flowcontrol.ServerParams lesTopics []discv5.Topic privateKey *ecdsa.PrivateKey quitSync chan struct{} @@ -98,7 +98,7 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) { srv.chtIndexer.Start(eth.BlockChain()) pm.server = srv - srv.defParams = &flowcontrol.ServerParams{ + srv.defParams = flowcontrol.ServerParams{ BufLimit: 300000000, MinRecharge: 50000, }