proxy.http.parser.parser 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.parser.parser.HttpParser(parser_type: int, enable_proxy_protocol: int = False)[source]#

Bases: object

HTTP request/response parser.

TODO: Make me zero-copy by using memoryview. Currently due to chunk/buffer handling we are not able to utilize memoryview efficiently.

For this to happen we must store buffer as List[memoryview] instead of raw bytes and update parser to work accordingly.

_get_body_or_chunks() Optional[bytes][source]#
_process_body(raw: memoryview) Tuple[bool, memoryview][source]#
_process_header(raw: bytes) None[source]#
_process_headers(raw: memoryview) Tuple[bool, memoryview][source]#

Returns False when no CRLF could be found in received bytes.

TODO: We should not return until parser reaches headers complete state or when there is no more data left to parse.

TODO: For protection against Slowloris attack, we must parse the request line and headers only after receiving end of header marker. This will also help make the parser even more stateless.

_process_line(raw: memoryview, allowed_url_schemes: Optional[List[bytes]] = None) Tuple[bool, memoryview][source]#
_set_line_attributes() None[source]#
add_header(key: bytes, value: bytes) bytes[source]#

Add/Update a header to internal data structure.

Returns key with which passed (key, value) tuple is available.

add_headers(headers: List[Tuple[bytes, bytes]]) None[source]#

Add/Update multiple headers to internal data structure

property body_expected: bool#

Returns true if content or chunked response is expected.

build(disable_headers: Optional[List[bytes]] = None, for_proxy: bool = False) bytes[source]#

Rebuild the request object.

build_response() bytes[source]#

Rebuild the response object.

property content_expected: bool#

Returns true if content-length is present and not 0.

del_header(header: bytes) None[source]#

Delete a header from internal data structure.

del_headers(headers: List[bytes]) None[source]#

Delete headers from internal data structure.

has_header(key: bytes) bool[source]#

Returns true if header key was found in payload.

header(key: bytes) bytes[source]#

Convenient method to return original header value from internal data structure.

property http_handler_protocol: int#

Returns HttpProtocols that this request belongs to.

property is_chunked_encoded: bool#

Returns true if transfer-encoding chunked is used.

property is_complete: bool#
property is_connection_upgrade: bool#

Returns true for websocket upgrade requests.

property is_http_1_1_keep_alive: bool#

Returns true for HTTP/1.1 keep-alive connections.

property is_https_tunnel: bool#

Returns true for HTTPS CONNECT tunnel request.

parse(raw: memoryview, allowed_url_schemes: Optional[List[bytes]] = None) None[source]#

Parses HTTP request out of raw bytes.

Check for HttpParser.state after parse has successfully returned.

classmethod request(raw: bytes, enable_proxy_protocol: int = False) proxy.http.parser.parser.T[source]#
classmethod response(raw: bytes) proxy.http.parser.parser.T[source]#
set_url(url: bytes, allowed_url_schemes: Optional[List[bytes]] = None) None[source]#

Given a request line, parses it and sets line attributes a.k.a. host, port, path.

update_body(body: bytes, content_type: bytes) None[source]#

This method must be used to update body after HTTP packet has been parsed.

Along with updating the body, this method also respects original request content encoding, transfer encoding settings.