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.

297 lines
8.5 KiB

  1. package consensus
  2. import (
  3. "sync"
  4. "testing"
  5. "time"
  6. "github.com/tendermint/tendermint/config/tendermint_test"
  7. . "github.com/tendermint/go-common"
  8. cfg "github.com/tendermint/go-config"
  9. "github.com/tendermint/go-events"
  10. "github.com/tendermint/go-p2p"
  11. "github.com/tendermint/tendermint/types"
  12. )
  13. func init() {
  14. config = tendermint_test.ResetConfig("consensus_byzantine_test")
  15. }
  16. //----------------------------------------------
  17. // byzantine failures
  18. // 4 validators. 1 is byzantine. The other three are partitioned into A (1 val) and B (2 vals).
  19. // byzantine validator sends conflicting proposals into A and B,
  20. // and prevotes/precommits on both of them.
  21. // B sees a commit, A doesn't.
  22. // Byzantine validator refuses to prevote.
  23. // Heal partition and ensure A sees the commit
  24. func TestByzantine(t *testing.T) {
  25. N := 4
  26. css := randConsensusNet(N, "consensus_byzantine_test", newMockTickerFunc(false), newCounter)
  27. // give the byzantine validator a normal ticker
  28. css[0].SetTimeoutTicker(NewTimeoutTicker())
  29. switches := make([]*p2p.Switch, N)
  30. for i := 0; i < N; i++ {
  31. switches[i] = p2p.NewSwitch(cfg.NewMapConfig(nil))
  32. }
  33. reactors := make([]p2p.Reactor, N)
  34. defer func() {
  35. for _, r := range reactors {
  36. if rr, ok := r.(*ByzantineReactor); ok {
  37. rr.reactor.Switch.Stop()
  38. } else {
  39. r.(*ConsensusReactor).Switch.Stop()
  40. }
  41. }
  42. }()
  43. eventChans := make([]chan interface{}, N)
  44. for i := 0; i < N; i++ {
  45. if i == 0 {
  46. css[i].privValidator = NewByzantinePrivValidator(css[i].privValidator.(*types.PrivValidator))
  47. // make byzantine
  48. css[i].decideProposal = func(j int) func(int, int) {
  49. return func(height, round int) {
  50. byzantineDecideProposalFunc(height, round, css[j], switches[j])
  51. }
  52. }(i)
  53. css[i].doPrevote = func(height, round int) {}
  54. }
  55. eventSwitch := events.NewEventSwitch()
  56. _, err := eventSwitch.Start()
  57. if err != nil {
  58. t.Fatalf("Failed to start switch: %v", err)
  59. }
  60. eventChans[i] = subscribeToEvent(eventSwitch, "tester", types.EventStringNewBlock(), 1)
  61. conR := NewConsensusReactor(css[i], true) // so we dont start the consensus states
  62. conR.SetEventSwitch(eventSwitch)
  63. var conRI p2p.Reactor
  64. conRI = conR
  65. if i == 0 {
  66. conRI = NewByzantineReactor(conR)
  67. }
  68. reactors[i] = conRI
  69. }
  70. p2p.MakeConnectedSwitches(N, func(i int, s *p2p.Switch) *p2p.Switch {
  71. // ignore new switch s, we already made ours
  72. switches[i].AddReactor("CONSENSUS", reactors[i])
  73. return switches[i]
  74. }, func(sws []*p2p.Switch, i, j int) {
  75. // the network starts partitioned with globally active adversary
  76. if i != 0 {
  77. return
  78. }
  79. p2p.Connect2Switches(sws, i, j)
  80. })
  81. // start the state machines
  82. byzR := reactors[0].(*ByzantineReactor)
  83. s := byzR.reactor.conS.GetState()
  84. byzR.reactor.SwitchToConsensus(s)
  85. for i := 1; i < N; i++ {
  86. cr := reactors[i].(*ConsensusReactor)
  87. cr.SwitchToConsensus(cr.conS.GetState())
  88. }
  89. // byz proposer sends one block to peers[0]
  90. // and the other block to peers[1] and peers[2].
  91. // note peers and switches order don't match.
  92. peers := switches[0].Peers().List()
  93. ind0 := getSwitchIndex(switches, peers[0])
  94. ind1 := getSwitchIndex(switches, peers[1])
  95. ind2 := getSwitchIndex(switches, peers[2])
  96. // connect the 2 peers in the larger partition
  97. p2p.Connect2Switches(switches, ind1, ind2)
  98. // wait for someone in the big partition to make a block
  99. select {
  100. case <-eventChans[ind2]:
  101. }
  102. log.Notice("A block has been committed. Healing partition")
  103. // connect the partitions
  104. p2p.Connect2Switches(switches, ind0, ind1)
  105. p2p.Connect2Switches(switches, ind0, ind2)
  106. // wait till everyone makes the first new block
  107. // (one of them already has)
  108. wg := new(sync.WaitGroup)
  109. wg.Add(2)
  110. for i := 1; i < N-1; i++ {
  111. go func(j int) {
  112. <-eventChans[j]
  113. wg.Done()
  114. }(i)
  115. }
  116. done := make(chan struct{})
  117. go func() {
  118. wg.Wait()
  119. close(done)
  120. }()
  121. tick := time.NewTicker(time.Second * 10)
  122. select {
  123. case <-done:
  124. case <-tick.C:
  125. for i, reactor := range reactors {
  126. t.Log(Fmt("Consensus Reactor %v", i))
  127. t.Log(Fmt("%v", reactor))
  128. }
  129. t.Fatalf("Timed out waiting for all validators to commit first block")
  130. }
  131. }
  132. //-------------------------------
  133. // byzantine consensus functions
  134. func byzantineDecideProposalFunc(height, round int, cs *ConsensusState, sw *p2p.Switch) {
  135. // byzantine user should create two proposals and try to split the vote.
  136. // Avoid sending on internalMsgQueue and running consensus state.
  137. // Create a new proposal block from state/txs from the mempool.
  138. block1, blockParts1 := cs.createProposalBlock()
  139. polRound, polBlockID := cs.Votes.POLInfo()
  140. proposal1 := types.NewProposal(height, round, blockParts1.Header(), polRound, polBlockID)
  141. cs.privValidator.SignProposal(cs.state.ChainID, proposal1) // byzantine doesnt err
  142. // Create a new proposal block from state/txs from the mempool.
  143. block2, blockParts2 := cs.createProposalBlock()
  144. polRound, polBlockID = cs.Votes.POLInfo()
  145. proposal2 := types.NewProposal(height, round, blockParts2.Header(), polRound, polBlockID)
  146. cs.privValidator.SignProposal(cs.state.ChainID, proposal2) // byzantine doesnt err
  147. block1Hash := block1.Hash()
  148. block2Hash := block2.Hash()
  149. // broadcast conflicting proposals/block parts to peers
  150. peers := sw.Peers().List()
  151. log.Notice("Byzantine: broadcasting conflicting proposals", "peers", len(peers))
  152. for i, peer := range peers {
  153. if i < len(peers)/2 {
  154. go sendProposalAndParts(height, round, cs, peer, proposal1, block1Hash, blockParts1)
  155. } else {
  156. go sendProposalAndParts(height, round, cs, peer, proposal2, block2Hash, blockParts2)
  157. }
  158. }
  159. }
  160. func sendProposalAndParts(height, round int, cs *ConsensusState, peer *p2p.Peer, proposal *types.Proposal, blockHash []byte, parts *types.PartSet) {
  161. // proposal
  162. msg := &ProposalMessage{Proposal: proposal}
  163. peer.Send(DataChannel, struct{ ConsensusMessage }{msg})
  164. // parts
  165. for i := 0; i < parts.Total(); i++ {
  166. part := parts.GetPart(i)
  167. msg := &BlockPartMessage{
  168. Height: height, // This tells peer that this part applies to us.
  169. Round: round, // This tells peer that this part applies to us.
  170. Part: part,
  171. }
  172. peer.Send(DataChannel, struct{ ConsensusMessage }{msg})
  173. }
  174. // votes
  175. cs.mtx.Lock()
  176. prevote, _ := cs.signVote(types.VoteTypePrevote, blockHash, parts.Header())
  177. precommit, _ := cs.signVote(types.VoteTypePrecommit, blockHash, parts.Header())
  178. cs.mtx.Unlock()
  179. peer.Send(VoteChannel, struct{ ConsensusMessage }{&VoteMessage{prevote}})
  180. peer.Send(VoteChannel, struct{ ConsensusMessage }{&VoteMessage{precommit}})
  181. }
  182. //----------------------------------------
  183. // byzantine consensus reactor
  184. type ByzantineReactor struct {
  185. Service
  186. reactor *ConsensusReactor
  187. }
  188. func NewByzantineReactor(conR *ConsensusReactor) *ByzantineReactor {
  189. return &ByzantineReactor{
  190. Service: conR,
  191. reactor: conR,
  192. }
  193. }
  194. func (br *ByzantineReactor) SetSwitch(s *p2p.Switch) { br.reactor.SetSwitch(s) }
  195. func (br *ByzantineReactor) GetChannels() []*p2p.ChannelDescriptor { return br.reactor.GetChannels() }
  196. func (br *ByzantineReactor) AddPeer(peer *p2p.Peer) {
  197. if !br.reactor.IsRunning() {
  198. return
  199. }
  200. // Create peerState for peer
  201. peerState := NewPeerState(peer)
  202. peer.Data.Set(types.PeerStateKey, peerState)
  203. // Send our state to peer.
  204. // If we're fast_syncing, broadcast a RoundStepMessage later upon SwitchToConsensus().
  205. if !br.reactor.fastSync {
  206. br.reactor.sendNewRoundStepMessage(peer)
  207. }
  208. }
  209. func (br *ByzantineReactor) RemovePeer(peer *p2p.Peer, reason interface{}) {
  210. br.reactor.RemovePeer(peer, reason)
  211. }
  212. func (br *ByzantineReactor) Receive(chID byte, peer *p2p.Peer, msgBytes []byte) {
  213. br.reactor.Receive(chID, peer, msgBytes)
  214. }
  215. //----------------------------------------
  216. // byzantine privValidator
  217. type ByzantinePrivValidator struct {
  218. Address []byte `json:"address"`
  219. types.Signer `json:"-"`
  220. mtx sync.Mutex
  221. }
  222. // Return a priv validator that will sign anything
  223. func NewByzantinePrivValidator(pv *types.PrivValidator) *ByzantinePrivValidator {
  224. return &ByzantinePrivValidator{
  225. Address: pv.Address,
  226. Signer: pv.Signer,
  227. }
  228. }
  229. func (privVal *ByzantinePrivValidator) GetAddress() []byte {
  230. return privVal.Address
  231. }
  232. func (privVal *ByzantinePrivValidator) SignVote(chainID string, vote *types.Vote) error {
  233. privVal.mtx.Lock()
  234. defer privVal.mtx.Unlock()
  235. // Sign
  236. vote.Signature = privVal.Sign(types.SignBytes(chainID, vote))
  237. return nil
  238. }
  239. func (privVal *ByzantinePrivValidator) SignProposal(chainID string, proposal *types.Proposal) error {
  240. privVal.mtx.Lock()
  241. defer privVal.mtx.Unlock()
  242. // Sign
  243. proposal.Signature = privVal.Sign(types.SignBytes(chainID, proposal))
  244. return nil
  245. }
  246. func (privVal *ByzantinePrivValidator) String() string {
  247. return Fmt("PrivValidator{%X}", privVal.Address)
  248. }