⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on Network monitoring, controls & Application development, testing, debugging.
2013-present by Abhinav Singh and contributors.
BSD, see LICENSE for more details.
- class proxy.core.base.tcp_server.BaseTcpServerHandler(*args: Any, **kwargs: Any)#
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>#
- _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.
- 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
connin non-blocking mode and initializes HTTP protocol plugins.