/usr/lib/python3/dist-packages/postgresql/python/socket.py is in python3-postgresql 1.1.0-2build2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
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 | ##
# .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
return getattr(err, 'strerror', '<strerror not present>')
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
|