Socket Helpers

Network socket helper module.

---

Data Structures And Constants

Typed Attributes

class easynetwork.lowlevel.socket.SocketAttribute

Bases: TypedAttributeSet

Typed attributes which can be used on an endpoint or a transport.

socket: ISocket = <object object>

socket.socket instance.

family: int = <object object>

the socket’s family, as returned by socket.socket.family.

sockname: _RetAddress = <object object>

the socket’s own address, result of socket.socket.getsockname().

peername: _RetAddress = <object object>

the remote address to which the socket is connected, result of socket.socket.getpeername().

class easynetwork.lowlevel.socket.INETSocketAttribute

Bases: SocketAttribute

Typed attributes which can be used on an endpoint or a transport.

family: Literal[_socket.AddressFamily.AF_INET, _socket.AddressFamily.AF_INET6] = <object object>

the socket’s family, as returned by socket.socket.family.

sockname: tuple[str, int] | tuple[str, int, int, int] = <object object>

the socket’s own address, result of socket.socket.getsockname().

peername: tuple[str, int] | tuple[str, int, int, int] = <object object>

the remote address to which the socket is connected, result of socket.socket.getpeername().

class easynetwork.lowlevel.socket.UNIXSocketAttribute

Bases: SocketAttribute

Typed attributes which can be used on an endpoint or a transport.

Added in version 1.1.

family: Literal[_socket.AddressFamily.AF_UNIX] = <object object>

the socket’s family, as returned by socket.socket.family.

sockname: RawUnixSocketAddress = <object object>

the socket’s own address, result of socket.socket.getsockname().

peername: RawUnixSocketAddress = <object object>

the remote address to which the socket is connected, result of socket.socket.getpeername().

class easynetwork.lowlevel.socket.TLSAttribute

Bases: TypedAttributeSet

Typed attributes which can be used on an endpoint or a transport.

sslcontext: SSLContext = <object object>

ssl.SSLContext instance.

peercert: _PeerCertRetDictType = <object object>

peer certificate; result of ssl.SSLSocket.getpeercert().

cipher: tuple[str, str, int] = <object object>

a three-value tuple containing the name of the cipher being used, the version of the SSL protocol that defines its use, and the number of secret bits being used; result of ssl.SSLSocket.cipher().

compression: str = <object object>

the compression algorithm being used as a string, or None if the connection isn’t compressed; result of ssl.SSLSocket.compression().

standard_compatible: bool = <object object>

True if this stream does (and expects) a closing TLS handshake when the stream is being closed.

tls_version: str = <object object>

the TLS protocol version (e.g. TLSv1.2).

Internet addresses

namedtuple easynetwork.lowlevel.socket.IPv4SocketAddress

Bases: NamedTuple

An internet (IPv4) socket address.

Fields:

0.  host (str) – Alias for field number 0

1.  port (int) – Alias for field number 1

for_connection() → tuple[str, int]

Returns:

A pair of (host, port)

Return type:

tuple[str, int]

static __new__(_cls, host: str, port: int)

Create new instance of IPv4SocketAddress(host, port)

namedtuple easynetwork.lowlevel.socket.IPv6SocketAddress

Bases: NamedTuple

An internet (IPv6) socket address.

Fields:

0.  host (str) – Alias for field number 0

1.  port (int) – Alias for field number 1

2.  flowinfo (int) – Alias for field number 2

3.  scope_id (int) – Alias for field number 3

for_connection() → tuple[str, int]

Returns:

A pair of (host, port)

Return type:

tuple[str, int]

static __new__(_cls, host: str, port: int, flowinfo: int = 0, scope_id: int = 0)

Create new instance of IPv6SocketAddress(host, port, flowinfo, scope_id)

type easynetwork.lowlevel.socket.SocketAddress = IPv4SocketAddress | IPv6SocketAddress

An internet socket address, either IPv4 or IPv6.

Unix-related

type easynetwork.lowlevel.socket.RawUnixSocketAddress = str | bytes

An address associated with a Unix socket.

final class easynetwork.lowlevel.socket.UnixSocketAddress

Bases: object

An address associated with a Unix socket.

Added in version 1.1.

str(addr)

Displays the Unix address in a similar way to that seen in /proc/net/unix.

