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, family=None, reuse_addr=None, transparent=False, queue_size=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.

Parameters:

  • addresses (List[Tuple[str, int]]) –

  • family (Optional[AddressFamily], default: None) –

  • reuse_addr (Optional[bool], default: None) –

  • transparent (bool, default: False) –

  • queue_size (int, default: 5) –

accept()

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

Return type:

Tuple[socket, Any]

close()

Close all registered sockets.

Return type:

None

filenos()

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

Return type:

List[int]

getsockname()

Return first registered socket’s own address.

Return type:

Any

getsockopt(level, optname, buflen=0)

Return first registered socket’s options.

Parameters:

  • level (int) –

  • optname (int) –

  • buflen (int, default: 0) –

Return type:

Union[int, bytes]

gettimeout()

Return first registered socket’s timeout.

Return type:

Optional[float]

setblocking(flag)

Set non/blocking mode for all registered sockets.

Parameters:

flag (bool) –

Return type:

None

setsockopt(level, optname, value, optlen)

Set option for all registered sockets.

Parameters:

  • level (int) –

  • optname (int) –

  • value (Union[int, bytes, None]) –

  • optlen (Optional[int]) –

Return type:

None

settimeout(timeout)

Set timeout for all registered sockets.

Parameters:

timeout (float) –

Return type:

None

shutdown(how)

Shut down all registered sockets.

Parameters:

how (int) –

Return type:

None

sshmitm.multisocket.create_server_sock(address, family=None, reuse_addr=None, transparent=False, queue_size=5, dual_stack=True)

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
Parameters:

  • address (Tuple[str, int]) –

  • family (Optional[AddressFamily], default: None) –

  • reuse_addr (Optional[bool], default: None) –

  • transparent (bool, default: False) –

  • queue_size (int, default: 5) –

  • dual_stack (bool, default: True) –

Return type:

socket

sshmitm.multisocket.has_dual_stack(sock=None)

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.

Parameters:

sock (Optional[socket], default: None) –

Return type:

bool