from __future__ import annotations
|
|
import os
|
import typing
|
|
# use http.client.HTTPException for consistency with non-emscripten
|
from http.client import HTTPException as HTTPException # noqa: F401
|
from http.client import ResponseNotReady
|
|
from ..._base_connection import _TYPE_BODY
|
from ...connection import HTTPConnection, ProxyConfig, port_by_scheme
|
from ...exceptions import TimeoutError
|
from ...response import BaseHTTPResponse
|
from ...util.connection import _TYPE_SOCKET_OPTIONS
|
from ...util.timeout import _DEFAULT_TIMEOUT, _TYPE_TIMEOUT
|
from ...util.url import Url
|
from .fetch import _RequestError, _TimeoutError, send_request, send_streaming_request
|
from .request import EmscriptenRequest
|
from .response import EmscriptenHttpResponseWrapper, EmscriptenResponse
|
|
if typing.TYPE_CHECKING:
|
from ..._base_connection import BaseHTTPConnection, BaseHTTPSConnection
|
|
|
class EmscriptenHTTPConnection:
|
default_port: typing.ClassVar[int] = port_by_scheme["http"]
|
default_socket_options: typing.ClassVar[_TYPE_SOCKET_OPTIONS]
|
|
timeout: None | (float)
|
|
host: str
|
port: int
|
blocksize: int
|
source_address: tuple[str, int] | None
|
socket_options: _TYPE_SOCKET_OPTIONS | None
|
|
proxy: Url | None
|
proxy_config: ProxyConfig | None
|
|
is_verified: bool = False
|
proxy_is_verified: bool | None = None
|
|
_response: EmscriptenResponse | None
|
|
def __init__(
|
self,
|
host: str,
|
port: int = 0,
|
*,
|
timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
|
source_address: tuple[str, int] | None = None,
|
blocksize: int = 8192,
|
socket_options: _TYPE_SOCKET_OPTIONS | None = None,
|
proxy: Url | None = None,
|
proxy_config: ProxyConfig | None = None,
|
) -> None:
|
self.host = host
|
self.port = port
|
self.timeout = timeout if isinstance(timeout, float) else 0.0
|
self.scheme = "http"
|
self._closed = True
|
self._response = None
|
# ignore these things because we don't
|
# have control over that stuff
|
self.proxy = None
|
self.proxy_config = None
|
self.blocksize = blocksize
|
self.source_address = None
|
self.socket_options = None
|
self.is_verified = False
|
|
def set_tunnel(
|
self,
|
host: str,
|
port: int | None = 0,
|
headers: typing.Mapping[str, str] | None = None,
|
scheme: str = "http",
|
) -> None:
|
pass
|
|
def connect(self) -> None:
|
pass
|
|
def request(
|
self,
|
method: str,
|
url: str,
|
body: _TYPE_BODY | None = None,
|
headers: typing.Mapping[str, str] | None = None,
|
# We know *at least* botocore is depending on the order of the
|
# first 3 parameters so to be safe we only mark the later ones
|
# as keyword-only to ensure we have space to extend.
|
*,
|
chunked: bool = False,
|
preload_content: bool = True,
|
decode_content: bool = True,
|
enforce_content_length: bool = True,
|
) -> None:
|
self._closed = False
|
if url.startswith("/"):
|
if self.port is not None:
|
port = f":{self.port}"
|
else:
|
port = ""
|
# no scheme / host / port included, make a full url
|
url = f"{self.scheme}://{self.host}{port}{url}"
|
request = EmscriptenRequest(
|
url=url,
|
method=method,
|
timeout=self.timeout if self.timeout else 0,
|
decode_content=decode_content,
|
)
|
request.set_body(body)
|
if headers:
|
for k, v in headers.items():
|
request.set_header(k, v)
|
self._response = None
|
try:
|
if not preload_content:
|
self._response = send_streaming_request(request)
|
if self._response is None:
|
self._response = send_request(request)
|
except _TimeoutError as e:
|
raise TimeoutError(e.message) from e
|
except _RequestError as e:
|
raise HTTPException(e.message) from e
|
|
def getresponse(self) -> BaseHTTPResponse:
|
if self._response is not None:
|
return EmscriptenHttpResponseWrapper(
|
internal_response=self._response,
|
url=self._response.request.url,
|
connection=self,
|
)
|
else:
|
raise ResponseNotReady()
|
|
def close(self) -> None:
|
self._closed = True
|
self._response = None
|
|
@property
|
def is_closed(self) -> bool:
|
"""Whether the connection either is brand new or has been previously closed.
|
If this property is True then both ``is_connected`` and ``has_connected_to_proxy``
|
properties must be False.
|
"""
|
return self._closed
|
|
@property
|
def is_connected(self) -> bool:
|
"""Whether the connection is actively connected to any origin (proxy or target)"""
|
return True
|
|
@property
|
def has_connected_to_proxy(self) -> bool:
|
"""Whether the connection has successfully connected to its proxy.
|
This returns False if no proxy is in use. Used to determine whether
|
errors are coming from the proxy layer or from tunnelling to the target origin.
|
"""
|
return False
|
|
|
class EmscriptenHTTPSConnection(EmscriptenHTTPConnection):
|
default_port = port_by_scheme["https"]
|
# all this is basically ignored, as browser handles https
|
cert_reqs: int | str | None = None
|
ca_certs: str | None = None
|
ca_cert_dir: str | None = None
|
ca_cert_data: None | str | bytes = None
|
cert_file: str | None
|
key_file: str | None
|
key_password: str | None
|
ssl_context: typing.Any | None
|
ssl_version: int | str | None = None
|
ssl_minimum_version: int | None = None
|
ssl_maximum_version: int | None = None
|
assert_hostname: None | str | typing.Literal[False]
|
assert_fingerprint: str | None = None
|
|
def __init__(
|
self,
|
host: str,
|
port: int = 0,
|
*,
|
timeout: _TYPE_TIMEOUT = _DEFAULT_TIMEOUT,
|
source_address: tuple[str, int] | None = None,
|
blocksize: int = 16384,
|
socket_options: (
|
None | _TYPE_SOCKET_OPTIONS
|
) = HTTPConnection.default_socket_options,
|
proxy: Url | None = None,
|
proxy_config: ProxyConfig | None = None,
|
cert_reqs: int | str | None = None,
|
assert_hostname: None | str | typing.Literal[False] = None,
|
assert_fingerprint: str | None = None,
|
server_hostname: str | None = None,
|
ssl_context: typing.Any | None = None,
|
ca_certs: str | None = None,
|
ca_cert_dir: str | None = None,
|
ca_cert_data: None | str | bytes = None,
|
ssl_minimum_version: int | None = None,
|
ssl_maximum_version: int | None = None,
|
ssl_version: int | str | None = None, # Deprecated
|
cert_file: str | None = None,
|
key_file: str | None = None,
|
key_password: str | None = None,
|
) -> None:
|
super().__init__(
|
host,
|
port=port,
|
timeout=timeout,
|
source_address=source_address,
|
blocksize=blocksize,
|
socket_options=socket_options,
|
proxy=proxy,
|
proxy_config=proxy_config,
|
)
|
self.scheme = "https"
|
|
self.key_file = key_file
|
self.cert_file = cert_file
|
self.key_password = key_password
|
self.ssl_context = ssl_context
|
self.server_hostname = server_hostname
|
self.assert_hostname = assert_hostname
|
self.assert_fingerprint = assert_fingerprint
|
self.ssl_version = ssl_version
|
self.ssl_minimum_version = ssl_minimum_version
|
self.ssl_maximum_version = ssl_maximum_version
|
self.ca_certs = ca_certs and os.path.expanduser(ca_certs)
|
self.ca_cert_dir = ca_cert_dir and os.path.expanduser(ca_cert_dir)
|
self.ca_cert_data = ca_cert_data
|
|
self.cert_reqs = None
|
|
# The browser will automatically verify all requests.
|
# We have no control over that setting.
|
self.is_verified = True
|
|
def set_cert(
|
self,
|
key_file: str | None = None,
|
cert_file: str | None = None,
|
cert_reqs: int | str | None = None,
|
key_password: str | None = None,
|
ca_certs: str | None = None,
|
assert_hostname: None | str | typing.Literal[False] = None,
|
assert_fingerprint: str | None = None,
|
ca_cert_dir: str | None = None,
|
ca_cert_data: None | str | bytes = None,
|
) -> None:
|
pass
|
|
|
# verify that this class implements BaseHTTP(s) connection correctly
|
if typing.TYPE_CHECKING:
|
_supports_http_protocol: BaseHTTPConnection = EmscriptenHTTPConnection("", 0)
|
_supports_https_protocol: BaseHTTPSConnection = EmscriptenHTTPSConnection("", 0)
|