Changed in version 1.2: Prior to this version, str(addr) used to return something like repr(addr).

Note

Returns <unnamed> for unnamed addresses.

Example

>>> from easynetwork.lowlevel.socket import UnixSocketAddress
>>> str(UnixSocketAddress.from_pathname("/tmp/sock"))
'/tmp/sock'
>>> str(UnixSocketAddress.from_abstract_name(b"hidden"))
'@hidden'
>>> str(UnixSocketAddress()) # Unnamed
'<unnamed>'

__init__() → None

Constructs an unnamed socket address:

>>> addr = UnixSocketAddress()
>>> addr
UnixSocketAddress(<unnamed>)
>>> addr.is_unnamed()
True

classmethod from_pathname(path: str | PathLike[str]) → Self

Constructs a UnixSocketAddress with the provided path.

Examples

>>> addr = UnixSocketAddress.from_pathname("/path/to/socket")
>>> addr
UnixSocketAddress('/path/to/socket' <pathname>)
>>> addr.as_pathname()
PosixPath('/path/to/socket')

Creating a UnixSocketAddress with a NULL byte results in an error.

>>> addr = UnixSocketAddress.from_pathname("/path/with/\0/bytes")
Traceback (most recent call last):
...
ValueError: paths must not contain interior null bytes

Creating a UnixSocketAddress with an empty string results in an unnamed socket address.

>>> addr = UnixSocketAddress.from_pathname("")
>>> addr
UnixSocketAddress(<unnamed>)
>>> addr.is_unnamed()
True

Return type:

Self

classmethod from_abstract_name(name: str | bytes) → Self

Creates a Unix socket address in the abstract namespace.

The abstract namespace is a Linux-specific extension that allows Unix sockets to be bound without creating an entry in the filesystem. Abstract sockets are unaffected by filesystem layout or permissions, and no cleanup is necessary when the socket is closed.

An abstract socket address name may contain any bytes, including zero.

Example

>>> addr = UnixSocketAddress.from_abstract_name("hidden")
>>> addr
UnixSocketAddress(b'hidden' <abstract>)
>>> addr.as_abstract_name()
b'hidden'

Availability: Linux

Return type:

Self

classmethod from_raw(addr: Any) → Self

Constructs a UnixSocketAddress from the raw addr supplied by the socket.

• the result of socket.socket.getsockname() or socket.socket.getpeername();

• the address retrieved by socket.socket.recvfrom();

• etc.

Return type:

Self

is_unnamed() → bool

Returns True if the address is unnamed.

Examples

A named address:

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> s.bind("/tmp/sock")
>>> addr = UnixSocketAddress.from_raw(s.getsockname())
>>> addr.is_unnamed()
False

An unnamed address:

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
>>> addr = UnixSocketAddress.from_raw(s.getsockname())
>>> addr.is_unnamed()
True

Return type:

bool

as_pathname() → Path | None

Returns the contents of this address if it is a pathname address.

Examples

With a pathname:

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> s.bind("/tmp/sock")
>>> addr = UnixSocketAddress.from_raw(s.getsockname())
>>> addr.as_pathname()
PosixPath('/tmp/sock')

Without a pathname:

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
>>> addr = UnixSocketAddress.from_raw(s.getsockname())
>>> print(addr.as_pathname())
None

Return type:

Path | None

as_abstract_name() → bytes | None

Returns the contents of this address if it is in the abstract namespace.

Example

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> s.bind(b"\0hidden")
>>> addr = UnixSocketAddress.from_raw(s.getsockname())
>>> addr.as_abstract_name()
b'hidden'

Availability: Linux

Return type:

bytes | None

as_raw() → str | bytes

Returns the raw representation of this address.

Example

>>> UnixSocketAddress.from_pathname("/tmp/sock").as_raw()
'/tmp/sock'
>>> UnixSocketAddress.from_abstract_name(b"hidden").as_raw()
b'\x00hidden'
>>> UnixSocketAddress().as_raw() # Unnamed
''

Return type:

str | bytes

namedtuple easynetwork.lowlevel.socket.UnixCredentials

Bases: NamedTuple

The Unix credentials tuple of the connected process.

Added in version 1.1.

Fields:

0.  pid (Optional[int]) – Process ID of the peer process.

1.  uid (int) – Effective user ID of the peer process.

