Source code for proxy.http.client

# -*- coding: utf-8 -*-
"""
    proxy.py
    ~~~~~~~~
    ⚡⚡⚡ Fast, Lightweight, Pluggable, TLS interception capable proxy server focused on
    Network monitoring, controls & Application development, testing, debugging.

    :copyright: (c) 2013-present by Abhinav Singh and contributors.
    :license: BSD, see LICENSE for more details.
"""
import ssl
import logging
from typing import Optional

from .parser import HttpParser, httpParserTypes
from ..common.types import TcpOrTlsSocket
from ..common.utils import build_http_request, new_socket_connection
from ..common.constants import (
    HTTPS_PROTO, DEFAULT_TIMEOUT, DEFAULT_SSL_CONTEXT_OPTIONS,
)


logger = logging.getLogger(__name__)


[docs]def client( host: bytes, port: int, path: bytes, method: bytes, body: Optional[bytes] = None, conn_close: bool = True, scheme: bytes = HTTPS_PROTO, timeout: float = DEFAULT_TIMEOUT, content_type: bytes = b'application/x-www-form-urlencoded', verify: bool = True, ) -> Optional[HttpParser]: """HTTP Client""" request = build_http_request( method=method, url=path, headers={ b'Host': host, b'Content-Type': content_type, }, body=body, conn_close=conn_close, ) try: conn = new_socket_connection((host.decode(), port)) except Exception: # logger.exception('Cannot establish connection', exc_info=exc) return None sock: TcpOrTlsSocket = conn if scheme == HTTPS_PROTO: try: ctx = ssl.SSLContext(protocol=ssl.PROTOCOL_TLS_CLIENT) ctx.options |= DEFAULT_SSL_CONTEXT_OPTIONS ctx.check_hostname = verify ctx.verify_mode = ssl.CERT_NONE if not verify else ssl.CERT_REQUIRED ctx.load_default_certs() sock = ctx.wrap_socket(conn, server_hostname=host.decode()) except Exception as exc: logger.exception('Unable to wrap', exc_info=exc) conn.close() return None parser = HttpParser(httpParserTypes.RESPONSE_PARSER) sock.settimeout(timeout) try: sock.sendall(request) while True: chunk = sock.recv(1024) if not chunk: break parser.parse(memoryview(chunk)) if parser.is_complete: break finally: sock.close() return parser