1 files changed, 156 insertions, 0 deletions
diff --git a/test/rpc-tests/test_framework/netutil.py b/test/rpc-tests/test_framework/netutil.py
new file mode 100644
index 0000000000..45d8e22d22
--- /dev/null
+++ b/test/rpc-tests/test_framework/netutil.py
@@ -0,0 +1,156 @@
+#!/usr/bin/env python3
+# Copyright (c) 2014-2016 The Bitcoin Core developers
+# Distributed under the MIT software license, see the accompanying
+# file COPYING or http://www.opensource.org/licenses/mit-license.php.
+"""Linux network utilities.
+
+Roughly based on http://voorloopnul.com/blog/a-python-netstat-in-less-than-100-lines-of-code/ by Ricardo Pascal
+"""
+
+import sys
+import socket
+import fcntl
+import struct
+import array
+import os
+from binascii import unhexlify, hexlify
+
+STATE_ESTABLISHED = '01'
+STATE_SYN_SENT  = '02'
+STATE_SYN_RECV = '03'
+STATE_FIN_WAIT1 = '04'
+STATE_FIN_WAIT2 = '05'
+STATE_TIME_WAIT = '06'
+STATE_CLOSE = '07'
+STATE_CLOSE_WAIT = '08'
+STATE_LAST_ACK = '09'
+STATE_LISTEN = '0A'
+STATE_CLOSING = '0B'
+
+def get_socket_inodes(pid):
+    '''
+    Get list of socket inodes for process pid.
+    '''
+    base = '/proc/%i/fd' % pid
+    inodes = []
+    for item in os.listdir(base):
+        target = os.readlink(os.path.join(base, item))
+        if target.startswith('socket:'):
+            inodes.append(int(target[8:-1]))
+    return inodes
+
+def _remove_empty(array):
+    return [x for x in array if x !='']
+
+def _convert_ip_port(array):
+    host,port = array.split(':')
+    # convert host from mangled-per-four-bytes form as used by kernel
+    host = unhexlify(host)
+    host_out = ''
+    for x in range(0, len(host) // 4):
+        (val,) = struct.unpack('=I', host[x*4:(x+1)*4])
+        host_out += '%08x' % val
+
+    return host_out,int(port,16)
+
+def netstat(typ='tcp'):
+    '''
+    Function to return a list with status of tcp connections at linux systems
+    To get pid of all network process running on system, you must run this script
+    as superuser
+    '''
+    with open('/proc/net/'+typ,'r',encoding='utf8') as f:
+        content = f.readlines()
+        content.pop(0)
+    result = []
+    for line in content:
+        line_array = _remove_empty(line.split(' '))     # Split lines and remove empty spaces.
+        tcp_id = line_array[0]
+        l_addr = _convert_ip_port(line_array[1])
+        r_addr = _convert_ip_port(line_array[2])
+        state = line_array[3]
+        inode = int(line_array[9])                      # Need the inode to match with process pid.
+        nline = [tcp_id, l_addr, r_addr, state, inode]
+        result.append(nline)
+    return result
+
+def get_bind_addrs(pid):
+    '''
+    Get bind addresses as (host,port) tuples for process pid.
+    '''
+    inodes = get_socket_inodes(pid)
+    bind_addrs = []
+    for conn in netstat('tcp') + netstat('tcp6'):
+        if conn[3] == STATE_LISTEN and conn[4] in inodes:
+            bind_addrs.append(conn[1])
+    return bind_addrs
+
+# from: http://code.activestate.com/recipes/439093/
+def all_interfaces():
+    '''
+    Return all interfaces that are up
+    '''
+    is_64bits = sys.maxsize > 2**32
+    struct_size = 40 if is_64bits else 32
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+    max_possible = 8 # initial value
+    while True:
+        bytes = max_possible * struct_size
+        names = array.array('B', b'\0' * bytes)
+        outbytes = struct.unpack('iL', fcntl.ioctl(
+            s.fileno(),
+            0x8912,  # SIOCGIFCONF
+            struct.pack('iL', bytes, names.buffer_info()[0])
+        ))[0]
+        if outbytes == bytes:
+            max_possible *= 2
+        else:
+            break
+    namestr = names.tostring()
+    return [(namestr[i:i+16].split(b'\0', 1)[0],
+             socket.inet_ntoa(namestr[i+20:i+24]))
+            for i in range(0, outbytes, struct_size)]
+
+def addr_to_hex(addr):
+    '''
+    Convert string IPv4 or IPv6 address to binary address as returned by
+    get_bind_addrs.
+    Very naive implementation that certainly doesn't work for all IPv6 variants.
+    '''
+    if '.' in addr: # IPv4
+        addr = [int(x) for x in addr.split('.')]
+    elif ':' in addr: # IPv6
+        sub = [[], []] # prefix, suffix
+        x = 0
+        addr = addr.split(':')
+        for i,comp in enumerate(addr):
+            if comp == '':
+                if i == 0 or i == (len(addr)-1): # skip empty component at beginning or end
+                    continue
+                x += 1 # :: skips to suffix
+                assert(x < 2)
+            else: # two bytes per component
+                val = int(comp, 16)
+                sub[x].append(val >> 8)
+                sub[x].append(val & 0xff)
+        nullbytes = 16 - len(sub[0]) - len(sub[1])
+        assert((x == 0 and nullbytes == 0) or (x == 1 and nullbytes > 0))
+        addr = sub[0] + ([0] * nullbytes) + sub[1]
+    else:
+        raise ValueError('Could not parse address %s' % addr)
+    return hexlify(bytearray(addr)).decode('ascii')
+
+def test_ipv6_local():
+    '''
+    Check for (local) IPv6 support.
+    '''
+    import socket
+    # By using SOCK_DGRAM this will not actually make a connection, but it will
+    # fail if there is no route to IPv6 localhost.
+    have_ipv6 = True
+    try:
+        s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
+        s.connect(('::1', 0))
+    except socket.error:
+        have_ipv6 = False
+    return have_ipv6