2.  gid (int) – Effective Group ID of the peer process.

static __new__(_cls, pid: int | None, uid: int, gid: int)

Create new instance of UnixCredentials(pid, uid, gid)

namedtuple easynetwork.lowlevel.socket.SocketCredential

Bases: NamedTuple

Unix credential tuple.

Availability: Linux, FreeBSD, NetBSD.

Added in version 1.2.

Fields:

0.  pid (int) – Process ID.

1.  uid (int) – Effective user ID.

2.  gid (int) – Effective Group ID.

static __new__(_cls, pid: int = 0, uid: int = 0, gid: int = 0)

Create new instance of SocketCredential(pid, uid, gid)

final class easynetwork.lowlevel.socket.SocketAncillary

Bases: object

A Unix socket Ancillary data object.

Added in version 1.2.

add_fds(fds: Iterable[int]) → None

Add file descriptors to the ancillary data.

Technically, that means this operation adds a control message with the level SOL_SOCKET and the type SCM_RIGHTS.

Parameters:

fds (Iterable[int]) – Any iterable of opened file descriptors.

add_creds(credentials: Iterable[SocketCredential]) → None

Add credentials to the ancillary data.

Parameters:

credentials (Iterable[SocketCredential]) – Any iterable of unix credentials.

Technically, that means this operation adds a control message with the level SOL_SOCKET and the type is equal to SCM_CREDENTIALS or SCM_CREDS or SCM_CREDS2.

Availability: Linux, FreeBSD, NetBSD.

messages() → Iterator[SCMRights | SCMCredentials]

Returns the iterator of the control messages.

Example:

s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.connect("/tmp/sock")
ancillary = SocketAncillary()

msg, ancdata, _, _ = s.recvmsg(1024, socket.CMSG_SPACE(4096))
ancillary.update_from_raw(ancdata)

for message in ancillary.messages():
    match message:
        case SCMRights(fds):
            for fd in fds:
                print(f"Received file descriptor: {fd}")
        case SCMCredentials(credentials):
            for ucred in credentials:
                print(f"Received unix credential: {ucred}")

Return type:

Iterator[SCMRights | SCMCredentials]

iter_fds() → Iterator[int]

Fast path to retrieve file descriptors within the messages.

Example:

s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.connect("/tmp/sock")
ancillary = SocketAncillary()

msg, ancdata, _, _ = s.recvmsg(1024, socket.CMSG_SPACE(4096))
ancillary.update_from_raw(ancdata)

for fd in ancillary.iter_fds():
    print(f"Received file descriptor: {fd}")

Return type:

Iterator[int]

update_from_raw(messages: Iterable[tuple[int, int, TypeAliasForwardRef('bytes | bytearray | memoryview | collections.abc.Buffer')]]) → None

Read raw socket control messages received from a unix socket.

Example

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> s.connect("/tmp/sock")
>>> ancillary = SocketAncillary()
>>> msg, ancdata, _, _ = s.recvmsg(1024, socket.CMSG_SPACE(4096))
>>> ancillary.update_from_raw(ancdata)
>>> list(ancillary.messages())
[SCMRights(fds=...)]

Parameters:

messages (Iterable[tuple[int, int, TypeAliasForwardRef('bytes | bytearray | memoryview | collections.abc.Buffer')]]) – Any iterable of socket control messages in the form (level, type, data).

Raises:

ExceptionGroup – Failed to parse some messages.

as_raw() → list[tuple[int, int, bytes]]

Create raw socket control messages to send to a unix socket.

Example

>>> import socket
>>> s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
>>> s.connect("/tmp/sock")
>>> ancillary = SocketAncillary()
>>> ancillary.add_fds([4, 5, 6])
>>> list(ancillary.messages())
[SCMRights(fds=...)]
>>> s.sendmsg([b"data"], ancillary.as_raw())

Returns:

A list of socket control messages in the form (level, type, data).

Return type:

list[tuple[int, int, bytes]]

clear() → None

Clears the ancillary data, removing all values.

type easynetwork.lowlevel.socket.SocketAncillaryMessages = SCMRights | SCMCredentials

Unix socket control messages.

final class easynetwork.lowlevel.socket.SCMCredentials

Bases: object

This control message contains unix credentials.

