FM / emscripten

/usr/lib/fixperms/venv/lib/python3.13/site-packages/urllib3/contrib/emscripten/

/usr/lib/fixperms/venv/lib/python3.13/site-packages/urllib3/contrib/emscripten

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
connection.py	8771	0644	edit dl rm
emscripten_fetch_worker.js	3655	0644	edit dl rm
fetch.py	23484	0644	edit dl rm
request.py	566	0644	edit dl rm
response.py	9507	0644	edit dl rm
__init__.py	733	0644	edit dl rm

Edit: /usr/lib/fixperms/venv/lib/python3.13/site-packages/urllib3/contrib/emscripten/connection.py (8771B)

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("/"):
            # no scheme / host / port included, make a full url
            url = f"{self.scheme}://{self.host}:{self.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

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

# 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)

cmd: