text: #!/usr/bin/env python3
  # Copyright (c) 2010 ArtForz -- public domain half-a-node
  # Copyright (c) 2012 Jeff Garzik
  # Copyright (c) 2010-2017 The Bitcoin Core developers
  # Distributed under the MIT software license, see the accompanying
  # file COPYING or http://www.opensource.org/licenses/mit-license.php.
  """Genesis P2P network half-a-node.
  
  This python code was modified from ArtForz' public domain  half-a-node, as
  found in the mini-node branch of http://github.com/jgarzik/pynode.
  
  P2PConnection: A low-level connection object to a node's P2P interface
  P2PInterface: A high-level interface object for communicating to a node over P2P"""
  import asyncore
  from collections import defaultdict
  from io import BytesIO
  import logging
  import socket
  import struct
  import sys
  import threading
  
  from test_framework.messages import *
  from test_framework.util import wait_until
  
  logger = logging.getLogger("TestFramework.mininode")
  
  MESSAGEMAP = {
      b"addr": msg_addr,
      b"block": msg_block,
      b"blocktxn": msg_blocktxn,
      b"cmpctblock": msg_cmpctblock,
      b"feefilter": msg_feefilter,
      b"getaddr": msg_getaddr,
      b"getblocks": msg_getblocks,
      b"getblocktxn": msg_getblocktxn,
      b"getdata": msg_getdata,
      b"getheaders": msg_getheaders,
      b"headers": msg_headers,
      b"inv": msg_inv,
      b"mempool": msg_mempool,
      b"ping": msg_ping,
      b"pong": msg_pong,
      b"reject": msg_reject,
      b"sendcmpct": msg_sendcmpct,
      b"sendheaders": msg_sendheaders,
      b"tx": msg_tx,
      b"verack": msg_verack,
      b"version": msg_version,
  }
  
  MAGIC_BYTES = {
      "mainnet": b"\xf9\xbe\xb4\xd9",   # mainnet
      "testnet3": b"\x0b\x11\x09\x07",  # testnet3
      "regtest": b"\xfa\xbf\xb5\xda",   # regtest
  }
  
  class P2PConnection(asyncore.dispatcher):
      """A low-level connection object to a node's P2P interface.
  
      This class is responsible for:
  
      - opening and closing the TCP connection to the node
      - reading bytes from and writing bytes to the socket
      - deserializing and serializing the P2P message header
      - logging messages as they are sent and received
  
      This class contains no logic for handing the P2P message payloads. It must be
      sub-classed and the on_message() callback overridden."""
  
      def __init__(self):
          # All P2PConnections must be created before starting the NetworkThread.
          # assert that the network thread is not running.
          assert not network_thread_running()
  
          super().__init__(map=mininode_socket_map)
  
      def peer_connect(self, dstaddr, dstport, net="regtest"):
          self.dstaddr = dstaddr
          self.dstport = dstport
          self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
          self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
          self.sendbuf = b""
          self.recvbuf = b""
          self.state = "connecting"
          self.network = net
          self.disconnect = False
  
          logger.info('Connecting to Genesis Node: %s:%d' % (self.dstaddr, self.dstport))
  
          try:
              self.connect((dstaddr, dstport))
          except:
              self.handle_close()
  
      def peer_disconnect(self):
          # Connection could have already been closed by other end.
          if self.state == "connected":
              self.disconnect_node()
  
      # Connection and disconnection methods
  
      def handle_connect(self):
          """asyncore callback when a connection is opened."""
          if self.state != "connected":
              logger.debug("Connected & Listening: %s:%d" % (self.dstaddr, self.dstport))
              self.state = "connected"
              self.on_open()
  
      def handle_close(self):
          """asyncore callback when a connection is closed."""
          logger.debug("Closing connection to: %s:%d" % (self.dstaddr, self.dstport))
          self.state = "closed"
          self.recvbuf = b""
          self.sendbuf = b""
          try:
              self.close()
          except:
              pass
          self.on_close()
  
      def disconnect_node(self):
          """Disconnect the p2p connection.
  
          Called by the test logic thread. Causes the p2p connection
          to be disconnected on the next iteration of the asyncore loop."""
          self.disconnect = True
  
      # Socket read methods
  
      def handle_read(self):
          """asyncore callback when data is read from the socket."""
          t = self.recv(8192)
          if len(t) > 0:
              self.recvbuf += t
              self._on_data()
  
      def _on_data(self):
          """Try to read P2P messages from the recv buffer.
  
          This method reads data from the buffer in a loop. It deserializes,
          parses and verifies the P2P header, then passes the P2P payload to
          the on_message callback for processing."""
          try:
              while True:
                  if len(self.recvbuf) < 4:
                      return
                  if self.recvbuf[:4] != MAGIC_BYTES[self.network]:
                      raise ValueError("got garbage %s" % repr(self.recvbuf))
                  if len(self.recvbuf) < 4 + 12 + 4 + 4:
                      return
                  command = self.recvbuf[4:4+12].split(b"\x00", 1)[0]
                  msglen = struct.unpack("<i", self.recvbuf[4+12:4+12+4])[0]
                  checksum = self.recvbuf[4+12+4:4+12+4+4]
                  if len(self.recvbuf) < 4 + 12 + 4 + 4 + msglen:
                      return
                  msg = self.recvbuf[4+12+4+4:4+12+4+4+msglen]
                  th = sha256(msg)
                  h = sha256(th)
                  if checksum != h[:4]:
                      raise ValueError("got bad checksum " + repr(self.recvbuf))
                  self.recvbuf = self.recvbuf[4+12+4+4+msglen:]
                  if command not in MESSAGEMAP:
                      raise ValueError("Received unknown command from %s:%d: '%s' %s" % (self.dstaddr, self.dstport, command, repr(msg)))
                  f = BytesIO(msg)
                  t = MESSAGEMAP[command]()
                  t.deserialize(f)
                  self._log_message("receive", t)
                  self.on_message(t)
          except Exception as e:
              logger.exception('Error reading message:', repr(e))
              raise
  
      def on_message(self, message):
          """Callback for processing a P2P payload. Must be overridden by derived class."""
          raise NotImplementedError
  
      # Socket write methods
  
      def writable(self):
          """asyncore method to determine whether the handle_write() callback should be called on the next loop."""
          with mininode_lock:
              pre_connection = self.state == "connecting"
              length = len(self.sendbuf)
          return (length > 0 or pre_connection)
  
      def handle_write(self):
          """asyncore callback when data should be written to the socket."""
          with mininode_lock:
              # asyncore does not expose socket connection, only the first read/write
              # event, thus we must check connection manually here to know when we
              # actually connect
              if self.state == "connecting":
                  self.handle_connect()
              if not self.writable():
                  return
  
              try:
                  sent = self.send(self.sendbuf)
              except:
                  self.handle_close()
                  return
              self.sendbuf = self.sendbuf[sent:]
  
      def send_message(self, message, pushbuf=False):
          """Send a P2P message over the socket.
  
          This method takes a P2P payload, builds the P2P header and adds
          the message to the send buffer to be sent over the socket."""
          if self.state != "connected" and not pushbuf:
              raise IOError('Not connected, no pushbuf')
          self._log_message("send", message)
          command = message.command
          data = message.serialize()
          tmsg = MAGIC_BYTES[self.network]
          tmsg += command
          tmsg += b"\x00" * (12 - len(command))
          tmsg += struct.pack("<I", len(data))
          th = sha256(data)
          h = sha256(th)
          tmsg += h[:4]
          tmsg += data
          with mininode_lock:
              if (len(self.sendbuf) == 0 and not pushbuf):
                  try:
                      sent = self.send(tmsg)
                      self.sendbuf = tmsg[sent:]
                  except BlockingIOError:
                      self.sendbuf = tmsg
              else:
                  self.sendbuf += tmsg
  
      # Class utility methods
  
      def _log_message(self, direction, msg):
          """Logs a message being sent or received over the connection."""
          if direction == "send":
              log_message = "Send message to "
          elif direction == "receive":
              log_message = "Received message from "
          log_message += "%s:%d: %s" % (self.dstaddr, self.dstport, repr(msg)[:500])
          if len(log_message) > 500:
              log_message += "... (msg truncated)"
          logger.debug(log_message)
  
  
  class P2PInterface(P2PConnection):
      """A high-level P2P interface class for communicating with a Genesis node.
  
      This class provides high-level callbacks for processing P2P message
      payloads, as well as convenience methods for interacting with the
      node over P2P.
  
      Individual testcases should subclass this and override the on_* methods
      if they want to alter message handling behaviour."""
      def __init__(self):
          super().__init__()
  
          # Track number of messages of each type received and the most recent
          # message of each type
          self.message_count = defaultdict(int)
          self.last_message = {}
  
          # A count of the number of ping messages we've sent to the node
          self.ping_counter = 1
  
          # The network services received from the peer
          self.nServices = 0
  
      def peer_connect(self, *args, services=NODE_NETWORK|NODE_WITNESS, send_version=True, **kwargs):
          super().peer_connect(*args, **kwargs)
  
          if send_version:
              # Send a version msg
              vt = msg_version()
              vt.nServices = services
              vt.addrTo.ip = self.dstaddr
              vt.addrTo.port = self.dstport
              vt.addrFrom.ip = "0.0.0.0"
              vt.addrFrom.port = 0
              self.send_message(vt, True)
  
      # Message receiving methods
  
      def on_message(self, message):
          """Receive message and dispatch message to appropriate callback.
  
          We keep a count of how many of each message type has been received
          and the most recent message of each type."""
          with mininode_lock:
              try:
                  command = message.command.decode('ascii')
                  self.message_count[command] += 1
                  self.last_message[command] = message
                  getattr(self, 'on_' + command)(message)
              except:
                  print("ERROR delivering %s (%s)" % (repr(message), sys.exc_info()[0]))
                  raise
  
      # Callback methods. Can be overridden by subclasses in individual test
      # cases to provide custom message handling behaviour.
  
      def on_open(self):
          pass
  
      def on_close(self):
          pass
  
      def on_addr(self, message): pass
      def on_block(self, message): pass
      def on_blocktxn(self, message): pass
      def on_cmpctblock(self, message): pass
      def on_feefilter(self, message): pass
      def on_getaddr(self, message): pass
      def on_getblocks(self, message): pass
      def on_getblocktxn(self, message): pass
      def on_getdata(self, message): pass
      def on_getheaders(self, message): pass
      def on_headers(self, message): pass
      def on_mempool(self, message): pass
      def on_pong(self, message): pass
      def on_reject(self, message): pass
      def on_sendcmpct(self, message): pass
      def on_sendheaders(self, message): pass
      def on_tx(self, message): pass
  
      def on_inv(self, message):
          want = msg_getdata()
          for i in message.inv:
              if i.type != 0:
                  want.inv.append(i)
          if len(want.inv):
              self.send_message(want)
  
      def on_ping(self, message):
          self.send_message(msg_pong(message.nonce))
  
      def on_verack(self, message):
          self.verack_received = True
  
      def on_version(self, message):
          assert message.nVersion >= MIN_VERSION_SUPPORTED, "Version {} received. Test framework only supports versions greater than {}".format(message.nVersion, MIN_VERSION_SUPPORTED)
          self.send_message(msg_verack())
          self.nServices = message.nServices
  
      # Connection helper methods
  
      def wait_for_disconnect(self, timeout=60):
          test_function = lambda: self.state != "connected"
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      # Message receiving helper methods
  
      def wait_for_block(self, blockhash, timeout=60):
          test_function = lambda: self.last_message.get("block") and self.last_message["block"].block.rehash() == blockhash
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      def wait_for_getdata(self, timeout=60):
          test_function = lambda: self.last_message.get("getdata")
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      def wait_for_getheaders(self, timeout=60):
          test_function = lambda: self.last_message.get("getheaders")
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      def wait_for_inv(self, expected_inv, timeout=60):
          """Waits for an INV message and checks that the first inv object in the message was as expected."""
          if len(expected_inv) > 1:
              raise NotImplementedError("wait_for_inv() will only verify the first inv object")
          test_function = lambda: self.last_message.get("inv") and \
                                  self.last_message["inv"].inv[0].type == expected_inv[0].type and \
                                  self.last_message["inv"].inv[0].hash == expected_inv[0].hash
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      def wait_for_verack(self, timeout=60):
          test_function = lambda: self.message_count["verack"]
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
  
      # Message sending helper functions
  
      def send_and_ping(self, message):
          self.send_message(message)
          self.sync_with_ping()
  
      # Sync up with the node
      def sync_with_ping(self, timeout=60):
          self.send_message(msg_ping(nonce=self.ping_counter))
          test_function = lambda: self.last_message.get("pong") and self.last_message["pong"].nonce == self.ping_counter
          wait_until(test_function, timeout=timeout, lock=mininode_lock)
          self.ping_counter += 1
  
  
  # Keep our own socket map for asyncore, so that we can track disconnects
  # ourselves (to workaround an issue with closing an asyncore socket when
  # using select)
  mininode_socket_map = dict()
  
  # One lock for synchronizing all data access between the networking thread (see
  # NetworkThread below) and the thread running the test logic.  For simplicity,
  # P2PConnection acquires this lock whenever delivering a message to a P2PInterface,
  # and whenever adding anything to the send buffer (in send_message()).  This
  # lock should be acquired in the thread running the test logic to synchronize
  # access to any data shared with the P2PInterface or P2PConnection.
  mininode_lock = threading.RLock()
  
  class NetworkThread(threading.Thread):
      def __init__(self):
          super().__init__(name="NetworkThread")
  
      def run(self):
          while mininode_socket_map:
              # We check for whether to disconnect outside of the asyncore
              # loop to workaround the behavior of asyncore when using
              # select
              disconnected = []
              for fd, obj in mininode_socket_map.items():
                  if obj.disconnect:
                      disconnected.append(obj)
              [obj.handle_close() for obj in disconnected]
              asyncore.loop(0.1, use_poll=True, map=mininode_socket_map, count=1)
          logger.debug("Network thread closing")
  
  def network_thread_start():
      """Start the network thread."""
      # Only one network thread may run at a time
      assert not network_thread_running()
  
      NetworkThread().start()
  
  def network_thread_running():
      """Return whether the network thread is running."""
      return any([thread.name == "NetworkThread" for thread in threading.enumerate()])
  
  def network_thread_join(timeout=10):
      """Wait timeout seconds for the network thread to terminate.
  
      Throw if the network thread doesn't terminate in timeout seconds."""
      network_threads = [thread for thread in threading.enumerate() if thread.name == "NetworkThread"]
      assert len(network_threads) <= 1
      for thread in network_threads:
          thread.join(timeout)
          assert not thread.is_alive()
  