The level is equal to SOL_SOCKET and the type is equal to SCM_CREDENTIALS or SCM_CREDS or SCM_CREDS2.

Availability: Linux, FreeBSD, NetBSD.

Added in version 1.2.

credentials: Iterator[SocketCredential]

Socket credential iterator.

final class easynetwork.lowlevel.socket.SCMRights

Bases: object

This control message contains file descriptors.

The level is equal to SOL_SOCKET and the type is equal to SCM_RIGHTS.

Added in version 1.2.

fds: Iterator[int]

File descriptor iterator.

Classes

protocol easynetwork.lowlevel.socket.SupportsSocketOptions

Bases: Protocol

Interface for an object which support getting and setting socket options.

This protocol is runtime checkable.

Classes that implement this protocol must have the following methods / attributes:

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

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

Similar to socket.socket.getsockopt().

abstractmethod setsockopt(level: int, optname: int, value: int | bytes, /) → None

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

Similar to socket.socket.setsockopt().

protocol easynetwork.lowlevel.socket.ISocket

Bases: SupportsSocketOptions, Protocol

Interface for an object which mirrors a socket.

This protocol is runtime checkable.

Classes that implement this protocol must have the following methods / attributes:

abstractmethod fileno() → int

Similar to socket.socket.fileno().

Return type:

int

abstractmethod get_inheritable() → bool

Similar to socket.socket.get_inheritable().

Return type:

bool

abstractmethod getpeername() → Any

Similar to socket.socket.getpeername().

Return type:

Any

abstractmethod getsockname() → Any

Similar to socket.socket.getsockname().

Return type:

Any

abstract property family: int

Similar to socket.socket.family.

abstract property type: int

Similar to socket.socket.type.

abstract property proto: int

Similar to socket.socket.proto.

final class easynetwork.lowlevel.socket.SocketProxy

Bases: object

A socket-like wrapper for exposing real transport sockets.

These objects can be safely returned by APIs like client.socket. All potentially disruptive operations (like socket.socket.close()) are banned.

__init__(socket: ISocket, *, lock: Callable[[], threading.Lock | threading.RLock] | None = None, runner: Callable[[Callable[[], Any]], Any] | None = None) → None

Parameters:

• socket (ISocket) – The socket-like object to wrap.

• lock (Callable[[], threading.Lock | threading.RLock] | None) – A callback function to use when a lock is required to gain access to the wrapped socket.

• runner (Callable[[Callable[[], Any]], Any] | None) – A callback function to use to execute the socket method.

Warning

If lock is ommitted, the proxy object is not thread-safe.

runner can be used for concurrent call management.

Example

Examples of how ISocket.fileno() would be called according to lock and runner values.

Neither lock nor runner:

return socket.fileno()

lock but no runner:

with lock():
    return socket.fileno()

runner but no lock:

return runner(socket.fileno)

Both lock and runner:

with lock():
    return runner(socket.fileno)

fileno() → int

Calls ISocket.fileno().

Return type:

int

get_inheritable() → bool

Calls ISocket.get_inheritable().

Return type:

bool

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

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

Calls ISocket.getsockopt.

setsockopt(level: int, optname: int, value: int | bytes, /) → None

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

Calls ISocket.setsockopt.

getpeername() → Any

Calls ISocket.getpeername().

Return type:

Any

getsockname() → Any

Calls ISocket.getsockname().

Return type:

Any

property family: int

The socket family.

property type: int

The socket type.

property proto: int

The socket protocol.

Functions

Internet addresses

easynetwork.lowlevel.socket.new_socket_address(addr: tuple[str, int], family: Literal[AddressFamily.AF_INET]) → IPv4SocketAddress

easynetwork.lowlevel.socket.new_socket_address(addr: tuple[str, int] | tuple[str, int, int, int], family: Literal[AddressFamily.AF_INET6]) → IPv6SocketAddress

easynetwork.lowlevel.socket.new_socket_address(addr: tuple[Any, ...], family: int) → IPv4SocketAddress | IPv6SocketAddress

Factory to create a SocketAddress from addr.

Example

