mirror of
https://github.com/fluencelabs/tendermint
synced 2025-04-25 06:42:16 +00:00
* OriginalAddr -> SocketAddr OriginalAddr records the originally dialed address for outbound peers, rather than the peer's self reported address. For inbound peers, it was nil. Here, we rename it to SocketAddr and for inbound peers, set it to the RemoteAddr of the connection. * use SocketAddr Numerous places in the code call peer.NodeInfo().NetAddress(). However, this call to NetAddress() may perform a DNS lookup if the reported NodeInfo.ListenAddr includes a name. Failure of this lookup returns a nil address, which can lead to panics in the code. Instead, call peer.SocketAddr() to return the static address of the connection. * remove nodeInfo.NetAddress() Expose `transport.NetAddress()`, a static result determined when the transport is created. Removing NetAddress() from the nodeInfo prevents accidental DNS lookups. * fixes from review * linter * fixes from review
This commit is contained in:
parent
1ecf814838
commit
882622ec10
@ -489,7 +489,7 @@ func NewNode(config *cfg.Config,
|
|||||||
addrBook := pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
|
addrBook := pex.NewAddrBook(config.P2P.AddrBookFile(), config.P2P.AddrBookStrict)
|
||||||
|
|
||||||
// Add ourselves to addrbook to prevent dialing ourselves
|
// Add ourselves to addrbook to prevent dialing ourselves
|
||||||
addrBook.AddOurAddress(nodeInfo.NetAddress())
|
addrBook.AddOurAddress(sw.NetAddress())
|
||||||
|
|
||||||
addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
|
addrBook.SetLogger(p2pLogger.With("book", config.P2P.AddrBookFile()))
|
||||||
if config.P2P.PexReactor {
|
if config.P2P.PexReactor {
|
||||||
|
@ -63,6 +63,6 @@ func (mp *Peer) Set(key string, value interface{}) {
|
|||||||
mp.kv[key] = value
|
mp.kv[key] = value
|
||||||
}
|
}
|
||||||
func (mp *Peer) RemoteIP() net.IP { return mp.ip }
|
func (mp *Peer) RemoteIP() net.IP { return mp.ip }
|
||||||
func (mp *Peer) OriginalAddr() *p2p.NetAddress { return mp.addr }
|
func (mp *Peer) SocketAddr() *p2p.NetAddress { return mp.addr }
|
||||||
func (mp *Peer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
func (mp *Peer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
||||||
func (mp *Peer) CloseConn() error { return nil }
|
func (mp *Peer) CloseConn() error { return nil }
|
||||||
|
@ -23,14 +23,8 @@ func MaxNodeInfoSize() int {
|
|||||||
// NodeInfo exposes basic info of a node
|
// NodeInfo exposes basic info of a node
|
||||||
// and determines if we're compatible.
|
// and determines if we're compatible.
|
||||||
type NodeInfo interface {
|
type NodeInfo interface {
|
||||||
nodeInfoAddress
|
|
||||||
nodeInfoTransport
|
|
||||||
}
|
|
||||||
|
|
||||||
// nodeInfoAddress exposes just the core info of a node.
|
|
||||||
type nodeInfoAddress interface {
|
|
||||||
ID() ID
|
ID() ID
|
||||||
NetAddress() *NetAddress
|
nodeInfoTransport
|
||||||
}
|
}
|
||||||
|
|
||||||
// nodeInfoTransport validates a nodeInfo and checks
|
// nodeInfoTransport validates a nodeInfo and checks
|
||||||
|
21
p2p/peer.go
21
p2p/peer.go
@ -29,7 +29,7 @@ type Peer interface {
|
|||||||
|
|
||||||
NodeInfo() NodeInfo // peer's info
|
NodeInfo() NodeInfo // peer's info
|
||||||
Status() tmconn.ConnectionStatus
|
Status() tmconn.ConnectionStatus
|
||||||
OriginalAddr() *NetAddress // original address for outbound peers
|
SocketAddr() *NetAddress // actual address of the socket
|
||||||
|
|
||||||
Send(byte, []byte) bool
|
Send(byte, []byte) bool
|
||||||
TrySend(byte, []byte) bool
|
TrySend(byte, []byte) bool
|
||||||
@ -46,7 +46,7 @@ type peerConn struct {
|
|||||||
persistent bool
|
persistent bool
|
||||||
conn net.Conn // source connection
|
conn net.Conn // source connection
|
||||||
|
|
||||||
originalAddr *NetAddress // nil for inbound connections
|
socketAddr *NetAddress
|
||||||
|
|
||||||
// cached RemoteIP()
|
// cached RemoteIP()
|
||||||
ip net.IP
|
ip net.IP
|
||||||
@ -55,14 +55,14 @@ type peerConn struct {
|
|||||||
func newPeerConn(
|
func newPeerConn(
|
||||||
outbound, persistent bool,
|
outbound, persistent bool,
|
||||||
conn net.Conn,
|
conn net.Conn,
|
||||||
originalAddr *NetAddress,
|
socketAddr *NetAddress,
|
||||||
) peerConn {
|
) peerConn {
|
||||||
|
|
||||||
return peerConn{
|
return peerConn{
|
||||||
outbound: outbound,
|
outbound: outbound,
|
||||||
persistent: persistent,
|
persistent: persistent,
|
||||||
conn: conn,
|
conn: conn,
|
||||||
originalAddr: originalAddr,
|
socketAddr: socketAddr,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -223,13 +223,12 @@ func (p *peer) NodeInfo() NodeInfo {
|
|||||||
return p.nodeInfo
|
return p.nodeInfo
|
||||||
}
|
}
|
||||||
|
|
||||||
// OriginalAddr returns the original address, which was used to connect with
|
// SocketAddr returns the address of the socket.
|
||||||
// the peer. Returns nil for inbound peers.
|
// For outbound peers, it's the address dialed (after DNS resolution).
|
||||||
func (p *peer) OriginalAddr() *NetAddress {
|
// For inbound peers, it's the address returned by the underlying connection
|
||||||
if p.peerConn.outbound {
|
// (not what's reported in the peer's NodeInfo).
|
||||||
return p.peerConn.originalAddr
|
func (p *peer) SocketAddr() *NetAddress {
|
||||||
}
|
return p.peerConn.socketAddr
|
||||||
return nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Status returns the peer's ConnectionStatus.
|
// Status returns the peer's ConnectionStatus.
|
||||||
|
@ -29,7 +29,7 @@ func (mp *mockPeer) IsPersistent() bool { return true }
|
|||||||
func (mp *mockPeer) Get(s string) interface{} { return s }
|
func (mp *mockPeer) Get(s string) interface{} { return s }
|
||||||
func (mp *mockPeer) Set(string, interface{}) {}
|
func (mp *mockPeer) Set(string, interface{}) {}
|
||||||
func (mp *mockPeer) RemoteIP() net.IP { return mp.ip }
|
func (mp *mockPeer) RemoteIP() net.IP { return mp.ip }
|
||||||
func (mp *mockPeer) OriginalAddr() *NetAddress { return nil }
|
func (mp *mockPeer) SocketAddr() *NetAddress { return nil }
|
||||||
func (mp *mockPeer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
func (mp *mockPeer) RemoteAddr() net.Addr { return &net.TCPAddr{IP: mp.ip, Port: 8800} }
|
||||||
func (mp *mockPeer) CloseConn() error { return nil }
|
func (mp *mockPeer) CloseConn() error { return nil }
|
||||||
|
|
||||||
|
@ -109,25 +109,27 @@ func testOutboundPeerConn(
|
|||||||
persistent bool,
|
persistent bool,
|
||||||
ourNodePrivKey crypto.PrivKey,
|
ourNodePrivKey crypto.PrivKey,
|
||||||
) (peerConn, error) {
|
) (peerConn, error) {
|
||||||
|
|
||||||
|
var pc peerConn
|
||||||
conn, err := testDial(addr, config)
|
conn, err := testDial(addr, config)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return peerConn{}, cmn.ErrorWrap(err, "Error creating peer")
|
return pc, cmn.ErrorWrap(err, "Error creating peer")
|
||||||
}
|
}
|
||||||
|
|
||||||
pc, err := testPeerConn(conn, config, true, persistent, ourNodePrivKey, addr)
|
pc, err = testPeerConn(conn, config, true, persistent, ourNodePrivKey, addr)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
if cerr := conn.Close(); cerr != nil {
|
if cerr := conn.Close(); cerr != nil {
|
||||||
return peerConn{}, cmn.ErrorWrap(err, cerr.Error())
|
return pc, cmn.ErrorWrap(err, cerr.Error())
|
||||||
}
|
}
|
||||||
return peerConn{}, err
|
return pc, err
|
||||||
}
|
}
|
||||||
|
|
||||||
// ensure dialed ID matches connection ID
|
// ensure dialed ID matches connection ID
|
||||||
if addr.ID != pc.ID() {
|
if addr.ID != pc.ID() {
|
||||||
if cerr := conn.Close(); cerr != nil {
|
if cerr := conn.Close(); cerr != nil {
|
||||||
return peerConn{}, cmn.ErrorWrap(err, cerr.Error())
|
return pc, cmn.ErrorWrap(err, cerr.Error())
|
||||||
}
|
}
|
||||||
return peerConn{}, ErrSwitchAuthenticationFailure{addr, pc.ID()}
|
return pc, ErrSwitchAuthenticationFailure{addr, pc.ID()}
|
||||||
}
|
}
|
||||||
|
|
||||||
return pc, nil
|
return pc, nil
|
||||||
|
@ -167,7 +167,7 @@ func (r *PEXReactor) AddPeer(p Peer) {
|
|||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
// inbound peer is its own source
|
// inbound peer is its own source
|
||||||
addr := p.NodeInfo().NetAddress()
|
addr := p.SocketAddr()
|
||||||
src := addr
|
src := addr
|
||||||
|
|
||||||
// add to book. dont RequestAddrs right away because
|
// add to book. dont RequestAddrs right away because
|
||||||
@ -309,7 +309,7 @@ func (r *PEXReactor) ReceiveAddrs(addrs []*p2p.NetAddress, src Peer) error {
|
|||||||
}
|
}
|
||||||
r.requestsSent.Delete(id)
|
r.requestsSent.Delete(id)
|
||||||
|
|
||||||
srcAddr := src.NodeInfo().NetAddress()
|
srcAddr := src.SocketAddr()
|
||||||
for _, netAddr := range addrs {
|
for _, netAddr := range addrs {
|
||||||
// Validate netAddr. Disconnect from a peer if it sends us invalid data.
|
// Validate netAddr. Disconnect from a peer if it sends us invalid data.
|
||||||
if netAddr == nil {
|
if netAddr == nil {
|
||||||
|
@ -96,7 +96,7 @@ func TestPEXReactorRunning(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
addOtherNodeAddrToAddrBook := func(switchIndex, otherSwitchIndex int) {
|
addOtherNodeAddrToAddrBook := func(switchIndex, otherSwitchIndex int) {
|
||||||
addr := switches[otherSwitchIndex].NodeInfo().NetAddress()
|
addr := switches[otherSwitchIndex].NetAddress()
|
||||||
books[switchIndex].AddAddress(addr, addr)
|
books[switchIndex].AddAddress(addr, addr)
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -127,7 +127,7 @@ func TestPEXReactorReceive(t *testing.T) {
|
|||||||
r.RequestAddrs(peer)
|
r.RequestAddrs(peer)
|
||||||
|
|
||||||
size := book.Size()
|
size := book.Size()
|
||||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||||
r.Receive(PexChannel, peer, msg)
|
r.Receive(PexChannel, peer, msg)
|
||||||
assert.Equal(t, size+1, book.Size())
|
assert.Equal(t, size+1, book.Size())
|
||||||
@ -184,7 +184,7 @@ func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
|
|||||||
assert.True(t, r.requestsSent.Has(id))
|
assert.True(t, r.requestsSent.Has(id))
|
||||||
assert.True(t, sw.Peers().Has(peer.ID()))
|
assert.True(t, sw.Peers().Has(peer.ID()))
|
||||||
|
|
||||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||||
|
|
||||||
// receive some addrs. should clear the request
|
// receive some addrs. should clear the request
|
||||||
@ -229,7 +229,7 @@ func TestCheckSeeds(t *testing.T) {
|
|||||||
badPeerConfig = &PEXReactorConfig{
|
badPeerConfig = &PEXReactorConfig{
|
||||||
Seeds: []string{"ed3dfd27bfc4af18f67a49862f04cc100696e84d@bad.network.addr:26657",
|
Seeds: []string{"ed3dfd27bfc4af18f67a49862f04cc100696e84d@bad.network.addr:26657",
|
||||||
"d824b13cb5d40fa1d8a614e089357c7eff31b670@anotherbad.network.addr:26657",
|
"d824b13cb5d40fa1d8a614e089357c7eff31b670@anotherbad.network.addr:26657",
|
||||||
seed.NodeInfo().NetAddress().String()},
|
seed.NetAddress().String()},
|
||||||
}
|
}
|
||||||
peer = testCreatePeerWithConfig(dir, 2, badPeerConfig)
|
peer = testCreatePeerWithConfig(dir, 2, badPeerConfig)
|
||||||
require.Nil(t, peer.Start())
|
require.Nil(t, peer.Start())
|
||||||
@ -263,12 +263,13 @@ func TestConnectionSpeedForPeerReceivedFromSeed(t *testing.T) {
|
|||||||
defer os.RemoveAll(dir) // nolint: errcheck
|
defer os.RemoveAll(dir) // nolint: errcheck
|
||||||
|
|
||||||
// 1. create peer
|
// 1. create peer
|
||||||
peer := testCreateDefaultPeer(dir, 1)
|
peerSwitch := testCreateDefaultPeer(dir, 1)
|
||||||
require.Nil(t, peer.Start())
|
require.Nil(t, peerSwitch.Start())
|
||||||
defer peer.Stop()
|
defer peerSwitch.Stop()
|
||||||
|
|
||||||
// 2. Create seed which knows about the peer
|
// 2. Create seed which knows about the peer
|
||||||
seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peer.NodeInfo().NetAddress()}, []*p2p.NetAddress{peer.NodeInfo().NetAddress()})
|
peerAddr := peerSwitch.NetAddress()
|
||||||
|
seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peerAddr}, []*p2p.NetAddress{peerAddr})
|
||||||
require.Nil(t, seed.Start())
|
require.Nil(t, seed.Start())
|
||||||
defer seed.Stop()
|
defer seed.Stop()
|
||||||
|
|
||||||
@ -295,7 +296,7 @@ func TestPEXReactorCrawlStatus(t *testing.T) {
|
|||||||
// Create a peer, add it to the peer set and the addrbook.
|
// Create a peer, add it to the peer set and the addrbook.
|
||||||
peer := p2p.CreateRandomPeer(false)
|
peer := p2p.CreateRandomPeer(false)
|
||||||
p2p.AddPeerToSwitch(pexR.Switch, peer)
|
p2p.AddPeerToSwitch(pexR.Switch, peer)
|
||||||
addr1 := peer.NodeInfo().NetAddress()
|
addr1 := peer.SocketAddr()
|
||||||
pexR.book.AddAddress(addr1, addr1)
|
pexR.book.AddAddress(addr1, addr1)
|
||||||
|
|
||||||
// Add a non-connected address to the book.
|
// Add a non-connected address to the book.
|
||||||
@ -359,7 +360,7 @@ func TestPEXReactorSeedModeFlushStop(t *testing.T) {
|
|||||||
reactor := switches[0].Reactors()["pex"].(*PEXReactor)
|
reactor := switches[0].Reactors()["pex"].(*PEXReactor)
|
||||||
peerID := switches[1].NodeInfo().ID()
|
peerID := switches[1].NodeInfo().ID()
|
||||||
|
|
||||||
err = switches[1].DialPeerWithAddress(switches[0].NodeInfo().NetAddress(), false)
|
err = switches[1].DialPeerWithAddress(switches[0].NetAddress(), false)
|
||||||
assert.NoError(t, err)
|
assert.NoError(t, err)
|
||||||
|
|
||||||
// sleep up to a second while waiting for the peer to send us a message.
|
// sleep up to a second while waiting for the peer to send us a message.
|
||||||
@ -397,7 +398,7 @@ func TestPEXReactorDoesNotAddPrivatePeersToAddrBook(t *testing.T) {
|
|||||||
pexR.RequestAddrs(peer)
|
pexR.RequestAddrs(peer)
|
||||||
|
|
||||||
size := book.Size()
|
size := book.Size()
|
||||||
addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
|
addrs := []*p2p.NetAddress{peer.SocketAddr()}
|
||||||
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
|
||||||
pexR.Receive(PexChannel, peer, msg)
|
pexR.Receive(PexChannel, peer, msg)
|
||||||
assert.Equal(t, size, book.Size())
|
assert.Equal(t, size, book.Size())
|
||||||
@ -414,7 +415,7 @@ func TestPEXReactorDialPeer(t *testing.T) {
|
|||||||
sw.SetAddrBook(book)
|
sw.SetAddrBook(book)
|
||||||
|
|
||||||
peer := mock.NewPeer(nil)
|
peer := mock.NewPeer(nil)
|
||||||
addr := peer.NodeInfo().NetAddress()
|
addr := peer.SocketAddr()
|
||||||
|
|
||||||
assert.Equal(t, 0, pexR.AttemptsToDial(addr))
|
assert.Equal(t, 0, pexR.AttemptsToDial(addr))
|
||||||
|
|
||||||
@ -547,7 +548,7 @@ func testCreateSeed(dir string, id int, knownAddrs, srcAddrs []*p2p.NetAddress)
|
|||||||
// Starting and stopping the peer is left to the caller
|
// Starting and stopping the peer is left to the caller
|
||||||
func testCreatePeerWithSeed(dir string, id int, seed *p2p.Switch) *p2p.Switch {
|
func testCreatePeerWithSeed(dir string, id int, seed *p2p.Switch) *p2p.Switch {
|
||||||
conf := &PEXReactorConfig{
|
conf := &PEXReactorConfig{
|
||||||
Seeds: []string{seed.NodeInfo().NetAddress().String()},
|
Seeds: []string{seed.NetAddress().String()},
|
||||||
}
|
}
|
||||||
return testCreatePeerWithConfig(dir, id, conf)
|
return testCreatePeerWithConfig(dir, id, conf)
|
||||||
}
|
}
|
||||||
|
@ -86,6 +86,12 @@ type Switch struct {
|
|||||||
metrics *Metrics
|
metrics *Metrics
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NetAddress returns the address the switch is listening on.
|
||||||
|
func (sw *Switch) NetAddress() *NetAddress {
|
||||||
|
addr := sw.transport.NetAddress()
|
||||||
|
return &addr
|
||||||
|
}
|
||||||
|
|
||||||
// SwitchOption sets an optional parameter on the Switch.
|
// SwitchOption sets an optional parameter on the Switch.
|
||||||
type SwitchOption func(*Switch)
|
type SwitchOption func(*Switch)
|
||||||
|
|
||||||
@ -289,13 +295,7 @@ func (sw *Switch) StopPeerForError(peer Peer, reason interface{}) {
|
|||||||
sw.stopAndRemovePeer(peer, reason)
|
sw.stopAndRemovePeer(peer, reason)
|
||||||
|
|
||||||
if peer.IsPersistent() {
|
if peer.IsPersistent() {
|
||||||
addr := peer.OriginalAddr()
|
go sw.reconnectToPeer(peer.SocketAddr())
|
||||||
if addr == nil {
|
|
||||||
// FIXME: persistent peers can't be inbound right now.
|
|
||||||
// self-reported address for inbound persistent peers
|
|
||||||
addr = peer.NodeInfo().NetAddress()
|
|
||||||
}
|
|
||||||
go sw.reconnectToPeer(addr)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -383,7 +383,7 @@ func (sw *Switch) SetAddrBook(addrBook AddrBook) {
|
|||||||
// like contributed to consensus.
|
// like contributed to consensus.
|
||||||
func (sw *Switch) MarkPeerAsGood(peer Peer) {
|
func (sw *Switch) MarkPeerAsGood(peer Peer) {
|
||||||
if sw.addrBook != nil {
|
if sw.addrBook != nil {
|
||||||
sw.addrBook.MarkGood(peer.NodeInfo().NetAddress())
|
sw.addrBook.MarkGood(peer.SocketAddr())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@ -400,7 +400,7 @@ func (sw *Switch) DialPeersAsync(addrBook AddrBook, peers []string, persistent b
|
|||||||
sw.Logger.Error("Error in peer's address", "err", err)
|
sw.Logger.Error("Error in peer's address", "err", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
ourAddr := sw.nodeInfo.NetAddress()
|
ourAddr := sw.NetAddress()
|
||||||
|
|
||||||
// TODO: this code feels like it's in the wrong place.
|
// TODO: this code feels like it's in the wrong place.
|
||||||
// The integration tests depend on the addrBook being saved
|
// The integration tests depend on the addrBook being saved
|
||||||
@ -536,7 +536,7 @@ func (sw *Switch) acceptRoutine() {
|
|||||||
if in >= sw.config.MaxNumInboundPeers {
|
if in >= sw.config.MaxNumInboundPeers {
|
||||||
sw.Logger.Info(
|
sw.Logger.Info(
|
||||||
"Ignoring inbound connection: already have enough inbound peers",
|
"Ignoring inbound connection: already have enough inbound peers",
|
||||||
"address", p.NodeInfo().NetAddress().String(),
|
"address", p.SocketAddr(),
|
||||||
"have", in,
|
"have", in,
|
||||||
"max", sw.config.MaxNumInboundPeers,
|
"max", sw.config.MaxNumInboundPeers,
|
||||||
)
|
)
|
||||||
@ -653,7 +653,7 @@ func (sw *Switch) addPeer(p Peer) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
p.SetLogger(sw.Logger.With("peer", p.NodeInfo().NetAddress()))
|
p.SetLogger(sw.Logger.With("peer", p.SocketAddr()))
|
||||||
|
|
||||||
// Handle the shut down case where the switch has stopped but we're
|
// Handle the shut down case where the switch has stopped but we're
|
||||||
// concurrently trying to add a peer.
|
// concurrently trying to add a peer.
|
||||||
|
@ -161,10 +161,6 @@ func assertMsgReceivedWithTimeout(t *testing.T, msgBytes []byte, channel byte, r
|
|||||||
|
|
||||||
func TestSwitchFiltersOutItself(t *testing.T) {
|
func TestSwitchFiltersOutItself(t *testing.T) {
|
||||||
s1 := MakeSwitch(cfg, 1, "127.0.0.1", "123.123.123", initSwitchFunc)
|
s1 := MakeSwitch(cfg, 1, "127.0.0.1", "123.123.123", initSwitchFunc)
|
||||||
// addr := s1.NodeInfo().NetAddress()
|
|
||||||
|
|
||||||
// // add ourselves like we do in node.go#427
|
|
||||||
// s1.addrBook.AddOurAddress(addr)
|
|
||||||
|
|
||||||
// simulate s1 having a public IP by creating a remote peer with the same ID
|
// simulate s1 having a public IP by creating a remote peer with the same ID
|
||||||
rp := &remotePeer{PrivKey: s1.nodeKey.PrivKey, Config: cfg}
|
rp := &remotePeer{PrivKey: s1.nodeKey.PrivKey, Config: cfg}
|
||||||
@ -498,7 +494,7 @@ func TestSwitchAcceptRoutine(t *testing.T) {
|
|||||||
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
||||||
remotePeers = append(remotePeers, rp)
|
remotePeers = append(remotePeers, rp)
|
||||||
rp.Start()
|
rp.Start()
|
||||||
c, err := rp.Dial(sw.NodeInfo().NetAddress())
|
c, err := rp.Dial(sw.NetAddress())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
// spawn a reading routine to prevent connection from closing
|
// spawn a reading routine to prevent connection from closing
|
||||||
go func(c net.Conn) {
|
go func(c net.Conn) {
|
||||||
@ -517,7 +513,7 @@ func TestSwitchAcceptRoutine(t *testing.T) {
|
|||||||
// 2. check we close new connections if we already have MaxNumInboundPeers peers
|
// 2. check we close new connections if we already have MaxNumInboundPeers peers
|
||||||
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
rp := &remotePeer{PrivKey: ed25519.GenPrivKey(), Config: cfg}
|
||||||
rp.Start()
|
rp.Start()
|
||||||
conn, err := rp.Dial(sw.NodeInfo().NetAddress())
|
conn, err := rp.Dial(sw.NetAddress())
|
||||||
require.NoError(t, err)
|
require.NoError(t, err)
|
||||||
// check conn is closed
|
// check conn is closed
|
||||||
one := make([]byte, 1)
|
one := make([]byte, 1)
|
||||||
@ -537,6 +533,10 @@ type errorTransport struct {
|
|||||||
acceptErr error
|
acceptErr error
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func (et errorTransport) NetAddress() NetAddress {
|
||||||
|
panic("not implemented")
|
||||||
|
}
|
||||||
|
|
||||||
func (et errorTransport) Accept(c peerConfig) (Peer, error) {
|
func (et errorTransport) Accept(c peerConfig) (Peer, error) {
|
||||||
return nil, et.acceptErr
|
return nil, et.acceptErr
|
||||||
}
|
}
|
||||||
|
@ -36,6 +36,7 @@ func CreateRandomPeer(outbound bool) *peer {
|
|||||||
p := &peer{
|
p := &peer{
|
||||||
peerConn: peerConn{
|
peerConn: peerConn{
|
||||||
outbound: outbound,
|
outbound: outbound,
|
||||||
|
socketAddr: netAddr,
|
||||||
},
|
},
|
||||||
nodeInfo: mockNodeInfo{netAddr},
|
nodeInfo: mockNodeInfo{netAddr},
|
||||||
mconn: &conn.MConnection{},
|
mconn: &conn.MConnection{},
|
||||||
@ -174,10 +175,15 @@ func MakeSwitch(
|
|||||||
PrivKey: ed25519.GenPrivKey(),
|
PrivKey: ed25519.GenPrivKey(),
|
||||||
}
|
}
|
||||||
nodeInfo := testNodeInfo(nodeKey.ID(), fmt.Sprintf("node%d", i))
|
nodeInfo := testNodeInfo(nodeKey.ID(), fmt.Sprintf("node%d", i))
|
||||||
|
addr, err := NewNetAddressString(
|
||||||
|
IDAddressString(nodeKey.ID(), nodeInfo.(DefaultNodeInfo).ListenAddr),
|
||||||
|
)
|
||||||
|
if err != nil {
|
||||||
|
panic(err)
|
||||||
|
}
|
||||||
|
|
||||||
t := NewMultiplexTransport(nodeInfo, nodeKey, MConnConfig(cfg))
|
t := NewMultiplexTransport(nodeInfo, nodeKey, MConnConfig(cfg))
|
||||||
|
|
||||||
addr := nodeInfo.NetAddress()
|
|
||||||
if err := t.Listen(*addr); err != nil {
|
if err := t.Listen(*addr); err != nil {
|
||||||
panic(err)
|
panic(err)
|
||||||
}
|
}
|
||||||
@ -214,7 +220,7 @@ func testPeerConn(
|
|||||||
cfg *config.P2PConfig,
|
cfg *config.P2PConfig,
|
||||||
outbound, persistent bool,
|
outbound, persistent bool,
|
||||||
ourNodePrivKey crypto.PrivKey,
|
ourNodePrivKey crypto.PrivKey,
|
||||||
originalAddr *NetAddress,
|
socketAddr *NetAddress,
|
||||||
) (pc peerConn, err error) {
|
) (pc peerConn, err error) {
|
||||||
conn := rawConn
|
conn := rawConn
|
||||||
|
|
||||||
@ -231,12 +237,7 @@ func testPeerConn(
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Only the information we already have
|
// Only the information we already have
|
||||||
return peerConn{
|
return newPeerConn(outbound, persistent, conn, socketAddr), nil
|
||||||
outbound: outbound,
|
|
||||||
persistent: persistent,
|
|
||||||
conn: conn,
|
|
||||||
originalAddr: originalAddr,
|
|
||||||
}, nil
|
|
||||||
}
|
}
|
||||||
|
|
||||||
//----------------------------------------------------------------
|
//----------------------------------------------------------------
|
||||||
|
@ -24,6 +24,7 @@ type IPResolver interface {
|
|||||||
// accept is the container to carry the upgraded connection and NodeInfo from an
|
// accept is the container to carry the upgraded connection and NodeInfo from an
|
||||||
// asynchronously running routine to the Accept method.
|
// asynchronously running routine to the Accept method.
|
||||||
type accept struct {
|
type accept struct {
|
||||||
|
netAddr *NetAddress
|
||||||
conn net.Conn
|
conn net.Conn
|
||||||
nodeInfo NodeInfo
|
nodeInfo NodeInfo
|
||||||
err error
|
err error
|
||||||
@ -47,6 +48,9 @@ type peerConfig struct {
|
|||||||
// the transport. Each transport is also responsible to filter establishing
|
// the transport. Each transport is also responsible to filter establishing
|
||||||
// peers specific to its domain.
|
// peers specific to its domain.
|
||||||
type Transport interface {
|
type Transport interface {
|
||||||
|
// Listening address.
|
||||||
|
NetAddress() NetAddress
|
||||||
|
|
||||||
// Accept returns a newly connected Peer.
|
// Accept returns a newly connected Peer.
|
||||||
Accept(peerConfig) (Peer, error)
|
Accept(peerConfig) (Peer, error)
|
||||||
|
|
||||||
@ -115,6 +119,7 @@ func MultiplexTransportResolver(resolver IPResolver) MultiplexTransportOption {
|
|||||||
// MultiplexTransport accepts and dials tcp connections and upgrades them to
|
// MultiplexTransport accepts and dials tcp connections and upgrades them to
|
||||||
// multiplexed peers.
|
// multiplexed peers.
|
||||||
type MultiplexTransport struct {
|
type MultiplexTransport struct {
|
||||||
|
netAddr NetAddress
|
||||||
listener net.Listener
|
listener net.Listener
|
||||||
|
|
||||||
acceptc chan accept
|
acceptc chan accept
|
||||||
@ -161,6 +166,11 @@ func NewMultiplexTransport(
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// NetAddress implements Transport.
|
||||||
|
func (mt *MultiplexTransport) NetAddress() NetAddress {
|
||||||
|
return mt.netAddr
|
||||||
|
}
|
||||||
|
|
||||||
// Accept implements Transport.
|
// Accept implements Transport.
|
||||||
func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
|
func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
|
||||||
select {
|
select {
|
||||||
@ -173,7 +183,7 @@ func (mt *MultiplexTransport) Accept(cfg peerConfig) (Peer, error) {
|
|||||||
|
|
||||||
cfg.outbound = false
|
cfg.outbound = false
|
||||||
|
|
||||||
return mt.wrapPeer(a.conn, a.nodeInfo, cfg, nil), nil
|
return mt.wrapPeer(a.conn, a.nodeInfo, cfg, a.netAddr), nil
|
||||||
case <-mt.closec:
|
case <-mt.closec:
|
||||||
return nil, ErrTransportClosed{}
|
return nil, ErrTransportClosed{}
|
||||||
}
|
}
|
||||||
@ -224,6 +234,7 @@ func (mt *MultiplexTransport) Listen(addr NetAddress) error {
|
|||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
|
|
||||||
|
mt.netAddr = addr
|
||||||
mt.listener = ln
|
mt.listener = ln
|
||||||
|
|
||||||
go mt.acceptPeers()
|
go mt.acceptPeers()
|
||||||
@ -258,15 +269,21 @@ func (mt *MultiplexTransport) acceptPeers() {
|
|||||||
var (
|
var (
|
||||||
nodeInfo NodeInfo
|
nodeInfo NodeInfo
|
||||||
secretConn *conn.SecretConnection
|
secretConn *conn.SecretConnection
|
||||||
|
netAddr *NetAddress
|
||||||
)
|
)
|
||||||
|
|
||||||
err := mt.filterConn(c)
|
err := mt.filterConn(c)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
secretConn, nodeInfo, err = mt.upgrade(c, nil)
|
secretConn, nodeInfo, err = mt.upgrade(c, nil)
|
||||||
|
if err == nil {
|
||||||
|
addr := c.RemoteAddr()
|
||||||
|
id := PubKeyToID(secretConn.RemotePubKey())
|
||||||
|
netAddr = NewNetAddress(id, addr)
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
select {
|
select {
|
||||||
case mt.acceptc <- accept{secretConn, nodeInfo, err}:
|
case mt.acceptc <- accept{netAddr, secretConn, nodeInfo, err}:
|
||||||
// Make the upgraded peer available.
|
// Make the upgraded peer available.
|
||||||
case <-mt.closec:
|
case <-mt.closec:
|
||||||
// Give up if the transport was closed.
|
// Give up if the transport was closed.
|
||||||
@ -426,14 +443,14 @@ func (mt *MultiplexTransport) wrapPeer(
|
|||||||
c net.Conn,
|
c net.Conn,
|
||||||
ni NodeInfo,
|
ni NodeInfo,
|
||||||
cfg peerConfig,
|
cfg peerConfig,
|
||||||
dialedAddr *NetAddress,
|
socketAddr *NetAddress,
|
||||||
) Peer {
|
) Peer {
|
||||||
|
|
||||||
peerConn := newPeerConn(
|
peerConn := newPeerConn(
|
||||||
cfg.outbound,
|
cfg.outbound,
|
||||||
cfg.persistent,
|
cfg.persistent,
|
||||||
c,
|
c,
|
||||||
dialedAddr,
|
socketAddr,
|
||||||
)
|
)
|
||||||
|
|
||||||
p := newPeer(
|
p := newPeer(
|
||||||
|
@ -8,6 +8,8 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/stretchr/testify/require"
|
||||||
|
|
||||||
"github.com/tendermint/tendermint/crypto/ed25519"
|
"github.com/tendermint/tendermint/crypto/ed25519"
|
||||||
"github.com/tendermint/tendermint/p2p/conn"
|
"github.com/tendermint/tendermint/p2p/conn"
|
||||||
)
|
)
|
||||||
@ -142,43 +144,23 @@ func TestTransportMultiplexConnFilterTimeout(t *testing.T) {
|
|||||||
|
|
||||||
func TestTransportMultiplexAcceptMultiple(t *testing.T) {
|
func TestTransportMultiplexAcceptMultiple(t *testing.T) {
|
||||||
mt := testSetupMultiplexTransport(t)
|
mt := testSetupMultiplexTransport(t)
|
||||||
|
id, addr := mt.nodeKey.ID(), mt.listener.Addr().String()
|
||||||
|
laddr, err := NewNetAddressStringWithOptionalID(IDAddressString(id, addr))
|
||||||
|
require.NoError(t, err)
|
||||||
|
|
||||||
var (
|
var (
|
||||||
seed = rand.New(rand.NewSource(time.Now().UnixNano()))
|
seed = rand.New(rand.NewSource(time.Now().UnixNano()))
|
||||||
errc = make(chan error, seed.Intn(64)+64)
|
nDialers = seed.Intn(64) + 64
|
||||||
|
errc = make(chan error, nDialers)
|
||||||
)
|
)
|
||||||
|
|
||||||
// Setup dialers.
|
// Setup dialers.
|
||||||
for i := 0; i < cap(errc); i++ {
|
for i := 0; i < nDialers; i++ {
|
||||||
go func() {
|
go testDialer(*laddr, errc)
|
||||||
var (
|
|
||||||
pv = ed25519.GenPrivKey()
|
|
||||||
dialer = newMultiplexTransport(
|
|
||||||
testNodeInfo(PubKeyToID(pv.PubKey()), defaultNodeName),
|
|
||||||
NodeKey{
|
|
||||||
PrivKey: pv,
|
|
||||||
},
|
|
||||||
)
|
|
||||||
)
|
|
||||||
addr, err := NewNetAddressStringWithOptionalID(IDAddressString(mt.nodeKey.ID(), mt.listener.Addr().String()))
|
|
||||||
if err != nil {
|
|
||||||
errc <- err
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
_, err = dialer.Dial(*addr, peerConfig{})
|
|
||||||
if err != nil {
|
|
||||||
errc <- err
|
|
||||||
return
|
|
||||||
}
|
|
||||||
|
|
||||||
// Signal that the connection was established.
|
|
||||||
errc <- nil
|
|
||||||
}()
|
|
||||||
}
|
}
|
||||||
|
|
||||||
// Catch connection errors.
|
// Catch connection errors.
|
||||||
for i := 0; i < cap(errc); i++ {
|
for i := 0; i < nDialers; i++ {
|
||||||
if err := <-errc; err != nil {
|
if err := <-errc; err != nil {
|
||||||
t.Fatal(err)
|
t.Fatal(err)
|
||||||
}
|
}
|
||||||
@ -216,6 +198,27 @@ func TestTransportMultiplexAcceptMultiple(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func testDialer(dialAddr NetAddress, errc chan error) {
|
||||||
|
var (
|
||||||
|
pv = ed25519.GenPrivKey()
|
||||||
|
dialer = newMultiplexTransport(
|
||||||
|
testNodeInfo(PubKeyToID(pv.PubKey()), defaultNodeName),
|
||||||
|
NodeKey{
|
||||||
|
PrivKey: pv,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
_, err := dialer.Dial(dialAddr, peerConfig{})
|
||||||
|
if err != nil {
|
||||||
|
errc <- err
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
// Signal that the connection was established.
|
||||||
|
errc <- nil
|
||||||
|
}
|
||||||
|
|
||||||
func TestTransportMultiplexAcceptNonBlocking(t *testing.T) {
|
func TestTransportMultiplexAcceptNonBlocking(t *testing.T) {
|
||||||
mt := testSetupMultiplexTransport(t)
|
mt := testSetupMultiplexTransport(t)
|
||||||
|
|
||||||
@ -591,6 +594,7 @@ func TestTransportHandshake(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// create listener
|
||||||
func testSetupMultiplexTransport(t *testing.T) *MultiplexTransport {
|
func testSetupMultiplexTransport(t *testing.T) *MultiplexTransport {
|
||||||
var (
|
var (
|
||||||
pv = ed25519.GenPrivKey()
|
pv = ed25519.GenPrivKey()
|
||||||
|
@ -218,7 +218,7 @@ func DumpConsensusState(ctx *rpctypes.Context) (*ctypes.ResultDumpConsensusState
|
|||||||
}
|
}
|
||||||
peerStates[i] = ctypes.PeerStateInfo{
|
peerStates[i] = ctypes.PeerStateInfo{
|
||||||
// Peer basic info.
|
// Peer basic info.
|
||||||
NodeAddress: peer.NodeInfo().NetAddress().String(),
|
NodeAddress: peer.SocketAddr().String(),
|
||||||
// Peer consensus state.
|
// Peer consensus state.
|
||||||
PeerState: peerStateJSON,
|
PeerState: peerStateJSON,
|
||||||
}
|
}
|
||||||
|
Loading…
x
Reference in New Issue
Block a user