exabgp.reactor.network.tcp.nagle() - Code Metrics - Inspection of "should help with #942 and change to connection cod..." - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( d75605...cff8bf )

by Thomas

created 2019-12-27 14:30 UTC

exabgp.reactor.network.tcp.nagle() A

↳ Parent: exabgp.reactor.network.tcp

Complexity

Conditions

Size

Total Lines	6
Code Lines	5

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	2
eloc	5
nop	2
dl	0
loc	6
rs	10
c	0
b	0
f	0

# encoding: utf-8
"""
tcp.py

Created by Thomas Mangin on 2013-07-13.
Copyright (c) 2013-2017 Exa Networks. All rights reserved.
License: 3-clause BSD. (See the COPYRIGHT file)
"""

import re
import base64
import socket
import select
import platform

from struct import pack,calcsize

from exabgp.util import bytes_ascii
from exabgp.util.errstr import errstr

from exabgp.protocol.family import AFI
from exabgp.protocol.ip import IP
from exabgp.reactor.network.error import errno
from exabgp.reactor.network.error import error

from exabgp.reactor.network.error import NotConnected
from exabgp.reactor.network.error import BindingError
from exabgp.reactor.network.error import MD5Error
from exabgp.reactor.network.error import NagleError
from exabgp.reactor.network.error import TTLError
from exabgp.reactor.network.error import AsyncError

from exabgp.logger import Logger


def create (afi):
	try:
		if afi == AFI.ipv4:
			io = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
		if afi == AFI.ipv6:
			io = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
		try:
			io.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

		except (socket.error,AttributeError):
			pass
		try:
			io.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1)  # pylint: disable=E1101
		except (socket.error,AttributeError):
			pass
	except socket.error:
		raise NotConnected('Could not create socket')
	return io


def bind (io, ip, afi):
	try:
		if afi == AFI.ipv4:
			io.bind((ip,0))
		if afi == AFI.ipv6:
			io.bind((ip,0,0,0))
	except socket.error as exc:
		raise BindingError('Could not bind to local ip %s - %s' % (ip,str(exc)))


def connect (io, ip, port, afi, md5):
	try:
		if afi == AFI.ipv4:
			io.connect((ip,port))
		if afi == AFI.ipv6:
			io.connect((ip,port,0,0))
	except socket.error as exc:
		if exc.errno == errno.EINPROGRESS:
			return
		if md5:
			raise NotConnected('Could not connect to peer %s:%d, check your MD5 password (%s)' % (ip,port,errstr(exc)))
		raise NotConnected('Could not connect to peer %s:%d (%s)' % (ip,port,errstr(exc)))


# http://lxr.free-electrons.com/source/include/uapi/linux/tcp.h#L197
#
# #define TCP_MD5SIG_MAXKEYLEN    80
#
# struct tcp_md5sig {
# 	struct __kernel_sockaddr_storage tcpm_addr;     /* address associated */  128
# 	__u16   __tcpm_pad1;                            /* zero */                  2
# 	__u16   tcpm_keylen;                            /* key length */            2
# 	__u32   __tcpm_pad2;                            /* zero */                  4
# 	__u8    tcpm_key[TCP_MD5SIG_MAXKEYLEN];         /* key (binary) */         80
# }
#
# #define _K_SS_MAXSIZE   128
#
# #define _K_SS_ALIGNSIZE (__alignof__ (struct sockaddr *))
# /* Implementation specific desired alignment */
#
# typedef unsigned short __kernel_sa_family_t;
#
# struct __kernel_sockaddr_storage {
# 	__kernel_sa_family_t    ss_family;              /* address family */
# 	/* Following field(s) are implementation specific */
# 	char    __data[_K_SS_MAXSIZE - sizeof(unsigned short)];
# 	/* space to achieve desired size, */
# 	/* _SS_MAXSIZE value minus size of ss_family */
# } __attribute__ ((aligned(_K_SS_ALIGNSIZE)));   /* force desired alignment */