>>> import socket
>>> new_socket_address(("127.0.0.1", 12345), socket.AF_INET)
IPv4SocketAddress(host='127.0.0.1', port=12345)
>>> new_socket_address(("::1", 12345), socket.AF_INET6)
IPv6SocketAddress(host='::1', port=12345, flowinfo=0, scope_id=0)
>>> new_socket_address(("::1", 12345, 12, 345), socket.AF_INET6)
IPv6SocketAddress(host='::1', port=12345, flowinfo=12, scope_id=345)
>>> new_socket_address(("127.0.0.1", 12345), socket.AF_APPLETALK)
Traceback (most recent call last):
...
ValueError: Unsupported address family <AddressFamily.AF_APPLETALK: 5>

Parameters:

• addr – The address in the form (host, port) or (host, port, flow, scope_id).

• family – The socket family.

Raises:

• ValueError – Invalid family.

• TypeError – Invalid addr.

Returns:

a SocketAddress named tuple.

TCP Options

easynetwork.lowlevel.socket.set_tcp_nodelay(sock: SupportsSocketOptions, state: bool) → None

Enables/Disable Nagle’s algorithm on a TCP socket.

This is equivalent to:

sock.setsockopt(IPPROTO_TCP, TCP_NODELAY, state)

except that if socket.TCP_NODELAY is not defined, it is silently ignored.

Parameters:

• sock (SupportsSocketOptions) – The socket.

• state (bool) – True to disable, False to enable.

Note

Modern operating systems enable it by default.

easynetwork.lowlevel.socket.set_tcp_keepalive(sock: SupportsSocketOptions, state: bool) → None

Enables/Disable keep-alive protocol on a TCP socket.

This is equivalent to:

sock.setsockopt(SOL_SOCKET, SO_KEEPALIVE, state)

except that if socket.SO_KEEPALIVE is not defined, it is silently ignored.

Parameters:

• sock (SupportsSocketOptions) – The socket.

• state (bool) – True to enable, False to disable.

Note

Modern operating systems enable it by default.

Socket Lingering

namedtuple easynetwork.lowlevel.socket.socket_linger

Bases: NamedTuple

The socket linger tuple returned by get_socket_linger().

Fields:

0.  enabled (bool) – Linger active.

1.  timeout (int) – How many seconds to linger for (if active).

static __new__(_cls, enabled: bool, timeout: int)

Create new instance of socket_linger(enabled, timeout)

easynetwork.lowlevel.socket.get_socket_linger_struct() → struct.Struct

Returns a Struct representation of the SO_LINGER structure. See socket(7) for details.

The format of the returned struct may vary depending on the operating system.

Return type:

struct.Struct

easynetwork.lowlevel.socket.get_socket_linger(sock: SupportsSocketOptions) → socket_linger

Gets socket linger.

This is equivalent to:

linger_struct = get_socket_linger_struct()
(enabled, timeout) = linger_struct.unpack(sock.getsockopt(SOL_SOCKET, SO_LINGER, linger_struct.size))

linger_struct is determined by the operating system. See get_socket_linger_struct() for details.

See the Unix manual page socket(7) for details.

Parameters:

sock (SupportsSocketOptions) – The socket.

Return type:

socket_linger

Note

Modern operating systems disable it by default.

See also

enable_socket_linger()

disable_socket_linger()

easynetwork.lowlevel.socket.enable_socket_linger(sock: SupportsSocketOptions, timeout: int) → None

Enables socket linger.

This is equivalent to:

linger_struct = get_socket_linger_struct()
sock.setsockopt(SOL_SOCKET, SO_LINGER, linger_struct.pack(1, timeout))

linger_struct is determined by the operating system. See get_socket_linger_struct() for details.

See the Unix manual page socket(7) for the meaning of the argument timeout.

Parameters:

• sock (SupportsSocketOptions) – The socket.

• timeout (int) – How many seconds to linger for.

Note

Modern operating systems disable it by default.

See also

disable_socket_linger()

easynetwork.lowlevel.socket.disable_socket_linger(sock: SupportsSocketOptions) → None

Disables socket linger.

This is equivalent to:

linger_struct = get_socket_linger_struct()
sock.setsockopt(SOL_SOCKET, SO_LINGER, linger_struct.pack(0, 0))

linger_struct is determined by the operating system. See get_socket_linger_struct() for details.

Parameters:

sock (SupportsSocketOptions) – The socket.

Note

Modern operating systems disable it by default.

See also

enable_socket_linger()