diff options
Diffstat (limited to 'eth')
| -rw-r--r-- | eth/backend.go | 437 | ||||
| -rw-r--r-- | eth/protocol.go | 380 | ||||
| -rw-r--r-- | eth/protocol_test.go | 218 |
3 files changed, 1035 insertions, 0 deletions
diff --git a/eth/backend.go b/eth/backend.go new file mode 100644 index 000000000..3f7f7c2cb --- /dev/null +++ b/eth/backend.go @@ -0,0 +1,437 @@ +package eth + +import ( + "crypto/ecdsa" + "fmt" + "io/ioutil" + "path" + "strings" + + "github.com/ethereum/ethash" + "github.com/ethereum/go-ethereum/accounts" + "github.com/ethereum/go-ethereum/blockpool" + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/core/vm" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/ethdb" + "github.com/ethereum/go-ethereum/event" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/miner" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/p2p/nat" + "github.com/ethereum/go-ethereum/whisper" +) + +var ( + servlogger = logger.NewLogger("SERV") + jsonlogger = logger.NewJsonLogger() + + defaultBootNodes = []*discover.Node{ + // ETH/DEV cmd/bootnode + discover.MustParseNode("enode://6cdd090303f394a1cac34ecc9f7cda18127eafa2a3a06de39f6d920b0e583e062a7362097c7c65ee490a758b442acd5c80c6fce4b148c6a391e946b45131365b@54.169.166.226:30303"), + // ETH/DEV cpp-ethereum (poc-8.ethdev.com) + discover.MustParseNode("enode://4a44599974518ea5b0f14c31c4463692ac0329cb84851f3435e6d1b18ee4eae4aa495f846a0fa1219bd58035671881d44423876e57db2abd57254d0197da0ebe@5.1.83.226:30303"), + } +) + +type Config struct { + Name string + ProtocolVersion int + NetworkId int + + DataDir string + LogFile string + LogLevel int + LogJSON string + VmDebug bool + + MaxPeers int + Port string + + // This should be a space-separated list of + // discovery node URLs. + BootNodes string + + // This key is used to identify the node on the network. + // If nil, an ephemeral key is used. + NodeKey *ecdsa.PrivateKey + + NAT nat.Interface + Shh bool + Dial bool + + MinerThreads int + AccountManager *accounts.Manager + + // NewDB is used to create databases. + // If nil, the default is to create leveldb databases on disk. + NewDB func(path string) (common.Database, error) +} + +func (cfg *Config) parseBootNodes() []*discover.Node { + if cfg.BootNodes == "" { + return defaultBootNodes + } + var ns []*discover.Node + for _, url := range strings.Split(cfg.BootNodes, " ") { + if url == "" { + continue + } + n, err := discover.ParseNode(url) + if err != nil { + servlogger.Errorf("Bootstrap URL %s: %v\n", url, err) + continue + } + ns = append(ns, n) + } + return ns +} + +func (cfg *Config) nodeKey() (*ecdsa.PrivateKey, error) { + // use explicit key from command line args if set + if cfg.NodeKey != nil { + return cfg.NodeKey, nil + } + // use persistent key if present + keyfile := path.Join(cfg.DataDir, "nodekey") + key, err := crypto.LoadECDSA(keyfile) + if err == nil { + return key, nil + } + // no persistent key, generate and store a new one + if key, err = crypto.GenerateKey(); err != nil { + return nil, fmt.Errorf("could not generate server key: %v", err) + } + if err := ioutil.WriteFile(keyfile, crypto.FromECDSA(key), 0600); err != nil { + servlogger.Errorln("could not persist nodekey: ", err) + } + return key, nil +} + +type Ethereum struct { + // Channel for shutting down the ethereum + shutdownChan chan bool + + // DB interfaces + blockDb common.Database // Block chain database + stateDb common.Database // State changes database + extraDb common.Database // Extra database (txs, etc) + + //*** SERVICES *** + // State manager for processing new blocks and managing the over all states + blockProcessor *core.BlockProcessor + txPool *core.TxPool + chainManager *core.ChainManager + blockPool *blockpool.BlockPool + accountManager *accounts.Manager + whisper *whisper.Whisper + pow *ethash.Ethash + + net *p2p.Server + eventMux *event.TypeMux + txSub event.Subscription + blockSub event.Subscription + miner *miner.Miner + + // logger logger.LogSystem + + Mining bool + DataDir string + version string + protocolVersion int + networkId int +} + +func New(config *Config) (*Ethereum, error) { + // Boostrap database + logger.New(config.DataDir, config.LogFile, config.LogLevel) + if len(config.LogJSON) > 0 { + logger.NewJSONsystem(config.DataDir, config.LogJSON) + } + + newdb := config.NewDB + if newdb == nil { + newdb = func(path string) (common.Database, error) { return ethdb.NewLDBDatabase(path) } + } + blockDb, err := newdb(path.Join(config.DataDir, "blockchain")) + if err != nil { + return nil, err + } + stateDb, err := newdb(path.Join(config.DataDir, "state")) + if err != nil { + return nil, err + } + extraDb, err := ethdb.NewLDBDatabase(path.Join(config.DataDir, "extra")) + + // Perform database sanity checks + d, _ := extraDb.Get([]byte("ProtocolVersion")) + protov := int(common.NewValue(d).Uint()) + if protov != config.ProtocolVersion && protov != 0 { + path := path.Join(config.DataDir, "blockchain") + return nil, fmt.Errorf("Database version mismatch. Protocol(%d / %d). `rm -rf %s`", protov, config.ProtocolVersion, path) + } + saveProtocolVersion(extraDb, config.ProtocolVersion) + servlogger.Infof("Protocol Version: %v, Network Id: %v", config.ProtocolVersion, config.NetworkId) + + eth := &Ethereum{ + shutdownChan: make(chan bool), + blockDb: blockDb, + stateDb: stateDb, + extraDb: extraDb, + eventMux: &event.TypeMux{}, + accountManager: config.AccountManager, + DataDir: config.DataDir, + version: config.Name, // TODO should separate from Name + protocolVersion: config.ProtocolVersion, + networkId: config.NetworkId, + } + + eth.chainManager = core.NewChainManager(blockDb, stateDb, eth.EventMux()) + eth.pow = ethash.New(eth.chainManager) + eth.txPool = core.NewTxPool(eth.EventMux()) + eth.blockProcessor = core.NewBlockProcessor(stateDb, extraDb, eth.pow, eth.txPool, eth.chainManager, eth.EventMux()) + eth.chainManager.SetProcessor(eth.blockProcessor) + eth.whisper = whisper.New() + eth.miner = miner.New(eth, eth.pow, config.MinerThreads) + + hasBlock := eth.chainManager.HasBlock + insertChain := eth.chainManager.InsertChain + td := eth.chainManager.Td() + eth.blockPool = blockpool.New(hasBlock, insertChain, eth.pow.Verify, eth.EventMux(), td) + + netprv, err := config.nodeKey() + if err != nil { + return nil, err + } + + ethProto := EthProtocol(config.ProtocolVersion, config.NetworkId, eth.txPool, eth.chainManager, eth.blockPool) + protocols := []p2p.Protocol{ethProto} + if config.Shh { + protocols = append(protocols, eth.whisper.Protocol()) + } + + eth.net = &p2p.Server{ + PrivateKey: netprv, + Name: config.Name, + MaxPeers: config.MaxPeers, + Protocols: protocols, + NAT: config.NAT, + NoDial: !config.Dial, + BootstrapNodes: config.parseBootNodes(), + } + if len(config.Port) > 0 { + eth.net.ListenAddr = ":" + config.Port + } + + vm.Debug = config.VmDebug + + return eth, nil +} + +type NodeInfo struct { + Name string + NodeUrl string + NodeID string + IP string + DiscPort int // UDP listening port for discovery protocol + TCPPort int // TCP listening port for RLPx + Td string + ListenAddr string +} + +func (s *Ethereum) NodeInfo() *NodeInfo { + node := s.net.Self() + + return &NodeInfo{ + Name: s.Name(), + NodeUrl: node.String(), + NodeID: node.ID.String(), + IP: node.IP.String(), + DiscPort: node.DiscPort, + TCPPort: node.TCPPort, + ListenAddr: s.net.ListenAddr, + Td: s.ChainManager().Td().String(), + } +} + +type PeerInfo struct { + ID string + Name string + Caps string + RemoteAddress string + LocalAddress string +} + +func newPeerInfo(peer *p2p.Peer) *PeerInfo { + var caps []string + for _, cap := range peer.Caps() { + caps = append(caps, cap.String()) + } + return &PeerInfo{ + ID: peer.ID().String(), + Name: peer.Name(), + Caps: strings.Join(caps, ", "), + RemoteAddress: peer.RemoteAddr().String(), + LocalAddress: peer.LocalAddr().String(), + } +} + +// PeersInfo returns an array of PeerInfo objects describing connected peers +func (s *Ethereum) PeersInfo() (peersinfo []*PeerInfo) { + for _, peer := range s.net.Peers() { + if peer != nil { + peersinfo = append(peersinfo, newPeerInfo(peer)) + } + } + return +} + +func (s *Ethereum) ResetWithGenesisBlock(gb *types.Block) { + s.chainManager.ResetWithGenesisBlock(gb) + s.pow.UpdateCache(true) +} + +func (s *Ethereum) StartMining() error { + cb, err := s.accountManager.Coinbase() + if err != nil { + servlogger.Errorf("Cannot start mining without coinbase: %v\n", err) + return fmt.Errorf("no coinbase: %v", err) + } + s.miner.Start(common.BytesToAddress(cb)) + return nil +} + +func (s *Ethereum) StopMining() { s.miner.Stop() } +func (s *Ethereum) IsMining() bool { return s.miner.Mining() } +func (s *Ethereum) Miner() *miner.Miner { return s.miner } + +// func (s *Ethereum) Logger() logger.LogSystem { return s.logger } +func (s *Ethereum) Name() string { return s.net.Name } +func (s *Ethereum) AccountManager() *accounts.Manager { return s.accountManager } +func (s *Ethereum) ChainManager() *core.ChainManager { return s.chainManager } +func (s *Ethereum) BlockProcessor() *core.BlockProcessor { return s.blockProcessor } +func (s *Ethereum) TxPool() *core.TxPool { return s.txPool } +func (s *Ethereum) BlockPool() *blockpool.BlockPool { return s.blockPool } +func (s *Ethereum) Whisper() *whisper.Whisper { return s.whisper } +func (s *Ethereum) EventMux() *event.TypeMux { return s.eventMux } +func (s *Ethereum) BlockDb() common.Database { return s.blockDb } +func (s *Ethereum) StateDb() common.Database { return s.stateDb } +func (s *Ethereum) ExtraDb() common.Database { return s.extraDb } +func (s *Ethereum) IsListening() bool { return true } // Always listening +func (s *Ethereum) PeerCount() int { return s.net.PeerCount() } +func (s *Ethereum) Peers() []*p2p.Peer { return s.net.Peers() } +func (s *Ethereum) MaxPeers() int { return s.net.MaxPeers } +func (s *Ethereum) Version() string { return s.version } +func (s *Ethereum) ProtocolVersion() int { return s.protocolVersion } +func (s *Ethereum) NetworkId() int { return s.networkId } + +// Start the ethereum +func (s *Ethereum) Start() error { + jsonlogger.LogJson(&logger.LogStarting{ + ClientString: s.net.Name, + ProtocolVersion: ProtocolVersion, + }) + + if s.net.MaxPeers > 0 { + err := s.net.Start() + if err != nil { + return err + } + } + + // Start services + s.txPool.Start() + s.blockPool.Start() + + if s.whisper != nil { + s.whisper.Start() + } + + // broadcast transactions + s.txSub = s.eventMux.Subscribe(core.TxPreEvent{}) + go s.txBroadcastLoop() + + // broadcast mined blocks + s.blockSub = s.eventMux.Subscribe(core.ChainHeadEvent{}) + go s.blockBroadcastLoop() + + servlogger.Infoln("Server started") + return nil +} + +func (s *Ethereum) StartForTest() { + jsonlogger.LogJson(&logger.LogStarting{ + ClientString: s.net.Name, + ProtocolVersion: ProtocolVersion, + }) + + // Start services + s.txPool.Start() + s.blockPool.Start() +} + +func (self *Ethereum) SuggestPeer(nodeURL string) error { + n, err := discover.ParseNode(nodeURL) + if err != nil { + return fmt.Errorf("invalid node URL: %v", err) + } + self.net.SuggestPeer(n) + return nil +} + +func (s *Ethereum) Stop() { + // Close the database + defer s.blockDb.Close() + defer s.stateDb.Close() + defer s.extraDb.Close() + + s.txSub.Unsubscribe() // quits txBroadcastLoop + s.blockSub.Unsubscribe() // quits blockBroadcastLoop + + s.txPool.Stop() + s.eventMux.Stop() + s.blockPool.Stop() + if s.whisper != nil { + s.whisper.Stop() + } + + servlogger.Infoln("Server stopped") + close(s.shutdownChan) +} + +// This function will wait for a shutdown and resumes main thread execution +func (s *Ethereum) WaitForShutdown() { + <-s.shutdownChan +} + +// now tx broadcasting is taken out of txPool +// handled here via subscription, efficiency? +func (self *Ethereum) txBroadcastLoop() { + // automatically stops if unsubscribe + for obj := range self.txSub.Chan() { + event := obj.(core.TxPreEvent) + self.net.Broadcast("eth", TxMsg, []*types.Transaction{event.Tx}) + } +} + +func (self *Ethereum) blockBroadcastLoop() { + // automatically stops if unsubscribe + for obj := range self.blockSub.Chan() { + switch ev := obj.(type) { + case core.ChainHeadEvent: + self.net.Broadcast("eth", NewBlockMsg, []interface{}{ev.Block, ev.Block.Td}) + } + } +} + +func saveProtocolVersion(db common.Database, protov int) { + d, _ := db.Get([]byte("ProtocolVersion")) + protocolVersion := common.NewValue(d).Uint() + + if protocolVersion == 0 { + db.Put([]byte("ProtocolVersion"), common.NewValue(protov).Bytes()) + } +} diff --git a/eth/protocol.go b/eth/protocol.go new file mode 100644 index 000000000..494c1c1bb --- /dev/null +++ b/eth/protocol.go @@ -0,0 +1,380 @@ +package eth + +import ( + "fmt" + "math/big" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/errs" + "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/rlp" +) + +const ( + ProtocolVersion = 58 + NetworkId = 0 + ProtocolLength = uint64(8) + ProtocolMaxMsgSize = 10 * 1024 * 1024 + maxHashes = 256 + maxBlocks = 64 +) + +// eth protocol message codes +const ( + StatusMsg = iota + GetTxMsg // unused + TxMsg + GetBlockHashesMsg + BlockHashesMsg + GetBlocksMsg + BlocksMsg + NewBlockMsg +) + +const ( + ErrMsgTooLarge = iota + ErrDecode + ErrInvalidMsgCode + ErrProtocolVersionMismatch + ErrNetworkIdMismatch + ErrGenesisBlockMismatch + ErrNoStatusMsg + ErrExtraStatusMsg + ErrSuspendedPeer +) + +var errorToString = map[int]string{ + ErrMsgTooLarge: "Message too long", + ErrDecode: "Invalid message", + ErrInvalidMsgCode: "Invalid message code", + ErrProtocolVersionMismatch: "Protocol version mismatch", + ErrNetworkIdMismatch: "NetworkId mismatch", + ErrGenesisBlockMismatch: "Genesis block mismatch", + ErrNoStatusMsg: "No status message", + ErrExtraStatusMsg: "Extra status message", + ErrSuspendedPeer: "Suspended peer", +} + +// ethProtocol represents the ethereum wire protocol +// instance is running on each peer +type ethProtocol struct { + txPool txPool + chainManager chainManager + blockPool blockPool + peer *p2p.Peer + id string + rw p2p.MsgReadWriter + errors *errs.Errors + protocolVersion int + networkId int +} + +// backend is the interface the ethereum protocol backend should implement +// used as an argument to EthProtocol +type txPool interface { + AddTransactions([]*types.Transaction) + GetTransactions() types.Transactions +} + +type chainManager interface { + GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) + GetBlock(hash common.Hash) (block *types.Block) + Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) +} + +type blockPool interface { + AddBlockHashes(next func() (common.Hash, bool), peerId string) + AddBlock(block *types.Block, peerId string) + AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool) + RemovePeer(peerId string) +} + +// message structs used for RLP serialization +type newBlockMsgData struct { + Block *types.Block + TD *big.Int +} + +type getBlockHashesMsgData struct { + Hash common.Hash + Amount uint64 +} + +type statusMsgData struct { + ProtocolVersion uint32 + NetworkId uint32 + TD *big.Int + CurrentBlock common.Hash + GenesisBlock common.Hash +} + +// main entrypoint, wrappers starting a server running the eth protocol +// use this constructor to attach the protocol ("class") to server caps +// the Dev p2p layer then runs the protocol instance on each peer +func EthProtocol(protocolVersion, networkId int, txPool txPool, chainManager chainManager, blockPool blockPool) p2p.Protocol { + return p2p.Protocol{ + Name: "eth", + Version: uint(protocolVersion), + Length: ProtocolLength, + Run: func(peer *p2p.Peer, rw p2p.MsgReadWriter) error { + return runEthProtocol(protocolVersion, networkId, txPool, chainManager, blockPool, peer, rw) + }, + } +} + +// the main loop that handles incoming messages +// note RemovePeer in the post-disconnect hook +func runEthProtocol(protocolVersion, networkId int, txPool txPool, chainManager chainManager, blockPool blockPool, peer *p2p.Peer, rw p2p.MsgReadWriter) (err error) { + id := peer.ID() + self := ðProtocol{ + txPool: txPool, + chainManager: chainManager, + blockPool: blockPool, + rw: rw, + peer: peer, + protocolVersion: protocolVersion, + networkId: networkId, + errors: &errs.Errors{ + Package: "ETH", + Errors: errorToString, + }, + id: fmt.Sprintf("%x", id[:8]), + } + + // handshake. + if err := self.handleStatus(); err != nil { + return err + } + defer self.blockPool.RemovePeer(self.id) + + // propagate existing transactions. new transactions appearing + // after this will be sent via broadcasts. + if err := p2p.Send(rw, TxMsg, txPool.GetTransactions()); err != nil { + return err + } + + // main loop. handle incoming messages. + for { + if err := self.handle(); err != nil { + return err + } + } +} + +func (self *ethProtocol) handle() error { + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + if msg.Size > ProtocolMaxMsgSize { + return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + // make sure that the payload has been fully consumed + defer msg.Discard() + + switch msg.Code { + case GetTxMsg: // ignore + case StatusMsg: + return self.protoError(ErrExtraStatusMsg, "") + + case TxMsg: + // TODO: rework using lazy RLP stream + var txs []*types.Transaction + if err := msg.Decode(&txs); err != nil { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + for _, tx := range txs { + jsonlogger.LogJson(&logger.EthTxReceived{ + TxHash: tx.Hash().Hex(), + RemoteId: self.peer.ID().String(), + }) + } + self.txPool.AddTransactions(txs) + + case GetBlockHashesMsg: + var request getBlockHashesMsgData + if err := msg.Decode(&request); err != nil { + return self.protoError(ErrDecode, "->msg %v: %v", msg, err) + } + + if request.Amount > maxHashes { + request.Amount = maxHashes + } + hashes := self.chainManager.GetBlockHashesFromHash(request.Hash, request.Amount) + return p2p.Send(self.rw, BlockHashesMsg, hashes) + + case BlockHashesMsg: + msgStream := rlp.NewStream(msg.Payload) + if _, err := msgStream.List(); err != nil { + return err + } + + var i int + iter := func() (hash common.Hash, ok bool) { + err := msgStream.Decode(&hash) + if err == rlp.EOL { + return common.Hash{}, false + } else if err != nil { + self.protoError(ErrDecode, "msg %v: after %v hashes : %v", msg, i, err) + return common.Hash{}, false + } + + i++ + return hash, true + } + self.blockPool.AddBlockHashes(iter, self.id) + + case GetBlocksMsg: + msgStream := rlp.NewStream(msg.Payload) + if _, err := msgStream.List(); err != nil { + return err + } + + var blocks []*types.Block + var i int + for { + i++ + var hash common.Hash + err := msgStream.Decode(&hash) + if err == rlp.EOL { + break + } else if err != nil { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + + block := self.chainManager.GetBlock(hash) + if block != nil { + blocks = append(blocks, block) + } + if i == maxBlocks { + break + } + } + return p2p.Send(self.rw, BlocksMsg, blocks) + + case BlocksMsg: + msgStream := rlp.NewStream(msg.Payload) + if _, err := msgStream.List(); err != nil { + return err + } + for { + var block types.Block + if err := msgStream.Decode(&block); err != nil { + if err == rlp.EOL { + break + } else { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + } + self.blockPool.AddBlock(&block, self.id) + } + + case NewBlockMsg: + var request newBlockMsgData + if err := msg.Decode(&request); err != nil { + return self.protoError(ErrDecode, "%v: %v", msg, err) + } + hash := request.Block.Hash() + _, chainHead, _ := self.chainManager.Status() + + jsonlogger.LogJson(&logger.EthChainReceivedNewBlock{ + BlockHash: hash.Hex(), + BlockNumber: request.Block.Number(), // this surely must be zero + ChainHeadHash: chainHead.Hex(), + BlockPrevHash: request.Block.ParentHash().Hex(), + RemoteId: self.peer.ID().String(), + }) + // to simplify backend interface adding a new block + // uses AddPeer followed by AddBlock only if peer is the best peer + // (or selected as new best peer) + if best, _ := self.blockPool.AddPeer(request.TD, hash, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect); best { + self.blockPool.AddBlock(request.Block, self.id) + } + + default: + return self.protoError(ErrInvalidMsgCode, "%v", msg.Code) + } + return nil +} + +func (self *ethProtocol) handleStatus() error { + if err := self.sendStatus(); err != nil { + return err + } + + // read and handle remote status + msg, err := self.rw.ReadMsg() + if err != nil { + return err + } + if msg.Code != StatusMsg { + return self.protoError(ErrNoStatusMsg, "first msg has code %x (!= %x)", msg.Code, StatusMsg) + } + if msg.Size > ProtocolMaxMsgSize { + return self.protoError(ErrMsgTooLarge, "%v > %v", msg.Size, ProtocolMaxMsgSize) + } + + var status statusMsgData + if err := msg.Decode(&status); err != nil { + return self.protoError(ErrDecode, "msg %v: %v", msg, err) + } + + _, _, genesisBlock := self.chainManager.Status() + + if status.GenesisBlock != genesisBlock { + return self.protoError(ErrGenesisBlockMismatch, "%x (!= %x)", status.GenesisBlock, genesisBlock) + } + + if int(status.NetworkId) != self.networkId { + return self.protoError(ErrNetworkIdMismatch, "%d (!= %d)", status.NetworkId, self.networkId) + } + + if int(status.ProtocolVersion) != self.protocolVersion { + return self.protoError(ErrProtocolVersionMismatch, "%d (!= %d)", status.ProtocolVersion, self.protocolVersion) + } + + _, suspended := self.blockPool.AddPeer(status.TD, status.CurrentBlock, self.id, self.requestBlockHashes, self.requestBlocks, self.protoErrorDisconnect) + if suspended { + return self.protoError(ErrSuspendedPeer, "") + } + + self.peer.Infof("Peer is [eth] capable (%d/%d). TD=%v H=%x\n", status.ProtocolVersion, status.NetworkId, status.TD, status.CurrentBlock[:4]) + + return nil +} + +func (self *ethProtocol) requestBlockHashes(from common.Hash) error { + self.peer.Debugf("fetching hashes (%d) %x...\n", maxHashes, from[0:4]) + return p2p.Send(self.rw, GetBlockHashesMsg, getBlockHashesMsgData{from, maxHashes}) +} + +func (self *ethProtocol) requestBlocks(hashes []common.Hash) error { + self.peer.Debugf("fetching %v blocks", len(hashes)) + return p2p.Send(self.rw, GetBlocksMsg, hashes) +} + +func (self *ethProtocol) protoError(code int, format string, params ...interface{}) (err *errs.Error) { + err = self.errors.New(code, format, params...) + err.Log(self.peer.Logger) + return +} + +func (self *ethProtocol) sendStatus() error { + td, currentBlock, genesisBlock := self.chainManager.Status() + return p2p.Send(self.rw, StatusMsg, &statusMsgData{ + ProtocolVersion: uint32(self.protocolVersion), + NetworkId: uint32(self.networkId), + TD: td, + CurrentBlock: currentBlock, + GenesisBlock: genesisBlock, + }) +} + +func (self *ethProtocol) protoErrorDisconnect(err *errs.Error) { + err.Log(self.peer.Logger) + if err.Fatal() { + self.peer.Disconnect(p2p.DiscSubprotocolError) + } +} diff --git a/eth/protocol_test.go b/eth/protocol_test.go new file mode 100644 index 000000000..8ca6d1be6 --- /dev/null +++ b/eth/protocol_test.go @@ -0,0 +1,218 @@ +package eth + +import ( + "log" + "math/big" + "os" + "testing" + "time" + + "github.com/ethereum/go-ethereum/common" + "github.com/ethereum/go-ethereum/core/types" + "github.com/ethereum/go-ethereum/crypto" + "github.com/ethereum/go-ethereum/errs" + ethlogger "github.com/ethereum/go-ethereum/logger" + "github.com/ethereum/go-ethereum/p2p" + "github.com/ethereum/go-ethereum/p2p/discover" +) + +var logsys = ethlogger.NewStdLogSystem(os.Stdout, log.LstdFlags, ethlogger.LogLevel(ethlogger.DebugDetailLevel)) + +var ini = false + +func logInit() { + if !ini { + ethlogger.AddLogSystem(logsys) + ini = true + } +} + +type testTxPool struct { + getTransactions func() []*types.Transaction + addTransactions func(txs []*types.Transaction) +} + +type testChainManager struct { + getBlockHashes func(hash common.Hash, amount uint64) (hashes []common.Hash) + getBlock func(hash common.Hash) *types.Block + status func() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) +} + +type testBlockPool struct { + addBlockHashes func(next func() (common.Hash, bool), peerId string) + addBlock func(block *types.Block, peerId string) (err error) + addPeer func(td *big.Int, currentBlock common.Hash, peerId string, requestHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool) + removePeer func(peerId string) +} + +func (self *testTxPool) AddTransactions(txs []*types.Transaction) { + if self.addTransactions != nil { + self.addTransactions(txs) + } +} + +func (self *testTxPool) GetTransactions() types.Transactions { return nil } + +func (self *testChainManager) GetBlockHashesFromHash(hash common.Hash, amount uint64) (hashes []common.Hash) { + if self.getBlockHashes != nil { + hashes = self.getBlockHashes(hash, amount) + } + return +} + +func (self *testChainManager) Status() (td *big.Int, currentBlock common.Hash, genesisBlock common.Hash) { + if self.status != nil { + td, currentBlock, genesisBlock = self.status() + } + return +} + +func (self *testChainManager) GetBlock(hash common.Hash) (block *types.Block) { + if self.getBlock != nil { + block = self.getBlock(hash) + } + return +} + +func (self *testBlockPool) AddBlockHashes(next func() (common.Hash, bool), peerId string) { + if self.addBlockHashes != nil { + self.addBlockHashes(next, peerId) + } +} + +func (self *testBlockPool) AddBlock(block *types.Block, peerId string) { + if self.addBlock != nil { + self.addBlock(block, peerId) + } +} + +func (self *testBlockPool) AddPeer(td *big.Int, currentBlock common.Hash, peerId string, requestBlockHashes func(common.Hash) error, requestBlocks func([]common.Hash) error, peerError func(*errs.Error)) (best bool, suspended bool) { + if self.addPeer != nil { + best, suspended = self.addPeer(td, currentBlock, peerId, requestBlockHashes, requestBlocks, peerError) + } + return +} + +func (self *testBlockPool) RemovePeer(peerId string) { + if self.removePeer != nil { + self.removePeer(peerId) + } +} + +func testPeer() *p2p.Peer { + var id discover.NodeID + pk := crypto.GenerateNewKeyPair().PublicKey + copy(id[:], pk) + return p2p.NewPeer(id, "test peer", []p2p.Cap{}) +} + +type ethProtocolTester struct { + p2p.MsgReadWriter // writing to the tester feeds the protocol + + quit chan error + pipe *p2p.MsgPipeRW // the protocol read/writes on this end + txPool *testTxPool // txPool + chainManager *testChainManager // chainManager + blockPool *testBlockPool // blockPool + t *testing.T +} + +func newEth(t *testing.T) *ethProtocolTester { + p1, p2 := p2p.MsgPipe() + return ðProtocolTester{ + MsgReadWriter: p1, + quit: make(chan error, 1), + pipe: p2, + txPool: &testTxPool{}, + chainManager: &testChainManager{}, + blockPool: &testBlockPool{}, + t: t, + } +} + +func (self *ethProtocolTester) reset() { + self.pipe.Close() + + p1, p2 := p2p.MsgPipe() + self.MsgReadWriter = p1 + self.pipe = p2 + self.quit = make(chan error, 1) +} + +func (self *ethProtocolTester) checkError(expCode int, delay time.Duration) (err error) { + var timer = time.After(delay) + select { + case err = <-self.quit: + case <-timer: + self.t.Errorf("no error after %v, expected %v", delay, expCode) + return + } + perr, ok := err.(*errs.Error) + if ok && perr != nil { + if code := perr.Code; code != expCode { + self.t.Errorf("expected protocol error (code %v), got %v (%v)", expCode, code, err) + } + } else { + self.t.Errorf("expected protocol error (code %v), got %v", expCode, err) + } + return +} + +func (self *ethProtocolTester) run() { + err := runEthProtocol(ProtocolVersion, NetworkId, self.txPool, self.chainManager, self.blockPool, testPeer(), self.pipe) + self.quit <- err +} + +func TestStatusMsgErrors(t *testing.T) { + logInit() + eth := newEth(t) + td := common.Big1 + currentBlock := common.Hash{1} + genesis := common.Hash{2} + eth.chainManager.status = func() (*big.Int, common.Hash, common.Hash) { return td, currentBlock, genesis } + go eth.run() + + tests := []struct { + code uint64 + data interface{} + wantErrorCode int + }{ + { + code: TxMsg, data: []interface{}{}, + wantErrorCode: ErrNoStatusMsg, + }, + { + code: StatusMsg, data: statusMsgData{10, NetworkId, td, currentBlock, genesis}, + wantErrorCode: ErrProtocolVersionMismatch, + }, + { + code: StatusMsg, data: statusMsgData{ProtocolVersion, 999, td, currentBlock, genesis}, + wantErrorCode: ErrNetworkIdMismatch, + }, + { + code: StatusMsg, data: statusMsgData{ProtocolVersion, NetworkId, td, currentBlock, common.Hash{3}}, + wantErrorCode: ErrGenesisBlockMismatch, + }, + } + for _, test := range tests { + // first outgoing msg should be StatusMsg. + err := p2p.ExpectMsg(eth, StatusMsg, &statusMsgData{ + ProtocolVersion: ProtocolVersion, + NetworkId: NetworkId, + TD: td, + CurrentBlock: currentBlock, + GenesisBlock: genesis, + }) + if err != nil { + t.Fatalf("incorrect outgoing status: %v", err) + } + + // the send call might hang until reset because + // the protocol might not read the payload. + go p2p.Send(eth, test.code, test.data) + eth.checkError(test.wantErrorCode, 1*time.Second) + + eth.reset() + go eth.run() + } +} |