proxy.http.proxy.server module#

proxy.py#

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

copyright
  1. 2013-present by Abhinav Singh and contributors.

license

BSD, see LICENSE for more details.

class proxy.http.proxy.server.HttpProxyPlugin(*args: Any, **kwargs: Any)[source]#

Bases: proxy.http.plugin.HttpProtocolHandlerPlugin

HttpProtocolHandler plugin which implements HttpProxy specifications.

_abc_impl = <_abc._abc_data object>#
_close_and_release() bool[source]#
access_log(log_attrs: Dict[str, Any]) None[source]#
client: HttpClientConnection#
connect_upstream() None[source]#
emit_request_complete() None[source]#
emit_response_chunk_received(chunk_size: int) None[source]#
emit_response_complete() None[source]#
emit_response_events(chunk_size: int) None[source]#
emit_response_headers_complete() None[source]#
flags: argparse.Namespace#
gen_ca_signed_certificate(cert_file_path: str, certificate: Dict[str, Any]) None[source]#

CA signing key (default) is used for generating a public key for common_name, if one already doesn’t exist. Using generated public key a CSR request is generated, which is then signed by CA key and secret. Again this process only happen if signed certificate doesn’t already exist.

returns signed certificate path.

generate_upstream_certificate(certificate: Dict[str, Any]) str[source]#
static generated_cert_file_path(ca_cert_dir: str, host: str) str[source]#
async get_descriptors() Tuple[List[int], List[int]][source]#

Implementations must return a list of descriptions that they wish to read from and write into.

handle_pipeline_response(raw: memoryview) None[source]#
intercept() Union[socket.socket, bool][source]#
lock = <unlocked _thread.lock object>#
on_client_connection_close() None[source]#

Client connection shutdown has been received, flush has been called, perform any cleanup work here.

on_client_data(raw: memoryview) None[source]#

Called only after original request has been completely received.

on_request_complete() Union[socket.socket, bool][source]#

Called right after client request parser has reached COMPLETE state.

on_response_chunk(chunk: List[memoryview]) List[memoryview][source]#

Handle data chunks as received from the server.

Return optionally modified chunk to return back to client.

pipeline_request: Optional[HttpParser]#
pipeline_response: Optional[HttpParser]#
plugins: Dict[str, HttpProxyBasePlugin]#
static protocols() List[int][source]#
async read_from_descriptors(r: List[int]) bool[source]#

Implementations must now read data over the socket.

request: HttpParser#
response: HttpParser#
start_time: float#
uid: str#
upstream: Optional[TcpServerConnection]#
wrap_client() bool[source]#
wrap_server() bool[source]#
async write_to_descriptors(w: List[int]) bool[source]#

Implementations must now write/flush data over the socket.

Note that buffer management is in-build into the connection classes. Hence implementations MUST call flush() here, to send any buffered data over the socket.