aboutsummaryrefslogtreecommitdiff
path: root/test/rpc-tests/test_framework/netutil.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/rpc-tests/test_framework/netutil.py')
-rw-r--r--test/rpc-tests/test_framework/netutil.py156
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