## # .python.socket - additional tools for working with sockets ## import sys import os import random import socket import errno __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', '') @property def _security_context(self): if self._security_context_ii is None: from ssl import SSLContext, PROTOCOL_TLS_CLIENT ctx = self._security_context_ii = SSLContext(PROTOCOL_TLS_CLIENT) ctx.check_hostname = False cf = self.socket_secure.get('certfile') kf = self.socket_secure.get('keyfile') if cf is not None: self._security_context_ii.load_cert_chain(cf, keyfile=kf) ca = self.socket_secure.get('ca_certs') if ca is not None: self._security_context_ii.load_verify_locations(ca) return self._security_context_ii def secure(self, socket: socket.socket): """ Secure a socket with SSL. """ return self._security_context.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, socket_connect, socket_secure = None, socket_security_context = None ): self._security_context_ii = socket_security_context self.socket_create = socket_create self.socket_connect = socket_connect self.socket_secure = socket_secure or {} def __str__(self): return 'socket' + repr(self.socket_connect) def find_available_port( interface = 'localhost', address_family = socket.AF_INET, ): """ Find an available port on the given interface for the given address family. """ port = None s = socket.socket(address_family, socket.SOCK_STREAM,) try: s.bind(('localhost', 0)) port = s.getsockname()[1] finally: s.close() return port