diff options
author | obscuren <geffobscura@gmail.com> | 2015-02-20 05:46:15 +0800 |
---|---|---|
committer | obscuren <geffobscura@gmail.com> | 2015-02-20 05:46:15 +0800 |
commit | 000658539136645e743d83d57a8106b2a9796b93 (patch) | |
tree | ca37fb68b6b7574d7c888968813598e68d301659 | |
parent | fa4cbad315609e41d88c59ecbce7c6c6169fc57a (diff) | |
parent | 639ac5c3daa253f36d41672165a5727d8eba207e (diff) | |
download | dexon-000658539136645e743d83d57a8106b2a9796b93.tar dexon-000658539136645e743d83d57a8106b2a9796b93.tar.gz dexon-000658539136645e743d83d57a8106b2a9796b93.tar.bz2 dexon-000658539136645e743d83d57a8106b2a9796b93.tar.lz dexon-000658539136645e743d83d57a8106b2a9796b93.tar.xz dexon-000658539136645e743d83d57a8106b2a9796b93.tar.zst dexon-000658539136645e743d83d57a8106b2a9796b93.zip |
Merge branch 'develop' of github.com-obscure:ethereum/go-ethereum into develop
-rw-r--r-- | cmd/ethereum/main.go | 1 | ||||
-rw-r--r-- | cmd/mist/assets/ext/mist.js | 14 | ||||
-rw-r--r-- | cmd/mist/assets/qml/main.qml | 9 | ||||
-rw-r--r-- | cmd/mist/assets/qml/views/browser.qml | 10 | ||||
-rw-r--r-- | cmd/mist/assets/qml/views/catalog.qml | 19 | ||||
-rw-r--r-- | cmd/mist/flags.go | 3 | ||||
-rw-r--r-- | cmd/mist/gui.go | 3 | ||||
-rw-r--r-- | cmd/mist/main.go | 27 | ||||
-rw-r--r-- | cmd/utils/cmd.go | 2 | ||||
-rw-r--r-- | eth/backend.go | 6 | ||||
-rw-r--r-- | logger/types.go | 573 | ||||
-rw-r--r-- | miner/miner.go | 4 | ||||
-rw-r--r-- | p2p/handshake.go (renamed from p2p/crypto.go) | 137 | ||||
-rw-r--r-- | p2p/handshake_test.go (renamed from p2p/crypto_test.go) | 63 | ||||
-rw-r--r-- | p2p/message.go | 2 | ||||
-rw-r--r-- | p2p/peer.go | 264 | ||||
-rw-r--r-- | p2p/peer_test.go | 105 | ||||
-rw-r--r-- | p2p/server.go | 66 | ||||
-rw-r--r-- | p2p/server_test.go | 12 |
19 files changed, 658 insertions, 662 deletions
diff --git a/cmd/ethereum/main.go b/cmd/ethereum/main.go index 1ffd4b6e9..1f1a0b761 100644 --- a/cmd/ethereum/main.go +++ b/cmd/ethereum/main.go @@ -67,6 +67,7 @@ func main() { DataDir: Datadir, LogFile: LogFile, LogLevel: LogLevel, + LogFormat: LogFormat, MaxPeers: MaxPeer, Port: OutboundPort, NAT: NAT, diff --git a/cmd/mist/assets/ext/mist.js b/cmd/mist/assets/ext/mist.js index 8734f8dc7..2fc38cdfa 100644 --- a/cmd/mist/assets/ext/mist.js +++ b/cmd/mist/assets/ext/mist.js @@ -20,16 +20,18 @@ console.log("loaded?"); document.onkeydown = function(evt) { + // This functions keeps track of keyboard inputs in order to allow copy, paste and other features + evt = evt || window.event; if (evt.ctrlKey && evt.keyCode == 67) { window.document.execCommand("copy"); - console.log("Ctrl-C"); } else if (evt.ctrlKey && evt.keyCode == 88) { window.document.execCommand("cut"); - console.log("Ctrl-X"); - } if (evt.ctrlKey && evt.keyCode == 86) { - console.log("Ctrl-V"); - } if (evt.ctrlKey && evt.keyCode == 90) { - console.log("Ctrl-Z"); + } else if (evt.ctrlKey && evt.keyCode == 86) { + window.document.execCommand("paste"); + } else if (evt.ctrlKey && evt.keyCode == 90) { + window.document.execCommand("undo"); + } else if (evt.ctrlKey && evt.shiftKey && evt.keyCode == 90) { + window.document.execCommand("redo"); } };
\ No newline at end of file diff --git a/cmd/mist/assets/qml/main.qml b/cmd/mist/assets/qml/main.qml index 5f7f05d83..937670bd2 100644 --- a/cmd/mist/assets/qml/main.qml +++ b/cmd/mist/assets/qml/main.qml @@ -131,7 +131,11 @@ ApplicationWindow { var existingDomain = matches && matches[1]; if (requestedDomain == existingDomain) { domainAlreadyOpen = true; - mainSplit.views[i].view.url = url; + + if (mainSplit.views[i].view.url != url){ + mainSplit.views[i].view.url = url; + } + activeView(mainSplit.views[i].view, mainSplit.views[i].menuItem); } } @@ -928,7 +932,8 @@ ApplicationWindow { model: peerModel TableViewColumn{width: 180; role: "addr" ; title: "Remote Address" } TableViewColumn{width: 280; role: "nodeID" ; title: "Node ID" } - TableViewColumn{width: 180; role: "caps" ; title: "Capabilities" } + TableViewColumn{width: 100; role: "name" ; title: "Name" } + TableViewColumn{width: 40; role: "caps" ; title: "Capabilities" } } } } diff --git a/cmd/mist/assets/qml/views/browser.qml b/cmd/mist/assets/qml/views/browser.qml index 54f5d755e..edecc8696 100644 --- a/cmd/mist/assets/qml/views/browser.qml +++ b/cmd/mist/assets/qml/views/browser.qml @@ -3,7 +3,7 @@ import QtQuick.Controls 1.0; import QtQuick.Controls.Styles 1.0 import QtQuick.Layouts 1.0; import QtWebEngine 1.0 -//import QtWebEngine.experimental 1.0 +import QtWebEngine.experimental 1.0 import QtQuick.Window 2.0; Rectangle { @@ -340,7 +340,7 @@ Rectangle { WebEngineView { objectName: "webView" id: webview - //experimental.settings.javascriptCanAccessClipboard: true + experimental.settings.javascriptCanAccessClipboard: true //experimental.settings.localContentCanAccessRemoteUrls: true anchors { left: parent.left @@ -399,7 +399,8 @@ Rectangle { onLoadingChanged: { if (loadRequest.status == WebEngineView.LoadSucceededStatus) { - webview.runJavaScript("document.title", function(pageTitle) { + + webview.runJavaScript("document.title", function(pageTitle) { menuItem.title = pageTitle; }); @@ -441,7 +442,8 @@ Rectangle { webview.runJavaScript(eth.readFile("bignumber.min.js")); webview.runJavaScript(eth.readFile("ethereum.js/dist/ethereum.js")); - + webview.runJavaScript(eth.readFile("mist.js")); + var cleanTitle = webview.url.toString() var matches = cleanTitle.match(/^[a-z]*\:\/\/([^\/?#]+)(?:[\/?#]|$)/i); var domain = matches && matches[1]; diff --git a/cmd/mist/assets/qml/views/catalog.qml b/cmd/mist/assets/qml/views/catalog.qml index 497d69ed1..29e133074 100644 --- a/cmd/mist/assets/qml/views/catalog.qml +++ b/cmd/mist/assets/qml/views/catalog.qml @@ -3,7 +3,7 @@ import QtQuick.Controls 1.0; import QtQuick.Controls.Styles 1.0 import QtQuick.Layouts 1.0; import QtWebEngine 1.0 -//import QtWebEngine.experimental 1.0 +import QtWebEngine.experimental 1.0 import QtQuick.Window 2.0; @@ -21,8 +21,6 @@ Rectangle { property alias windowTitle: webview.title property alias webView: webview - - property var cleanPath: false property var open: function(url) { if(!window.cleanPath) { @@ -66,9 +64,6 @@ Rectangle { } } - Component.onCompleted: { - } - Item { objectName: "root" id: root @@ -85,7 +80,7 @@ Rectangle { property var domain: "ethereum-dapp-catalog.meteor.com" url: protocol + domain - //experimental.settings.javascriptCanAccessClipboard: true + experimental.settings.javascriptCanAccessClipboard: true onJavaScriptConsoleMessage: { @@ -112,11 +107,11 @@ Rectangle { } } - // onLoadingChanged: { - // if (loadRequest.status == WebEngineView.LoadSucceededStatus) { - // webview.runJavaScript(eth.readFile("mist.js")); - // } - // } + onLoadingChanged: { + if (loadRequest.status == WebEngineView.LoadSucceededStatus) { + webview.runJavaScript(eth.readFile("mist.js")); + } + } } diff --git a/cmd/mist/flags.go b/cmd/mist/flags.go index eb280f71b..d9487de9e 100644 --- a/cmd/mist/flags.go +++ b/cmd/mist/flags.go @@ -63,6 +63,7 @@ var ( DebugFile string LogLevel int VmType int + MinerThreads int ) // flags specific to gui client @@ -137,6 +138,8 @@ func Init() { flag.StringVar(&BootNodes, "bootnodes", "", "space-separated node URLs for discovery bootstrap") flag.IntVar(&MaxPeer, "maxpeer", 30, "maximum desired peers") + flag.IntVar(&MinerThreads, "minerthreads", runtime.NumCPU(), "number of miner threads") + flag.Parse() var err error diff --git a/cmd/mist/gui.go b/cmd/mist/gui.go index afddeb8f6..4af0cff43 100644 --- a/cmd/mist/gui.go +++ b/cmd/mist/gui.go @@ -453,7 +453,7 @@ NumGC: %d )) } -type qmlpeer struct{ Addr, NodeID, Caps string } +type qmlpeer struct{ Addr, NodeID, Name, Caps string } type peersByID []*qmlpeer @@ -468,6 +468,7 @@ func (gui *Gui) setPeerInfo() { qpeers[i] = &qmlpeer{ NodeID: p.ID().String(), Addr: p.RemoteAddr().String(), + Name: p.Name(), Caps: fmt.Sprint(p.Caps()), } } diff --git a/cmd/mist/main.go b/cmd/mist/main.go index 14f561e99..d41aa34bf 100644 --- a/cmd/mist/main.go +++ b/cmd/mist/main.go @@ -52,19 +52,20 @@ func run() error { config := utils.InitConfig(VmType, ConfigFile, Datadir, "ETH") ethereum, err := eth.New(ð.Config{ - Name: p2p.MakeName(ClientIdentifier, Version), - KeyStore: KeyStore, - DataDir: Datadir, - LogFile: LogFile, - LogLevel: LogLevel, - MaxPeers: MaxPeer, - Port: OutboundPort, - NAT: NAT, - Shh: true, - BootNodes: BootNodes, - NodeKey: NodeKey, - KeyRing: KeyRing, - Dial: true, + Name: p2p.MakeName(ClientIdentifier, Version), + KeyStore: KeyStore, + DataDir: Datadir, + LogFile: LogFile, + LogLevel: LogLevel, + MaxPeers: MaxPeer, + Port: OutboundPort, + NAT: NAT, + Shh: true, + BootNodes: BootNodes, + NodeKey: NodeKey, + KeyRing: KeyRing, + Dial: true, + MinerThreads: MinerThreads, }) if err != nil { mainlogger.Fatalln(err) diff --git a/cmd/utils/cmd.go b/cmd/utils/cmd.go index ecb847fc3..d252f3ab2 100644 --- a/cmd/utils/cmd.go +++ b/cmd/utils/cmd.go @@ -225,7 +225,7 @@ func StartMining(ethereum *eth.Ethereum) bool { go func() { clilogger.Infoln("Start mining") if gminer == nil { - gminer = miner.New(addr, ethereum) + gminer = miner.New(addr, ethereum, 4) } gminer.Start() }() diff --git a/eth/backend.go b/eth/backend.go index d109ab98e..b57e9fd69 100644 --- a/eth/backend.go +++ b/eth/backend.go @@ -53,6 +53,8 @@ type Config struct { Shh bool Dial bool + MinerThreads int + KeyManager *crypto.KeyManager } @@ -153,7 +155,7 @@ func New(config *Config) (*Ethereum, error) { eth.blockProcessor = core.NewBlockProcessor(db, eth.txPool, eth.chainManager, eth.EventMux()) eth.chainManager.SetProcessor(eth.blockProcessor) eth.whisper = whisper.New() - eth.miner = miner.New(keyManager.Address(), eth) + eth.miner = miner.New(keyManager.Address(), eth, config.MinerThreads) hasBlock := eth.chainManager.HasBlock insertChain := eth.chainManager.InsertChain @@ -209,9 +211,7 @@ func (s *Ethereum) Coinbase() []byte { return nil } // TODO func (s *Ethereum) Start() error { jsonlogger.LogJson(ðlogger.LogStarting{ ClientString: s.net.Name, - Coinbase: ethutil.Bytes2Hex(s.KeyManager().Address()), ProtocolVersion: ProtocolVersion, - LogEvent: ethlogger.LogEvent{Guid: ethutil.Bytes2Hex(crypto.FromECDSAPub(&s.net.PrivateKey.PublicKey))}, }) err := s.net.Start() diff --git a/logger/types.go b/logger/types.go index 419382231..7ab4a2b8c 100644 --- a/logger/types.go +++ b/logger/types.go @@ -7,7 +7,6 @@ import ( type utctime8601 struct{} func (utctime8601) MarshalJSON() ([]byte, error) { - // FIX This should be re-formated for proper ISO 8601 return []byte(`"` + time.Now().UTC().Format(time.RFC3339Nano)[:26] + `Z"`), nil } @@ -16,14 +15,13 @@ type JsonLog interface { } type LogEvent struct { - Guid string `json:"guid"` - Ts utctime8601 `json:"ts"` + // Guid string `json:"guid"` + Ts utctime8601 `json:"ts"` // Level string `json:"level"` } type LogStarting struct { - ClientString string `json:"version_string"` - Coinbase string `json:"coinbase"` + ClientString string `json:"client_impl"` ProtocolVersion int `json:"eth_version"` LogEvent } @@ -32,17 +30,6 @@ func (l *LogStarting) EventName() string { return "starting" } -type P2PConnecting struct { - RemoteId string `json:"remote_id"` - RemoteEndpoint string `json:"remote_endpoint"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PConnecting) EventName() string { - return "p2p.connecting" -} - type P2PConnected struct { RemoteId string `json:"remote_id"` RemoteAddress string `json:"remote_addr"` @@ -55,17 +42,6 @@ func (l *P2PConnected) EventName() string { return "p2p.connected" } -type P2PHandshaked struct { - RemoteCapabilities []string `json:"remote_capabilities"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PHandshaked) EventName() string { - return "p2p.handshaked" -} - type P2PDisconnected struct { NumConnections int `json:"num_connections"` RemoteId string `json:"remote_id"` @@ -76,247 +52,46 @@ func (l *P2PDisconnected) EventName() string { return "p2p.disconnected" } -type P2PDisconnecting struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PDisconnecting) EventName() string { - return "p2p.disconnecting" -} - -type P2PDisconnectingBadHandshake struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PDisconnectingBadHandshake) EventName() string { - return "p2p.disconnecting.bad_handshake" -} - -type P2PDisconnectingBadProtocol struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PDisconnectingBadProtocol) EventName() string { - return "p2p.disconnecting.bad_protocol" -} - -type P2PDisconnectingReputation struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PDisconnectingReputation) EventName() string { - return "p2p.disconnecting.reputation" -} - -type P2PDisconnectingDHT struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PDisconnectingDHT) EventName() string { - return "p2p.disconnecting.dht" -} - -type P2PEthDisconnectingBadBlock struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PEthDisconnectingBadBlock) EventName() string { - return "p2p.eth.disconnecting.bad_block" -} - -type P2PEthDisconnectingBadTx struct { - Reason string `json:"reason"` - RemoteId string `json:"remote_id"` - NumConnections int `json:"num_connections"` - LogEvent -} - -func (l *P2PEthDisconnectingBadTx) EventName() string { - return "p2p.eth.disconnecting.bad_tx" -} - -type EthNewBlockMined struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockHexRlp string `json:"block_hexrlp"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockMined) EventName() string { - return "eth.newblock.mined" -} - -type EthNewBlockBroadcasted struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockBroadcasted) EventName() string { - return "eth.newblock.broadcasted" -} - -type EthNewBlockReceived struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockReceived) EventName() string { - return "eth.newblock.received" -} - -type EthNewBlockIsKnown struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockIsKnown) EventName() string { - return "eth.newblock.is_known" -} - -type EthNewBlockIsNew struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockIsNew) EventName() string { - return "eth.newblock.is_new" -} - -type EthNewBlockMissingParent struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockMissingParent) EventName() string { - return "eth.newblock.missing_parent" -} - -type EthNewBlockIsInvalid struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockIsInvalid) EventName() string { - return "eth.newblock.is_invalid" -} - -type EthNewBlockChainIsOlder struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockChainIsOlder) EventName() string { - return "eth.newblock.chain.is_older" -} - -type EthNewBlockChainIsCanonical struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` - LogEvent -} - -func (l *EthNewBlockChainIsCanonical) EventName() string { - return "eth.newblock.chain.is_cannonical" -} - -type EthNewBlockChainNotCanonical struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` +type EthMinerNewBlock struct { + BlockHash string `json:"block_hash"` + BlockNumber int `json:"block_number"` + ChainHeadHash string `json:"chain_head_hash"` + BlockPrevHash string `json:"block_prev_hash"` LogEvent } -func (l *EthNewBlockChainNotCanonical) EventName() string { - return "eth.newblock.chain.not_cannonical" +func (l *EthMinerNewBlock) EventName() string { + return "eth.miner.new_block" } -type EthNewBlockChainSwitched struct { - BlockNumber int `json:"block_number"` - HeadHash string `json:"head_hash"` - OldHeadHash string `json:"old_head_hash"` - BlockHash string `json:"block_hash"` - BlockDifficulty int `json:"block_difficulty"` - BlockPrevHash string `json:"block_prev_hash"` +type EthChainReceivedNewBlock struct { + BlockHash string `json:"block_hash"` + BlockNumber int `json:"block_number"` + ChainHeadHash string `json:"chain_head_hash"` + BlockPrevHash string `json:"block_prev_hash"` + RemoteId int `json:"remote_id"` LogEvent } -func (l *EthNewBlockChainSwitched) EventName() string { - return "eth.newblock.chain.switched" +func (l *EthChainReceivedNewBlock) EventName() string { + return "eth.chain.received.new_block" } -type EthTxCreated struct { - TxHash string `json:"tx_hash"` - TxSender string `json:"tx_sender"` - TxAddress string `json:"tx_address"` - TxHexRLP string `json:"tx_hexrlp"` - TxNonce int `json:"tx_nonce"` +type EthChainNewHead struct { + BlockHash string `json:"block_hash"` + BlockNumber int `json:"block_number"` + ChainHeadHash string `json:"chain_head_hash"` + BlockPrevHash string `json:"block_prev_hash"` LogEvent } -func (l *EthTxCreated) EventName() string { - return "eth.tx.created" +func (l *EthChainNewHead) EventName() string { + return "eth.chain.new_head" } type EthTxReceived struct { - TxHash string `json:"tx_hash"` - TxAddress string `json:"tx_address"` - TxHexRLP string `json:"tx_hexrlp"` - RemoteId string `json:"remote_id"` - TxNonce int `json:"tx_nonce"` + TxHash string `json:"tx_hash"` + RemoteId string `json:"remote_id"` LogEvent } @@ -324,39 +99,261 @@ func (l *EthTxReceived) EventName() string { return "eth.tx.received" } -type EthTxBroadcasted struct { - TxHash string `json:"tx_hash"` - TxSender string `json:"tx_sender"` - TxAddress string `json:"tx_address"` - TxNonce int `json:"tx_nonce"` - LogEvent -} - -func (l *EthTxBroadcasted) EventName() string { - return "eth.tx.broadcasted" -} - -type EthTxValidated struct { - TxHash string `json:"tx_hash"` - TxSender string `json:"tx_sender"` - TxAddress string `json:"tx_address"` - TxNonce int `json:"tx_nonce"` - LogEvent -} - -func (l *EthTxValidated) EventName() string { - return "eth.tx.validated" -} - -type EthTxIsInvalid struct { - TxHash string `json:"tx_hash"` - TxSender string `json:"tx_sender"` - TxAddress string `json:"tx_address"` - Reason string `json:"reason"` - TxNonce int `json:"tx_nonce"` - LogEvent -} - -func (l *EthTxIsInvalid) EventName() string { - return "eth.tx.is_invalid" -} +// +// +// The types below are legacy and need to be converted to new format or deleted +// +// + +// type P2PConnecting struct { +// RemoteId string `json:"remote_id"` +// RemoteEndpoint string `json:"remote_endpoint"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PConnecting) EventName() string { +// return "p2p.connecting" +// } + +// type P2PHandshaked struct { +// RemoteCapabilities []string `json:"remote_capabilities"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PHandshaked) EventName() string { +// return "p2p.handshaked" +// } + +// type P2PDisconnecting struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PDisconnecting) EventName() string { +// return "p2p.disconnecting" +// } + +// type P2PDisconnectingBadHandshake struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PDisconnectingBadHandshake) EventName() string { +// return "p2p.disconnecting.bad_handshake" +// } + +// type P2PDisconnectingBadProtocol struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PDisconnectingBadProtocol) EventName() string { +// return "p2p.disconnecting.bad_protocol" +// } + +// type P2PDisconnectingReputation struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PDisconnectingReputation) EventName() string { +// return "p2p.disconnecting.reputation" +// } + +// type P2PDisconnectingDHT struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PDisconnectingDHT) EventName() string { +// return "p2p.disconnecting.dht" +// } + +// type P2PEthDisconnectingBadBlock struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PEthDisconnectingBadBlock) EventName() string { +// return "p2p.eth.disconnecting.bad_block" +// } + +// type P2PEthDisconnectingBadTx struct { +// Reason string `json:"reason"` +// RemoteId string `json:"remote_id"` +// NumConnections int `json:"num_connections"` +// LogEvent +// } + +// func (l *P2PEthDisconnectingBadTx) EventName() string { +// return "p2p.eth.disconnecting.bad_tx" +// } + +// type EthNewBlockBroadcasted struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockBroadcasted) EventName() string { +// return "eth.newblock.broadcasted" +// } + +// type EthNewBlockIsKnown struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockIsKnown) EventName() string { +// return "eth.newblock.is_known" +// } + +// type EthNewBlockIsNew struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockIsNew) EventName() string { +// return "eth.newblock.is_new" +// } + +// type EthNewBlockMissingParent struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockMissingParent) EventName() string { +// return "eth.newblock.missing_parent" +// } + +// type EthNewBlockIsInvalid struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockIsInvalid) EventName() string { +// return "eth.newblock.is_invalid" +// } + +// type EthNewBlockChainIsOlder struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockChainIsOlder) EventName() string { +// return "eth.newblock.chain.is_older" +// } + +// type EthNewBlockChainIsCanonical struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockChainIsCanonical) EventName() string { +// return "eth.newblock.chain.is_cannonical" +// } + +// type EthNewBlockChainNotCanonical struct { +// BlockNumber int `json:"block_number"` +// HeadHash string `json:"head_hash"` +// BlockHash string `json:"block_hash"` +// BlockDifficulty int `json:"block_difficulty"` +// BlockPrevHash string `json:"block_prev_hash"` +// LogEvent +// } + +// func (l *EthNewBlockChainNotCanonical) EventName() string { +// return "eth.newblock.chain.not_cannonical" +// } + +// type EthTxCreated struct { +// TxHash string `json:"tx_hash"` +// TxSender string `json:"tx_sender"` +// TxAddress string `json:"tx_address"` +// TxHexRLP string `json:"tx_hexrlp"` +// TxNonce int `json:"tx_nonce"` +// LogEvent +// } + +// func (l *EthTxCreated) EventName() string { +// return "eth.tx.created" +// } + +// type EthTxBroadcasted struct { +// TxHash string `json:"tx_hash"` +// TxSender string `json:"tx_sender"` +// TxAddress string `json:"tx_address"` +// TxNonce int `json:"tx_nonce"` +// LogEvent +// } + +// func (l *EthTxBroadcasted) EventName() string { +// return "eth.tx.broadcasted" +// } + +// type EthTxValidated struct { +// TxHash string `json:"tx_hash"` +// TxSender string `json:"tx_sender"` +// TxAddress string `json:"tx_address"` +// TxNonce int `json:"tx_nonce"` +// LogEvent +// } + +// func (l *EthTxValidated) EventName() string { +// return "eth.tx.validated" +// } + +// type EthTxIsInvalid struct { +// TxHash string `json:"tx_hash"` +// TxSender string `json:"tx_sender"` +// TxAddress string `json:"tx_address"` +// Reason string `json:"reason"` +// TxNonce int `json:"tx_nonce"` +// LogEvent +// } + +// func (l *EthTxIsInvalid) EventName() string { +// return "eth.tx.is_invalid" +// } diff --git a/miner/miner.go b/miner/miner.go index 27afcf684..0cc2361c8 100644 --- a/miner/miner.go +++ b/miner/miner.go @@ -20,13 +20,13 @@ type Miner struct { mining bool } -func New(coinbase []byte, eth core.Backend) *Miner { +func New(coinbase []byte, eth core.Backend, minerThreads int) *Miner { miner := &Miner{ Coinbase: coinbase, worker: newWorker(coinbase, eth), } - for i := 0; i < 4; i++ { + for i := 0; i < minerThreads; i++ { miner.worker.register(NewCpuMiner(i, ezp.New())) } diff --git a/p2p/crypto.go b/p2p/handshake.go index 7e4b43712..614711eaf 100644 --- a/p2p/crypto.go +++ b/p2p/handshake.go @@ -1,21 +1,20 @@ package p2p import ( - // "binary" "crypto/ecdsa" "crypto/rand" + "errors" "fmt" "io" + "net" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto/ecies" "github.com/ethereum/go-ethereum/crypto/secp256k1" - ethlogger "github.com/ethereum/go-ethereum/logger" "github.com/ethereum/go-ethereum/p2p/discover" + "github.com/ethereum/go-ethereum/rlp" ) -var clogger = ethlogger.NewLogger("CRYPTOID") - const ( sskLen = 16 // ecies.MaxSharedKeyLength(pubKey) / 2 sigLen = 65 // elliptic S256 @@ -30,26 +29,76 @@ const ( rHSLen = authRespLen + eciesBytes // size of the final ECIES payload sent as receiver's handshake ) -type hexkey []byte +type conn struct { + *frameRW + *protoHandshake +} -func (self hexkey) String() string { - return fmt.Sprintf("(%d) %x", len(self), []byte(self)) +func newConn(fd net.Conn, hs *protoHandshake) *conn { + return &conn{newFrameRW(fd, msgWriteTimeout), hs} } -func encHandshake(conn io.ReadWriter, prv *ecdsa.PrivateKey, dial *discover.Node) ( - remoteID discover.NodeID, - sessionToken []byte, - err error, -) { +// encHandshake represents information about the remote end +// of a connection that is negotiated during the encryption handshake. +type encHandshake struct { + ID discover.NodeID + IngressMAC []byte + EgressMAC []byte + Token []byte +} + +// protoHandshake is the RLP structure of the protocol handshake. +type protoHandshake struct { + Version uint64 + Name string + Caps []Cap + ListenPort uint64 + ID discover.NodeID +} + +// setupConn starts a protocol session on the given connection. +// It runs the encryption handshake and the protocol handshake. +// If dial is non-nil, the connection the local node is the initiator. +func setupConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, dial *discover.Node) (*conn, error) { if dial == nil { - var remotePubkey []byte - sessionToken, remotePubkey, err = inboundEncHandshake(conn, prv, nil) - copy(remoteID[:], remotePubkey) + return setupInboundConn(fd, prv, our) } else { - remoteID = dial.ID - sessionToken, err = outboundEncHandshake(conn, prv, remoteID[:], nil) + return setupOutboundConn(fd, prv, our, dial) + } +} + +func setupInboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake) (*conn, error) { + // var remotePubkey []byte + // sessionToken, remotePubkey, err = inboundEncHandshake(fd, prv, nil) + // copy(remoteID[:], remotePubkey) + + rw := newFrameRW(fd, msgWriteTimeout) + rhs, err := readProtocolHandshake(rw, our) + if err != nil { + return nil, err + } + if err := writeProtocolHandshake(rw, our); err != nil { + return nil, fmt.Errorf("protocol write error: %v", err) + } + return &conn{rw, rhs}, nil +} + +func setupOutboundConn(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, dial *discover.Node) (*conn, error) { + // remoteID = dial.ID + // sessionToken, err = outboundEncHandshake(fd, prv, remoteID[:], nil) + + rw := newFrameRW(fd, msgWriteTimeout) + if err := writeProtocolHandshake(rw, our); err != nil { + return nil, fmt.Errorf("protocol write error: %v", err) } - return remoteID, sessionToken, err + rhs, err := readProtocolHandshake(rw, our) + if err != nil { + return nil, fmt.Errorf("protocol handshake read error: %v", err) + } + if rhs.ID != dial.ID { + return nil, errors.New("dialed node id mismatch") + } + return &conn{rw, rhs}, nil } // outboundEncHandshake negotiates a session token on conn. @@ -66,18 +115,9 @@ func outboundEncHandshake(conn io.ReadWriter, prvKey *ecdsa.PrivateKey, remotePu if err != nil { return nil, err } - if sessionToken != nil { - clogger.Debugf("session-token: %v", hexkey(sessionToken)) - } - - clogger.Debugf("initiator-nonce: %v", hexkey(initNonce)) - clogger.Debugf("initiator-random-private-key: %v", hexkey(crypto.FromECDSA(randomPrivKey))) - randomPublicKeyS, _ := exportPublicKey(&randomPrivKey.PublicKey) - clogger.Debugf("initiator-random-public-key: %v", hexkey(randomPublicKeyS)) if _, err = conn.Write(auth); err != nil { return nil, err } - clogger.Debugf("initiator handshake: %v", hexkey(auth)) response := make([]byte, rHSLen) if _, err = io.ReadFull(conn, response); err != nil { @@ -88,9 +128,6 @@ func outboundEncHandshake(conn io.ReadWriter, prvKey *ecdsa.PrivateKey, remotePu return nil, err } - clogger.Debugf("receiver-nonce: %v", hexkey(recNonce)) - remoteRandomPubKeyS, _ := exportPublicKey(remoteRandomPubKey) - clogger.Debugf("receiver-random-public-key: %v", hexkey(remoteRandomPubKeyS)) return newSession(initNonce, recNonce, randomPrivKey, remoteRandomPubKey) } @@ -221,12 +258,9 @@ func inboundEncHandshake(conn io.ReadWriter, prvKey *ecdsa.PrivateKey, sessionTo if err != nil { return nil, nil, err } - clogger.Debugf("receiver-nonce: %v", hexkey(recNonce)) - clogger.Debugf("receiver-random-priv-key: %v", hexkey(crypto.FromECDSA(randomPrivKey))) if _, err = conn.Write(response); err != nil { return nil, nil, err } - clogger.Debugf("receiver handshake:\n%v", hexkey(response)) token, err = newSession(initNonce, recNonce, randomPrivKey, remoteRandomPubKey) return token, remotePubKey, err } @@ -361,3 +395,40 @@ func xor(one, other []byte) (xor []byte) { } return xor } + +func writeProtocolHandshake(w MsgWriter, our *protoHandshake) error { + return EncodeMsg(w, handshakeMsg, our.Version, our.Name, our.Caps, our.ListenPort, our.ID[:]) +} + +func readProtocolHandshake(r MsgReader, our *protoHandshake) (*protoHandshake, error) { + // read and handle remote handshake + msg, err := r.ReadMsg() + if err != nil { + return nil, err + } + if msg.Code == discMsg { + // disconnect before protocol handshake is valid according to the + // spec and we send it ourself if Server.addPeer fails. + var reason DiscReason + rlp.Decode(msg.Payload, &reason) + return nil, discRequestedError(reason) + } + if msg.Code != handshakeMsg { + return nil, fmt.Errorf("expected handshake, got %x", msg.Code) + } + if msg.Size > baseProtocolMaxMsgSize { + return nil, fmt.Errorf("message too big (%d > %d)", msg.Size, baseProtocolMaxMsgSize) + } + var hs protoHandshake + if err := msg.Decode(&hs); err != nil { + return nil, err + } + // validate handshake info + if hs.Version != our.Version { + return nil, newPeerError(errP2PVersionMismatch, "required version %d, received %d\n", baseProtocolVersion, hs.Version) + } + if (hs.ID == discover.NodeID{}) { + return nil, newPeerError(errPubkeyInvalid, "missing") + } + return &hs, nil +} diff --git a/p2p/crypto_test.go b/p2p/handshake_test.go index 6cf0f4818..06c6a6932 100644 --- a/p2p/crypto_test.go +++ b/p2p/handshake_test.go @@ -5,10 +5,12 @@ import ( "crypto/ecdsa" "crypto/rand" "net" + "reflect" "testing" "github.com/ethereum/go-ethereum/crypto" "github.com/ethereum/go-ethereum/crypto/ecies" + "github.com/ethereum/go-ethereum/p2p/discover" ) func TestPublicKeyEncoding(t *testing.T) { @@ -91,14 +93,14 @@ func testCryptoHandshake(prv0, prv1 *ecdsa.PrivateKey, sessionToken []byte, t *t if err != nil { t.Errorf("%v", err) } - t.Logf("-> %v", hexkey(auth)) + // t.Logf("-> %v", hexkey(auth)) // receiver reads auth and responds with response response, remoteRecNonce, remoteInitNonce, _, remoteRandomPrivKey, remoteInitRandomPubKey, err := authResp(auth, sessionToken, prv1) if err != nil { t.Errorf("%v", err) } - t.Logf("<- %v\n", hexkey(response)) + // t.Logf("<- %v\n", hexkey(response)) // initiator reads receiver's response and the key exchange completes recNonce, remoteRandomPubKey, _, err := completeHandshake(response, prv0) @@ -132,7 +134,7 @@ func testCryptoHandshake(prv0, prv1 *ecdsa.PrivateKey, sessionToken []byte, t *t } } -func TestHandshake(t *testing.T) { +func TestEncHandshake(t *testing.T) { defer testlog(t).detach() prv0, _ := crypto.GenerateKey() @@ -165,3 +167,58 @@ func TestHandshake(t *testing.T) { t.Error("session token mismatch") } } + +func TestSetupConn(t *testing.T) { + prv0, _ := crypto.GenerateKey() + prv1, _ := crypto.GenerateKey() + node0 := &discover.Node{ + ID: discover.PubkeyID(&prv0.PublicKey), + IP: net.IP{1, 2, 3, 4}, + TCPPort: 33, + } + node1 := &discover.Node{ + ID: discover.PubkeyID(&prv1.PublicKey), + IP: net.IP{5, 6, 7, 8}, + TCPPort: 44, + } + hs0 := &protoHandshake{ + Version: baseProtocolVersion, + ID: node0.ID, + Caps: []Cap{{"a", 0}, {"b", 2}}, + } + hs1 := &protoHandshake{ + Version: baseProtocolVersion, + ID: node1.ID, + Caps: []Cap{{"c", 1}, {"d", 3}}, + } + fd0, fd1 := net.Pipe() + + done := make(chan struct{}) + go func() { + defer close(done) + conn0, err := setupConn(fd0, prv0, hs0, node1) + if err != nil { + t.Errorf("outbound side error: %v", err) + return + } + if conn0.ID != node1.ID { + t.Errorf("outbound conn id mismatch: got %v, want %v", conn0.ID, node1.ID) + } + if !reflect.DeepEqual(conn0.Caps, hs1.Caps) { + t.Errorf("outbound caps mismatch: got %v, want %v", conn0.Caps, hs1.Caps) + } + }() + + conn1, err := setupConn(fd1, prv1, hs1, nil) + if err != nil { + t.Fatalf("inbound side error: %v", err) + } + if conn1.ID != node0.ID { + t.Errorf("inbound conn id mismatch: got %v, want %v", conn1.ID, node0.ID) + } + if !reflect.DeepEqual(conn1.Caps, hs0.Caps) { + t.Errorf("inbound caps mismatch: got %v, want %v", conn1.Caps, hs0.Caps) + } + + <-done +} diff --git a/p2p/message.go b/p2p/message.go index 07916f7b3..7adad4b09 100644 --- a/p2p/message.go +++ b/p2p/message.go @@ -197,7 +197,7 @@ func (rw *frameRW) ReadMsg() (msg Msg, err error) { return msg, err } if !bytes.HasPrefix(start, magicToken) { - return msg, fmt.Errorf("bad magic token %x", start[:4], magicToken) + return msg, fmt.Errorf("bad magic token %x", start[:4]) } size := binary.BigEndian.Uint32(start[4:]) diff --git a/p2p/peer.go b/p2p/peer.go index fd5bec7d5..fb027c834 100644 --- a/p2p/peer.go +++ b/p2p/peer.go @@ -21,6 +21,7 @@ const ( baseProtocolMaxMsgSize = 10 * 1024 * 1024 disconnectGracePeriod = 2 * time.Second + pingInterval = 15 * time.Second ) const ( @@ -33,37 +34,14 @@ const ( peersMsg = 0x05 ) -// handshake is the RLP structure of the protocol handshake. -type handshake struct { - Version uint64 - Name string - Caps []Cap - ListenPort uint64 - NodeID discover.NodeID -} - // Peer represents a connected remote node. type Peer struct { // Peers have all the log methods. // Use them to display messages related to the peer. *logger.Logger - infoMu sync.Mutex - name string - caps []Cap - - ourID, remoteID *discover.NodeID - ourName string - - rw *frameRW - - // These fields maintain the running protocols. - protocols []Protocol - runlock sync.RWMutex // protects running - running map[string]*proto - - // disables protocol handshake, for testing - noHandshake bool + rw *conn + running map[string]*protoRW protoWG sync.WaitGroup protoErr chan error @@ -73,36 +51,27 @@ type Peer struct { // NewPeer returns a peer for testing purposes. func NewPeer(id discover.NodeID, name string, caps []Cap) *Peer { - conn, _ := net.Pipe() - peer := newPeer(conn, nil, "", nil, &id) - peer.setHandshakeInfo(name, caps) + pipe, _ := net.Pipe() + conn := newConn(pipe, &protoHandshake{ID: id, Name: name, Caps: caps}) + peer := newPeer(conn, nil) close(peer.closed) // ensures Disconnect doesn't block return peer } // ID returns the node's public key. func (p *Peer) ID() discover.NodeID { - return *p.remoteID + return p.rw.ID } // Name returns the node name that the remote node advertised. func (p *Peer) Name() string { - // this needs a lock because the information is part of the - // protocol handshake. - p.infoMu.Lock() - name := p.name - p.infoMu.Unlock() - return name + return p.rw.Name } // Caps returns the capabilities (supported subprotocols) of the remote peer. func (p *Peer) Caps() []Cap { - // this needs a lock because the information is part of the - // protocol handshake. - p.infoMu.Lock() - caps := p.caps - p.infoMu.Unlock() - return caps + // TODO: maybe return copy + return p.rw.Caps } // RemoteAddr returns the remote address of the network connection. @@ -126,30 +95,20 @@ func (p *Peer) Disconnect(reason DiscReason) { // String implements fmt.Stringer. func (p *Peer) String() string { - return fmt.Sprintf("Peer %.8x %v", p.remoteID[:], p.RemoteAddr()) + return fmt.Sprintf("Peer %.8x %v", p.rw.ID[:], p.RemoteAddr()) } -func newPeer(conn net.Conn, protocols []Protocol, ourName string, ourID, remoteID *discover.NodeID) *Peer { - logtag := fmt.Sprintf("Peer %.8x %v", remoteID[:], conn.RemoteAddr()) - return &Peer{ - Logger: logger.NewLogger(logtag), - rw: newFrameRW(conn, msgWriteTimeout), - ourID: ourID, - ourName: ourName, - remoteID: remoteID, - protocols: protocols, - running: make(map[string]*proto), - disc: make(chan DiscReason), - protoErr: make(chan error), - closed: make(chan struct{}), +func newPeer(conn *conn, protocols []Protocol) *Peer { + logtag := fmt.Sprintf("Peer %.8x %v", conn.ID[:], conn.RemoteAddr()) + p := &Peer{ + Logger: logger.NewLogger(logtag), + rw: conn, + running: matchProtocols(protocols, conn.Caps, conn), + disc: make(chan DiscReason), + protoErr: make(chan error), + closed: make(chan struct{}), } -} - -func (p *Peer) setHandshakeInfo(name string, caps []Cap) { - p.infoMu.Lock() - p.name = name - p.caps = caps - p.infoMu.Unlock() + return p } func (p *Peer) run() DiscReason { @@ -157,29 +116,36 @@ func (p *Peer) run() DiscReason { defer p.closeProtocols() defer close(p.closed) + p.startProtocols() go func() { readErr <- p.readLoop() }() - if !p.noHandshake { - if err := writeProtocolHandshake(p.rw, p.ourName, *p.ourID, p.protocols); err != nil { - p.DebugDetailf("Protocol handshake error: %v\n", err) - p.rw.Close() - return DiscProtocolError - } - } + ping := time.NewTicker(pingInterval) + defer ping.Stop() // Wait for an error or disconnect. var reason DiscReason - select { - case err := <-readErr: - // We rely on protocols to abort if there is a write error. It - // might be more robust to handle them here as well. - p.DebugDetailf("Read error: %v\n", err) - p.rw.Close() - return DiscNetworkError - - case err := <-p.protoErr: - reason = discReasonForError(err) - case reason = <-p.disc: +loop: + for { + select { + case <-ping.C: + go func() { + if err := EncodeMsg(p.rw, pingMsg, nil); err != nil { + p.protoErr <- err + return + } + }() + case err := <-readErr: + // We rely on protocols to abort if there is a write error. It + // might be more robust to handle them here as well. + p.DebugDetailf("Read error: %v\n", err) + p.rw.Close() + return DiscNetworkError + case err := <-p.protoErr: + reason = discReasonForError(err) + break loop + case reason = <-p.disc: + break loop + } } p.politeDisconnect(reason) @@ -206,11 +172,6 @@ func (p *Peer) politeDisconnect(reason DiscReason) { } func (p *Peer) readLoop() error { - if !p.noHandshake { - if err := readProtocolHandshake(p, p.rw); err != nil { - return err - } - } for { msg, err := p.rw.ReadMsg() if err != nil { @@ -249,105 +210,51 @@ func (p *Peer) handle(msg Msg) error { return nil } -func readProtocolHandshake(p *Peer, rw MsgReadWriter) error { - // read and handle remote handshake - msg, err := rw.ReadMsg() - if err != nil { - return err - } - if msg.Code == discMsg { - // disconnect before protocol handshake is valid according to the - // spec and we send it ourself if Server.addPeer fails. - var reason DiscReason - rlp.Decode(msg.Payload, &reason) - return discRequestedError(reason) - } - if msg.Code != handshakeMsg { - return newPeerError(errProtocolBreach, "expected handshake, got %x", msg.Code) - } - if msg.Size > baseProtocolMaxMsgSize { - return newPeerError(errInvalidMsg, "message too big") - } - var hs handshake - if err := msg.Decode(&hs); err != nil { - return err - } - // validate handshake info - if hs.Version != baseProtocolVersion { - return newPeerError(errP2PVersionMismatch, "required version %d, received %d\n", - baseProtocolVersion, hs.Version) - } - if hs.NodeID == *p.remoteID { - return newPeerError(errPubkeyForbidden, "node ID mismatch") - } - // TODO: remove Caps with empty name - p.setHandshakeInfo(hs.Name, hs.Caps) - p.startSubprotocols(hs.Caps) - return nil -} - -func writeProtocolHandshake(w MsgWriter, name string, id discover.NodeID, ps []Protocol) error { - var caps []interface{} - for _, proto := range ps { - caps = append(caps, proto.cap()) - } - return EncodeMsg(w, handshakeMsg, baseProtocolVersion, name, caps, 0, id) -} - -// startProtocols starts matching named subprotocols. -func (p *Peer) startSubprotocols(caps []Cap) { +// matchProtocols creates structures for matching named subprotocols. +func matchProtocols(protocols []Protocol, caps []Cap, rw MsgReadWriter) map[string]*protoRW { sort.Sort(capsByName(caps)) - p.runlock.Lock() - defer p.runlock.Unlock() offset := baseProtocolLength + result := make(map[string]*protoRW) outer: for _, cap := range caps { - for _, proto := range p.protocols { - if proto.Name == cap.Name && - proto.Version == cap.Version && - p.running[cap.Name] == nil { - p.running[cap.Name] = p.startProto(offset, proto) + for _, proto := range protocols { + if proto.Name == cap.Name && proto.Version == cap.Version && result[cap.Name] == nil { + result[cap.Name] = &protoRW{Protocol: proto, offset: offset, in: make(chan Msg), w: rw} offset += proto.Length continue outer } } } + return result } -func (p *Peer) startProto(offset uint64, impl Protocol) *proto { - p.DebugDetailf("Starting protocol %s/%d\n", impl.Name, impl.Version) - rw := &proto{ - name: impl.Name, - in: make(chan Msg), - offset: offset, - maxcode: impl.Length, - w: p.rw, +func (p *Peer) startProtocols() { + for _, proto := range p.running { + proto := proto + p.DebugDetailf("Starting protocol %s/%d\n", proto.Name, proto.Version) + p.protoWG.Add(1) + go func() { + err := proto.Run(p, proto) + if err == nil { + p.DebugDetailf("Protocol %s/%d returned\n", proto.Name, proto.Version) + err = errors.New("protocol returned") + } else { + p.DebugDetailf("Protocol %s/%d error: %v\n", proto.Name, proto.Version, err) + } + select { + case p.protoErr <- err: + case <-p.closed: + } + p.protoWG.Done() + }() } - p.protoWG.Add(1) - go func() { - err := impl.Run(p, rw) - if err == nil { - p.DebugDetailf("Protocol %s/%d returned\n", impl.Name, impl.Version) - err = errors.New("protocol returned") - } else { - p.DebugDetailf("Protocol %s/%d error: %v\n", impl.Name, impl.Version, err) - } - select { - case p.protoErr <- err: - case <-p.closed: - } - p.protoWG.Done() - }() - return rw } // getProto finds the protocol responsible for handling // the given message code. -func (p *Peer) getProto(code uint64) (*proto, error) { - p.runlock.RLock() - defer p.runlock.RUnlock() +func (p *Peer) getProto(code uint64) (*protoRW, error) { for _, proto := range p.running { - if code >= proto.offset && code < proto.offset+proto.maxcode { + if code >= proto.offset && code < proto.offset+proto.Length { return proto, nil } } @@ -355,46 +262,43 @@ func (p *Peer) getProto(code uint64) (*proto, error) { } func (p *Peer) closeProtocols() { - p.runlock.RLock() for _, p := range p.running { close(p.in) } - p.runlock.RUnlock() p.protoWG.Wait() } // writeProtoMsg sends the given message on behalf of the given named protocol. // this exists because of Server.Broadcast. func (p *Peer) writeProtoMsg(protoName string, msg Msg) error { - p.runlock.RLock() proto, ok := p.running[protoName] - p.runlock.RUnlock() if !ok { return fmt.Errorf("protocol %s not handled by peer", protoName) } - if msg.Code >= proto.maxcode { + if msg.Code >= proto.Length { return newPeerError(errInvalidMsgCode, "code %x is out of range for protocol %q", msg.Code, protoName) } msg.Code += proto.offset return p.rw.WriteMsg(msg) } -type proto struct { - name string - in chan Msg - maxcode, offset uint64 - w MsgWriter +type protoRW struct { + Protocol + + in chan Msg + offset uint64 + w MsgWriter } -func (rw *proto) WriteMsg(msg Msg) error { - if msg.Code >= rw.maxcode { +func (rw *protoRW) WriteMsg(msg Msg) error { + if msg.Code >= rw.Length { return newPeerError(errInvalidMsgCode, "not handled") } msg.Code += rw.offset return rw.w.WriteMsg(msg) } -func (rw *proto) ReadMsg() (Msg, error) { +func (rw *protoRW) ReadMsg() (Msg, error) { msg, ok := <-rw.in if !ok { return msg, io.EOF diff --git a/p2p/peer_test.go b/p2p/peer_test.go index 68c9910a2..a1260adbd 100644 --- a/p2p/peer_test.go +++ b/p2p/peer_test.go @@ -6,11 +6,9 @@ import ( "io/ioutil" "net" "reflect" - "sort" "testing" "time" - "github.com/ethereum/go-ethereum/p2p/discover" "github.com/ethereum/go-ethereum/rlp" ) @@ -23,6 +21,7 @@ var discard = Protocol{ if err != nil { return err } + fmt.Printf("discarding %d\n", msg.Code) if err = msg.Discard(); err != nil { return err } @@ -30,13 +29,20 @@ var discard = Protocol{ }, } -func testPeer(noHandshake bool, protos []Protocol) (*frameRW, *Peer, <-chan DiscReason) { - conn1, conn2 := net.Pipe() - peer := newPeer(conn1, protos, "name", &discover.NodeID{}, &discover.NodeID{}) - peer.noHandshake = noHandshake +func testPeer(protos []Protocol) (*conn, *Peer, <-chan DiscReason) { + fd1, fd2 := net.Pipe() + hs1 := &protoHandshake{ID: randomID(), Version: baseProtocolVersion} + hs2 := &protoHandshake{ID: randomID(), Version: baseProtocolVersion} + for _, p := range protos { + hs1.Caps = append(hs1.Caps, p.cap()) + hs2.Caps = append(hs2.Caps, p.cap()) + } + + peer := newPeer(newConn(fd1, hs1), protos) errc := make(chan DiscReason, 1) go func() { errc <- peer.run() }() - return newFrameRW(conn2, msgWriteTimeout), peer, errc + + return newConn(fd2, hs2), peer, errc } func TestPeerProtoReadMsg(t *testing.T) { @@ -61,9 +67,8 @@ func TestPeerProtoReadMsg(t *testing.T) { }, } - rw, peer, errc := testPeer(true, []Protocol{proto}) + rw, _, errc := testPeer([]Protocol{proto}) defer rw.Close() - peer.startSubprotocols([]Cap{proto.cap()}) EncodeMsg(rw, baseProtocolLength+2, 1) EncodeMsg(rw, baseProtocolLength+3, 2) @@ -100,9 +105,8 @@ func TestPeerProtoReadLargeMsg(t *testing.T) { }, } - rw, peer, errc := testPeer(true, []Protocol{proto}) + rw, _, errc := testPeer([]Protocol{proto}) defer rw.Close() - peer.startSubprotocols([]Cap{proto.cap()}) EncodeMsg(rw, 18, make([]byte, msgsize)) select { @@ -130,9 +134,8 @@ func TestPeerProtoEncodeMsg(t *testing.T) { return nil }, } - rw, peer, _ := testPeer(true, []Protocol{proto}) + rw, _, _ := testPeer([]Protocol{proto}) defer rw.Close() - peer.startSubprotocols([]Cap{proto.cap()}) if err := expectMsg(rw, 17, []string{"foo", "bar"}); err != nil { t.Error(err) @@ -142,9 +145,8 @@ func TestPeerProtoEncodeMsg(t *testing.T) { func TestPeerWriteForBroadcast(t *testing.T) { defer testlog(t).detach() - rw, peer, peerErr := testPeer(true, []Protocol{discard}) + rw, peer, peerErr := testPeer([]Protocol{discard}) defer rw.Close() - peer.startSubprotocols([]Cap{discard.cap()}) // test write errors if err := peer.writeProtoMsg("b", NewMsg(3)); err == nil { @@ -160,7 +162,7 @@ func TestPeerWriteForBroadcast(t *testing.T) { read := make(chan struct{}) go func() { if err := expectMsg(rw, 16, nil); err != nil { - t.Error() + t.Error(err) } close(read) }() @@ -179,7 +181,7 @@ func TestPeerWriteForBroadcast(t *testing.T) { func TestPeerPing(t *testing.T) { defer testlog(t).detach() - rw, _, _ := testPeer(true, nil) + rw, _, _ := testPeer(nil) defer rw.Close() if err := EncodeMsg(rw, pingMsg); err != nil { t.Fatal(err) @@ -192,7 +194,7 @@ func TestPeerPing(t *testing.T) { func TestPeerDisconnect(t *testing.T) { defer testlog(t).detach() - rw, _, disc := testPeer(true, nil) + rw, _, disc := testPeer(nil) defer rw.Close() if err := EncodeMsg(rw, discMsg, DiscQuitting); err != nil { t.Fatal(err) @@ -206,73 +208,6 @@ func TestPeerDisconnect(t *testing.T) { } } -func TestPeerHandshake(t *testing.T) { - defer testlog(t).detach() - - // remote has two matching protocols: a and c - remote := NewPeer(randomID(), "", []Cap{{"a", 1}, {"b", 999}, {"c", 3}}) - remoteID := randomID() - remote.ourID = &remoteID - remote.ourName = "remote peer" - - start := make(chan string) - stop := make(chan struct{}) - run := func(p *Peer, rw MsgReadWriter) error { - name := rw.(*proto).name - if name != "a" && name != "c" { - t.Errorf("protocol %q should not be started", name) - } else { - start <- name - } - <-stop - return nil - } - protocols := []Protocol{ - {Name: "a", Version: 1, Length: 1, Run: run}, - {Name: "b", Version: 2, Length: 1, Run: run}, - {Name: "c", Version: 3, Length: 1, Run: run}, - {Name: "d", Version: 4, Length: 1, Run: run}, - } - rw, p, disc := testPeer(false, protocols) - p.remoteID = remote.ourID - defer rw.Close() - - // run the handshake - remoteProtocols := []Protocol{protocols[0], protocols[2]} - if err := writeProtocolHandshake(rw, "remote peer", remoteID, remoteProtocols); err != nil { - t.Fatalf("handshake write error: %v", err) - } - if err := readProtocolHandshake(remote, rw); err != nil { - t.Fatalf("handshake read error: %v", err) - } - - // check that all protocols have been started - var started []string - for i := 0; i < 2; i++ { - select { - case name := <-start: - started = append(started, name) - case <-time.After(100 * time.Millisecond): - } - } - sort.Strings(started) - if !reflect.DeepEqual(started, []string{"a", "c"}) { - t.Errorf("wrong protocols started: %v", started) - } - - // check that metadata has been set - if p.ID() != remoteID { - t.Errorf("peer has wrong node ID: got %v, want %v", p.ID(), remoteID) - } - if p.Name() != remote.ourName { - t.Errorf("peer has wrong node name: got %q, want %q", p.Name(), remote.ourName) - } - - close(stop) - expectMsg(rw, discMsg, nil) - t.Logf("disc reason: %v", <-disc) -} - func TestNewPeer(t *testing.T) { name := "nodename" caps := []Cap{{"foo", 2}, {"bar", 3}} diff --git a/p2p/server.go b/p2p/server.go index 35b584a27..3ea2538d1 100644 --- a/p2p/server.go +++ b/p2p/server.go @@ -5,7 +5,6 @@ import ( "crypto/ecdsa" "errors" "fmt" - "io" "net" "runtime" "sync" @@ -23,6 +22,7 @@ const ( ) var srvlog = logger.NewLogger("P2P Server") +var srvjslog = logger.NewJsonLogger() // MakeName creates a node name that follows the ethereum convention // for such names. It adds the operation system name and Go runtime version @@ -83,9 +83,11 @@ type Server struct { // Hooks for testing. These are useful because we can inhibit // the whole protocol stack. - handshakeFunc + setupFunc newPeerHook + ourHandshake *protoHandshake + lock sync.RWMutex running bool listener net.Listener @@ -99,7 +101,7 @@ type Server struct { peerConnect chan *discover.Node } -type handshakeFunc func(io.ReadWriter, *ecdsa.PrivateKey, *discover.Node) (discover.NodeID, []byte, error) +type setupFunc func(net.Conn, *ecdsa.PrivateKey, *protoHandshake, *discover.Node) (*conn, error) type newPeerHook func(*Peer) // Peers returns all connected peers. @@ -159,7 +161,7 @@ func (srv *Server) Start() (err error) { } srvlog.Infoln("Starting Server") - // initialize all the fields + // static fields if srv.PrivateKey == nil { return fmt.Errorf("Server.PrivateKey must be set to a non-nil key") } @@ -169,25 +171,32 @@ func (srv *Server) Start() (err error) { srv.quit = make(chan struct{}) srv.peers = make(map[discover.NodeID]*Peer) srv.peerConnect = make(chan *discover.Node) - - if srv.handshakeFunc == nil { - srv.handshakeFunc = encHandshake + if srv.setupFunc == nil { + srv.setupFunc = setupConn } if srv.Blacklist == nil { srv.Blacklist = NewBlacklist() } + + // node table + ntab, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT) + if err != nil { + return err + } + srv.ntab = ntab + + // handshake + srv.ourHandshake = &protoHandshake{Version: baseProtocolVersion, Name: srv.Name, ID: ntab.Self()} + for _, p := range srv.Protocols { + srv.ourHandshake.Caps = append(srv.ourHandshake.Caps, p.cap()) + } + + // listen/dial if srv.ListenAddr != "" { if err := srv.startListening(); err != nil { return err } } - - // dial stuff - dt, err := discover.ListenUDP(srv.PrivateKey, srv.ListenAddr, srv.NAT) - if err != nil { - return err - } - srv.ntab = dt if srv.Dialer == nil { srv.Dialer = &net.Dialer{Timeout: defaultDialTimeout} } @@ -347,30 +356,41 @@ func (srv *Server) findPeers() { } } -func (srv *Server) startPeer(conn net.Conn, dest *discover.Node) { +func (srv *Server) startPeer(fd net.Conn, dest *discover.Node) { // TODO: handle/store session token - conn.SetDeadline(time.Now().Add(handshakeTimeout)) - remoteID, _, err := srv.handshakeFunc(conn, srv.PrivateKey, dest) + fd.SetDeadline(time.Now().Add(handshakeTimeout)) + conn, err := srv.setupFunc(fd, srv.PrivateKey, srv.ourHandshake, dest) if err != nil { - conn.Close() - srvlog.Debugf("Encryption Handshake with %v failed: %v", conn.RemoteAddr(), err) + fd.Close() + srvlog.Debugf("Handshake with %v failed: %v", fd.RemoteAddr(), err) return } - ourID := srv.ntab.Self() - p := newPeer(conn, srv.Protocols, srv.Name, &ourID, &remoteID) - if ok, reason := srv.addPeer(remoteID, p); !ok { + p := newPeer(conn, srv.Protocols) + if ok, reason := srv.addPeer(conn.ID, p); !ok { srvlog.DebugDetailf("Not adding %v (%v)\n", p, reason) p.politeDisconnect(reason) return } + srvlog.Debugf("Added %v\n", p) + srvjslog.LogJson(&logger.P2PConnected{ + RemoteId: fmt.Sprintf("%x", conn.ID[:]), + RemoteAddress: conn.RemoteAddr().String(), + RemoteVersionString: conn.Name, + NumConnections: srv.PeerCount(), + }) if srv.newPeerHook != nil { srv.newPeerHook(p) } discreason := p.run() srv.removePeer(p) + srvlog.Debugf("Removed %v (%v)\n", p, discreason) + srvjslog.LogJson(&logger.P2PDisconnected{ + RemoteId: fmt.Sprintf("%x", conn.ID[:]), + NumConnections: srv.PeerCount(), + }) } func (srv *Server) addPeer(id discover.NodeID, p *Peer) (bool, DiscReason) { @@ -394,7 +414,7 @@ func (srv *Server) addPeer(id discover.NodeID, p *Peer) (bool, DiscReason) { func (srv *Server) removePeer(p *Peer) { srv.lock.Lock() - delete(srv.peers, *p.remoteID) + delete(srv.peers, p.ID()) srv.lock.Unlock() srv.peerWG.Done() } diff --git a/p2p/server_test.go b/p2p/server_test.go index aa2b3d243..c109fffb9 100644 --- a/p2p/server_test.go +++ b/p2p/server_test.go @@ -21,8 +21,12 @@ func startTestServer(t *testing.T, pf newPeerHook) *Server { ListenAddr: "127.0.0.1:0", PrivateKey: newkey(), newPeerHook: pf, - handshakeFunc: func(io.ReadWriter, *ecdsa.PrivateKey, *discover.Node) (id discover.NodeID, st []byte, err error) { - return randomID(), nil, err + setupFunc: func(fd net.Conn, prv *ecdsa.PrivateKey, our *protoHandshake, dial *discover.Node) (*conn, error) { + id := randomID() + return &conn{ + frameRW: newFrameRW(fd, msgWriteTimeout), + protoHandshake: &protoHandshake{ID: id, Version: baseProtocolVersion}, + }, nil }, } if err := server.Start(); err != nil { @@ -116,9 +120,7 @@ func TestServerBroadcast(t *testing.T) { var connected sync.WaitGroup srv := startTestServer(t, func(p *Peer) { - p.protocols = []Protocol{discard} - p.startSubprotocols([]Cap{discard.cap()}) - p.noHandshake = true + p.running = matchProtocols([]Protocol{discard}, []Cap{discard.cap()}, p.rw) connected.Done() }) defer srv.Stop() |