1 #!/usr/bin/env python3
2 # Copyright (c) 2014-present The Bitcoin Core developers
3 # Distributed under the MIT software license, see the accompanying
4 # file COPYING or http://www.opensource.org/licenses/mit-license.php.
5 """Linux, macOS, and BSD network utilities.
6 7 Roughly based on https://web.archive.org/web/20190424172231/http://voorloopnul.com/blog/a-python-netstat-in-less-than-100-lines-of-code/ by Ricardo Pascal
8 """
9 10 import sys
11 import socket
12 import struct
13 import array
14 import os
15 16 # Easily unreachable address. Attempts to connect to it will stay within the machine.
17 # Used to avoid non-loopback traffic or DNS queries.
18 UNREACHABLE_PROXY_ARG = '-proxy=127.0.0.1:1'
19 20 # STATE_ESTABLISHED = '01'
21 # STATE_SYN_SENT = '02'
22 # STATE_SYN_RECV = '03'
23 # STATE_FIN_WAIT1 = '04'
24 # STATE_FIN_WAIT2 = '05'
25 # STATE_TIME_WAIT = '06'
26 # STATE_CLOSE = '07'
27 # STATE_CLOSE_WAIT = '08'
28 # STATE_LAST_ACK = '09'
29 STATE_LISTEN = '0A'
30 # STATE_CLOSING = '0B'
31 32 # Address manager size constants as defined in addrman_impl.h
33 ADDRMAN_NEW_BUCKET_COUNT = 1 << 10
34 ADDRMAN_TRIED_BUCKET_COUNT = 1 << 8
35 ADDRMAN_BUCKET_SIZE = 1 << 6
36 37 def get_socket_inodes(pid):
38 '''
39 Get list of socket inodes for process pid.
40 '''
41 base = '/proc/%i/fd' % pid
42 inodes = []
43 for item in os.listdir(base):
44 try:
45 target = os.readlink(os.path.join(base, item))
46 if target.startswith('socket:'):
47 inodes.append(int(target[8:-1]))
48 except FileNotFoundError:
49 pass
50 return inodes
51 52 def _remove_empty(array):
53 return [x for x in array if x !='']
54 55 def _convert_ip_port(array):
56 host,port = array.split(':')
57 # convert host from mangled-per-four-bytes form as used by kernel
58 host = bytes.fromhex(host)
59 host_out = ''
60 for x in range(0, len(host) // 4):
61 (val,) = struct.unpack('=I', host[x*4:(x+1)*4])
62 host_out += '%08x' % val
63 64 return host_out,int(port,16)
65 66 def netstat(typ='tcp'):
67 '''
68 Function to return a list with status of tcp connections at linux systems
69 To get pid of all network process running on system, you must run this script
70 as superuser
71 '''
72 with open('/proc/net/'+typ,'r') as f:
73 content = f.readlines()
74 content.pop(0)
75 result = []
76 for line in content:
77 line_array = _remove_empty(line.split(' ')) # Split lines and remove empty spaces.
78 tcp_id = line_array[0]
79 l_addr = _convert_ip_port(line_array[1])
80 r_addr = _convert_ip_port(line_array[2])
81 state = line_array[3]
82 inode = int(line_array[9]) # Need the inode to match with process pid.
83 nline = [tcp_id, l_addr, r_addr, state, inode]
84 result.append(nline)
85 return result
86 87 def get_bind_addrs(pid):
88 '''
89 Get bind addresses as (host,port) tuples for process pid.
90 '''
91 if sys.platform == 'linux':
92 inodes = get_socket_inodes(pid)
93 bind_addrs = []
94 for conn in netstat('tcp') + netstat('tcp6'):
95 if conn[3] == STATE_LISTEN and conn[4] in inodes:
96 bind_addrs.append(conn[1])
97 return bind_addrs
98 elif sys.platform.startswith(("darwin", "freebsd", "netbsd", "openbsd")):
99 import re
100 import subprocess
101 output = subprocess.check_output(["lsof",
102 *(["-Di"] if sys.platform.startswith("freebsd") else []), # Ignore device cache to avoid stderr warnings.
103 "-nP", # Keep hosts and ports numeric.
104 "-a", # Require all filters to match.
105 "-p", str(pid), # Limit results to the target pid.
106 "-iTCP", # Only inspect TCP sockets.
107 "-sTCP:LISTEN", # Only keep listening sockets.
108 "-Ftn", # Emit machine-readable type and name fields.
109 ], text=True)
110 return [
111 (addr_to_hex(("::" if sock_type == "IPv6" else "0.0.0.0") if host == "*" else host.strip("[]")), int(port))
112 for sock_type, host, port in re.findall(r"t(IPv[46])\nn(\*|\[.+?]|[^:]+):(\d+)", output)
113 ]
114 else:
115 raise NotImplementedError(f"get_bind_addrs is not supported on {sys.platform}")
116 117 def all_interfaces():
118 '''
119 Return all IPv4 interfaces that are up.
120 '''
121 if sys.platform == 'linux':
122 import fcntl # Linux only, so only import when required
123 124 is_64bits = sys.maxsize > 2**32
125 struct_size = 40 if is_64bits else 32
126 s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
127 max_possible = 8 # initial value
128 while True:
129 bytes = max_possible * struct_size
130 names = array.array('B', b'\0' * bytes)
131 outbytes = struct.unpack('iL', fcntl.ioctl(
132 s.fileno(),
133 0x8912, # SIOCGIFCONF
134 struct.pack('iL', bytes, names.buffer_info()[0])
135 ))[0]
136 if outbytes == bytes:
137 max_possible *= 2
138 else:
139 break
140 namestr = names.tobytes()
141 return [(namestr[i:i+16].split(b'\0', 1)[0],
142 socket.inet_ntoa(namestr[i+20:i+24]))
143 for i in range(0, outbytes, struct_size)]
144 elif sys.platform.startswith(("darwin", "freebsd", "netbsd", "openbsd")):
145 import re
146 import subprocess
147 output = subprocess.check_output(["ifconfig", "-au"], text=True)
148 return [
149 (m["iface"].encode(), ip)
150 for m in re.finditer(r"(?m)^(?P<iface>\S+):(?P<block>[^\n]*(?:\n[ \t]+[^\n]*)*)", output)
151 for ip in re.findall(r"inet (\S+)", m["block"])
152 ]
153 else:
154 raise NotImplementedError(f"all_interfaces is not supported on {sys.platform}")
155 156 def addr_to_hex(addr):
157 '''
158 Convert string IPv4 or IPv6 address to binary address as returned by
159 get_bind_addrs.
160 Very naive implementation that certainly doesn't work for all IPv6 variants.
161 '''
162 if '.' in addr: # IPv4
163 addr = [int(x) for x in addr.split('.')]
164 elif ':' in addr: # IPv6
165 sub = [[], []] # prefix, suffix
166 x = 0
167 addr = addr.split(':')
168 for i,comp in enumerate(addr):
169 if comp == '':
170 if i == 0 or i == (len(addr)-1): # skip empty component at beginning or end
171 continue
172 x += 1 # :: skips to suffix
173 assert x < 2
174 else: # two bytes per component
175 val = int(comp, 16)
176 sub[x].append(val >> 8)
177 sub[x].append(val & 0xff)
178 nullbytes = 16 - len(sub[0]) - len(sub[1])
179 assert (x == 0 and nullbytes == 0) or (x == 1 and nullbytes > 0)
180 addr = sub[0] + ([0] * nullbytes) + sub[1]
181 else:
182 raise ValueError('Could not parse address %s' % addr)
183 return bytearray(addr).hex()
184 185 def test_ipv6_local():
186 '''
187 Check for (local) IPv6 support.
188 '''
189 # By using SOCK_DGRAM this will not actually make a connection, but it will
190 # fail if there is no route to IPv6 localhost.
191 have_ipv6 = True
192 try:
193 s = socket.socket(socket.AF_INET6, socket.SOCK_DGRAM)
194 s.connect(('::1', 1))
195 except socket.error:
196 have_ipv6 = False
197 return have_ipv6
198 199 def test_unix_socket():
200 '''Return True if UNIX sockets are available on this platform.'''
201 try:
202 socket.AF_UNIX
203 except AttributeError:
204 return False
205 else:
206 return True
207 208 def format_addr_port(addr, port):
209 '''Return either "addr:port" or "[addr]:port" based on whether addr looks like an IPv6 address.'''
210 if ":" in addr:
211 return f"[{addr}]:{port}"
212 else:
213 return f"{addr}:{port}"
214 215 def format_sock(sock, *, local):
216 '''
217 Format either local or remote side of a socket to a human readable string, e.g.
218 1.2.3.4:8333 or
219 [11:22::33]:8333 or
220 /path/to/socket or
221 @abstract-socket
222 '''
223 try:
224 if local:
225 name = sock.getsockname()
226 else:
227 name = sock.getpeername()
228 except Exception:
229 return "n/a"
230 231 if sock.family == socket.AF_INET:
232 return f"{name[0]}:{name[1]}"
233 234 if sock.family == socket.AF_INET6:
235 return f"[{name[0]}]:{name[1]}"
236 237 if sock.family == socket.AF_UNIX:
238 if isinstance(name, bytes):
239 name = name.decode(errors="backslashreplace")
240 if name.startswith("\0"):
241 return f"@{name[1:]}"
242 return name
243 244 return str(name)
245 246 247 def set_ephemeral_port_range(sock):
248 '''On FreeBSD, set socket to use the high ephemeral port range (49152-65535).
249 250 FreeBSD's default ephemeral port range (10000-65535) overlaps with the test
251 framework's static port range starting at TEST_RUNNER_PORT_MIN (default=11000).
252 Using IP_PORTRANGE_HIGH avoids this overlap when binding to port 0 for dynamic
253 port allocation.
254 '''
255 if sys.platform.startswith('freebsd'):
256 # Constants from FreeBSD's netinet/in.h and netinet6/in6.h
257 IP_PORTRANGE = 19
258 IPV6_PORTRANGE = 14
259 IP_PORTRANGE_HIGH = 1 # Same value for both IPv4 and IPv6
260 if sock.family == socket.AF_INET6:
261 sock.setsockopt(socket.IPPROTO_IPV6, IPV6_PORTRANGE, IP_PORTRANGE_HIGH)
262 else:
263 sock.setsockopt(socket.IPPROTO_IP, IP_PORTRANGE, IP_PORTRANGE_HIGH)
264