aboutsummaryrefslogtreecommitdiffstats
path: root/les
diff options
context:
space:
mode:
Diffstat (limited to 'les')
-rw-r--r--les/backend.go31
-rw-r--r--les/fetcher.go300
-rw-r--r--les/fetcher_test.go155
-rw-r--r--les/handler.go47
-rw-r--r--les/handler_test.go2
-rw-r--r--les/helper_test.go14
-rw-r--r--les/odr.go5
-rw-r--r--les/peer.go46
-rw-r--r--les/peer_test.go297
-rw-r--r--les/server.go38
-rw-r--r--les/serverpool.go101
-rw-r--r--les/txrelay.go2
-rw-r--r--les/ulc.go39
-rw-r--r--les/ulc_test.go239
14 files changed, 1156 insertions, 160 deletions
diff --git a/les/backend.go b/les/backend.go
index d0db71019..cd99f8f81 100644
--- a/les/backend.go
+++ b/les/backend.go
@@ -109,8 +109,12 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
bloomIndexer: eth.NewBloomIndexer(chainDb, params.BloomBitsBlocksClient, params.HelperTrieConfirmations),
}
+ var trustedNodes []string
+ if leth.config.ULC != nil {
+ trustedNodes = leth.config.ULC.TrustedServers
+ }
leth.relay = NewLesTxRelay(peers, leth.reqDist)
- leth.serverPool = newServerPool(chainDb, quitSync, &leth.wg)
+ leth.serverPool = newServerPool(chainDb, quitSync, &leth.wg, trustedNodes)
leth.retriever = newRetrieveManager(peers, leth.reqDist, leth.serverPool)
leth.odr = NewLesOdr(chainDb, light.DefaultClientIndexerConfig, leth.retriever)
@@ -136,10 +140,33 @@ func New(ctx *node.ServiceContext, config *eth.Config) (*LightEthereum, error) {
}
leth.txPool = light.NewTxPool(leth.chainConfig, leth.blockchain, leth.relay)
- if leth.protocolManager, err = NewProtocolManager(leth.chainConfig, light.DefaultClientIndexerConfig, true, config.NetworkId, leth.eventMux, leth.engine, leth.peers, leth.blockchain, nil, chainDb, leth.odr, leth.relay, leth.serverPool, quitSync, &leth.wg); err != nil {
+
+ if leth.protocolManager, err = NewProtocolManager(
+ leth.chainConfig,
+ light.DefaultClientIndexerConfig,
+ true,
+ config.NetworkId,
+ leth.eventMux,
+ leth.engine,
+ leth.peers,
+ leth.blockchain,
+ nil,
+ chainDb,
+ leth.odr,
+ leth.relay,
+ leth.serverPool,
+ quitSync,
+ &leth.wg,
+ config.ULC); err != nil {
return nil, err
}
+
+ if leth.protocolManager.isULCEnabled() {
+ log.Warn("Ultra light client is enabled", "trustedNodes", len(leth.protocolManager.ulc.trustedKeys), "minTrustedFraction", leth.protocolManager.ulc.minTrustedFraction)
+ leth.blockchain.DisableCheckFreq()
+ }
leth.ApiBackend = &LesApiBackend{leth, nil}
+
gpoParams := config.GPO
if gpoParams.Default == nil {
gpoParams.Default = config.MinerGasPrice
diff --git a/les/fetcher.go b/les/fetcher.go
index 2615f69df..aa3101af7 100644
--- a/les/fetcher.go
+++ b/les/fetcher.go
@@ -43,7 +43,7 @@ const (
type lightFetcher struct {
pm *ProtocolManager
odr *LesOdr
- chain *light.LightChain
+ chain lightChain
lock sync.Mutex // lock protects access to the fetcher's internal state variables except sent requests
maxConfirmedTd *big.Int
@@ -52,11 +52,19 @@ type lightFetcher struct {
syncing bool
syncDone chan *peer
- reqMu sync.RWMutex // reqMu protects access to sent header fetch requests
- requested map[uint64]fetchRequest
- deliverChn chan fetchResponse
- timeoutChn chan uint64
- requestChn chan bool // true if initiated from outside
+ reqMu sync.RWMutex // reqMu protects access to sent header fetch requests
+ requested map[uint64]fetchRequest
+ deliverChn chan fetchResponse
+ timeoutChn chan uint64
+ requestChn chan bool // true if initiated from outside
+ lastTrustedHeader *types.Header
+}
+
+// lightChain extends the BlockChain interface by locking.
+type lightChain interface {
+ BlockChain
+ LockChain()
+ UnlockChain()
}
// fetcherPeerInfo holds fetcher-specific information about each active peer
@@ -145,6 +153,7 @@ func (f *lightFetcher) syncLoop() {
reqID uint64
syncing bool
)
+
if !f.syncing && !(newAnnounce && s) {
rq, reqID, syncing = f.nextRequest()
}
@@ -227,7 +236,6 @@ func (f *lightFetcher) registerPeer(p *peer) {
f.lock.Lock()
defer f.lock.Unlock()
-
f.peers[p] = &fetcherPeerInfo{nodeByHash: make(map[common.Hash]*fetcherTreeNode)}
}
@@ -280,8 +288,10 @@ func (f *lightFetcher) announce(p *peer, head *announceData) {
fp.nodeCnt = 0
fp.nodeByHash = make(map[common.Hash]*fetcherTreeNode)
}
+ // check if the node count is too high to add new nodes, discard oldest ones if necessary
if n != nil {
- // check if the node count is too high to add new nodes, discard oldest ones if necessary
+ // n is now the reorg common ancestor, add a new branch of nodes
+ // check if the node count is too high to add new nodes
locked := false
for uint64(fp.nodeCnt)+head.Number-n.number > maxNodeCount && fp.root != nil {
if !locked {
@@ -325,6 +335,7 @@ func (f *lightFetcher) announce(p *peer, head *announceData) {
fp.nodeByHash[n.hash] = n
}
}
+
if n == nil {
// could not find reorg common ancestor or had to delete entire tree, a new root and a resync is needed
if fp.root != nil {
@@ -411,25 +422,13 @@ func (f *lightFetcher) requestedID(reqID uint64) bool {
// to be downloaded starting from the head backwards is also returned
func (f *lightFetcher) nextRequest() (*distReq, uint64, bool) {
var (
- bestHash common.Hash
- bestAmount uint64
+ bestHash common.Hash
+ bestAmount uint64
+ bestTd *big.Int
+ bestSyncing bool
)
- bestTd := f.maxConfirmedTd
- bestSyncing := false
+ bestHash, bestAmount, bestTd, bestSyncing = f.findBestRequest()
- for p, fp := range f.peers {
- for hash, n := range fp.nodeByHash {
- if !f.checkKnownNode(p, n) && !n.requested && (bestTd == nil || n.td.Cmp(bestTd) >= 0) {
- amount := f.requestAmount(p, n)
- if bestTd == nil || n.td.Cmp(bestTd) > 0 || amount < bestAmount {
- bestHash = hash
- bestAmount = amount
- bestTd = n.td
- bestSyncing = fp.bestConfirmed == nil || fp.root == nil || !f.checkKnownNode(p, fp.root)
- }
- }
- }
- }
if bestTd == f.maxConfirmedTd {
return nil, 0, false
}
@@ -437,72 +436,140 @@ func (f *lightFetcher) nextRequest() (*distReq, uint64, bool) {
var rq *distReq
reqID := genReqID()
if bestSyncing {
- rq = &distReq{
- getCost: func(dp distPeer) uint64 {
- return 0
- },
- canSend: func(dp distPeer) bool {
- p := dp.(*peer)
- f.lock.Lock()
- defer f.lock.Unlock()
-
- fp := f.peers[p]
- return fp != nil && fp.nodeByHash[bestHash] != nil
- },
- request: func(dp distPeer) func() {
- go func() {
- p := dp.(*peer)
- p.Log().Debug("Synchronisation started")
- f.pm.synchronise(p)
- f.syncDone <- p
- }()
- return nil
- },
- }
+ rq = f.newFetcherDistReqForSync(bestHash)
} else {
- rq = &distReq{
- getCost: func(dp distPeer) uint64 {
- p := dp.(*peer)
- return p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
- },
- canSend: func(dp distPeer) bool {
+ rq = f.newFetcherDistReq(bestHash, reqID, bestAmount)
+ }
+ return rq, reqID, bestSyncing
+}
+
+// findBestRequest finds the best head to request that has been announced by but not yet requested from a known peer.
+// It also returns the announced Td (which should be verified after fetching the head),
+// the necessary amount to request and whether a downloader sync is necessary instead of a normal header request.
+func (f *lightFetcher) findBestRequest() (bestHash common.Hash, bestAmount uint64, bestTd *big.Int, bestSyncing bool) {
+ bestTd = f.maxConfirmedTd
+ bestSyncing = false
+
+ for p, fp := range f.peers {
+ for hash, n := range fp.nodeByHash {
+ if f.checkKnownNode(p, n) || n.requested {
+ continue
+ }
+
+ //if ulc mode is disabled, isTrustedHash returns true
+ amount := f.requestAmount(p, n)
+ if (bestTd == nil || n.td.Cmp(bestTd) > 0 || amount < bestAmount) && (f.isTrustedHash(hash) || f.maxConfirmedTd.Int64() == 0) {
+ bestHash = hash
+ bestTd = n.td
+ bestAmount = amount
+ bestSyncing = fp.bestConfirmed == nil || fp.root == nil || !f.checkKnownNode(p, fp.root)
+ }
+ }
+ }
+ return
+}
+
+// isTrustedHash checks if the block can be trusted by the minimum trusted fraction.
+func (f *lightFetcher) isTrustedHash(hash common.Hash) bool {
+ if !f.pm.isULCEnabled() {
+ return true
+ }
+
+ var numAgreed int
+ for p, fp := range f.peers {
+ if !p.isTrusted {
+ continue
+ }
+ if _, ok := fp.nodeByHash[hash]; !ok {
+ continue
+ }
+
+ numAgreed++
+ }
+
+ return 100*numAgreed/len(f.pm.ulc.trustedKeys) >= f.pm.ulc.minTrustedFraction
+}
+
+func (f *lightFetcher) newFetcherDistReqForSync(bestHash common.Hash) *distReq {
+ return &distReq{
+ getCost: func(dp distPeer) uint64 {
+ return 0
+ },
+ canSend: func(dp distPeer) bool {
+ p := dp.(*peer)
+ f.lock.Lock()
+ defer f.lock.Unlock()
+
+ if p.isOnlyAnnounce {
+ return false
+ }
+
+ fp := f.peers[p]
+ return fp != nil && fp.nodeByHash[bestHash] != nil
+ },
+ request: func(dp distPeer) func() {
+ if f.pm.isULCEnabled() {
+ //keep last trusted header before sync
+ f.setLastTrustedHeader(f.chain.CurrentHeader())
+ }
+ go func() {
p := dp.(*peer)
- f.lock.Lock()
- defer f.lock.Unlock()
+ p.Log().Debug("Synchronisation started")
+ f.pm.synchronise(p)
+ f.syncDone <- p
+ }()
+ return nil
+ },
+ }
+}
- fp := f.peers[p]
- if fp == nil {
- return false
- }
+// newFetcherDistReq creates a new request for the distributor.
+func (f *lightFetcher) newFetcherDistReq(bestHash common.Hash, reqID uint64, bestAmount uint64) *distReq {
+ return &distReq{
+ getCost: func(dp distPeer) uint64 {
+ p := dp.(*peer)
+ return p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
+ },
+ canSend: func(dp distPeer) bool {
+ p := dp.(*peer)
+ f.lock.Lock()
+ defer f.lock.Unlock()
+
+ if p.isOnlyAnnounce {
+ return false
+ }
+
+ fp := f.peers[p]
+ if fp == nil {
+ return false
+ }
+ n := fp.nodeByHash[bestHash]
+ return n != nil && !n.requested
+ },
+ request: func(dp distPeer) func() {
+ p := dp.(*peer)
+ f.lock.Lock()
+ fp := f.peers[p]
+ if fp != nil {
n := fp.nodeByHash[bestHash]
- return n != nil && !n.requested
- },
- request: func(dp distPeer) func() {
- p := dp.(*peer)
- f.lock.Lock()
- fp := f.peers[p]
- if fp != nil {
- n := fp.nodeByHash[bestHash]
- if n != nil {
- n.requested = true
- }
+ if n != nil {
+ n.requested = true
}
- f.lock.Unlock()
-
- cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
- p.fcServer.QueueRequest(reqID, cost)
- f.reqMu.Lock()
- f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}
- f.reqMu.Unlock()
- go func() {
- time.Sleep(hardRequestTimeout)
- f.timeoutChn <- reqID
- }()
- return func() { p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true) }
- },
- }
+ }
+ f.lock.Unlock()
+
+ cost := p.GetRequestCost(GetBlockHeadersMsg, int(bestAmount))
+ p.fcServer.QueueRequest(reqID, cost)
+ f.reqMu.Lock()
+ f.requested[reqID] = fetchRequest{hash: bestHash, amount: bestAmount, peer: p, sent: mclock.Now()}
+ f.reqMu.Unlock()
+ go func() {
+ time.Sleep(hardRequestTimeout)
+ f.timeoutChn <- reqID
+ }()
+ return func() { p.RequestHeadersByHash(reqID, cost, bestHash, int(bestAmount), 0, true) }
+ },
}
- return rq, reqID, bestSyncing
}
// deliverHeaders delivers header download request responses for processing
@@ -520,6 +587,7 @@ func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) boo
for i, header := range resp.headers {
headers[int(req.amount)-1-i] = header
}
+
if _, err := f.chain.InsertHeaderChain(headers, 1); err != nil {
if err == consensus.ErrFutureBlock {
return true
@@ -544,6 +612,7 @@ func (f *lightFetcher) processResponse(req fetchRequest, resp fetchResponse) boo
// downloaded and validated batch or headers
func (f *lightFetcher) newHeaders(headers []*types.Header, tds []*big.Int) {
var maxTd *big.Int
+
for p, fp := range f.peers {
if !f.checkAnnouncedHeaders(fp, headers, tds) {
p.Log().Debug("Inconsistent announcement")
@@ -553,6 +622,7 @@ func (f *lightFetcher) newHeaders(headers []*types.Header, tds []*big.Int) {
maxTd = fp.confirmedTd
}
}
+
if maxTd != nil {
f.updateMaxConfirmedTd(maxTd)
}
@@ -640,22 +710,72 @@ func (f *lightFetcher) checkSyncedHeaders(p *peer) {
p.Log().Debug("Unknown peer to check sync headers")
return
}
+
n := fp.lastAnnounced
var td *big.Int
+
+ var h *types.Header
+ if f.pm.isULCEnabled() {
+ var unapprovedHashes []common.Hash
+ // Overwrite last announced for ULC mode
+ h, unapprovedHashes = f.lastTrustedTreeNode(p)
+ //rollback untrusted blocks
+ f.chain.Rollback(unapprovedHashes)
+ //overwrite to last trusted
+ n = fp.nodeByHash[h.Hash()]
+ }
+
+ //find last valid block
for n != nil {
if td = f.chain.GetTd(n.hash, n.number); td != nil {
break
}
n = n.parent
}
- // now n is the latest downloaded header after syncing
+
+ // Now n is the latest downloaded/approved header after syncing
if n == nil {
p.Log().Debug("Synchronisation failed")
go f.pm.removePeer(p.id)
- } else {
- header := f.chain.GetHeader(n.hash, n.number)
- f.newHeaders([]*types.Header{header}, []*big.Int{td})
+ return
}
+ header := f.chain.GetHeader(n.hash, n.number)
+ f.newHeaders([]*types.Header{header}, []*big.Int{td})
+}
+
+// lastTrustedTreeNode return last approved treeNode and a list of unapproved hashes
+func (f *lightFetcher) lastTrustedTreeNode(p *peer) (*types.Header, []common.Hash) {
+ unapprovedHashes := make([]common.Hash, 0)
+ current := f.chain.CurrentHeader()
+
+ if f.lastTrustedHeader == nil {
+ return current, unapprovedHashes
+ }
+
+ canonical := f.chain.CurrentHeader()
+ if canonical.Number.Uint64() > f.lastTrustedHeader.Number.Uint64() {
+ canonical = f.chain.GetHeaderByNumber(f.lastTrustedHeader.Number.Uint64())
+ }
+ commonAncestor := rawdb.FindCommonAncestor(f.pm.chainDb, canonical, f.lastTrustedHeader)
+ if commonAncestor == nil {
+ log.Error("Common ancestor of last trusted header and canonical header is nil", "canonical hash", canonical.Hash(), "trusted hash", f.lastTrustedHeader.Hash())
+ return current, unapprovedHashes
+ }
+
+ for current.Hash() == commonAncestor.Hash() {
+ if f.isTrustedHash(current.Hash()) {
+ break
+ }
+ unapprovedHashes = append(unapprovedHashes, current.Hash())
+ current = f.chain.GetHeader(current.ParentHash, current.Number.Uint64()-1)
+ }
+ return current, unapprovedHashes
+}
+
+func (f *lightFetcher) setLastTrustedHeader(h *types.Header) {
+ f.lock.Lock()
+ defer f.lock.Unlock()
+ f.lastTrustedHeader = h
}
// checkKnownNode checks if a block tree node is known (downloaded and validated)
@@ -747,6 +867,7 @@ func (f *lightFetcher) updateMaxConfirmedTd(td *big.Int) {
if f.lastUpdateStats != nil {
f.lastUpdateStats.next = newEntry
}
+
f.lastUpdateStats = newEntry
for p := range f.peers {
f.checkUpdateStats(p, newEntry)
@@ -769,6 +890,7 @@ func (f *lightFetcher) checkUpdateStats(p *peer, newEntry *updateStatsEntry) {
p.Log().Debug("Unknown peer to check update stats")
return
}
+
if newEntry != nil && fp.firstUpdateStats == nil {
fp.firstUpdateStats = newEntry
}
diff --git a/les/fetcher_test.go b/les/fetcher_test.go
new file mode 100644
index 000000000..4ae7f6d04
--- /dev/null
+++ b/les/fetcher_test.go
@@ -0,0 +1,155 @@
+package les
+
+import (
+ "math/big"
+ "testing"
+
+ "net"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/crypto"
+ "github.com/ethereum/go-ethereum/p2p"
+ "github.com/ethereum/go-ethereum/p2p/enode"
+)
+
+func TestFetcherULCPeerSelector(t *testing.T) {
+
+ var (
+ id1 enode.ID = newNodeID(t).ID()
+ id2 enode.ID = newNodeID(t).ID()
+ id3 enode.ID = newNodeID(t).ID()
+ id4 enode.ID = newNodeID(t).ID()
+ )
+
+ ftn1 := &fetcherTreeNode{
+ hash: common.HexToHash("1"),
+ td: big.NewInt(1),
+ }
+ ftn2 := &fetcherTreeNode{
+ hash: common.HexToHash("2"),
+ td: big.NewInt(2),
+ parent: ftn1,
+ }
+ ftn3 := &fetcherTreeNode{
+ hash: common.HexToHash("3"),
+ td: big.NewInt(3),
+ parent: ftn2,
+ }
+ lf := lightFetcher{
+ pm: &ProtocolManager{
+ ulc: &ulc{
+ trustedKeys: map[string]struct{}{
+ id1.String(): {},
+ id2.String(): {},
+ id3.String(): {},
+ id4.String(): {},
+ },
+ minTrustedFraction: 70,
+ },
+ },
+ maxConfirmedTd: ftn1.td,
+
+ peers: map[*peer]*fetcherPeerInfo{
+ {
+ id: "peer1",
+ Peer: p2p.NewPeer(id1, "peer1", []p2p.Cap{}),
+ isTrusted: true,
+ }: {
+ nodeByHash: map[common.Hash]*fetcherTreeNode{
+ ftn1.hash: ftn1,
+ ftn2.hash: ftn2,
+ },
+ },
+ {
+ Peer: p2p.NewPeer(id2, "peer2", []p2p.Cap{}),
+ id: "peer2",
+ isTrusted: true,
+ }: {
+ nodeByHash: map[common.Hash]*fetcherTreeNode{
+ ftn1.hash: ftn1,
+ ftn2.hash: ftn2,
+ },
+ },
+ {
+ id: "peer3",
+ Peer: p2p.NewPeer(id3, "peer3", []p2p.Cap{}),
+ isTrusted: true,
+ }: {
+ nodeByHash: map[common.Hash]*fetcherTreeNode{
+ ftn1.hash: ftn1,
+ ftn2.hash: ftn2,
+ ftn3.hash: ftn3,
+ },
+ },
+ {
+ id: "peer4",
+ Peer: p2p.NewPeer(id4, "peer4", []p2p.Cap{}),
+ isTrusted: true,
+ }: {
+ nodeByHash: map[common.Hash]*fetcherTreeNode{
+ ftn1.hash: ftn1,
+ },
+ },
+ },
+ chain: &lightChainStub{
+ tds: map[common.Hash]*big.Int{},
+ headers: map[common.Hash]*types.Header{
+ ftn1.hash: {},
+ ftn2.hash: {},
+ ftn3.hash: {},
+ },
+ },
+ }
+ bestHash, bestAmount, bestTD, sync := lf.findBestRequest()
+
+ if bestTD == nil {
+ t.Fatal("Empty result")
+ }
+
+ if bestTD.Cmp(ftn2.td) != 0 {
+ t.Fatal("bad td", bestTD)
+ }
+ if bestHash != ftn2.hash {
+ t.Fatal("bad hash", bestTD)
+ }
+
+ _, _ = bestAmount, sync
+}
+
+type lightChainStub struct {
+ BlockChain
+ tds map[common.Hash]*big.Int
+ headers map[common.Hash]*types.Header
+ insertHeaderChainAssertFunc func(chain []*types.Header, checkFreq int) (int, error)
+}
+
+func (l *lightChainStub) GetHeader(hash common.Hash, number uint64) *types.Header {
+ if h, ok := l.headers[hash]; ok {
+ return h
+ }
+
+ return nil
+}
+
+func (l *lightChainStub) LockChain() {}
+func (l *lightChainStub) UnlockChain() {}
+
+func (l *lightChainStub) GetTd(hash common.Hash, number uint64) *big.Int {
+ if td, ok := l.tds[hash]; ok {
+ return td
+ }
+ return nil
+}
+
+func (l *lightChainStub) InsertHeaderChain(chain []*types.Header, checkFreq int) (int, error) {
+ return l.insertHeaderChainAssertFunc(chain, checkFreq)
+}
+
+func newNodeID(t *testing.T) *enode.Node {
+ key, err := crypto.GenerateKey()
+ if err != nil {
+ t.Fatal("generate key err:", err)
+ }
+ return enode.NewV4(&key.PublicKey, net.IP{}, 35000, 35000)
+}
diff --git a/les/handler.go b/les/handler.go
index 19ccbcd2b..46a1ed2d7 100644
--- a/les/handler.go
+++ b/les/handler.go
@@ -33,6 +33,7 @@ import (
"github.com/ethereum/go-ethereum/core/rawdb"
"github.com/ethereum/go-ethereum/core/state"
"github.com/ethereum/go-ethereum/core/types"
+ "github.com/ethereum/go-ethereum/eth"
"github.com/ethereum/go-ethereum/eth/downloader"
"github.com/ethereum/go-ethereum/ethdb"
"github.com/ethereum/go-ethereum/event"
@@ -119,12 +120,29 @@ type ProtocolManager struct {
// wait group is used for graceful shutdowns during downloading
// and processing
- wg *sync.WaitGroup
+ wg *sync.WaitGroup
+ ulc *ulc
}
// NewProtocolManager returns a new ethereum sub protocol manager. The Ethereum sub protocol manages peers capable
// with the ethereum network.
-func NewProtocolManager(chainConfig *params.ChainConfig, indexerConfig *light.IndexerConfig, lightSync bool, networkId uint64, mux *event.TypeMux, engine consensus.Engine, peers *peerSet, blockchain BlockChain, txpool txPool, chainDb ethdb.Database, odr *LesOdr, txrelay *LesTxRelay, serverPool *serverPool, quitSync chan struct{}, wg *sync.WaitGroup) (*ProtocolManager, error) {
+func NewProtocolManager(
+ chainConfig *params.ChainConfig,
+ indexerConfig *light.IndexerConfig,
+ lightSync bool,
+ networkId uint64,
+ mux *event.TypeMux,
+ engine consensus.Engine,
+ peers *peerSet,
+ blockchain BlockChain,
+ txpool txPool,
+ chainDb ethdb.Database,
+ odr *LesOdr,
+ txrelay *LesTxRelay,
+ serverPool *serverPool,
+ quitSync chan struct{},
+ wg *sync.WaitGroup,
+ ulcConfig *eth.ULCConfig) (*ProtocolManager, error) {
// Create the protocol manager with the base fields
manager := &ProtocolManager{
lightSync: lightSync,
@@ -149,6 +167,10 @@ func NewProtocolManager(chainConfig *params.ChainConfig, indexerConfig *light.In
manager.reqDist = odr.retriever.dist
}
+ if ulcConfig != nil {
+ manager.ulc = newULC(ulcConfig)
+ }
+
removePeer := manager.removePeer
if disableClientRemovePeer {
removePeer = func(id string) {}
@@ -234,7 +256,11 @@ func (pm *ProtocolManager) runPeer(version uint, p *p2p.Peer, rw p2p.MsgReadWrit
}
func (pm *ProtocolManager) newPeer(pv int, nv uint64, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
- return newPeer(pv, nv, p, newMeteredMsgWriter(rw))
+ var isTrusted bool
+ if pm.isULCEnabled() {
+ isTrusted = pm.ulc.isTrusted(p.ID())
+ }
+ return newPeer(pv, nv, isTrusted, p, newMeteredMsgWriter(rw))
}
// handle is the callback invoked to manage the life cycle of a les peer. When
@@ -276,6 +302,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
if rw, ok := p.rw.(*meteredMsgReadWriter); ok {
rw.Init(p.version)
}
+
// Register the peer locally
if err := pm.peers.Register(p); err != nil {
p.Log().Error("Light Ethereum peer registration failed", "err", err)
@@ -287,6 +314,7 @@ func (pm *ProtocolManager) handle(p *peer) error {
}
pm.removePeer(p.id)
}()
+
// Register the peer in the downloader. If the downloader considers it banned, we disconnect
if pm.lightSync {
p.lock.Lock()
@@ -371,16 +399,15 @@ func (pm *ProtocolManager) handleMsg(p *peer) error {
// Block header query, collect the requested headers and reply
case AnnounceMsg:
p.Log().Trace("Received announce message")
- if p.requestAnnounceType == announceTypeNone {
+ if p.announceType == announceTypeNone {
return errResp(ErrUnexpectedResponse, "")
}
-
var req announceData
if err := msg.Decode(&req); err != nil {
return errResp(ErrDecode, "%v: %v", msg, err)
}
- if p.requestAnnounceType == announceTypeSigned {
+ if p.announceType == announceTypeSigned {
if err := req.checkSignature(p.ID()); err != nil {
p.Log().Trace("Invalid announcement signature", "err", err)
return err
@@ -1175,6 +1202,14 @@ func (pm *ProtocolManager) txStatus(hashes []common.Hash) []txStatus {
return stats
}
+// isULCEnabled returns true if we can use ULC
+func (pm *ProtocolManager) isULCEnabled() bool {
+ if pm.ulc == nil || len(pm.ulc.trustedKeys) == 0 {
+ return false
+ }
+ return true
+}
+
// downloaderPeerNotify implements peerSetNotify
type downloaderPeerNotify ProtocolManager
diff --git a/les/handler_test.go b/les/handler_test.go
index 43be7f41b..72ba266b3 100644
--- a/les/handler_test.go
+++ b/les/handler_test.go
@@ -494,7 +494,7 @@ func TestGetBloombitsProofs(t *testing.T) {
func TestTransactionStatusLes2(t *testing.T) {
db := ethdb.NewMemDatabase()
- pm := newTestProtocolManagerMust(t, false, 0, nil, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, 0, nil, nil, nil, db, nil)
chain := pm.blockchain.(*core.BlockChain)
config := core.DefaultTxPoolConfig
config.Journal = ""
diff --git a/les/helper_test.go b/les/helper_test.go
index b46d41f17..02b1668c8 100644
--- a/les/helper_test.go
+++ b/les/helper_test.go
@@ -146,7 +146,7 @@ func testRCL() RequestCostList {
// newTestProtocolManager creates a new protocol manager for testing purposes,
// with the given number of blocks already known, potential notification
// channels for different events and relative chain indexers array.
-func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) (*ProtocolManager, error) {
+func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database, ulcConfig *eth.ULCConfig) (*ProtocolManager, error) {
var (
evmux = new(event.TypeMux)
engine = ethash.NewFaker()
@@ -176,7 +176,7 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
if lightSync {
indexConfig = light.TestClientIndexerConfig
}
- pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup))
+ pm, err := NewProtocolManager(gspec.Config, indexConfig, lightSync, NetworkId, evmux, engine, peers, chain, nil, db, odr, nil, nil, make(chan struct{}), new(sync.WaitGroup), ulcConfig)
if err != nil {
return nil, err
}
@@ -200,8 +200,8 @@ func newTestProtocolManager(lightSync bool, blocks int, generator func(int, *cor
// with the given number of blocks already known, potential notification
// channels for different events and relative chain indexers array. In case of an error, the constructor force-
// fails the test.
-func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database) *ProtocolManager {
- pm, err := newTestProtocolManager(lightSync, blocks, generator, odr, peers, db)
+func newTestProtocolManagerMust(t *testing.T, lightSync bool, blocks int, generator func(int, *core.BlockGen), odr *LesOdr, peers *peerSet, db ethdb.Database, ulcConfig *eth.ULCConfig) *ProtocolManager {
+ pm, err := newTestProtocolManager(lightSync, blocks, generator, odr, peers, db, ulcConfig)
if err != nil {
t.Fatalf("Failed to create protocol manager: %v", err)
}
@@ -343,7 +343,7 @@ func newServerEnv(t *testing.T, blocks int, protocol int, waitIndexers func(*cor
db := ethdb.NewMemDatabase()
cIndexer, bIndexer, btIndexer := testIndexers(db, nil, light.TestServerIndexerConfig)
- pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, nil, db)
+ pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, nil, db, nil)
peer, _ := newTestPeer(t, "peer", protocol, pm, true)
cIndexer.Start(pm.blockchain.(*core.BlockChain))
@@ -383,8 +383,8 @@ func newClientServerEnv(t *testing.T, blocks int, protocol int, waitIndexers fun
lcIndexer, lbIndexer, lbtIndexer := testIndexers(ldb, odr, light.TestClientIndexerConfig)
odr.SetIndexers(lcIndexer, lbtIndexer, lbIndexer)
- pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, peers, db)
- lpm := newTestProtocolManagerMust(t, true, 0, nil, odr, lPeers, ldb)
+ pm := newTestProtocolManagerMust(t, false, blocks, testChainGen, nil, peers, db, nil)
+ lpm := newTestProtocolManagerMust(t, true, 0, nil, odr, lPeers, ldb, nil)
startIndexers := func(clientMode bool, pm *ProtocolManager) {
if clientMode {
diff --git a/les/odr.go b/les/odr.go
index 9def05a67..f7592354d 100644
--- a/les/odr.go
+++ b/les/odr.go
@@ -109,7 +109,10 @@ func (odr *LesOdr) Retrieve(ctx context.Context, req light.OdrRequest) (err erro
},
canSend: func(dp distPeer) bool {
p := dp.(*peer)
- return lreq.CanSend(p)
+ if !p.isOnlyAnnounce {
+ return lreq.CanSend(p)
+ }
+ return false
},
request: func(dp distPeer) func() {
p := dp.(*peer)
diff --git a/les/peer.go b/les/peer.go
index 678384f0e..9ae94b20f 100644
--- a/les/peer.go
+++ b/les/peer.go
@@ -56,7 +56,7 @@ type peer struct {
version int // Protocol version negotiated
network uint64 // Network ID being on
- announceType, requestAnnounceType uint64
+ announceType uint64
id string
@@ -74,9 +74,12 @@ type peer struct {
fcServer *flowcontrol.ServerNode // nil if the peer is client only
fcServerParams *flowcontrol.ServerParams
fcCosts requestCostTable
+
+ isTrusted bool
+ isOnlyAnnounce bool
}
-func newPeer(version int, network uint64, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
+func newPeer(version int, network uint64, isTrusted bool, p *p2p.Peer, rw p2p.MsgReadWriter) *peer {
id := p.ID()
return &peer{
@@ -86,6 +89,7 @@ func newPeer(version int, network uint64, p *p2p.Peer, rw p2p.MsgReadWriter) *pe
network: network,
id: fmt.Sprintf("%x", id[:8]),
announceChn: make(chan announceData, 20),
+ isTrusted: isTrusted,
}
}
@@ -401,23 +405,32 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
send = send.add("headNum", headNum)
send = send.add("genesisHash", genesis)
if server != nil {
- send = send.add("serveHeaders", nil)
- send = send.add("serveChainSince", uint64(0))
- send = send.add("serveStateSince", uint64(0))
- send = send.add("txRelay", nil)
+ if !server.onlyAnnounce {
+ //only announce server. It sends only announse requests
+ send = send.add("serveHeaders", nil)
+ send = send.add("serveChainSince", uint64(0))
+ send = send.add("serveStateSince", uint64(0))
+ send = send.add("txRelay", nil)
+ }
send = send.add("flowControl/BL", server.defParams.BufLimit)
send = send.add("flowControl/MRR", server.defParams.MinRecharge)
list := server.fcCostStats.getCurrentList()
send = send.add("flowControl/MRC", list)
p.fcCosts = list.decode()
} else {
- p.requestAnnounceType = announceTypeSimple // set to default until "very light" client mode is implemented
- send = send.add("announceType", p.requestAnnounceType)
+ //on client node
+ p.announceType = announceTypeSimple
+ if p.isTrusted {
+ p.announceType = announceTypeSigned
+ }
+ send = send.add("announceType", p.announceType)
}
+
recvList, err := p.sendReceiveHandshake(send)
if err != nil {
return err
}
+
recv := recvList.decode()
var rGenesis, rHash common.Hash
@@ -452,25 +465,33 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
if int(rVersion) != p.version {
return errResp(ErrProtocolVersionMismatch, "%d (!= %d)", rVersion, p.version)
}
+
if server != nil {
// until we have a proper peer connectivity API, allow LES connection to other servers
/*if recv.get("serveStateSince", nil) == nil {
return errResp(ErrUselessPeer, "wanted client, got server")
}*/
if recv.get("announceType", &p.announceType) != nil {
+ //set default announceType on server side
p.announceType = announceTypeSimple
}
p.fcClient = flowcontrol.NewClientNode(server.fcManager, server.defParams)
} else {
+ //mark OnlyAnnounce server if "serveHeaders", "serveChainSince", "serveStateSince" or "txRelay" fields don't exist
if recv.get("serveChainSince", nil) != nil {
- return errResp(ErrUselessPeer, "peer cannot serve chain")
+ p.isOnlyAnnounce = true
}
if recv.get("serveStateSince", nil) != nil {
- return errResp(ErrUselessPeer, "peer cannot serve state")
+ p.isOnlyAnnounce = true
}
if recv.get("txRelay", nil) != nil {
- return errResp(ErrUselessPeer, "peer cannot relay transactions")
+ p.isOnlyAnnounce = true
}
+
+ if p.isOnlyAnnounce && !p.isTrusted {
+ return errResp(ErrUselessPeer, "peer cannot serve requests")
+ }
+
params := &flowcontrol.ServerParams{}
if err := recv.get("flowControl/BL", &params.BufLimit); err != nil {
return err
@@ -486,7 +507,6 @@ func (p *peer) Handshake(td *big.Int, head common.Hash, headNum uint64, genesis
p.fcServer = flowcontrol.NewServerNode(params)
p.fcCosts = MRC.decode()
}
-
p.headInfo = &announceData{Td: rTd, Hash: rHash, Number: rNum}
return nil
}
@@ -576,8 +596,10 @@ func (ps *peerSet) Unregister(id string) error {
for _, n := range peers {
n.unregisterPeer(p)
}
+
p.sendQueue.quit()
p.Peer.Disconnect(p2p.DiscUselessPeer)
+
return nil
}
}
diff --git a/les/peer_test.go b/les/peer_test.go
new file mode 100644
index 000000000..0ba9cbfd5
--- /dev/null
+++ b/les/peer_test.go
@@ -0,0 +1,297 @@
+package les
+
+import (
+ "math/big"
+ "testing"
+
+ "github.com/ethereum/go-ethereum/common"
+ "github.com/ethereum/go-ethereum/les/flowcontrol"
+ "github.com/ethereum/go-ethereum/p2p"
+ "github.com/ethereum/go-ethereum/p2p/enode"
+ "github.com/ethereum/go-ethereum/rlp"
+)
+
+const (
+ test_networkid = 10
+ protocol_version = 2123
+)
+
+var (
+ hash = common.HexToHash("some string")
+ genesis = common.HexToHash("genesis hash")
+ headNum = uint64(1234)
+ td = big.NewInt(123)
+)
+
+//ulc connects to trusted peer and send announceType=announceTypeSigned
+func TestPeerHandshakeSetAnnounceTypeToAnnounceTypeSignedForTrustedPeer(t *testing.T) {
+
+ var id enode.ID = newNodeID(t).ID()
+
+ //peer to connect(on ulc side)
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ isTrusted: true,
+ rw: &rwStub{
+ WriteHook: func(recvList keyValueList) {
+ //checking that ulc sends to peer allowedRequests=onlyAnnounceRequests and announceType = announceTypeSigned
+ recv := recvList.decode()
+ var reqType uint64
+
+ err := recv.get("announceType", &reqType)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if reqType != announceTypeSigned {
+ t.Fatal("Expected announceTypeSigned")
+ }
+ },
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("serveHeaders", nil)
+ l = l.add("serveChainSince", uint64(0))
+ l = l.add("serveStateSince", uint64(0))
+ l = l.add("txRelay", nil)
+ l = l.add("flowControl/BL", uint64(0))
+ l = l.add("flowControl/MRR", uint64(0))
+ l = l.add("flowControl/MRC", RequestCostList{})
+
+ return l
+ },
+ },
+ network: test_networkid,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, nil)
+ if err != nil {
+ t.Fatalf("Handshake error: %s", err)
+ }
+
+ if p.announceType != announceTypeSigned {
+ t.Fatal("Incorrect announceType")
+ }
+}
+
+func TestPeerHandshakeAnnounceTypeSignedForTrustedPeersPeerNotInTrusted(t *testing.T) {
+ var id enode.ID = newNodeID(t).ID()
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ rw: &rwStub{
+ WriteHook: func(recvList keyValueList) {
+ //checking that ulc sends to peer allowedRequests=noRequests and announceType != announceTypeSigned
+ recv := recvList.decode()
+ var reqType uint64
+
+ err := recv.get("announceType", &reqType)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if reqType == announceTypeSigned {
+ t.Fatal("Expected not announceTypeSigned")
+ }
+ },
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("serveHeaders", nil)
+ l = l.add("serveChainSince", uint64(0))
+ l = l.add("serveStateSince", uint64(0))
+ l = l.add("txRelay", nil)
+ l = l.add("flowControl/BL", uint64(0))
+ l = l.add("flowControl/MRR", uint64(0))
+ l = l.add("flowControl/MRC", RequestCostList{})
+
+ return l
+ },
+ },
+ network: test_networkid,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if p.announceType == announceTypeSigned {
+ t.Fatal("Incorrect announceType")
+ }
+}
+
+func TestPeerHandshakeDefaultAllRequests(t *testing.T) {
+ var id enode.ID = newNodeID(t).ID()
+
+ s := generateLesServer()
+
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ rw: &rwStub{
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("announceType", uint64(announceTypeSigned))
+ l = l.add("allowedRequests", uint64(0))
+
+ return l
+ },
+ },
+ network: test_networkid,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, s)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if p.isOnlyAnnounce {
+ t.Fatal("Incorrect announceType")
+ }
+}
+
+func TestPeerHandshakeServerSendOnlyAnnounceRequestsHeaders(t *testing.T) {
+ var id enode.ID = newNodeID(t).ID()
+
+ s := generateLesServer()
+ s.onlyAnnounce = true
+
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ rw: &rwStub{
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("announceType", uint64(announceTypeSigned))
+
+ return l
+ },
+ WriteHook: func(l keyValueList) {
+ for _, v := range l {
+ if v.Key == "serveHeaders" ||
+ v.Key == "serveChainSince" ||
+ v.Key == "serveStateSince" ||
+ v.Key == "txRelay" {
+ t.Fatalf("%v exists", v.Key)
+ }
+ }
+ },
+ },
+ network: test_networkid,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, s)
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+func TestPeerHandshakeClientReceiveOnlyAnnounceRequestsHeaders(t *testing.T) {
+ var id enode.ID = newNodeID(t).ID()
+
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ rw: &rwStub{
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("flowControl/BL", uint64(0))
+ l = l.add("flowControl/MRR", uint64(0))
+ l = l.add("flowControl/MRC", RequestCostList{})
+
+ l = l.add("announceType", uint64(announceTypeSigned))
+
+ return l
+ },
+ },
+ network: test_networkid,
+ isTrusted: true,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ if !p.isOnlyAnnounce {
+ t.Fatal("isOnlyAnnounce must be true")
+ }
+}
+
+func TestPeerHandshakeClientReturnErrorOnUselessPeer(t *testing.T) {
+ var id enode.ID = newNodeID(t).ID()
+
+ p := peer{
+ Peer: p2p.NewPeer(id, "test peer", []p2p.Cap{}),
+ version: protocol_version,
+ rw: &rwStub{
+ ReadHook: func(l keyValueList) keyValueList {
+ l = l.add("flowControl/BL", uint64(0))
+ l = l.add("flowControl/MRR", uint64(0))
+ l = l.add("flowControl/MRC", RequestCostList{})
+
+ l = l.add("announceType", uint64(announceTypeSigned))
+
+ return l
+ },
+ },
+ network: test_networkid,
+ }
+
+ err := p.Handshake(td, hash, headNum, genesis, nil)
+ if err == nil {
+ t.FailNow()
+ }
+}
+
+func generateLesServer() *LesServer {
+ s := &LesServer{
+ defParams: &flowcontrol.ServerParams{
+ BufLimit: uint64(300000000),
+ MinRecharge: uint64(50000),
+ },
+ fcManager: flowcontrol.NewClientManager(1, 2, 3),
+ fcCostStats: &requestCostStats{
+ stats: make(map[uint64]*linReg, len(reqList)),
+ },
+ }
+ for _, code := range reqList {
+ s.fcCostStats.stats[code] = &linReg{cnt: 100}
+ }
+ return s
+}
+
+type rwStub struct {
+ ReadHook func(l keyValueList) keyValueList
+ WriteHook func(l keyValueList)
+}
+
+func (s *rwStub) ReadMsg() (p2p.Msg, error) {
+ payload := keyValueList{}
+ payload = payload.add("protocolVersion", uint64(protocol_version))
+ payload = payload.add("networkId", uint64(test_networkid))
+ payload = payload.add("headTd", td)
+ payload = payload.add("headHash", hash)
+ payload = payload.add("headNum", headNum)
+ payload = payload.add("genesisHash", genesis)
+
+ if s.ReadHook != nil {
+ payload = s.ReadHook(payload)
+ }
+
+ size, p, err := rlp.EncodeToReader(payload)
+ if err != nil {
+ return p2p.Msg{}, err
+ }
+
+ return p2p.Msg{
+ Size: uint32(size),
+ Payload: p,
+ }, nil
+}
+
+func (s *rwStub) WriteMsg(m p2p.Msg) error {
+ recvList := keyValueList{}
+ if err := m.Decode(&recvList); err != nil {
+ return err
+ }
+
+ if s.WriteHook != nil {
+ s.WriteHook(recvList)
+ }
+
+ return nil
+}
diff --git a/les/server.go b/les/server.go
index 2fa0456d6..2ded3c184 100644
--- a/les/server.go
+++ b/les/server.go
@@ -41,17 +41,34 @@ import (
type LesServer struct {
lesCommons
- fcManager *flowcontrol.ClientManager // nil if our node is client only
- fcCostStats *requestCostStats
- defParams *flowcontrol.ServerParams
- lesTopics []discv5.Topic
- privateKey *ecdsa.PrivateKey
- quitSync chan struct{}
+ fcManager *flowcontrol.ClientManager // nil if our node is client only
+ fcCostStats *requestCostStats
+ defParams *flowcontrol.ServerParams
+ lesTopics []discv5.Topic
+ privateKey *ecdsa.PrivateKey
+ quitSync chan struct{}
+ onlyAnnounce bool
}
func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
quitSync := make(chan struct{})
- pm, err := NewProtocolManager(eth.BlockChain().Config(), light.DefaultServerIndexerConfig, false, config.NetworkId, eth.EventMux(), eth.Engine(), newPeerSet(), eth.BlockChain(), eth.TxPool(), eth.ChainDb(), nil, nil, nil, quitSync, new(sync.WaitGroup))
+ pm, err := NewProtocolManager(
+ eth.BlockChain().Config(),
+ light.DefaultServerIndexerConfig,
+ false,
+ config.NetworkId,
+ eth.EventMux(),
+ eth.Engine(),
+ newPeerSet(),
+ eth.BlockChain(),
+ eth.TxPool(),
+ eth.ChainDb(),
+ nil,
+ nil,
+ nil,
+ quitSync,
+ new(sync.WaitGroup),
+ config.ULC)
if err != nil {
return nil, err
}
@@ -70,8 +87,9 @@ func NewLesServer(eth *eth.Ethereum, config *eth.Config) (*LesServer, error) {
bloomTrieIndexer: light.NewBloomTrieIndexer(eth.ChainDb(), nil, params.BloomBitsBlocks, params.BloomTrieFrequency),
protocolManager: pm,
},
- quitSync: quitSync,
- lesTopics: lesTopics,
+ quitSync: quitSync,
+ lesTopics: lesTopics,
+ onlyAnnounce: config.OnlyAnnounce,
}
logger := log.New()
@@ -289,10 +307,8 @@ func (s *requestCostStats) getCurrentList() RequestCostList {
defer s.lock.Unlock()
list := make(RequestCostList, len(reqList))
- //fmt.Println("RequestCostList")
for idx, code := range reqList {
b, m := s.stats[code].calc()
- //fmt.Println(code, s.stats[code].cnt, b/1000000, m/1000000)
if m < 0 {
b += m
m = 0
diff --git a/les/serverpool.go b/les/serverpool.go
index 52b54b371..3f4d0a1d9 100644
--- a/les/serverpool.go
+++ b/les/serverpool.go
@@ -126,22 +126,22 @@ type serverPool struct {
discNodes chan *enode.Node
discLookups chan bool
+ trustedNodes map[enode.ID]*enode.Node
entries map[enode.ID]*poolEntry
timeout, enableRetry chan *poolEntry
adjustStats chan poolStatAdjust
- connCh chan *connReq
- disconnCh chan *disconnReq
- registerCh chan *registerReq
-
knownQueue, newQueue poolEntryQueue
knownSelect, newSelect *weightedRandomSelect
knownSelected, newSelected int
fastDiscover bool
+ connCh chan *connReq
+ disconnCh chan *disconnReq
+ registerCh chan *registerReq
}
// newServerPool creates a new serverPool instance
-func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *serverPool {
+func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup, trustedNodes []string) *serverPool {
pool := &serverPool{
db: db,
quit: quit,
@@ -156,7 +156,9 @@ func newServerPool(db ethdb.Database, quit chan struct{}, wg *sync.WaitGroup) *s
knownSelect: newWeightedRandomSelect(),
newSelect: newWeightedRandomSelect(),
fastDiscover: true,
+ trustedNodes: parseTrustedNodes(trustedNodes),
}
+
pool.knownQueue = newPoolEntryQueue(maxKnownEntries, pool.removeEntry)
pool.newQueue = newPoolEntryQueue(maxNewEntries, pool.removeEntry)
return pool
@@ -168,6 +170,7 @@ func (pool *serverPool) start(server *p2p.Server, topic discv5.Topic) {
pool.dbKey = append([]byte("serverPool/"), []byte(topic)...)
pool.wg.Add(1)
pool.loadNodes()
+ pool.connectToTrustedNodes()
if pool.server.DiscV5 != nil {
pool.discSetPeriod = make(chan time.Duration, 1)
@@ -337,8 +340,10 @@ func (pool *serverPool) eventLoop() {
}
case node := <-pool.discNodes:
- entry := pool.findOrNewNode(node)
- pool.updateCheckDial(entry)
+ if pool.trustedNodes[node.ID()] == nil {
+ entry := pool.findOrNewNode(node)
+ pool.updateCheckDial(entry)
+ }
case conv := <-pool.discLookups:
if conv {
@@ -355,29 +360,34 @@ func (pool *serverPool) eventLoop() {
}
case req := <-pool.connCh:
- // Handle peer connection requests.
- entry := pool.entries[req.p.ID()]
- if entry == nil {
- entry = pool.findOrNewNode(req.node)
- }
- if entry.state == psConnected || entry.state == psRegistered {
+ if pool.trustedNodes[req.p.ID()] != nil {
+ // ignore trusted nodes
req.result <- nil
- continue
- }
- pool.connWg.Add(1)
- entry.peer = req.p
- entry.state = psConnected
- addr := &poolEntryAddress{
- ip: req.node.IP(),
- port: uint16(req.node.TCP()),
- lastSeen: mclock.Now(),
+ } else {
+ // Handle peer connection requests.
+ entry := pool.entries[req.p.ID()]
+ if entry == nil {
+ entry = pool.findOrNewNode(req.node)
+ }
+ if entry.state == psConnected || entry.state == psRegistered {
+ req.result <- nil
+ continue
+ }
+ pool.connWg.Add(1)
+ entry.peer = req.p
+ entry.state = psConnected
+ addr := &poolEntryAddress{
+ ip: req.node.IP(),
+ port: uint16(req.node.TCP()),
+ lastSeen: mclock.Now(),
+ }
+ entry.lastConnected = addr
+ entry.addr = make(map[string]*poolEntryAddress)
+ entry.addr[addr.strKey()] = addr
+ entry.addrSelect = *newWeightedRandomSelect()
+ entry.addrSelect.update(addr)
+ req.result <- entry
}
- entry.lastConnected = addr
- entry.addr = make(map[string]*poolEntryAddress)
- entry.addr[addr.strKey()] = addr
- entry.addrSelect = *newWeightedRandomSelect()
- entry.addrSelect.update(addr)
- req.result <- entry
case req := <-pool.registerCh:
// Handle peer registration requests.
@@ -470,9 +480,40 @@ func (pool *serverPool) loadNodes() {
"response", fmt.Sprintf("%v/%v", time.Duration(e.responseStats.avg), e.responseStats.weight),
"timeout", fmt.Sprintf("%v/%v", e.timeoutStats.avg, e.timeoutStats.weight))
pool.entries[e.node.ID()] = e
- pool.knownQueue.setLatest(e)
- pool.knownSelect.update((*knownEntry)(e))
+ if pool.trustedNodes[e.node.ID()] == nil {
+ pool.knownQueue.setLatest(e)
+ pool.knownSelect.update((*knownEntry)(e))
+ }
+ }
+}
+
+// connectToTrustedNodes adds trusted server nodes as static trusted peers.
+//
+// Note: trusted nodes are not handled by the server pool logic, they are not
+// added to either the known or new selection pools. They are connected/reconnected
+// by p2p.Server whenever possible.
+func (pool *serverPool) connectToTrustedNodes() {
+ //connect to trusted nodes
+ for _, node := range pool.trustedNodes {
+ pool.server.AddTrustedPeer(node)
+ pool.server.AddPeer(node)
+ log.Debug("Added trusted node", "id", node.ID().String())
+ }
+}
+
+// parseTrustedNodes returns valid and parsed enodes
+func parseTrustedNodes(trustedNodes []string) map[enode.ID]*enode.Node {
+ nodes := make(map[enode.ID]*enode.Node)
+
+ for _, node := range trustedNodes {
+ node, err := enode.ParseV4(node)
+ if err != nil {
+ log.Warn("Trusted node URL invalid", "enode", node, "err", err)
+ continue
+ }
+ nodes[node.ID()] = node
}
+ return nodes
}
// saveNodes saves known nodes and their statistics into the database. Nodes are
diff --git a/les/txrelay.go b/les/txrelay.go
index 7a02cc837..6d22856f9 100644
--- a/les/txrelay.go
+++ b/les/txrelay.go
@@ -121,7 +121,7 @@ func (self *LesTxRelay) send(txs types.Transactions, count int) {
return peer.GetRequestCost(SendTxMsg, len(ll))
},
canSend: func(dp distPeer) bool {
- return dp.(*peer) == pp
+ return !dp.(*peer).isOnlyAnnounce && dp.(*peer) == pp
},
request: func(dp distPeer) func() {
peer := dp.(*peer)
diff --git a/les/ulc.go b/les/ulc.go
new file mode 100644
index 000000000..d9f7dc76c
--- /dev/null
+++ b/les/ulc.go
@@ -0,0 +1,39 @@
+package les
+
+import (
+ "fmt"
+
+ "github.com/ethereum/go-ethereum/eth"
+ "github.com/ethereum/go-ethereum/p2p/enode"
+)
+
+type ulc struct {
+ trustedKeys map[string]struct{}
+ minTrustedFraction int
+}
+
+func newULC(ulcConfig *eth.ULCConfig) *ulc {
+ if ulcConfig == nil {
+ return nil
+ }
+
+ m := make(map[string]struct{}, len(ulcConfig.TrustedServers))
+ for _, id := range ulcConfig.TrustedServers {
+ node, err := enode.ParseV4(id)
+ if err != nil {
+ fmt.Println("node:", id, " err:", err)
+ continue
+ }
+ m[node.ID().String()] = struct{}{}
+ }
+
+ return &ulc{m, ulcConfig.MinTrustedFraction}
+}
+
+func (u *ulc) isTrusted(p enode.ID) bool {
+ if u.trustedKeys == nil {
+ return false
+ }
+ _, ok := u.trustedKeys[p.String()]
+ return ok
+}
diff --git a/les/ulc_test.go b/les/ulc_test.go
new file mode 100644
index 000000000..3b95e6368
--- /dev/null
+++ b/les/ulc_test.go
@@ -0,0 +1,239 @@
+package les
+
+import (
+ "fmt"
+ "reflect"
+ "testing"
+ "time"
+
+ "net"
+
+ "crypto/ecdsa"
+ "math/big"
+
+ "github.com/ethereum/go-ethereum/core"
+ "github.com/ethereum/go-ethereum/crypto"
+ "github.com/ethereum/go-ethereum/eth"
+ "github.com/ethereum/go-ethereum/ethdb"
+ "github.com/ethereum/go-ethereum/light"
+ "github.com/ethereum/go-ethereum/p2p"
+ "github.com/ethereum/go-ethereum/p2p/enode"
+)
+
+func TestULCSyncWithOnePeer(t *testing.T) {
+ f := newFullPeerPair(t, 1, 4, testChainGen)
+ ulcConfig := &eth.ULCConfig{
+ MinTrustedFraction: 100,
+ TrustedServers: []string{f.Node.String()},
+ }
+
+ l := newLightPeer(t, ulcConfig)
+
+ if reflect.DeepEqual(f.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Hash()) {
+ t.Fatal("blocks are equal")
+ }
+
+ _, _, err := connectPeers(f, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ l.PM.fetcher.lock.Lock()
+ l.PM.fetcher.nextRequest()
+ l.PM.fetcher.lock.Unlock()
+
+ if !reflect.DeepEqual(f.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Hash()) {
+ t.Fatal("sync doesn't work")
+ }
+}
+
+func TestULCReceiveAnnounce(t *testing.T) {
+ f := newFullPeerPair(t, 1, 4, testChainGen)
+ ulcConfig := &eth.ULCConfig{
+ MinTrustedFraction: 100,
+ TrustedServers: []string{f.Node.String()},
+ }
+
+ l := newLightPeer(t, ulcConfig)
+ fPeer, lPeer, err := connectPeers(f, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ l.PM.synchronise(fPeer)
+
+ //check that the sync is finished correctly
+ if !reflect.DeepEqual(f.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Hash()) {
+ t.Fatal("sync doesn't work")
+ }
+
+ l.PM.peers.lock.Lock()
+ if len(l.PM.peers.peers) == 0 {
+ t.Fatal("peer list should not be empty")
+ }
+ l.PM.peers.lock.Unlock()
+
+ time.Sleep(time.Second)
+ //send a signed announce message(payload doesn't matter)
+ td := f.PM.blockchain.GetTd(l.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Number.Uint64())
+ announce := announceData{
+ Number: l.PM.blockchain.CurrentHeader().Number.Uint64() + 1,
+ Td: td.Add(td, big.NewInt(1)),
+ }
+ announce.sign(f.Key)
+ lPeer.SendAnnounce(announce)
+}
+
+func TestULCShouldNotSyncWithTwoPeersOneHaveEmptyChain(t *testing.T) {
+ f1 := newFullPeerPair(t, 1, 4, testChainGen)
+ f2 := newFullPeerPair(t, 2, 0, nil)
+ ulcConf := &ulc{minTrustedFraction: 100, trustedKeys: make(map[string]struct{})}
+ ulcConf.trustedKeys[f1.Node.ID().String()] = struct{}{}
+ ulcConf.trustedKeys[f2.Node.ID().String()] = struct{}{}
+ ulcConfig := &eth.ULCConfig{
+ MinTrustedFraction: 100,
+ TrustedServers: []string{f1.Node.String(), f2.Node.String()},
+ }
+ l := newLightPeer(t, ulcConfig)
+ l.PM.ulc.minTrustedFraction = 100
+
+ _, _, err := connectPeers(f1, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, _, err = connectPeers(f2, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ l.PM.fetcher.lock.Lock()
+ l.PM.fetcher.nextRequest()
+ l.PM.fetcher.lock.Unlock()
+
+ if reflect.DeepEqual(f2.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Hash()) {
+ t.Fatal("Incorrect hash: second peer has empty chain")
+ }
+}
+
+func TestULCShouldNotSyncWithThreePeersOneHaveEmptyChain(t *testing.T) {
+ f1 := newFullPeerPair(t, 1, 3, testChainGen)
+ f2 := newFullPeerPair(t, 2, 4, testChainGen)
+ f3 := newFullPeerPair(t, 3, 0, nil)
+
+ ulcConfig := &eth.ULCConfig{
+ MinTrustedFraction: 60,
+ TrustedServers: []string{f1.Node.String(), f2.Node.String(), f3.Node.String()},
+ }
+
+ l := newLightPeer(t, ulcConfig)
+ _, _, err := connectPeers(f1, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ _, _, err = connectPeers(f2, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ _, _, err = connectPeers(f3, l, 2)
+ if err != nil {
+ t.Fatal(err)
+ }
+
+ l.PM.fetcher.lock.Lock()
+ l.PM.fetcher.nextRequest()
+ l.PM.fetcher.lock.Unlock()
+
+ if !reflect.DeepEqual(f1.PM.blockchain.CurrentHeader().Hash(), l.PM.blockchain.CurrentHeader().Hash()) {
+ t.Fatal("Incorrect hash")
+ }
+}
+
+type pairPeer struct {
+ Name string
+ Node *enode.Node
+ PM *ProtocolManager
+ Key *ecdsa.PrivateKey
+}
+
+func connectPeers(full, light pairPeer, version int) (*peer, *peer, error) {
+ // Create a message pipe to communicate through
+ app, net := p2p.MsgPipe()
+
+ peerLight := full.PM.newPeer(version, NetworkId, p2p.NewPeer(light.Node.ID(), light.Name, nil), net)
+ peerFull := light.PM.newPeer(version, NetworkId, p2p.NewPeer(full.Node.ID(), full.Name, nil), app)
+
+ // Start the peerLight on a new thread
+ errc1 := make(chan error, 1)
+ errc2 := make(chan error, 1)
+ go func() {
+ select {
+ case light.PM.newPeerCh <- peerFull:
+ errc1 <- light.PM.handle(peerFull)
+ case <-light.PM.quitSync:
+ errc1 <- p2p.DiscQuitting
+ }
+ }()
+ go func() {
+ select {
+ case full.PM.newPeerCh <- peerLight:
+ errc2 <- full.PM.handle(peerLight)
+ case <-full.PM.quitSync:
+ errc2 <- p2p.DiscQuitting
+ }
+ }()
+
+ select {
+ case <-time.After(time.Millisecond * 100):
+ case err := <-errc1:
+ return nil, nil, fmt.Errorf("peerLight handshake error: %v", err)
+ case err := <-errc2:
+ return nil, nil, fmt.Errorf("peerFull handshake error: %v", err)
+ }
+
+ return peerFull, peerLight, nil
+}
+
+// newFullPeerPair creates node with full sync mode
+func newFullPeerPair(t *testing.T, index int, numberOfblocks int, chainGen func(int, *core.BlockGen)) pairPeer {
+ db := ethdb.NewMemDatabase()
+
+ pmFull := newTestProtocolManagerMust(t, false, numberOfblocks, chainGen, nil, nil, db, nil)
+
+ peerPairFull := pairPeer{
+ Name: "full node",
+ PM: pmFull,
+ }
+ key, err := crypto.GenerateKey()
+ if err != nil {
+ t.Fatal("generate key err:", err)
+ }
+ peerPairFull.Key = key
+ peerPairFull.Node = enode.NewV4(&key.PublicKey, net.ParseIP("127.0.0.1"), 35000, 35000)
+ return peerPairFull
+}
+
+// newLightPeer creates node with light sync mode
+func newLightPeer(t *testing.T, ulcConfig *eth.ULCConfig) pairPeer {
+ peers := newPeerSet()
+ dist := newRequestDistributor(peers, make(chan struct{}))
+ rm := newRetrieveManager(peers, dist, nil)
+ ldb := ethdb.NewMemDatabase()
+
+ odr := NewLesOdr(ldb, light.DefaultClientIndexerConfig, rm)
+
+ pmLight := newTestProtocolManagerMust(t, true, 0, nil, odr, peers, ldb, ulcConfig)
+ peerPairLight := pairPeer{
+ Name: "ulc node",
+ PM: pmLight,
+ }
+
+ key, err := crypto.GenerateKey()
+ if err != nil {
+ t.Fatal("generate key err:", err)
+ }
+ peerPairLight.Key = key
+ peerPairLight.Node = enode.NewV4(&key.PublicKey, net.IP{}, 35000, 35000)
+ return peerPairLight
+}
generated by cgit v1.2.3 (git 2.25.1) at 2024-08-26 03:17:37 +0800