You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

549 lines
15 KiB

  1. package pex
  2. import (
  3. "fmt"
  4. "io/ioutil"
  5. "net"
  6. "os"
  7. "path/filepath"
  8. "testing"
  9. "time"
  10. "github.com/stretchr/testify/assert"
  11. "github.com/stretchr/testify/require"
  12. crypto "github.com/tendermint/tendermint/crypto"
  13. "github.com/tendermint/tendermint/crypto/ed25519"
  14. cmn "github.com/tendermint/tendermint/libs/common"
  15. "github.com/tendermint/tendermint/libs/log"
  16. "github.com/tendermint/tendermint/config"
  17. "github.com/tendermint/tendermint/p2p"
  18. "github.com/tendermint/tendermint/p2p/conn"
  19. )
  20. var (
  21. cfg *config.P2PConfig
  22. )
  23. func init() {
  24. cfg = config.DefaultP2PConfig()
  25. cfg.PexReactor = true
  26. cfg.AllowDuplicateIP = true
  27. }
  28. func TestPEXReactorBasic(t *testing.T) {
  29. r, book := createReactor(&PEXReactorConfig{})
  30. defer teardownReactor(book)
  31. assert.NotNil(t, r)
  32. assert.NotEmpty(t, r.GetChannels())
  33. }
  34. func TestPEXReactorAddRemovePeer(t *testing.T) {
  35. r, book := createReactor(&PEXReactorConfig{})
  36. defer teardownReactor(book)
  37. size := book.Size()
  38. peer := p2p.CreateRandomPeer(false)
  39. r.AddPeer(peer)
  40. assert.Equal(t, size+1, book.Size())
  41. r.RemovePeer(peer, "peer not available")
  42. outboundPeer := p2p.CreateRandomPeer(true)
  43. r.AddPeer(outboundPeer)
  44. assert.Equal(t, size+1, book.Size(), "outbound peers should not be added to the address book")
  45. r.RemovePeer(outboundPeer, "peer not available")
  46. }
  47. // --- FAIL: TestPEXReactorRunning (11.10s)
  48. // pex_reactor_test.go:411: expected all switches to be connected to at
  49. // least one peer (switches: 0 => {outbound: 1, inbound: 0}, 1 =>
  50. // {outbound: 0, inbound: 1}, 2 => {outbound: 0, inbound: 0}, )
  51. //
  52. // EXPLANATION: peers are getting rejected because in switch#addPeer we check
  53. // if any peer (who we already connected to) has the same IP. Even though local
  54. // peers have different IP addresses, they all have the same underlying remote
  55. // IP: 127.0.0.1.
  56. //
  57. func TestPEXReactorRunning(t *testing.T) {
  58. N := 3
  59. switches := make([]*p2p.Switch, N)
  60. // directory to store address books
  61. dir, err := ioutil.TempDir("", "pex_reactor")
  62. require.Nil(t, err)
  63. defer os.RemoveAll(dir) // nolint: errcheck
  64. books := make([]*addrBook, N)
  65. logger := log.TestingLogger()
  66. // create switches
  67. for i := 0; i < N; i++ {
  68. switches[i] = p2p.MakeSwitch(cfg, i, "testing", "123.123.123", func(i int, sw *p2p.Switch) *p2p.Switch {
  69. books[i] = NewAddrBook(filepath.Join(dir, fmt.Sprintf("addrbook%d.json", i)), false)
  70. books[i].SetLogger(logger.With("pex", i))
  71. sw.SetAddrBook(books[i])
  72. sw.SetLogger(logger.With("pex", i))
  73. r := NewPEXReactor(books[i], &PEXReactorConfig{})
  74. r.SetLogger(logger.With("pex", i))
  75. r.SetEnsurePeersPeriod(250 * time.Millisecond)
  76. sw.AddReactor("pex", r)
  77. return sw
  78. })
  79. }
  80. addOtherNodeAddrToAddrBook := func(switchIndex, otherSwitchIndex int) {
  81. addr := switches[otherSwitchIndex].NodeInfo().NetAddress()
  82. books[switchIndex].AddAddress(addr, addr)
  83. }
  84. addOtherNodeAddrToAddrBook(0, 1)
  85. addOtherNodeAddrToAddrBook(1, 0)
  86. addOtherNodeAddrToAddrBook(2, 1)
  87. for _, sw := range switches {
  88. err := sw.Start() // start switch and reactors
  89. require.Nil(t, err)
  90. }
  91. assertPeersWithTimeout(t, switches, 10*time.Millisecond, 10*time.Second, N-1)
  92. // stop them
  93. for _, s := range switches {
  94. s.Stop()
  95. }
  96. }
  97. func TestPEXReactorReceive(t *testing.T) {
  98. r, book := createReactor(&PEXReactorConfig{})
  99. defer teardownReactor(book)
  100. peer := p2p.CreateRandomPeer(false)
  101. // we have to send a request to receive responses
  102. r.RequestAddrs(peer)
  103. size := book.Size()
  104. addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
  105. msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
  106. r.Receive(PexChannel, peer, msg)
  107. assert.Equal(t, size+1, book.Size())
  108. msg = cdc.MustMarshalBinaryBare(&pexRequestMessage{})
  109. r.Receive(PexChannel, peer, msg) // should not panic.
  110. }
  111. func TestPEXReactorRequestMessageAbuse(t *testing.T) {
  112. r, book := createReactor(&PEXReactorConfig{})
  113. defer teardownReactor(book)
  114. sw := createSwitchAndAddReactors(r)
  115. sw.SetAddrBook(book)
  116. peer := newMockPeer()
  117. p2p.AddPeerToSwitch(sw, peer)
  118. assert.True(t, sw.Peers().Has(peer.ID()))
  119. id := string(peer.ID())
  120. msg := cdc.MustMarshalBinaryBare(&pexRequestMessage{})
  121. // first time creates the entry
  122. r.Receive(PexChannel, peer, msg)
  123. assert.True(t, r.lastReceivedRequests.Has(id))
  124. assert.True(t, sw.Peers().Has(peer.ID()))
  125. // next time sets the last time value
  126. r.Receive(PexChannel, peer, msg)
  127. assert.True(t, r.lastReceivedRequests.Has(id))
  128. assert.True(t, sw.Peers().Has(peer.ID()))
  129. // third time is too many too soon - peer is removed
  130. r.Receive(PexChannel, peer, msg)
  131. assert.False(t, r.lastReceivedRequests.Has(id))
  132. assert.False(t, sw.Peers().Has(peer.ID()))
  133. }
  134. func TestPEXReactorAddrsMessageAbuse(t *testing.T) {
  135. r, book := createReactor(&PEXReactorConfig{})
  136. defer teardownReactor(book)
  137. sw := createSwitchAndAddReactors(r)
  138. sw.SetAddrBook(book)
  139. peer := newMockPeer()
  140. p2p.AddPeerToSwitch(sw, peer)
  141. assert.True(t, sw.Peers().Has(peer.ID()))
  142. id := string(peer.ID())
  143. // request addrs from the peer
  144. r.RequestAddrs(peer)
  145. assert.True(t, r.requestsSent.Has(id))
  146. assert.True(t, sw.Peers().Has(peer.ID()))
  147. addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
  148. msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
  149. // receive some addrs. should clear the request
  150. r.Receive(PexChannel, peer, msg)
  151. assert.False(t, r.requestsSent.Has(id))
  152. assert.True(t, sw.Peers().Has(peer.ID()))
  153. // receiving more addrs causes a disconnect
  154. r.Receive(PexChannel, peer, msg)
  155. assert.False(t, sw.Peers().Has(peer.ID()))
  156. }
  157. func TestCheckSeeds(t *testing.T) {
  158. // directory to store address books
  159. dir, err := ioutil.TempDir("", "pex_reactor")
  160. require.Nil(t, err)
  161. defer os.RemoveAll(dir) // nolint: errcheck
  162. // 1. test creating peer with no seeds works
  163. peer := testCreateDefaultPeer(dir, 0)
  164. require.Nil(t, peer.Start())
  165. peer.Stop()
  166. // 2. create seed
  167. seed := testCreateSeed(dir, 1, []*p2p.NetAddress{}, []*p2p.NetAddress{})
  168. // 3. test create peer with online seed works
  169. peer = testCreatePeerWithSeed(dir, 2, seed)
  170. require.Nil(t, peer.Start())
  171. peer.Stop()
  172. // 4. test create peer with all seeds having unresolvable DNS fails
  173. badPeerConfig := &PEXReactorConfig{
  174. Seeds: []string{"ed3dfd27bfc4af18f67a49862f04cc100696e84d@bad.network.addr:26657",
  175. "d824b13cb5d40fa1d8a614e089357c7eff31b670@anotherbad.network.addr:26657"},
  176. }
  177. peer = testCreatePeerWithConfig(dir, 2, badPeerConfig)
  178. require.Error(t, peer.Start())
  179. peer.Stop()
  180. // 5. test create peer with one good seed address succeeds
  181. badPeerConfig = &PEXReactorConfig{
  182. Seeds: []string{"ed3dfd27bfc4af18f67a49862f04cc100696e84d@bad.network.addr:26657",
  183. "d824b13cb5d40fa1d8a614e089357c7eff31b670@anotherbad.network.addr:26657",
  184. seed.NodeInfo().NetAddress().String()},
  185. }
  186. peer = testCreatePeerWithConfig(dir, 2, badPeerConfig)
  187. require.Nil(t, peer.Start())
  188. peer.Stop()
  189. }
  190. func TestPEXReactorUsesSeedsIfNeeded(t *testing.T) {
  191. // directory to store address books
  192. dir, err := ioutil.TempDir("", "pex_reactor")
  193. require.Nil(t, err)
  194. defer os.RemoveAll(dir) // nolint: errcheck
  195. // 1. create seed
  196. seed := testCreateSeed(dir, 0, []*p2p.NetAddress{}, []*p2p.NetAddress{})
  197. require.Nil(t, seed.Start())
  198. defer seed.Stop()
  199. // 2. create usual peer with only seed configured.
  200. peer := testCreatePeerWithSeed(dir, 1, seed)
  201. require.Nil(t, peer.Start())
  202. defer peer.Stop()
  203. // 3. check that the peer connects to seed immediately
  204. assertPeersWithTimeout(t, []*p2p.Switch{peer}, 10*time.Millisecond, 3*time.Second, 1)
  205. }
  206. func TestConnectionSpeedForPeerReceivedFromSeed(t *testing.T) {
  207. // directory to store address books
  208. dir, err := ioutil.TempDir("", "pex_reactor")
  209. require.Nil(t, err)
  210. defer os.RemoveAll(dir) // nolint: errcheck
  211. // 1. create peer
  212. peer := testCreateDefaultPeer(dir, 1)
  213. require.Nil(t, peer.Start())
  214. defer peer.Stop()
  215. // 2. Create seed which knows about the peer
  216. seed := testCreateSeed(dir, 2, []*p2p.NetAddress{peer.NodeInfo().NetAddress()}, []*p2p.NetAddress{peer.NodeInfo().NetAddress()})
  217. require.Nil(t, seed.Start())
  218. defer seed.Stop()
  219. // 3. create another peer with only seed configured.
  220. secondPeer := testCreatePeerWithSeed(dir, 3, seed)
  221. require.Nil(t, secondPeer.Start())
  222. defer secondPeer.Stop()
  223. // 4. check that the second peer connects to seed immediately
  224. assertPeersWithTimeout(t, []*p2p.Switch{secondPeer}, 10*time.Millisecond, 3*time.Second, 1)
  225. // 5. check that the second peer connects to the first peer immediately
  226. assertPeersWithTimeout(t, []*p2p.Switch{secondPeer}, 10*time.Millisecond, 1*time.Second, 2)
  227. }
  228. func TestPEXReactorCrawlStatus(t *testing.T) {
  229. pexR, book := createReactor(&PEXReactorConfig{SeedMode: true})
  230. defer teardownReactor(book)
  231. // Seed/Crawler mode uses data from the Switch
  232. sw := createSwitchAndAddReactors(pexR)
  233. sw.SetAddrBook(book)
  234. // Create a peer, add it to the peer set and the addrbook.
  235. peer := p2p.CreateRandomPeer(false)
  236. p2p.AddPeerToSwitch(pexR.Switch, peer)
  237. addr1 := peer.NodeInfo().NetAddress()
  238. pexR.book.AddAddress(addr1, addr1)
  239. // Add a non-connected address to the book.
  240. _, addr2 := p2p.CreateRoutableAddr()
  241. pexR.book.AddAddress(addr2, addr1)
  242. // Get some peerInfos to crawl
  243. peerInfos := pexR.getPeersToCrawl()
  244. // Make sure it has the proper number of elements
  245. assert.Equal(t, 2, len(peerInfos))
  246. // TODO: test
  247. }
  248. func TestPEXReactorDoesNotAddPrivatePeersToAddrBook(t *testing.T) {
  249. peer := p2p.CreateRandomPeer(false)
  250. pexR, book := createReactor(&PEXReactorConfig{})
  251. book.AddPrivateIDs([]string{string(peer.NodeInfo().ID)})
  252. defer teardownReactor(book)
  253. // we have to send a request to receive responses
  254. pexR.RequestAddrs(peer)
  255. size := book.Size()
  256. addrs := []*p2p.NetAddress{peer.NodeInfo().NetAddress()}
  257. msg := cdc.MustMarshalBinaryBare(&pexAddrsMessage{Addrs: addrs})
  258. pexR.Receive(PexChannel, peer, msg)
  259. assert.Equal(t, size, book.Size())
  260. pexR.AddPeer(peer)
  261. assert.Equal(t, size, book.Size())
  262. }
  263. func TestPEXReactorDialPeer(t *testing.T) {
  264. pexR, book := createReactor(&PEXReactorConfig{})
  265. defer teardownReactor(book)
  266. sw := createSwitchAndAddReactors(pexR)
  267. sw.SetAddrBook(book)
  268. peer := newMockPeer()
  269. addr := peer.NodeInfo().NetAddress()
  270. assert.Equal(t, 0, pexR.AttemptsToDial(addr))
  271. // 1st unsuccessful attempt
  272. pexR.dialPeer(addr)
  273. assert.Equal(t, 1, pexR.AttemptsToDial(addr))
  274. // 2nd unsuccessful attempt
  275. pexR.dialPeer(addr)
  276. // must be skipped because it is too early
  277. assert.Equal(t, 1, pexR.AttemptsToDial(addr))
  278. if !testing.Short() {
  279. time.Sleep(3 * time.Second)
  280. // 3rd attempt
  281. pexR.dialPeer(addr)
  282. assert.Equal(t, 2, pexR.AttemptsToDial(addr))
  283. }
  284. }
  285. type mockPeer struct {
  286. *cmn.BaseService
  287. pubKey crypto.PubKey
  288. addr *p2p.NetAddress
  289. outbound, persistent bool
  290. }
  291. func newMockPeer() mockPeer {
  292. _, netAddr := p2p.CreateRoutableAddr()
  293. mp := mockPeer{
  294. addr: netAddr,
  295. pubKey: ed25519.GenPrivKey().PubKey(),
  296. }
  297. mp.BaseService = cmn.NewBaseService(nil, "MockPeer", mp)
  298. mp.Start()
  299. return mp
  300. }
  301. func (mp mockPeer) ID() p2p.ID { return mp.addr.ID }
  302. func (mp mockPeer) IsOutbound() bool { return mp.outbound }
  303. func (mp mockPeer) IsPersistent() bool { return mp.persistent }
  304. func (mp mockPeer) NodeInfo() p2p.NodeInfo {
  305. return p2p.NodeInfo{
  306. ID: mp.addr.ID,
  307. ListenAddr: mp.addr.DialString(),
  308. }
  309. }
  310. func (mockPeer) RemoteIP() net.IP { return net.ParseIP("127.0.0.1") }
  311. func (mockPeer) Status() conn.ConnectionStatus { return conn.ConnectionStatus{} }
  312. func (mockPeer) Send(byte, []byte) bool { return false }
  313. func (mockPeer) TrySend(byte, []byte) bool { return false }
  314. func (mockPeer) Set(string, interface{}) {}
  315. func (mockPeer) Get(string) interface{} { return nil }
  316. func (mockPeer) OriginalAddr() *p2p.NetAddress { return nil }
  317. func assertPeersWithTimeout(
  318. t *testing.T,
  319. switches []*p2p.Switch,
  320. checkPeriod, timeout time.Duration,
  321. nPeers int,
  322. ) {
  323. var (
  324. ticker = time.NewTicker(checkPeriod)
  325. remaining = timeout
  326. )
  327. for {
  328. select {
  329. case <-ticker.C:
  330. // check peers are connected
  331. allGood := true
  332. for _, s := range switches {
  333. outbound, inbound, _ := s.NumPeers()
  334. if outbound+inbound < nPeers {
  335. allGood = false
  336. break
  337. }
  338. }
  339. remaining -= checkPeriod
  340. if remaining < 0 {
  341. remaining = 0
  342. }
  343. if allGood {
  344. return
  345. }
  346. case <-time.After(remaining):
  347. numPeersStr := ""
  348. for i, s := range switches {
  349. outbound, inbound, _ := s.NumPeers()
  350. numPeersStr += fmt.Sprintf("%d => {outbound: %d, inbound: %d}, ", i, outbound, inbound)
  351. }
  352. t.Errorf(
  353. "expected all switches to be connected to at least %d peer(s) (switches: %s)",
  354. nPeers, numPeersStr,
  355. )
  356. return
  357. }
  358. }
  359. }
  360. // Creates a peer with the provided config
  361. func testCreatePeerWithConfig(dir string, id int, config *PEXReactorConfig) *p2p.Switch {
  362. peer := p2p.MakeSwitch(
  363. cfg,
  364. id,
  365. "127.0.0.1",
  366. "123.123.123",
  367. func(i int, sw *p2p.Switch) *p2p.Switch {
  368. book := NewAddrBook(filepath.Join(dir, fmt.Sprintf("addrbook%d.json", id)), false)
  369. book.SetLogger(log.TestingLogger())
  370. sw.SetAddrBook(book)
  371. sw.SetLogger(log.TestingLogger())
  372. r := NewPEXReactor(
  373. book,
  374. config,
  375. )
  376. r.SetLogger(log.TestingLogger())
  377. sw.AddReactor("pex", r)
  378. return sw
  379. },
  380. )
  381. return peer
  382. }
  383. // Creates a peer with the default config
  384. func testCreateDefaultPeer(dir string, id int) *p2p.Switch {
  385. return testCreatePeerWithConfig(dir, id, &PEXReactorConfig{})
  386. }
  387. // Creates a seed which knows about the provided addresses / source address pairs.
  388. // Starting and stopping the seed is left to the caller
  389. func testCreateSeed(dir string, id int, knownAddrs, srcAddrs []*p2p.NetAddress) *p2p.Switch {
  390. seed := p2p.MakeSwitch(
  391. cfg,
  392. id,
  393. "127.0.0.1",
  394. "123.123.123",
  395. func(i int, sw *p2p.Switch) *p2p.Switch {
  396. book := NewAddrBook(filepath.Join(dir, "addrbookSeed.json"), false)
  397. book.SetLogger(log.TestingLogger())
  398. for j := 0; j < len(knownAddrs); j++ {
  399. book.AddAddress(knownAddrs[j], srcAddrs[j])
  400. book.MarkGood(knownAddrs[j])
  401. }
  402. sw.SetAddrBook(book)
  403. sw.SetLogger(log.TestingLogger())
  404. r := NewPEXReactor(book, &PEXReactorConfig{})
  405. r.SetLogger(log.TestingLogger())
  406. sw.AddReactor("pex", r)
  407. return sw
  408. },
  409. )
  410. return seed
  411. }
  412. // Creates a peer which knows about the provided seed.
  413. // Starting and stopping the peer is left to the caller
  414. func testCreatePeerWithSeed(dir string, id int, seed *p2p.Switch) *p2p.Switch {
  415. conf := &PEXReactorConfig{
  416. Seeds: []string{seed.NodeInfo().NetAddress().String()},
  417. }
  418. return testCreatePeerWithConfig(dir, id, conf)
  419. }
  420. func createReactor(conf *PEXReactorConfig) (r *PEXReactor, book *addrBook) {
  421. // directory to store address book
  422. dir, err := ioutil.TempDir("", "pex_reactor")
  423. if err != nil {
  424. panic(err)
  425. }
  426. book = NewAddrBook(filepath.Join(dir, "addrbook.json"), true)
  427. book.SetLogger(log.TestingLogger())
  428. r = NewPEXReactor(book, conf)
  429. r.SetLogger(log.TestingLogger())
  430. return
  431. }
  432. func teardownReactor(book *addrBook) {
  433. err := os.RemoveAll(filepath.Dir(book.FilePath()))
  434. if err != nil {
  435. panic(err)
  436. }
  437. }
  438. func createSwitchAndAddReactors(reactors ...p2p.Reactor) *p2p.Switch {
  439. sw := p2p.MakeSwitch(cfg, 0, "127.0.0.1", "123.123.123", func(i int, sw *p2p.Switch) *p2p.Switch { return sw })
  440. sw.SetLogger(log.TestingLogger())
  441. for _, r := range reactors {
  442. sw.AddReactor(r.String(), r)
  443. r.SetSwitch(sw)
  444. }
  445. return sw
  446. }