socks_router.models

Attributes

SOCKS_VERSION

PackingSequence

RecursiveMapping

PACKABLE_DEFERRED_FORMAT

PACKABLE_VARIABLE_LENGTH_DECLARATION_FORMAT

Address

RoutingEntry

RoutingTable

Socks5Addresses

Socks5AddressTypes

Upstream

Classes

Packable

Base class for protocol classes.

SupportsUnbytes

Base class for protocol classes.

SocketAddress

IPv4

IPv6

Host

Socks5Method

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-3

Socks5Command

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4

Socks5AddressType

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4

Socks5MethodSelectionRequest

Socks5 Header.

Socks5MethodSelectionResponse

Socks5 Method Selection Response

Socks5Address

Socks5Request

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4

Socks5ReplyType

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-6

Socks5Reply

Socks5 Reply

Socks5State

Enum where members are also (and must be) strings

Pattern

UpstreamScheme

Enum where members are also (and must be) strings

UpstreamAddress

SSHUpstream

ProxyUpstream

RetryOptions

ApplicationContext

Module Contents

SOCKS_VERSION: Literal[5] = 5
PackingSequence: TypeAlias = str | tuple[str, str]
RecursiveMapping: TypeAlias = Mapping[K, V | RecursiveMapping[K, V]]
PACKABLE_DEFERRED_FORMAT: Final[str] = '&'
PACKABLE_VARIABLE_LENGTH_DECLARATION_FORMAT: Final[str] = '%*'
class Packable[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...
classmethod __pack_format__() str[source]
class SupportsUnbytes[source]

Bases: Protocol

Base class for protocol classes.

Protocol classes are defined as:

class Proto(Protocol):
    def meth(self) -> int:
        ...

Such classes are primarily used with static type checkers that recognize structural subtyping (static duck-typing).

For example:

class C:
    def meth(self) -> int:
        return 0

def func(x: Proto) -> int:
    return x.meth()

func(C())  # Passes static type check

See PEP 544 for details. Protocol classes decorated with @typing.runtime_checkable act as simple-minded runtime protocols that check only the presence of given attributes, ignoring their type signatures. Protocol classes can be generic, they are defined as:

class GenProto[T](Protocol):
    def meth(self) -> T:
        ...
classmethod __unbytes__(input: bytes) Self[source]
Abstractmethod:

class SocketAddress[source]
address: Any
port: Annotated[Optional[int], !H] = None
__str__()[source]

Return str(self).

property sockaddr: tuple[str, int]
with_port(port: int | None) Self[source]
with_default_port(port: int) Self[source]
property url_literal: str
class IPv4(address: str | IPv4, *argv, **kwargs)[source]

Bases: SocketAddress

address: IPv4
class IPv4Address(address)[source]

Bases: ipaddress.IPv4Address

Represent and manipulate single IPv4 Addresses.

classmethod __pack_format__() str[source]
__bytes__() bytes[source]
classmethod __unbytes__(input: bytes) Self[source]
class IPv6(address: str | IPv6, *argv, **kwargs)[source]

Bases: SocketAddress

address: IPv6
class IPv6Address(address)[source]

Bases: ipaddress.IPv6Address

Represent and manipulate single IPv6 Addresses.

classmethod __pack_format__() str[source]
__bytes__() bytes[source]
classmethod __unbytes__(input: bytes) Self[source]
property url_literal: str

//www.ietf.org/rfc/rfc2732.txt

Type:

SEE

Type:

https

class Host[source]

Bases: SocketAddress

address: Annotated[str, !B%*s]
Address: TypeAlias = IPv4 | IPv6 | Host
class Socks5Method[source]

Bases: enum.IntEnum

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-3

NO_AUTHENTICATION_REQUIRED = 0
GSSAPI = 1
USERNAME_PASSWORD = 2
NO_ACCEPTABLE_METHODS = 255
classmethod __pack_format__() str[source]
class Socks5Command[source]

Bases: enum.IntEnum

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4

CONNECT = 1
BIND = 2
UDP_ASSOCIATE = 3
classmethod __pack_format__() str[source]
class Socks5AddressType[source]

Bases: enum.IntEnum

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4

IPv4 = 1
DOMAINNAME = 3
IPv6 = 4
classmethod __pack_format__() str[source]
class Socks5MethodSelectionRequest[source]

Socks5 Header. SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-3 Header —— | version | method_count | methods | | 1 byte | 1 byte | [method_count] bytes |

version: Annotated[int, !B]
methods: Annotated[list[int], !B%*B]
class Socks5MethodSelectionResponse[source]

Socks5 Method Selection Response Method Selection Response ————————- | version | method | | 1 byte | 1 byte |

version: Annotated[int, !B]
method: Socks5Method
class Socks5Address[source]
classmethod address_type(type: Literal[Socks5AddressType]) Type[IPv4][source]
classmethod address_type(type: Literal[Socks5AddressType]) Type[Host]
classmethod address_type(type: Literal[Socks5AddressType]) Type[IPv6]
type: Socks5AddressType
sockaddr: Annotated[Address, &, Socks5Address.type, address_type]
classmethod from_address(address: Address) Self[source]
class Socks5Request[source]

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-4 Request ——- | version | cmd | rsv | atyp | dst.addr | dst.port | | 1 byte | 1 byte | 0x00 | 1 byte | 4-255 bytes | 2 bytes |

version: Annotated[int, !B]
command: Socks5Command
reserved: Annotated[int, !B]
destination: Socks5Address
class Socks5ReplyType[source]

Bases: enum.IntEnum

SEE: https://datatracker.ietf.org/doc/html/rfc1928#section-6

SUCCEEDED = 0
GENERAL_SOCKS_SERVER_FAILURE = 1
CONNECTION_NOT_ALLOWED_BY_RULESET = 2
NETWORK_UNREACHABLE = 3
HOST_UNREACHABLE = 4
CONNECTION_REFUSED = 5
TTL_EXPIRED = 6
COMMAND_NOT_SUPPORTED = 7
ADDRESS_TYPE_NOT_SUPPORTED = 8
classmethod __pack_format__() str[source]
property message: str
class Socks5Reply[source]

Socks5 Reply Reply —– | version | reply | rsv | atyp | dst.addr | dst.port | | 1 byte | 1 byte | 0x00 | 1 byte | 4-255 bytes | 2 bytes |

version: Annotated[int, !B]
reply: Socks5ReplyType
reserved: Annotated[int, !B] = 0
server_bound_address: Socks5Address
class Socks5State[source]

Bases: enum.StrEnum

Enum where members are also (and must be) strings

LISTEN
HANDSHAKE
REQUEST
ESTABLISHED
CLOSED
class Pattern[source]
address: str
is_positive_match: bool = True
__str__()[source]

Return str(self).

class UpstreamScheme[source]

Bases: enum.StrEnum

Enum where members are also (and must be) strings

SSH
SOCKS5
SOCKS5H
property default_port
class UpstreamAddress[source]

Bases: object

scheme: UpstreamScheme
address: Address
__str__()[source]

Return str(self).

with_default_port(port: int | None = None) Self[source]
RoutingEntry: TypeAlias = list[Pattern]
RoutingTable: TypeAlias = Mapping[UpstreamAddress, RoutingEntry]
Socks5Addresses: collections.abc.Mapping[Socks5AddressType, type[Address]]
Socks5AddressTypes
class SSHUpstream[source]
ssh_client: subprocess.Popen
proxy_server: Address
classmethod create(upstream: Address, proxy_server: Address, ssh_options: dict | None = None) Self[source]
class ProxyUpstream[source]
proxy_server: Address
Upstream: TypeAlias = SSHUpstream | ProxyUpstream
class RetryOptions[source]
tries: int
delay: float = 1
max_delay: float | None = None
backoff: float = 1
jitter: float = 0
classmethod exponential_backoff(*argv, backoff=2, **kwargs)[source]
class ApplicationContext[source]
name: str = 'socks-router'
routing_table: RoutingTable
ssh_connection_timeout: int = 10
remote_socket_timeout: float | None = 10
proxy_poll_socket_timeout: float = 0.1
proxy_retry_options: RetryOptions
mutex: threading.Lock
upstreams: collections.abc.MutableMapping[UpstreamAddress, Upstream]
is_terminating: bool = False