proxy.core.base package

Submodules

• proxy.core.base.tcp_server module
  • proxy.py
• proxy.core.base.tcp_tunnel module
  • proxy.py
• proxy.core.base.tcp_upstream module
  • proxy.py

Module contents

proxy.py

⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on Network monitoring, controls & Application development, testing, debugging.

copyright

3. 2013-present by Abhinav Singh and contributors.

license

BSD, see LICENSE for more details.

class proxy.core.base.BaseTcpServerHandler(*args: Any, **kwargs: Any)

Bases: proxy.core.work.work.Work[proxy.core.base.tcp_server.T]

BaseTcpServerHandler implements Work interface.

BaseTcpServerHandler lifecycle is controlled by Threadless core using asyncio. If you want to also support threaded mode, also implement the optional run() method from Work class.

An instance of BaseTcpServerHandler is created for each client connection. BaseTcpServerHandler ensures that server is always ready to accept new data from the client. It also ensures, client is ready to accept new data before flushing data to it.

Most importantly, BaseTcpServerHandler ensures that pending buffers to the client are flushed before connection is closed.

Implementations must provide:

a. handle_data(data: memoryview) implementation
b. Optionally, also implement other Work method
   e.g. initialize, is_inactive, shutdown

_abc_impl = <_abc._abc_data object>

_encryption_enabled() → bool

_optionally_wrap_socket(conn: socket.socket) → Union[ssl.SSLSocket, socket.socket]

Attempts to wrap accepted client connection using provided certificates.

Shutdown and closes client connection upon error.

abstract static create(*args: Any) → proxy.core.base.tcp_server.T

Implementations are responsible for creation of work objects from incoming args. This helps keep work core agnostic to creation of externally defined work class objects.

async get_events() → Dict[int, int]

Return sockets and events (read or write) that we are interested in.

abstract handle_data(data: memoryview) → Optional[bool]

Optionally return True to close client connection.

async handle_events(readables: List[int], writables: List[int]) → bool

Return True to shutdown work.

async handle_readables(readables: List[int]) → bool

async handle_writables(writables: List[int]) → bool

initialize() → None

Optionally upgrades connection to HTTPS, sets conn in non-blocking mode and initializes HTTP protocol plugins.

class proxy.core.base.BaseTcpTunnelHandler(*args: Any, **kwargs: Any)

Bases: proxy.core.base.tcp_server.BaseTcpServerHandler[proxy.core.connection.client.TcpClientConnection]

BaseTcpTunnelHandler build on-top of BaseTcpServerHandler work class.

On-top of BaseTcpServerHandler implementation, BaseTcpTunnelHandler introduces an upstream TcpServerConnection and adds it to the core event loop when needed.

Currently, implementations must call connect_upstream from within handle_data. See HttpsConnectTunnelHandler for example usage.

_abc_impl = <_abc._abc_data object>

connect_upstream() → None

static create(*args: Any) → proxy.core.connection.client.TcpClientConnection

Implementations are responsible for creation of work objects from incoming args. This helps keep work core agnostic to creation of externally defined work class objects.

async get_events() → Dict[int, int]

Return sockets and events (read or write) that we are interested in.

abstract handle_data(data: memoryview) → Optional[bool]

Optionally return True to close client connection.

async handle_events(readables: List[int], writables: List[int]) → bool

Return True to shutdown work.

initialize() → None

Optionally upgrades connection to HTTPS, sets conn in non-blocking mode and initializes HTTP protocol plugins.

shutdown() → None

Implementation must close any opened resources here and call super().shutdown().

class proxy.core.base.TcpUpstreamConnectionHandler(*args: Any, **kwargs: Any)

Bases: abc.ABC

TcpUpstreamConnectionHandler can be used to insert an upstream server connection lifecycle.

Call initialize_upstream to initialize the upstream connection object. Then, directly use self.upstream object within your class.

See ProxyPoolPlugin for example usage.

_abc_impl = <_abc._abc_data object>

async get_descriptors() → Tuple[List[int], List[int]]

abstract handle_upstream_data(raw: memoryview) → None

initialize_upstream(addr: str, port: int) → None

async read_from_descriptors(r: List[int]) → bool

async write_to_descriptors(w: List[int]) → bool