def MD5 (io, ip, port, md5, md5_base64):
	platform_os = platform.system()
	if platform_os == 'FreeBSD':
		if md5:
			if md5 != 'kernel':
				raise MD5Error(
					'FreeBSD requires that you set your MD5 key via ipsec.conf.\n'
					'Something like:\n'
					'flush;\n'
					'add <local ip> <peer ip> tcp 0x1000 -A tcp-md5 "password";'
					)
			try:
				TCP_MD5SIG = 0x10
				io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, 1)
			except socket.error:
				raise MD5Error(
					'FreeBSD requires that you rebuild your kernel to enable TCP MD5 Signatures:\n'
					'options         IPSEC\n'
					'options         TCP_SIGNATURE\n'
					'device          crypto\n'
				)
	elif platform_os == 'Linux':
		try:
			md5_bytes = None
			if md5:
				if md5_base64 is True:
					try:
						md5_bytes = base64.b64decode(md5)
					except TypeError:
						raise MD5Error("Failed to decode base 64 encoded PSK")
				elif md5_base64 is None and not re.match('.*[^a-f0-9].*', md5):  # auto
					options = [md5+'==', md5+'=', md5]
					for md5 in options:
						try:
							md5_bytes = base64.b64decode(md5)
							break
						except TypeError:
							pass

			# __kernel_sockaddr_storage
			n_af   = IP.toaf(ip)
			n_addr = IP.pton(ip)
			n_port = socket.htons(port)

			# pack 'x' is padding, so we want the struct
			# Do not use '!' for the pack, the network (big) endian switch in
			# struct.pack is fighting against inet_pton and htons (note the n)

			if IP.toafi(ip) == AFI.ipv4:
				# SS_MAXSIZE is 128 but addr_family, port and ipaddr (8 bytes total) are written independently of the padding
				SS_MAXSIZE_PADDING = 128 - calcsize('HH4s')  # 8
				sockaddr = pack('HH4s%dx' % SS_MAXSIZE_PADDING, socket.AF_INET, n_port, n_addr)
			else:
				SS_MAXSIZE_PADDING = 128 - calcsize('HI16sI')  # 28
				SIN6_FLOWINFO = 0
				SIN6_SCOPE_ID = 0
				sockaddr = pack('HHI16sI%dx' % SS_MAXSIZE_PADDING, n_af, n_port, SIN6_FLOWINFO, n_addr, SIN6_SCOPE_ID)

			TCP_MD5SIG_MAXKEYLEN = 80
			TCP_MD5SIG = 14

			if md5_bytes:
				key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, len(md5_bytes), md5_bytes)
				io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)
			elif md5:
				md5_bytes = bytes_ascii(md5)
				key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, len(md5_bytes), md5_bytes)
				io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)
			# else:
			# 	key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, 0, b'')
			# 	io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)

		except socket.error as exc:
			if exc.errno != errno.ENOENT:
				raise MD5Error('This linux machine does not support TCP_MD5SIG, you can not use MD5 (%s)' % errstr(exc))
	elif md5:
		raise MD5Error('ExaBGP has no MD5 support for %s' % platform_os)


def nagle (io, ip):
	try:
		# diable Nagle's algorithm (no grouping of packets)
		io.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
	except (socket.error,AttributeError):
		raise NagleError("Could not disable nagle's algorithm for %s" % ip)


def TTL (io, ip, ttl):
	# None (ttl-security unset) or zero (maximum TTL) is the same thing
	if ttl:
		try:
			io.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
		except socket.error as exc:
			raise TTLError('This OS does not support IP_TTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))


def TTLv6 (io, ip, ttl):
	if ttl:
		try:
			io.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_UNICAST_HOPS, ttl)
		except socket.error as exc:
			raise TTLError('This OS does not support unicast_hops (ttl-security) for %s (%s)' % (ip,errstr(exc)))


