socks_router.router

Attributes

pysocks_socks5_error

SOCKS_VERSION

Address

RoutingTable

Socks5AddressTypes

CHUNK_SIZE

logger

Classes

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

Socks5Request

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

Socks5Address

Socks5ReplyType

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

Socks5Reply

Socks5 Reply

Socks5State

Enum where members are also (and must be) strings

IPv4

IPv6

Host

Pattern

UpstreamScheme

Enum where members are also (and must be) strings

UpstreamAddress

SSHUpstream

ProxyUpstream

RetryOptions

ApplicationContext

SocksRouter

Mix-in class to handle each request in a new thread.

SocksRouterRequestHandler

Define self.rfile and self.wfile for stream sockets.

Functions

parse_sockaddr(→ socks_router.models.Address)

read_socket(→ read_socket.T)

write_socket(→ None)

free_port(→ tuple[str, int])

create_socket(→ socks.socksocket)

with_proxy(→ socks.socksocket)

poll_socket(destination[, timeout])

resolve_address(→ socks_router.models.Address)

create_remote(→ socks.socksocket)

connect_remote(→ socks.socksocket)

exchange_loop(client, remote[, chunk_size, timeout])

match_upstream(...)

Module Contents

parse_sockaddr(sockaddr: tuple[parse_sockaddr.S, int]) → socks_router.models.Address[source]
pysocks_socks5_error: parsec.Parser[tuple[socks_router.models.Socks5ReplyType, str]]
SOCKS_VERSION: Literal[5] = 5
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 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 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 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
Address: TypeAlias = IPv4 | IPv6 | Host
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]
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]
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
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
RoutingTable: TypeAlias = Mapping[UpstreamAddress, RoutingEntry]
Socks5AddressTypes
read_socket(sock: socket.socket, type: type[read_socket.T], format: str | None = None) → read_socket.T[source]
write_socket(sock: socket.socket, instance: write_socket.T, format: str | None = None, type: type[write_socket.T] | None = None) → None[source]
free_port(address: str = '') → tuple[str, int][source]
CHUNK_SIZE = 4096
logger
create_socket(type: socks_router.models.Socks5AddressType) → socks.socksocket[source]
with_proxy(socket: socks.socksocket, proxy_server: socks_router.models.Address | None = None) → socks.socksocket[source]
poll_socket(destination: socks_router.models.Address, timeout: float = 0.1)[source]
resolve_address(address: socks_router.models.Address, logger: logging.Logger = logger, **kwargs) → socks_router.models.Address[source]
create_remote(address: socks_router.models.Address, proxy_server: socks_router.models.Address | None = None) → socks.socksocket[source]
connect_remote(destination: socks_router.models.Address, proxy_server: socks_router.models.Address | None = None, remote_socket_timeout: float | None = None, proxy_poll_socket_timeout: float = 0.1, proxy_retry_options: socks_router.models.RetryOptions | None = None, logger: logging.Logger = logger) → socks.socksocket[source]
exchange_loop(client: socket.socket, remote: socket.socket, chunk_size: int = CHUNK_SIZE, timeout: float | None = None)[source]
match_upstream(routing_table: socks_router.models.RoutingTable, destination: socks_router.models.Address) → socks_router.models.UpstreamAddress | None[source]
class SocksRouter(*argv, context: socks_router.models.ApplicationContext | None = None, address_family: socket.AddressFamily = socket.AF_INET, **kwargs)[source]

Bases: socketserver.ThreadingTCPServer

Mix-in class to handle each request in a new thread.

allow_reuse_address = True
daemon_threads = True
block_on_close = True
context: socks_router.models.ApplicationContext
logger: logging.Logger
property address: socks_router.models.Address
server_activate() → None[source]

Called by constructor to activate the server.

May be overridden.

get_request()[source]

Get the request and client address from the socket.

May be overridden.

shutdown_request(request: socket.socket | tuple[bytes, socket.socket]) → None[source]

Called to shutdown and close an individual request.

shutdown() → None[source]

Stops the serve_forever loop.

Blocks until the loop has finished. This must be called while serve_forever() is running in another thread, or it will deadlock.

class SocksRouterRequestHandler(request, client_address, server)[source]

Bases: socketserver.StreamRequestHandler

Define self.rfile and self.wfile for stream sockets.

server: SocksRouter
state: socks_router.models.Socks5State
remote: socks.socksocket | None = None
property logger
acquire_upstream(destination: socks_router.models.Address) → socks_router.models.UpstreamAddress | None[source]
handshake()[source]
reply(type: socks_router.models.Socks5ReplyType)[source]
connect_remote(destination: socks_router.models.Address) → socks.socksocket[source]
handle_request()[source]
exchange()[source]
setup()[source]
handle()[source]

Handle incoming connections

finish()[source]