Python p2p testing framework

mininode.py provides a framework for connecting to a bitcoin node over the p2p
network. NodeConn is the main object that manages connectivity to a node and
provides callbacks; the interface for those callbacks is defined by NodeConnCB.
Defined also are all data structures from bitcoin core that pass on the network
(CBlock, CTransaction, etc), along with de-/serialization functions.

maxblocksinflight.py is an example test using this framework that tests whether
a node is limiting the maximum number of in-flight block requests.

This also adds support to util.py for specifying the binary to use when
starting nodes (for tests that compare the behavior of different bitcoind
versions), and adds maxblocksinflight.py to the pull tester.

3 files changed, 1353 insertions, 4 deletions

diff --git a/qa/rpc-tests/maxblocksinflight.py b/qa/rpc-tests/maxblocksinflight.py
new file mode 100755
index 0000000000..4bfe203990
--- /dev/null
+++ b/qa/rpc-tests/maxblocksinflight.py
@@ -0,0 +1,99 @@
+#!/usr/bin/env python2
+#
+# Distributed under the MIT/X11 software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+#
+
+from mininode import *
+from test_framework import BitcoinTestFramework
+from util import *
+import logging
+
+'''
+In this test we connect to one node over p2p, send it numerous inv's, and
+compare the resulting number of getdata requests to a max allowed value.  We
+test for exceeding 128 blocks in flight, which was the limit an 0.9 client will
+reach. [0.10 clients shouldn't request more than 16 from a single peer.]
+'''
+MAX_REQUESTS = 128
+
+class TestManager(NodeConnCB):
+    # set up NodeConnCB callbacks, overriding base class
+    def on_getdata(self, conn, message):
+        self.log.debug("got getdata %s" % repr(message))
+        # Log the requests
+        for inv in message.inv:
+            if inv.hash not in self.blockReqCounts:
+                self.blockReqCounts[inv.hash] = 0
+            self.blockReqCounts[inv.hash] += 1
+
+    def on_close(self, conn):
+        if not self.disconnectOkay:
+            raise EarlyDisconnectError(0)
+
+    def __init__(self):
+        NodeConnCB.__init__(self)
+        self.log = logging.getLogger("BlockRelayTest")
+        self.create_callback_map()
+
+    def add_new_connection(self, connection):
+        self.connection = connection
+        self.blockReqCounts = {}
+        self.disconnectOkay = False
+
+    def run(self):
+        try:
+            fail = False
+            self.connection.rpc.generate(1) # Leave IBD
+
+            numBlocksToGenerate = [ 8, 16, 128, 1024 ]
+            for count in range(len(numBlocksToGenerate)):
+                current_invs = []
+                for i in range(numBlocksToGenerate[count]):
+                    current_invs.append(CInv(2, random.randrange(0, 1<<256)))
+                    if len(current_invs) >= 50000:
+                        self.connection.send_message(msg_inv(current_invs))
+                        current_invs = []
+                if len(current_invs) > 0:
+                    self.connection.send_message(msg_inv(current_invs))
+                
+                # Wait and see how many blocks were requested
+                time.sleep(2)
+
+                total_requests = 0
+                for key in self.blockReqCounts:
+                    total_requests += self.blockReqCounts[key]
+                    if self.blockReqCounts[key] > 1:
+                        raise AssertionError("Error, test failed: block %064x requested more than once" % key)
+                if total_requests > MAX_REQUESTS:
+                    raise AssertionError("Error, too many blocks (%d) requested" % total_requests)
+                print "Round %d: success (total requests: %d)" % (count, total_requests)
+        except AssertionError as e:
+            print "TEST FAILED: ", e.args
+
+        self.disconnectOkay = True
+        self.connection.disconnect_node()
+
+        
+class MaxBlocksInFlightTest(BitcoinTestFramework):
+    def add_options(self, parser):
+        parser.add_option("--testbinary", dest="testbinary", default="bitcoind",
+                          help="Binary to test max block requests behavior")
+
+    def setup_chain(self):
+        print "Initializing test directory "+self.options.tmpdir
+        initialize_chain_clean(self.options.tmpdir, 1)
+
+    def setup_network(self):
+        self.nodes = start_nodes(1, self.options.tmpdir, 
+                                 extra_args=[['-debug', '-whitelist=127.0.0.1']],
+                                 binary=[self.options.testbinary])
+
+    def run_test(self):
+        test = TestManager()
+        test.add_new_connection(NodeConn('127.0.0.1', p2p_port(0), self.nodes[0], test))
+        NetworkThread().start()  # Start up network handling in another thread
+        test.run()
+
+if __name__ == '__main__':
+    MaxBlocksInFlightTest().main()
diff --git a/qa/rpc-tests/mininode.py b/qa/rpc-tests/mininode.py
new file mode 100755
index 0000000000..c5c1bcfbbe
--- /dev/null
+++ b/qa/rpc-tests/mininode.py
@@ -0,0 +1,1247 @@
+# mininode.py - Bitcoin P2P network half-a-node
+#
+# Distributed under the MIT/X11 software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+#
+# 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.
+#
+# NodeConn: an object which manages p2p connectivity to a bitcoin node
+# NodeConnCB: a base class that describes the interface for receiving
+#             callbacks with network messages from a NodeConn
+# CBlock, CTransaction, CBlockHeader, CTxIn, CTxOut, etc....:
+#     data structures that should map to corresponding structures in
+#     bitcoin/primitives
+# msg_block, msg_tx, msg_headers, etc.:
+#     data structures that represent network messages
+# ser_*, deser_*: functions that handle serialization/deserialization
+
+
+import struct
+import socket
+import asyncore
+import binascii
+import time
+import sys
+import random
+import cStringIO
+import hashlib
+from threading import Lock
+from threading import Thread
+import logging
+import copy
+
+BIP0031_VERSION = 60000
+MY_VERSION = 60001  # past bip-31 for ping/pong
+MY_SUBVERSION = "/python-mininode-tester:0.0.1/"
+
+MAX_INV_SZ = 50000
+
+# Serialization/deserialization tools
+def sha256(s):
+    return hashlib.new('sha256', s).digest()
+
+
+def hash256(s):
+    return sha256(sha256(s))
+
+
+def deser_string(f):
+    nit = struct.unpack("<B", f.read(1))[0]
+    if nit == 253:
+        nit = struct.unpack("<H", f.read(2))[0]
+    elif nit == 254:
+        nit = struct.unpack("<I", f.read(4))[0]
+    elif nit == 255:
+        nit = struct.unpack("<Q", f.read(8))[0]
+    return f.read(nit)
+
+
+def ser_string(s):
+    if len(s) < 253:
+        return chr(len(s)) + s
+    elif len(s) < 0x10000:
+        return chr(253) + struct.pack("<H", len(s)) + s
+    elif len(s) < 0x100000000L:
+        return chr(254) + struct.pack("<I", len(s)) + s
+    return chr(255) + struct.pack("<Q", len(s)) + s
+
+
+def deser_uint256(f):
+    r = 0L
+    for i in xrange(8):
+        t = struct.unpack("<I", f.read(4))[0]
+        r += t << (i * 32)
+    return r
+
+
+def ser_uint256(u):
+    rs = ""
+    for i in xrange(8):
+        rs += struct.pack("<I", u & 0xFFFFFFFFL)
+        u >>= 32
+    return rs
+
+
+def uint256_from_str(s):
+    r = 0L
+    t = struct.unpack("<IIIIIIII", s[:32])
+    for i in xrange(8):
+        r += t[i] << (i * 32)
+    return r
+
+
+def uint256_from_compact(c):
+    nbytes = (c >> 24) & 0xFF
+    v = (c & 0xFFFFFFL) << (8 * (nbytes - 3))
+    return v
+
+
+def deser_vector(f, c):
+    nit = struct.unpack("<B", f.read(1))[0]
+    if nit == 253:
+        nit = struct.unpack("<H", f.read(2))[0]
+    elif nit == 254:
+        nit = struct.unpack("<I", f.read(4))[0]
+    elif nit == 255:
+        nit = struct.unpack("<Q", f.read(8))[0]
+    r = []
+    for i in xrange(nit):
+        t = c()
+        t.deserialize(f)
+        r.append(t)
+    return r
+
+
+def ser_vector(l):
+    r = ""
+    if len(l) < 253:
+        r = chr(len(l))
+    elif len(l) < 0x10000:
+        r = chr(253) + struct.pack("<H", len(l))
+    elif len(l) < 0x100000000L:
+        r = chr(254) + struct.pack("<I", len(l))
+    else:
+        r = chr(255) + struct.pack("<Q", len(l))
+    for i in l:
+        r += i.serialize()
+    return r
+
+
+def deser_uint256_vector(f):
+    nit = struct.unpack("<B", f.read(1))[0]
+    if nit == 253:
+        nit = struct.unpack("<H", f.read(2))[0]
+    elif nit == 254:
+        nit = struct.unpack("<I", f.read(4))[0]
+    elif nit == 255:
+        nit = struct.unpack("<Q", f.read(8))[0]
+    r = []
+    for i in xrange(nit):
+        t = deser_uint256(f)
+        r.append(t)
+    return r
+
+
+def ser_uint256_vector(l):
+    r = ""
+    if len(l) < 253:
+        r = chr(len(l))
+    elif len(l) < 0x10000:
+        r = chr(253) + struct.pack("<H", len(l))
+    elif len(l) < 0x100000000L:
+        r = chr(254) + struct.pack("<I", len(l))
+    else:
+        r = chr(255) + struct.pack("<Q", len(l))
+    for i in l:
+        r += ser_uint256(i)
+    return r
+
+
+def deser_string_vector(f):
+    nit = struct.unpack("<B", f.read(1))[0]
+    if nit == 253:
+        nit = struct.unpack("<H", f.read(2))[0]
+    elif nit == 254:
+        nit = struct.unpack("<I", f.read(4))[0]
+    elif nit == 255:
+        nit = struct.unpack("<Q", f.read(8))[0]
+    r = []
+    for i in xrange(nit):
+        t = deser_string(f)
+        r.append(t)
+    return r
+
+
+def ser_string_vector(l):
+    r = ""
+    if len(l) < 253:
+        r = chr(len(l))
+    elif len(l) < 0x10000:
+        r = chr(253) + struct.pack("<H", len(l))
+    elif len(l) < 0x100000000L:
+        r = chr(254) + struct.pack("<I", len(l))
+    else:
+        r = chr(255) + struct.pack("<Q", len(l))
+    for sv in l:
+        r += ser_string(sv)
+    return r
+
+
+def deser_int_vector(f):
+    nit = struct.unpack("<B", f.read(1))[0]
+    if nit == 253:
+        nit = struct.unpack("<H", f.read(2))[0]
+    elif nit == 254:
+        nit = struct.unpack("<I", f.read(4))[0]
+    elif nit == 255:
+        nit = struct.unpack("<Q", f.read(8))[0]
+    r = []
+    for i in xrange(nit):
+        t = struct.unpack("<i", f.read(4))[0]
+        r.append(t)
+    return r
+
+
+def ser_int_vector(l):
+    r = ""
+    if len(l) < 253:
+        r = chr(len(l))
+    elif len(l) < 0x10000:
+        r = chr(253) + struct.pack("<H", len(l))
+    elif len(l) < 0x100000000L:
+        r = chr(254) + struct.pack("<I", len(l))
+    else:
+        r = chr(255) + struct.pack("<Q", len(l))
+    for i in l:
+        r += struct.pack("<i", i)
+    return r
+
+
+# Objects that map to bitcoind objects, which can be serialized/deserialized
+
+class CAddress(object):
+    def __init__(self):
+        self.nServices = 1
+        self.pchReserved = "\x00" * 10 + "\xff" * 2
+        self.ip = "0.0.0.0"
+        self.port = 0
+
+    def deserialize(self, f):
+        self.nServices = struct.unpack("<Q", f.read(8))[0]
+        self.pchReserved = f.read(12)
+        self.ip = socket.inet_ntoa(f.read(4))
+        self.port = struct.unpack(">H", f.read(2))[0]
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<Q", self.nServices)
+        r += self.pchReserved
+        r += socket.inet_aton(self.ip)
+        r += struct.pack(">H", self.port)
+        return r
+
+    def __repr__(self):
+        return "CAddress(nServices=%i ip=%s port=%i)" % (self.nServices,
+                                                         self.ip, self.port)
+
+
+class CInv(object):
+    typemap = {
+        0: "Error",
+        1: "TX",
+        2: "Block"}
+
+    def __init__(self, t=0, h=0L):
+        self.type = t
+        self.hash = h
+
+    def deserialize(self, f):
+        self.type = struct.unpack("<i", f.read(4))[0]
+        self.hash = deser_uint256(f)
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.type)
+        r += ser_uint256(self.hash)
+        return r
+
+    def __repr__(self):
+        return "CInv(type=%s hash=%064x)" \
+            % (self.typemap[self.type], self.hash)
+
+
+class CBlockLocator(object):
+    def __init__(self):
+        self.nVersion = MY_VERSION
+        self.vHave = []
+
+    def deserialize(self, f):
+        self.nVersion = struct.unpack("<i", f.read(4))[0]
+        self.vHave = deser_uint256_vector(f)
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.nVersion)
+        r += ser_uint256_vector(self.vHave)
+        return r
+
+    def __repr__(self):
+        return "CBlockLocator(nVersion=%i vHave=%s)" \
+            % (self.nVersion, repr(self.vHave))
+
+
+class COutPoint(object):
+    def __init__(self, hash=0, n=0):
+        self.hash = hash
+        self.n = n
+
+    def deserialize(self, f):
+        self.hash = deser_uint256(f)
+        self.n = struct.unpack("<I", f.read(4))[0]
+
+    def serialize(self):
+        r = ""
+        r += ser_uint256(self.hash)
+        r += struct.pack("<I", self.n)
+        return r
+
+    def __repr__(self):
+        return "COutPoint(hash=%064x n=%i)" % (self.hash, self.n)
+
+
+class CTxIn(object):
+    def __init__(self, outpoint=None, scriptSig="", nSequence=0):
+        if outpoint is None:
+            self.prevout = COutPoint()
+        else:
+            self.prevout = outpoint
+        self.scriptSig = scriptSig
+        self.nSequence = nSequence
+
+    def deserialize(self, f):
+        self.prevout = COutPoint()
+        self.prevout.deserialize(f)
+        self.scriptSig = deser_string(f)
+        self.nSequence = struct.unpack("<I", f.read(4))[0]
+
+    def serialize(self):
+        r = ""
+        r += self.prevout.serialize()
+        r += ser_string(self.scriptSig)
+        r += struct.pack("<I", self.nSequence)
+        return r
+
+    def __repr__(self):
+        return "CTxIn(prevout=%s scriptSig=%s nSequence=%i)" \
+            % (repr(self.prevout), binascii.hexlify(self.scriptSig),
+               self.nSequence)
+
+
+class CTxOut(object):
+    def __init__(self, nValue=0, scriptPubKey=""):
+        self.nValue = nValue
+        self.scriptPubKey = scriptPubKey
+
+    def deserialize(self, f):
+        self.nValue = struct.unpack("<q", f.read(8))[0]
+        self.scriptPubKey = deser_string(f)
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<q", self.nValue)
+        r += ser_string(self.scriptPubKey)
+        return r
+
+    def __repr__(self):
+        return "CTxOut(nValue=%i.%08i scriptPubKey=%s)" \
+            % (self.nValue // 100000000, self.nValue % 100000000,
+               binascii.hexlify(self.scriptPubKey))
+
+
+class CTransaction(object):
+    def __init__(self, tx=None):
+        if tx is None:
+            self.nVersion = 1
+            self.vin = []
+            self.vout = []
+            self.nLockTime = 0
+            self.sha256 = None
+            self.hash = None
+        else:
+            self.nVersion = tx.nVersion
+            self.vin = copy.deepcopy(tx.vin)
+            self.vout = copy.deepcopy(tx.vout)
+            self.nLockTime = tx.nLockTime
+            self.sha256 = None
+            self.hash = None
+
+    def deserialize(self, f):
+        self.nVersion = struct.unpack("<i", f.read(4))[0]
+        self.vin = deser_vector(f, CTxIn)
+        self.vout = deser_vector(f, CTxOut)
+        self.nLockTime = struct.unpack("<I", f.read(4))[0]
+        self.sha256 = None
+        self.hash = None
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.nVersion)
+        r += ser_vector(self.vin)
+        r += ser_vector(self.vout)
+        r += struct.pack("<I", self.nLockTime)
+        return r
+
+    def rehash(self):
+        self.sha256 = None
+        self.calc_sha256()
+
+    def calc_sha256(self):
+        if self.sha256 is None:
+            self.sha256 = uint256_from_str(hash256(self.serialize()))
+        self.hash = hash256(self.serialize())[::-1].encode('hex_codec')
+
+    def is_valid(self):
+        self.calc_sha256()
+        for tout in self.vout:
+            if tout.nValue < 0 or tout.nValue > 21000000L * 100000000L:
+                return False
+        return True
+
+    def __repr__(self):
+        return "CTransaction(nVersion=%i vin=%s vout=%s nLockTime=%i)" \
+            % (self.nVersion, repr(self.vin), repr(self.vout), self.nLockTime)
+
+
+class CBlockHeader(object):
+    def __init__(self, header=None):
+        if header is None:
+            self.set_null()
+        else:
+            self.nVersion = header.nVersion
+            self.hashPrevBlock = header.hashPrevBlock
+            self.hashMerkleRoot = header.hashMerkleRoot
+            self.nTime = header.nTime
+            self.nBits = header.nBits
+            self.nNonce = header.nNonce
+            self.sha256 = header.sha256
+            self.hash = header.hash
+            self.calc_sha256()
+
+    def set_null(self):
+        self.nVersion = 1
+        self.hashPrevBlock = 0
+        self.hashMerkleRoot = 0
+        self.nTime = 0
+        self.nBits = 0
+        self.nNonce = 0
+        self.sha256 = None
+        self.hash = None
+
+    def deserialize(self, f):
+        self.nVersion = struct.unpack("<i", f.read(4))[0]
+        self.hashPrevBlock = deser_uint256(f)
+        self.hashMerkleRoot = deser_uint256(f)
+        self.nTime = struct.unpack("<I", f.read(4))[0]
+        self.nBits = struct.unpack("<I", f.read(4))[0]
+        self.nNonce = struct.unpack("<I", f.read(4))[0]
+        self.sha256 = None
+        self.hash = None
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.nVersion)
+        r += ser_uint256(self.hashPrevBlock)
+        r += ser_uint256(self.hashMerkleRoot)
+        r += struct.pack("<I", self.nTime)
+        r += struct.pack("<I", self.nBits)
+        r += struct.pack("<I", self.nNonce)
+        return r
+
+    def calc_sha256(self):
+        if self.sha256 is None:
+            r = ""
+            r += struct.pack("<i", self.nVersion)
+            r += ser_uint256(self.hashPrevBlock)
+            r += ser_uint256(self.hashMerkleRoot)
+            r += struct.pack("<I", self.nTime)
+            r += struct.pack("<I", self.nBits)
+            r += struct.pack("<I", self.nNonce)
+            self.sha256 = uint256_from_str(hash256(r))
+            self.hash = hash256(r)[::-1].encode('hex_codec')
+
+    def rehash(self):
+        self.sha256 = None
+        self.calc_sha256()
+        return self.sha256
+
+    def __repr__(self):
+        return "CBlockHeader(nVersion=%i hashPrevBlock=%064x hashMerkleRoot=%064x nTime=%s nBits=%08x nNonce=%08x)" \
+            % (self.nVersion, self.hashPrevBlock, self.hashMerkleRoot,
+               time.ctime(self.nTime), self.nBits, self.nNonce)
+
+
+class CBlock(CBlockHeader):
+    def __init__(self, header=None):
+        super(CBlock, self).__init__(header)
+        self.vtx = []
+
+    def deserialize(self, f):
+        super(CBlock, self).deserialize(f)
+        self.vtx = deser_vector(f, CTransaction)
+
+    def serialize(self):
+        r = ""
+        r += super(CBlock, self).serialize()
+        r += ser_vector(self.vtx)
+        return r
+
+    def calc_merkle_root(self):
+        hashes = []
+        for tx in self.vtx:
+            tx.calc_sha256()
+            hashes.append(ser_uint256(tx.sha256))
+        while len(hashes) > 1:
+            newhashes = []
+            for i in xrange(0, len(hashes), 2):
+                i2 = min(i+1, len(hashes)-1)
+                newhashes.append(hash256(hashes[i] + hashes[i2]))
+            hashes = newhashes
+        return uint256_from_str(hashes[0])
+
+    def is_valid(self):
+        self.calc_sha256()
+        target = uint256_from_compact(self.nBits)
+        if self.sha256 > target:
+            return False
+        for tx in self.vtx:
+            if not tx.is_valid():
+                return False
+        if self.calc_merkle_root() != self.hashMerkleRoot:
+            return False
+        return True
+
+    def solve(self):
+        self.calc_sha256()
+        target = uint256_from_compact(self.nBits)
+        while self.sha256 > target:
+            self.nNonce += 1
+            self.rehash()
+
+    def __repr__(self):
+        return "CBlock(nVersion=%i hashPrevBlock=%064x hashMerkleRoot=%064x nTime=%s nBits=%08x nNonce=%08x vtx=%s)" \
+            % (self.nVersion, self.hashPrevBlock, self.hashMerkleRoot,
+               time.ctime(self.nTime), self.nBits, self.nNonce, repr(self.vtx))
+
+
+class CUnsignedAlert(object):
+    def __init__(self):
+        self.nVersion = 1
+        self.nRelayUntil = 0
+        self.nExpiration = 0
+        self.nID = 0
+        self.nCancel = 0
+        self.setCancel = []
+        self.nMinVer = 0
+        self.nMaxVer = 0
+        self.setSubVer = []
+        self.nPriority = 0
+        self.strComment = ""
+        self.strStatusBar = ""
+        self.strReserved = ""
+
+    def deserialize(self, f):
+        self.nVersion = struct.unpack("<i", f.read(4))[0]
+        self.nRelayUntil = struct.unpack("<q", f.read(8))[0]
+        self.nExpiration = struct.unpack("<q", f.read(8))[0]
+        self.nID = struct.unpack("<i", f.read(4))[0]
+        self.nCancel = struct.unpack("<i", f.read(4))[0]
+        self.setCancel = deser_int_vector(f)
+        self.nMinVer = struct.unpack("<i", f.read(4))[0]
+        self.nMaxVer = struct.unpack("<i", f.read(4))[0]
+        self.setSubVer = deser_string_vector(f)
+        self.nPriority = struct.unpack("<i", f.read(4))[0]
+        self.strComment = deser_string(f)
+        self.strStatusBar = deser_string(f)
+        self.strReserved = deser_string(f)
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.nVersion)
+        r += struct.pack("<q", self.nRelayUntil)
+        r += struct.pack("<q", self.nExpiration)
+        r += struct.pack("<i", self.nID)
+        r += struct.pack("<i", self.nCancel)
+        r += ser_int_vector(self.setCancel)
+        r += struct.pack("<i", self.nMinVer)
+        r += struct.pack("<i", self.nMaxVer)
+        r += ser_string_vector(self.setSubVer)
+        r += struct.pack("<i", self.nPriority)
+        r += ser_string(self.strComment)
+        r += ser_string(self.strStatusBar)
+        r += ser_string(self.strReserved)
+        return r
+
+    def __repr__(self):
+        return "CUnsignedAlert(nVersion %d, nRelayUntil %d, nExpiration %d, nID %d, nCancel %d, nMinVer %d, nMaxVer %d, nPriority %d, strComment %s, strStatusBar %s, strReserved %s)" \
+            % (self.nVersion, self.nRelayUntil, self.nExpiration, self.nID,
+               self.nCancel, self.nMinVer, self.nMaxVer, self.nPriority,
+               self.strComment, self.strStatusBar, self.strReserved)
+
+
+class CAlert(object):
+    def __init__(self):
+        self.vchMsg = ""
+        self.vchSig = ""
+
+    def deserialize(self, f):
+        self.vchMsg = deser_string(f)
+        self.vchSig = deser_string(f)
+
+    def serialize(self):
+        r = ""
+        r += ser_string(self.vchMsg)
+        r += ser_string(self.vchSig)
+        return r
+
+    def __repr__(self):
+        return "CAlert(vchMsg.sz %d, vchSig.sz %d)" \
+            % (len(self.vchMsg), len(self.vchSig))
+
+
+# Objects that correspond to messages on the wire
+class msg_version(object):
+    command = "version"
+
+    def __init__(self):
+        self.nVersion = MY_VERSION
+        self.nServices = 1
+        self.nTime = time.time()
+        self.addrTo = CAddress()
+        self.addrFrom = CAddress()
+        self.nNonce = random.getrandbits(64)
+        self.strSubVer = MY_SUBVERSION
+        self.nStartingHeight = -1
+
+    def deserialize(self, f):
+        self.nVersion = struct.unpack("<i", f.read(4))[0]
+        if self.nVersion == 10300:
+            self.nVersion = 300
+        self.nServices = struct.unpack("<Q", f.read(8))[0]
+        self.nTime = struct.unpack("<q", f.read(8))[0]
+        self.addrTo = CAddress()
+        self.addrTo.deserialize(f)
+        if self.nVersion >= 106:
+            self.addrFrom = CAddress()
+            self.addrFrom.deserialize(f)
+            self.nNonce = struct.unpack("<Q", f.read(8))[0]
+            self.strSubVer = deser_string(f)
+            if self.nVersion >= 209:
+                self.nStartingHeight = struct.unpack("<i", f.read(4))[0]
+            else:
+                self.nStartingHeight = None
+        else:
+            self.addrFrom = None
+            self.nNonce = None
+            self.strSubVer = None
+            self.nStartingHeight = None
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<i", self.nVersion)
+        r += struct.pack("<Q", self.nServices)
+        r += struct.pack("<q", self.nTime)
+        r += self.addrTo.serialize()
+        r += self.addrFrom.serialize()
+        r += struct.pack("<Q", self.nNonce)
+        r += ser_string(self.strSubVer)
+        r += struct.pack("<i", self.nStartingHeight)
+        return r
+
+    def __repr__(self):
+        return 'msg_version(nVersion=%i nServices=%i nTime=%s addrTo=%s addrFrom=%s nNonce=0x%016X strSubVer=%s nStartingHeight=%i)' \
+            % (self.nVersion, self.nServices, time.ctime(self.nTime),
+               repr(self.addrTo), repr(self.addrFrom), self.nNonce,
+               self.strSubVer, self.nStartingHeight)
+
+
+class msg_verack(object):
+    command = "verack"
+
+    def __init__(self):
+        pass
+
+    def deserialize(self, f):
+        pass
+
+    def serialize(self):
+        return ""
+
+    def __repr__(self):
+        return "msg_verack()"
+
+
+class msg_addr(object):
+    command = "addr"
+
+    def __init__(self):
+        self.addrs = []
+
+    def deserialize(self, f):
+        self.addrs = deser_vector(f, CAddress)
+
+    def serialize(self):
+        return ser_vector(self.addrs)
+
+    def __repr__(self):
+        return "msg_addr(addrs=%s)" % (repr(self.addrs))
+
+
+class msg_alert(object):
+    command = "alert"
+
+    def __init__(self):
+        self.alert = CAlert()
+
+    def deserialize(self, f):
+        self.alert = CAlert()
+        self.alert.deserialize(f)
+
+    def serialize(self):
+        r = ""
+        r += self.alert.serialize()
+        return r
+
+    def __repr__(self):
+        return "msg_alert(alert=%s)" % (repr(self.alert), )
+
+
+class msg_inv(object):
+    command = "inv"
+
+    def __init__(self, inv=None):
+        if inv is None:
+            self.inv = []
+        else:
+            self.inv = inv
+
+    def deserialize(self, f):
+        self.inv = deser_vector(f, CInv)
+
+    def serialize(self):
+        return ser_vector(self.inv)
+
+    def __repr__(self):
+        return "msg_inv(inv=%s)" % (repr(self.inv))
+
+
+class msg_getdata(object):
+    command = "getdata"
+
+    def __init__(self):
+        self.inv = []
+
+    def deserialize(self, f):
+        self.inv = deser_vector(f, CInv)
+
+    def serialize(self):
+        return ser_vector(self.inv)
+
+    def __repr__(self):
+        return "msg_getdata(inv=%s)" % (repr(self.inv))
+
+
+class msg_getblocks(object):
+    command = "getblocks"
+
+    def __init__(self):
+        self.locator = CBlockLocator()
+        self.hashstop = 0L
+
+    def deserialize(self, f):
+        self.locator = CBlockLocator()
+        self.locator.deserialize(f)
+        self.hashstop = deser_uint256(f)
+
+    def serialize(self):
+        r = ""
+        r += self.locator.serialize()
+        r += ser_uint256(self.hashstop)
+        return r
+
+    def __repr__(self):
+        return "msg_getblocks(locator=%s hashstop=%064x)" \
+            % (repr(self.locator), self.hashstop)
+
+
+class msg_tx(object):
+    command = "tx"
+
+    def __init__(self, tx=CTransaction()):
+        self.tx = tx
+
+    def deserialize(self, f):
+        self.tx.deserialize(f)
+
+    def serialize(self):
+        return self.tx.serialize()
+
+    def __repr__(self):
+        return "msg_tx(tx=%s)" % (repr(self.tx))
+
+
+class msg_block(object):
+    command = "block"
+
+    def __init__(self, block=None):
+        if block is None:
+            self.block = CBlock()
+        else:
+            self.block = block
+
+    def deserialize(self, f):
+        self.block.deserialize(f)
+
+    def serialize(self):
+        return self.block.serialize()
+
+    def __repr__(self):
+        return "msg_block(block=%s)" % (repr(self.block))
+
+
+class msg_getaddr(object):
+    command = "getaddr"
+
+    def __init__(self):
+        pass
+
+    def deserialize(self, f):
+        pass
+
+    def serialize(self):
+        return ""
+
+    def __repr__(self):
+        return "msg_getaddr()"
+
+
+class msg_ping_prebip31(object):
+    command = "ping"
+
+    def __init__(self):
+        pass
+
+    def deserialize(self, f):
+        pass
+
+    def serialize(self):
+        return ""
+
+    def __repr__(self):
+        return "msg_ping() (pre-bip31)"
+
+
+class msg_ping(object):
+    command = "ping"
+
+    def __init__(self, nonce=0L):
+        self.nonce = nonce
+
+    def deserialize(self, f):
+        self.nonce = struct.unpack("<Q", f.read(8))[0]
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<Q", self.nonce)
+        return r
+
+    def __repr__(self):
+        return "msg_ping(nonce=%08x)" % self.nonce
+
+
+class msg_pong(object):
+    command = "pong"
+
+    def __init__(self, nonce=0L):
+        self.nonce = nonce
+
+    def deserialize(self, f):
+        self.nonce = struct.unpack("<Q", f.read(8))[0]
+
+    def serialize(self):
+        r = ""
+        r += struct.pack("<Q", self.nonce)
+        return r
+
+    def __repr__(self):
+        return "msg_pong(nonce=%08x)" % self.nonce
+
+
+class msg_mempool(object):
+    command = "mempool"
+
+    def __init__(self):
+        pass
+
+    def deserialize(self, f):
+        pass
+
+    def serialize(self):
+        return ""
+
+    def __repr__(self):
+        return "msg_mempool()"
+
+
+# getheaders message has
+# number of entries
+# vector of hashes
+# hash_stop (hash of last desired block header, 0 to get as many as possible)
+class msg_getheaders(object):
+    command = "getheaders"
+
+    def __init__(self):
+        self.locator = CBlockLocator()
+        self.hashstop = 0L
+
+    def deserialize(self, f):
+        self.locator = CBlockLocator()
+        self.locator.deserialize(f)
+        self.hashstop = deser_uint256(f)
+
+    def serialize(self):
+        r = ""
+        r += self.locator.serialize()
+        r += ser_uint256(self.hashstop)
+        return r
+
+    def __repr__(self):
+        return "msg_getheaders(locator=%s, stop=%064x)" \
+            % (repr(self.locator), self.hashstop)
+
+
+# headers message has
+# <count> <vector of block headers>
+class msg_headers(object):
+    command = "headers"
+
+    def __init__(self):
+        self.headers = []
+
+    def deserialize(self, f):
+        # comment in bitcoind indicates these should be deserialized as blocks
+        blocks = deser_vector(f, CBlock)
+        for x in blocks:
+            self.headers.append(CBlockHeader(x))
+
+    def serialize(self):
+        blocks = [CBlock(x) for x in self.headers]
+        return ser_vector(blocks)
+
+    def __repr__(self):
+        return "msg_headers(headers=%s)" % repr(self.headers)
+
+
+class msg_reject(object):
+    command = "reject"
+
+    def __init__(self):
+        self.message = ""
+        self.code = ""
+        self.reason = ""
+        self.data = 0L
+
+    def deserialize(self, f):
+        self.message = deser_string(f)
+        self.code = struct.unpack("<B", f.read(1))[0]
+        self.reason = deser_string(f)
+        if (self.message == "block" or self.message == "tx"):
+            self.data = deser_uint256(f)
+
+    def serialize(self):
+        r = ser_string(self.message)
+        r += struct.pack("<B", self.code)
+        r += ser_string(self.reason)
+        if (self.message == "block" or self.message == "tx"):
+            r += ser_uint256(self.data)
+        return r
+
+    def __repr__(self):
+        return "msg_reject: %s %d %s [%064x]" \
+            % (self.message, self.code, self.reason, self.data)
+
+
+# This is what a callback should look like for NodeConn
+# Reimplement the on_* functions to provide handling for events
+class NodeConnCB(object):
+    def __init__(self):
+        # Acquire on all callbacks -- overkill for now since asyncore is
+        # single-threaded, but may be useful for synchronizing access to
+        # member variables in derived classes.
+        self.cbLock = Lock()
+        self.verack_received = False
+
+    # Derived classes should call this function once to set the message map
+    # which associates the derived classes' functions to incoming messages
+    def create_callback_map(self):
+        self.cbmap = {
+            "version": self.on_version,
+            "verack": self.on_verack,
+            "addr": self.on_addr,
+            "alert": self.on_alert,
+            "inv": self.on_inv,
+            "getdata": self.on_getdata,
+            "getblocks": self.on_getblocks,
+            "tx": self.on_tx,
+            "block": self.on_block,
+            "getaddr": self.on_getaddr,
+            "ping": self.on_ping,
+            "pong": self.on_pong,
+            "headers": self.on_headers,
+            "getheaders": self.on_getheaders,
+            "reject": self.on_reject,
+            "mempool": self.on_mempool
+        }
+
+    def deliver(self, conn, message):
+        with self.cbLock:
+            try:
+                self.cbmap[message.command](conn, message)
+            except:
+                print "ERROR delivering %s (%s)" % (repr(message),
+                                                    sys.exc_info()[0])
+
+    def on_version(self, conn, message):
+        if message.nVersion >= 209:
+            conn.send_message(msg_verack())
+        conn.ver_send = min(MY_VERSION, message.nVersion)
+        if message.nVersion < 209:
+            conn.ver_recv = conn.ver_send
+
+    def on_verack(self, conn, message):
+        conn.ver_recv = conn.ver_send
+        self.verack_received = True
+
+    def on_inv(self, conn, message):
+        want = msg_getdata()
+        for i in message.inv:
+            if i.type != 0:
+                want.inv.append(i)
+        if len(want.inv):
+            conn.send_message(want)
+
+    def on_addr(self, conn, message): pass
+    def on_alert(self, conn, message): pass
+    def on_getdata(self, conn, message): pass
+    def on_getblocks(self, conn, message): pass
+    def on_tx(self, conn, message): pass
+    def on_block(self, conn, message): pass
+    def on_getaddr(self, conn, message): pass
+    def on_headers(self, conn, message): pass
+    def on_getheaders(self, conn, message): pass
+    def on_ping(self, conn, message):
+        if conn.ver_send > BIP0031_VERSION:
+            conn.send_message(msg_pong(message.nonce))
+    def on_reject(self, conn, message): pass
+    def on_close(self, conn): pass
+    def on_mempool(self, conn): pass
+    def on_pong(self, conn, message): pass
+
+
+# The actual NodeConn class
+# This class provides an interface for a p2p connection to a specified node
+class NodeConn(asyncore.dispatcher):
+    messagemap = {
+        "version": msg_version,
+        "verack": msg_verack,
+        "addr": msg_addr,
+        "alert": msg_alert,
+        "inv": msg_inv,
+        "getdata": msg_getdata,
+        "getblocks": msg_getblocks,
+        "tx": msg_tx,
+        "block": msg_block,
+        "getaddr": msg_getaddr,
+        "ping": msg_ping,
+        "pong": msg_pong,
+        "headers": msg_headers,
+        "getheaders": msg_getheaders,
+        "reject": msg_reject,
+        "mempool": msg_mempool
+    }
+    MAGIC_BYTES = {
+        "mainnet": "\xf9\xbe\xb4\xd9",   # mainnet
+        "testnet3": "\x0b\x11\x09\x07",  # testnet3
+        "regtest": "\xfa\xbf\xb5\xda"    # regtest
+    }
+
+    def __init__(self, dstaddr, dstport, rpc, callback, net="regtest"):
+        asyncore.dispatcher.__init__(self)
+        self.log = logging.getLogger("NodeConn(%s:%d)" % (dstaddr, dstport))
+        self.dstaddr = dstaddr
+        self.dstport = dstport
+        self.create_socket(socket.AF_INET, socket.SOCK_STREAM)
+        self.sendbuf = ""
+        self.recvbuf = ""
+        self.ver_send = 209
+        self.ver_recv = 209
+        self.last_sent = 0
+        self.state = "connecting"
+        self.network = net
+        self.cb = callback
+        self.sendbufLock = Lock()  # for protecting the sendbuffer
+        self.disconnect = False
+
+        # stuff version msg into sendbuf
+        vt = msg_version()
+        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)
+        print 'MiniNode: Connecting to Bitcoin Node IP # ' + dstaddr + ':' \
+            + str(dstport)
+
+        try:
+            self.connect((dstaddr, dstport))
+        except:
+            self.handle_close()
+        self.rpc = rpc
+
+    def show_debug_msg(self, msg):
+        self.log.debug(msg)
+
+    def handle_connect(self):
+        self.show_debug_msg("MiniNode: Connected & Listening: \n")
+        self.state = "connected"
+
+    def handle_close(self):
+        self.show_debug_msg("MiniNode: Closing Connection to %s:%d... "
+                            % (self.dstaddr, self.dstport))
+        self.state = "closed"
+        self.recvbuf = ""
+        self.sendbuf = ""
+        try:
+            self.close()
+        except:
+            pass
+        self.cb.on_close(self)
+
+    def handle_read(self):
+        try:
+            t = self.recv(8192)
+            if len(t) > 0:
+                self.recvbuf += t
+                self.got_data()
+        except:
+            pass
+
+    def readable(self):
+        return True
+
+    def writable(self):
+        if self.disconnect:
+            self.handle_close()
+            return False
+        else:
+            self.sendbufLock.acquire()
+            length = len(self.sendbuf)
+            self.sendbufLock.release()
+            return (length > 0)
+
+    def handle_write(self):
+        self.sendbufLock.acquire()
+        try:
+            sent = self.send(self.sendbuf)
+        except:
+            self.handle_close()
+            return
+        self.sendbuf = self.sendbuf[sent:]
+        self.sendbufLock.release()
+
+    def got_data(self):
+        while True:
+            if len(self.recvbuf) < 4:
+                return
+            if self.recvbuf[:4] != self.MAGIC_BYTES[self.network]:
+                raise ValueError("got garbage %s" % repr(self.recvbuf))
+            if self.ver_recv < 209:
+                if len(self.recvbuf) < 4 + 12 + 4:
+                    return
+                command = self.recvbuf[4:4+12].split("\x00", 1)[0]
+                msglen = struct.unpack("<i", self.recvbuf[4+12:4+12+4])[0]
+                checksum = None
+                if len(self.recvbuf) < 4 + 12 + 4 + msglen:
+                    return
+                msg = self.recvbuf[4+12+4:4+12+4+msglen]
+                self.recvbuf = self.recvbuf[4+12+4+msglen:]
+            else:
+                if len(self.recvbuf) < 4 + 12 + 4 + 4:
+                    return
+                command = self.recvbuf[4:4+12].split("\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 in self.messagemap:
+                f = cStringIO.StringIO(msg)
+                t = self.messagemap[command]()
+                t.deserialize(f)
+                self.got_message(t)
+            else:
+                self.show_debug_msg("Unknown command: '" + command + "' " +
+                                    repr(msg))
+
+    def send_message(self, message, pushbuf=False):
+        if self.state != "connected" and not pushbuf:
+            return
+        self.sendbufLock.acquire()
+        self.show_debug_msg("Send %s" % repr(message))
+        command = message.command
+        data = message.serialize()
+        tmsg = self.MAGIC_BYTES[self.network]
+        tmsg += command
+        tmsg += "\x00" * (12 - len(command))
+        tmsg += struct.pack("<I", len(data))
+        if self.ver_send >= 209:
+            th = sha256(data)
+            h = sha256(th)
+            tmsg += h[:4]
+        tmsg += data
+        self.sendbuf += tmsg
+        self.last_sent = time.time()
+        self.sendbufLock.release()
+
+    def got_message(self, message):
+        if message.command == "version":
+            if message.nVersion <= BIP0031_VERSION:
+                self.messagemap['ping'] = msg_ping_prebip31
+        if self.last_sent + 30 * 60 < time.time():
+            self.send_message(self.messagemap['ping']())
+        self.show_debug_msg("Recv %s" % repr(message))
+        self.cb.deliver(self, message)
+
+    def disconnect_node(self):
+        self.disconnect = True
+        self.send_message(self.messagemap['ping']())
+
+
+class NetworkThread(Thread):
+    def run(self):
+        asyncore.loop(0.1, True)
+
+
+# An exception we can raise if we detect a potential disconnect
+# (p2p or rpc) before the test is complete
+class EarlyDisconnectError(Exception):
+    def __init__(self, value):
+        self.value = value
+
+    def __str__(self):
+        return repr(self.value)
diff --git a/qa/rpc-tests/util.py b/qa/rpc-tests/util.py
index cf789f48e2..385215a088 100644
--- a/qa/rpc-tests/util.py
+++ b/qa/rpc-tests/util.py
@@ -158,12 +158,14 @@ def _rpchost_to_args(rpchost):
         rv += ['-rpcport=' + rpcport]
     return rv
 
-def start_node(i, dirname, extra_args=None, rpchost=None, timewait=None):
+def start_node(i, dirname, extra_args=None, rpchost=None, timewait=None, binary=None):
     """
     Start a bitcoind and return RPC connection to it
     """
     datadir = os.path.join(dirname, "node"+str(i))
-    args = [ os.getenv("BITCOIND", "bitcoind"), "-datadir="+datadir, "-keypool=1", "-discover=0", "-rest" ]
+    if binary is None:
+        binary = os.getenv("BITCOIND", "bitcoind")
+    args = [ binary, "-datadir="+datadir, "-keypool=1", "-discover=0", "-rest" ]
     if extra_args is not None: args.extend(extra_args)
     bitcoind_processes[i] = subprocess.Popen(args)
     devnull = open("/dev/null", "w+")
@@ -179,12 +181,13 @@ def start_node(i, dirname, extra_args=None, rpchost=None, timewait=None):
     proxy.url = url # store URL on proxy for info
     return proxy
 
-def start_nodes(num_nodes, dirname, extra_args=None, rpchost=None):
+def start_nodes(num_nodes, dirname, extra_args=None, rpchost=None, binary=None):
     """
     Start multiple bitcoinds, return RPC connections to them
     """
     if extra_args is None: extra_args = [ None for i in range(num_nodes) ]
-    return [ start_node(i, dirname, extra_args[i], rpchost) for i in range(num_nodes) ]
+    if binary is None: binary = [ None for i in range(num_nodes) ]
+    return [ start_node(i, dirname, extra_args[i], rpchost, binary=binary[i]) for i in range(num_nodes) ]
 
 def log_filename(dirname, n_node, logname):
     return os.path.join(dirname, "node"+str(n_node), "regtest", logname)