def MIN_TTL (io, ip, ttl):
	# None (ttl-security unset) or zero (maximum TTL) is the same thing
	if ttl:
		try:
			io.setsockopt(socket.IPPROTO_IP, socket.IP_MINTTL, ttl)
		except socket.error as exc:
			raise TTLError('This OS does not support IP_MINTTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))
		except AttributeError:
			pass

		try:
			io.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
		except socket.error as exc:
			raise TTLError('This OS does not support IP_MINTTL or IP_TTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))


def asynchronous(io, ip):
	try:
		io.setblocking(0)
	except socket.error as exc:
		raise AsyncError('could not set socket non-blocking for %s (%s)' % (ip,errstr(exc)))


def ready (io):
	logger = Logger()

	poller = select.poll()
	poller.register(io, select.POLLOUT | select.POLLNVAL | select.POLLERR)

	found = False

	while True:
		try:
			for _, event in poller.poll(0):
				if event & select.POLLOUT or event & select.POLLIN:
					found = True
				elif event & select.POLLHUP or event & select.POLLRDHUP:
					yield False, 'could not connect, retrying'
					return
				elif event & select.POLLERR or event & select.POLLNVAL:
					yield False, 'connect attempt failed, issue with reading on the network, retrying'
					return

			if found:
				err = io.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
				if not err:
					yield True, 'connection established'
					return
				elif err in error.block:
					yield False, 'connect attempt failed, retrying, reason %s' % errno.errorcode[err]
					return
			yield False, 'waiting for socket to become ready'
		except select.error as err:
			yield False, 'error, retrying %s' % str(err)
			return

1			# encoding: utf-8
2			"""
3			tcp.py
4
5			Created by Thomas Mangin on 2013-07-13.
6			Copyright (c) 2013-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			import re
11			import base64
12			import socket
13			import select
14			import platform
15
16			from struct import pack,calcsize
17
18			from exabgp.util import bytes_ascii
19			from exabgp.util.errstr import errstr
20
21			from exabgp.protocol.family import AFI
22			from exabgp.protocol.ip import IP
23			from exabgp.reactor.network.error import errno
24			from exabgp.reactor.network.error import error
25
26			from exabgp.reactor.network.error import NotConnected
27			from exabgp.reactor.network.error import BindingError
28			from exabgp.reactor.network.error import MD5Error
29			from exabgp.reactor.network.error import NagleError
30			from exabgp.reactor.network.error import TTLError
31			from exabgp.reactor.network.error import AsyncError
32
33			from exabgp.logger import Logger
34
35
36			def create (afi):
37			try:
38			if afi == AFI.ipv4:
39			io = socket.socket(socket.AF_INET, socket.SOCK_STREAM, socket.IPPROTO_TCP)
40			if afi == AFI.ipv6:
41			io = socket.socket(socket.AF_INET6, socket.SOCK_STREAM, socket.IPPROTO_TCP)
42			try:
43			io.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
			0 ignored issues – show introduced 2019-07-17 07:30 UTC by Report Bug Copy Issue Report The variable `io` does not seem to be defined in case `afi == AFI.ipv4` on line `38` is `False`. Are you sure this can never be the case? Loading history...
44			except (socket.error,AttributeError):
45			pass
46			try:
47			io.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) # pylint: disable=E1101
48			except (socket.error,AttributeError):
49			pass
50			except socket.error:
51			raise NotConnected('Could not create socket')
52			return io
53
54
55			def bind (io, ip, afi):
56			try:
57			if afi == AFI.ipv4:
58			io.bind((ip,0))
59			if afi == AFI.ipv6:
60			io.bind((ip,0,0,0))
61			except socket.error as exc:
62			raise BindingError('Could not bind to local ip %s - %s' % (ip,str(exc)))
63
64
65			def connect (io, ip, port, afi, md5):
66			try:
67			if afi == AFI.ipv4:
68			io.connect((ip,port))
69			if afi == AFI.ipv6:
70			io.connect((ip,port,0,0))
71			except socket.error as exc:
72			if exc.errno == errno.EINPROGRESS:
73			return
74			if md5:
75			raise NotConnected('Could not connect to peer %s:%d, check your MD5 password (%s)' % (ip,port,errstr(exc)))
76			raise NotConnected('Could not connect to peer %s:%d (%s)' % (ip,port,errstr(exc)))
77
78
79			# http://lxr.free-electrons.com/source/include/uapi/linux/tcp.h#L197
80			#
81			# #define TCP_MD5SIG_MAXKEYLEN 80
82			#
83			# struct tcp_md5sig {
84			# struct __kernel_sockaddr_storage tcpm_addr; /* address associated */ 128
85			# __u16 __tcpm_pad1; /* zero */ 2
86			# __u16 tcpm_keylen; /* key length */ 2
87			# __u32 __tcpm_pad2; /* zero */ 4
88			# __u8 tcpm_key[TCP_MD5SIG_MAXKEYLEN]; /* key (binary) */ 80
89			# }
90			#
91			# #define _K_SS_MAXSIZE 128
92			#
93			# #define _K_SS_ALIGNSIZE (__alignof__ (struct sockaddr *))
94			# /* Implementation specific desired alignment */
95			#
96			# typedef unsigned short __kernel_sa_family_t;
97			#
98			# struct __kernel_sockaddr_storage {
99			# __kernel_sa_family_t ss_family; /* address family */
100			# /* Following field(s) are implementation specific */
101			# char __data[_K_SS_MAXSIZE - sizeof(unsigned short)];
102			# /* space to achieve desired size, */
103			# /* _SS_MAXSIZE value minus size of ss_family */
104			# } __attribute__ ((aligned(_K_SS_ALIGNSIZE))); /* force desired alignment */
105
106			def MD5 (io, ip, port, md5, md5_base64):
107			platform_os = platform.system()
108			if platform_os == 'FreeBSD':
109			if md5:
110			if md5 != 'kernel':
111			raise MD5Error(
112			'FreeBSD requires that you set your MD5 key via ipsec.conf.\n'
113			'Something like:\n'
114			'flush;\n'
115			'add <local ip> <peer ip> tcp 0x1000 -A tcp-md5 "password";'
116			)
117			try:
118			TCP_MD5SIG = 0x10
119			io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, 1)
120			except socket.error:
121			raise MD5Error(
122			'FreeBSD requires that you rebuild your kernel to enable TCP MD5 Signatures:\n'
123			'options IPSEC\n'
124			'options TCP_SIGNATURE\n'
125			'device crypto\n'
126			)
127			elif platform_os == 'Linux':
128			try:
129			md5_bytes = None
130			if md5:
131			if md5_base64 is True:
132			try:
133			md5_bytes = base64.b64decode(md5)
134			except TypeError:
135			raise MD5Error("Failed to decode base 64 encoded PSK")
136			elif md5_base64 is None and not re.match('.[^a-f0-9].', md5): # auto
137			options = [md5+'==', md5+'=', md5]
138			for md5 in options:
139			try:
140			md5_bytes = base64.b64decode(md5)
141			break
142			except TypeError:
143			pass
144
145			# __kernel_sockaddr_storage
146			n_af = IP.toaf(ip)
147			n_addr = IP.pton(ip)
148			n_port = socket.htons(port)
149
150			# pack 'x' is padding, so we want the struct
151			# Do not use '!' for the pack, the network (big) endian switch in
152			# struct.pack is fighting against inet_pton and htons (note the n)
153
154			if IP.toafi(ip) == AFI.ipv4:
155			# SS_MAXSIZE is 128 but addr_family, port and ipaddr (8 bytes total) are written independently of the padding
156			SS_MAXSIZE_PADDING = 128 - calcsize('HH4s') # 8
157			sockaddr = pack('HH4s%dx' % SS_MAXSIZE_PADDING, socket.AF_INET, n_port, n_addr)
158			else:
159			SS_MAXSIZE_PADDING = 128 - calcsize('HI16sI') # 28
160			SIN6_FLOWINFO = 0
161			SIN6_SCOPE_ID = 0
162			sockaddr = pack('HHI16sI%dx' % SS_MAXSIZE_PADDING, n_af, n_port, SIN6_FLOWINFO, n_addr, SIN6_SCOPE_ID)
163
164			TCP_MD5SIG_MAXKEYLEN = 80
165			TCP_MD5SIG = 14
166
167			if md5_bytes:
168			key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, len(md5_bytes), md5_bytes)
169			io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)
170			elif md5:
171			md5_bytes = bytes_ascii(md5)
172			key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, len(md5_bytes), md5_bytes)
173			io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)
174			# else:
175			# key = pack('2xH4x%ds' % TCP_MD5SIG_MAXKEYLEN, 0, b'')
176			# io.setsockopt(socket.IPPROTO_TCP, TCP_MD5SIG, sockaddr + key)
177
178			except socket.error as exc:
179			if exc.errno != errno.ENOENT:
180			raise MD5Error('This linux machine does not support TCP_MD5SIG, you can not use MD5 (%s)' % errstr(exc))
181			elif md5:
182			raise MD5Error('ExaBGP has no MD5 support for %s' % platform_os)
183
184
185			def nagle (io, ip):
186			try:
187			# diable Nagle's algorithm (no grouping of packets)
188			io.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
189			except (socket.error,AttributeError):
190			raise NagleError("Could not disable nagle's algorithm for %s" % ip)
191
192
193			def TTL (io, ip, ttl):
194			# None (ttl-security unset) or zero (maximum TTL) is the same thing
195			if ttl:
196			try:
197			io.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
198			except socket.error as exc:
199			raise TTLError('This OS does not support IP_TTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))
200
201
202			def TTLv6 (io, ip, ttl):
203			if ttl:
204			try:
205			io.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_UNICAST_HOPS, ttl)
206			except socket.error as exc:
207			raise TTLError('This OS does not support unicast_hops (ttl-security) for %s (%s)' % (ip,errstr(exc)))
208
209
210			def MIN_TTL (io, ip, ttl):
211			# None (ttl-security unset) or zero (maximum TTL) is the same thing
212			if ttl:
213			try:
214			io.setsockopt(socket.IPPROTO_IP, socket.IP_MINTTL, ttl)
215			except socket.error as exc:
216			raise TTLError('This OS does not support IP_MINTTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))
217			except AttributeError:
218			pass
219
220			try:
221			io.setsockopt(socket.IPPROTO_IP, socket.IP_TTL, ttl)
222			except socket.error as exc:
223			raise TTLError('This OS does not support IP_MINTTL or IP_TTL (ttl-security) for %s (%s)' % (ip,errstr(exc)))
224
225
226			def asynchronous(io, ip):
227			try:
228			io.setblocking(0)
229			except socket.error as exc:
230			raise AsyncError('could not set socket non-blocking for %s (%s)' % (ip,errstr(exc)))
231
232
233			def ready (io):
234			logger = Logger()
235
236			poller = select.poll()
237			poller.register(io, select.POLLOUT \| select.POLLNVAL \| select.POLLERR)
238
239			found = False
240
241			while True:
242			try:
243			for _, event in poller.poll(0):
244			if event & select.POLLOUT or event & select.POLLIN:
245			found = True
246			elif event & select.POLLHUP or event & select.POLLRDHUP:
247			yield False, 'could not connect, retrying'
248			return
249			elif event & select.POLLERR or event & select.POLLNVAL:
250			yield False, 'connect attempt failed, issue with reading on the network, retrying'
251			return
252
253			if found:
254			err = io.getsockopt(socket.SOL_SOCKET, socket.SO_ERROR)
255			if not err:
256			yield True, 'connection established'
257			return
258			elif err in error.block:
259			yield False, 'connect attempt failed, retrying, reason %s' % errno.errorcode[err]
260			return
261			yield False, 'waiting for socket to become ready'
262			except select.error as err:
263			yield False, 'error, retrying %s' % str(err)
264			return

Exa-Networks / exabgp

Push — master ( d75605...cff8bf )

exabgp.reactor.network.tcp.nagle() A

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like