proxy.common.utils 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.

proxy.common.utils.build_http_header(k: bytes, v: bytes) bytes[source]#

Build and return a HTTP header line for use in raw packet.

proxy.common.utils.build_http_pkt(line: List[bytes], headers: Optional[Dict[bytes, bytes]] = None, body: Optional[bytes] = None, conn_close: bool = False) bytes[source]#

Build and returns a HTTP request or response packet.

proxy.common.utils.build_http_request(method: bytes, url: bytes, protocol_version: bytes = b'HTTP/1.1', content_type: Optional[bytes] = None, headers: Optional[Dict[bytes, bytes]] = None, body: Optional[bytes] = None, conn_close: bool = False, no_ua: bool = False) bytes[source]#

Build and returns a HTTP request packet.

proxy.common.utils.build_http_response(status_code: int, protocol_version: bytes = b'HTTP/1.1', reason: Optional[bytes] = None, headers: Optional[Dict[bytes, bytes]] = None, body: Optional[bytes] = None, conn_close: bool = False, no_cl: bool = False) bytes[source]#

Build and returns a HTTP response packet.

proxy.common.utils.build_websocket_handshake_request(key: bytes, method: bytes = b'GET', url: bytes = b'/', host: bytes = b'localhost') bytes[source]#

Build and returns a Websocket handshake request packet.

Parameters
  • key – Sec-WebSocket-Key header value.

  • method – HTTP method.

  • url – Websocket request path.

proxy.common.utils.build_websocket_handshake_response(accept: bytes) bytes[source]#

Build and returns a Websocket handshake response packet.

Parameters

accept – Sec-WebSocket-Accept header value

proxy.common.utils.bytes_(s: Any, encoding: str = 'utf-8', errors: str = 'strict') Any[source]#

Utility to ensure binary-like usability.

If s is type str or int, return s.encode(encoding, errors), otherwise return s as it is.

proxy.common.utils.find_http_line(raw: bytes) Tuple[Optional[bytes], bytes][source]#

Find and returns first line ending in CRLF along with following buffer.

If no ending CRLF is found, line is None.

proxy.common.utils.get_available_port() int[source]#

Finds and returns an available port on the system.

proxy.common.utils.is_py2() bool[source]#

Exists only to avoid mocking sys.version_info in tests.

proxy.common.utils.is_threadless(threadless: bool, threaded: bool) bool[source]#
proxy.common.utils.new_socket_connection(addr: Tuple[str, int], timeout: float = 10.0, source_address: Optional[Tuple[str, int]] = None) socket.socket[source]#
proxy.common.utils.set_open_file_limit(soft_limit: int) None[source]#

Configure open file description soft limit on supported OS.

class proxy.common.utils.socket_connection(addr: Tuple[str, int])[source]#

Bases: contextlib.ContextDecorator

Same as new_socket_connection but as a context manager and decorator.

proxy.common.utils.text_(s: Any, encoding: str = 'utf-8', errors: str = 'strict') Any[source]#

Utility to ensure text-like usability.

If s is of type bytes or int, return s.decode(encoding, errors), otherwise return s as it is.

proxy.common.utils.tls_interception_enabled(flags: argparse.Namespace) bool[source]#
proxy.common.utils.wrap_socket(conn: socket.socket, keyfile: str, certfile: str, cafile: Optional[str] = None) ssl.SSLSocket[source]#

Use this to upgrade server_side socket to TLS.