sshmitm.multisocket module

Utility functions to create server sockets able to listen on both IPv4 and IPv6.

Source: https://code.activestate.com/recipes/578504-server-supporting-ipv4-and-ipv6/ Inspired by: http://bugs.python.org/issue17561

Expected usage:

>>> sock = create_server_sock(("", 8000))
>>> if not has_dual_stack(sock):
...     sock.close()
...     sock = MultipleSocketsListener([("0.0.0.0", 8000), ("::", 8000)])
>>>

From here on you have a socket which listens on port 8000, all interfaces, serving both IPv4 and IPv6. You can start accepting new connections as usual:

>>> while True:
...     conn, addr = sock.accept()
...     # handle new connection

Supports UNIX, Windows, non-blocking sockets and socket timeouts. Works with Python >= 2.6 and 3.X.

class sshmitm.multisocket.MultipleSocketsListener(addresses: List[Tuple[str, int]], family: Optional[AddressFamily] = None, reuse_addr: Optional[bool] = None, transparent: bool = False, queue_size: int = 5)

Bases: object

Listen on multiple addresses specified as a list of (host, port) tuples. Useful to listen on both IPv4 and IPv6 on those systems where a dual stack is not supported natively (Windows and many UNIXes).

The returned instance is a socket-like object which can be used to accept() new connections, as with a common socket. Calls like settimeout() and setsockopt() will be applied to all sockets. Calls like gettimeout() or getsockopt() will refer to the first socket in the list.

accept() → Tuple[socket, Any]

Accept a connection from the first socket which is ready to do so.

close() → None

Close all registered sockets.

filenos() → List[int]

Return sockets’ file descriptors as a list of integers. This is useful with select().

getsockname() → Any

Return first registered socket’s own address.

getsockopt(level: int, optname: int) → int

getsockopt(level: int, optname: int, buflen: int) → bytes

Return first registered socket’s options.

gettimeout() → Optional[float]

Return first registered socket’s timeout.

setblocking(flag: bool) → None

Set non/blocking mode for all registered sockets.

setsockopt(level: int, optname: int, value: Union[int, bytes], optlen: None) → None

setsockopt(level: int, optname: int, value: None, optlen: int) → None

Set option for all registered sockets.

settimeout(timeout: float) → None

Set timeout for all registered sockets.

shutdown(how: int) → None

Shut down all registered sockets.

sshmitm.multisocket.create_server_sock(address: Tuple[str, int], family: Optional[AddressFamily] = None, reuse_addr: Optional[bool] = None, transparent: bool = False, queue_size: int = 5, dual_stack: bool = True) → socket

Convenience function which creates a TCP server bound to address and return the socket object.

Internally it takes care of choosing the right address family (IPv4 or IPv6) depending on the host specified in address (a (host, port) tuple. If host is an empty string or None all interfaces are assumed and if dual stack is supported by kernel the socket will be able to listen for both IPv4 and IPv6 connections.

family can be set to either AF_INET or AF_INET6 to force the socket to use IPv4 or IPv6. If not set it will be determined from host.

reuse_addr tells the kernel to reuse a local socket in TIME_WAIT state, without waiting for its natural timeout to expire. If not set will default to True on POSIX.

queue_size is the maximum number of queued connections passed to listen() (defaults to 5).

If dual_stack if True it will force the socket to listen on both IPv4 and IPv6 connections (defaults to True on all platforms natively supporting this functionality).

The returned socket can be used to accept() new connections as in:

>>> server = create_server_sock((None, 8000))
>>> while True:
...     sock, addr = server.accept()
...     # handle new sock connection

sshmitm.multisocket.has_dual_stack(sock: Optional[socket] = None) → bool

Return True if kernel allows creating a socket which is able to listen for both IPv4 and IPv6 connections. If sock is provided the check is made against it.