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