Repository navigation
Expand file tree
/
Copy pathsocket.py
More file actions
119 lines (103 loc) · 3.03 KB
/
Copy pathsocket.py
File metadata and controls
119 lines (103 loc) · 3.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
##
# .python.socket - additional tools for working with sockets
##
import sys
import os
import random
import socket
import math
import errno
import ssl
__all__ = ['find_available_port', 'SocketFactory']
class SocketFactory(object):
"""
Object used to create a socket and connect it.
This is, more or less, a specialized partial() for socket creation.
Additionally, it provides methods and attributes for abstracting
exception management on socket operation.
"""
timeout_exception = socket.timeout
fatal_exception = socket.error
try_again_exception = socket.error
def timed_out(self, err) -> bool:
return err.__class__ is self.timeout_exception
@staticmethod
def try_again(err, codes = (errno.EAGAIN, errno.EINTR, errno.EWOULDBLOCK, errno.ETIMEDOUT)) -> bool:
"""
Does the error indicate that the operation should be tried again?
More importantly, the connection is *not* dead.
"""
errno = getattr(err, 'errno', None)
if errno is None:
return False
return errno in codes
@classmethod
def fatal_exception_message(typ, err) -> (str, None):
"""
If the exception was fatal to the connection,
what message should be given to the user?
"""
if typ.try_again(err):
return None
strerr = getattr(err, 'strerror', None)
if strerr is not None:
return str(strerr)
return os.strerror(err.errno)
def secure(self, socket : socket.socket) -> ssl.SSLSocket:
"secure a socket with SSL"
if self.socket_secure is not None:
return ssl.wrap_socket(socket, **self.socket_secure)
else:
return ssl.wrap_socket(socket)
def __call__(self, timeout = None):
s = socket.socket(*self.socket_create)
try:
s.settimeout(float(timeout) if timeout is not None else None)
s.connect(self.socket_connect)
s.settimeout(None)
except Exception:
s.close()
raise
return s
def __init__(self,
socket_create : "positional parameters given to socket.socket()",
socket_connect : "parameter given to socket.connect()",
socket_secure : "keywords given to ssl.wrap_socket" = None,
):
self.socket_create = socket_create
self.socket_connect = socket_connect
self.socket_secure = socket_secure
def __str__(self):
return 'socket' + repr(self.socket_connect)
def find_available_port(
interface : "attempt to bind to interface" = 'localhost',
address_family : "address family to use (default: AF_INET)" = socket.AF_INET,
limit : "Number tries to make before giving up" = 1024,
port_range = (6600, 56600)
) -> (int, None):
"""
Find an available port on the given interface for the given address family.
Returns a port number that was successfully bound to or `None` if the
attempt limit was reached.
"""
i = 0
while i < limit:
i += 1
port = (
math.floor(
random.random() * (port_range[1] - port_range[0])
) + port_range[0]
)
s = socket.socket(address_family, socket.SOCK_STREAM,)
try:
s.bind(('localhost', port))
s.close()
except socket.error as e:
s.close()
if e.errno in (errno.EACCES, errno.EADDRINUSE, errno.EINTR):
# try again
continue
break
else:
port = None
return port