Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. from ._base import BaseNode, NodeApiResponse from ._base_async import BaseAsyncNode from ._http_aiohttp import AiohttpHttpNode from ._http_httpx import HttpxAsyncHttpNode from ._http_requests import RequestsHttpNode from ._http_urllib3 import Urllib3HttpNode __all__ = [ "AiohttpHttpNode", "BaseNode", "BaseAsyncNode", "NodeApiResponse", "RequestsHttpNode", "Urllib3HttpNode", "HttpxAsyncHttpNode", ]	1.19 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 _base.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import asyncio import logging import os import ssl from typing import Any, ClassVar, List, NamedTuple, Optional, Tuple, Union from .._models import ApiResponseMeta, HttpHeaders, NodeConfig from .._utils import is_ipaddress from .._version import __version__ from ..client_utils import DEFAULT, DefaultType _logger = logging.getLogger("elastic_transport.node") _logger.propagate = False # This logger is very verbose so disable propogation. DEFAULT_CA_CERTS: Optional[str] = None DEFAULT_USER_AGENT = f"elastic-transport-python/{__version__}" RERAISE_EXCEPTIONS = (RecursionError, asyncio.CancelledError) BUILTIN_EXCEPTIONS = ( ValueError, KeyError, NameError, AttributeError, LookupError, AssertionError, IndexError, MemoryError, RuntimeError, SystemError, TypeError, ) HTTP_STATUS_REASONS = { 200: "OK", 201: "Created", 202: "Accepted", 204: "No Content", 205: "Reset Content", 206: "Partial Content", 400: "Bad Request", 401: "Unauthorized", 402: "Payment Required", 403: "Forbidden", 404: "Not Found", 405: "Method Not Allowed", 406: "Not Acceptable", 407: "Proxy Authentication Required", 408: "Request Timeout", 409: "Conflict", 410: "Gone", 411: "Length Required", 412: "Precondition Failed", 413: "Content Too Large", 414: "URI Too Long", 415: "Unsupported Media Type", 429: "Too Many Requests", 500: "Internal Server Error", 501: "Not Implemented", 502: "Bad Gateway", 503: "Service Unavailable", 504: "Gateway Timeout", } try: import certifi DEFAULT_CA_CERTS = certifi.where() except ImportError: # pragma: nocover pass class NodeApiResponse(NamedTuple): meta: ApiResponseMeta body: bytes class BaseNode: """ Class responsible for maintaining a connection to a node. It holds persistent node pool to it and it's main interface (``perform_request``) is thread-safe. :arg config: :class:`~elastic_transport.NodeConfig` instance """ _CLIENT_META_HTTP_CLIENT: ClassVar[Tuple[str, str]] def __init__(self, config: NodeConfig): self._config = config self._headers: HttpHeaders = self.config.headers.copy() # type: ignore[attr-defined] self.headers.setdefault("connection", "keep-alive") self.headers.setdefault("user-agent", DEFAULT_USER_AGENT) self._http_compress = bool(config.http_compress or False) if config.http_compress: self.headers["accept-encoding"] = "gzip" self._scheme = config.scheme self._host = config.host self._port = config.port self._path_prefix = ( ("/" + config.path_prefix.strip("/")) if config.path_prefix else "" ) @property def config(self) -> NodeConfig: return self._config @property def headers(self) -> HttpHeaders: return self._headers @property def scheme(self) -> str: return self._scheme @property def host(self) -> str: return self._host @property def port(self) -> int: return self._port @property def path_prefix(self) -> str: return self._path_prefix def __repr__(self) -> str: return f"<{self.__class__.__name__}({self.base_url})>" def __lt__(self, other: object) -> bool: if not isinstance(other, BaseNode): return NotImplemented return id(self) < id(other) def __eq__(self, other: object) -> bool: if not isinstance(other, BaseNode): return NotImplemented return self.__hash__() == other.__hash__() def __ne__(self, other: object) -> bool: if not isinstance(other, BaseNode): return NotImplemented return not self == other def __hash__(self) -> int: return hash((str(type(self).__name__), self.config)) @property def base_url(self) -> str: return "".join( [ self.scheme, "://", # IPv6 must be wrapped by [...] "[%s]" % self.host if ":" in self.host else self.host, ":%s" % self.port if self.port is not None else "", self.path_prefix, ] ) def perform_request( self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: # pragma: nocover """Constructs and sends an HTTP request and parses the HTTP response. :param method: HTTP method :param target: HTTP request target, typically path+query :param body: Optional HTTP request body encoded as bytes :param headers: Optional HTTP headers to send in addition to the headers already configured. :param request_timeout: Amount of time to wait for the first response bytes to arrive before raising a :class:`elastic_transport.ConnectionTimeout` error. :raises: :class:`elastic_transport.ConnectionError`, :class:`elastic_transport.ConnectionTimeout`, :class:`elastic_transport.TlsError` :rtype: Tuple[ApiResponseMeta, bytes] :returns: Metadata about the request+response and the raw decompressed bytes from the HTTP response body. """ raise NotImplementedError() def close(self) -> None: # pragma: nocover pass def _log_request( self, method: str, target: str, headers: Optional[HttpHeaders], body: Optional[bytes], meta: Optional[ApiResponseMeta] = None, response: Optional[bytes] = None, exception: Optional[Exception] = None, ) -> None: if _logger.hasHandlers(): http_version = meta.http_version if meta else "?.?" lines = ["> %s %s HTTP/%s"] log_args: List[Any] = [method, target, http_version] if headers: for header, value in sorted(headers._dict_hide_auth().items()): lines.append(f"> {header.title()}: {value}") if body is not None: try: body_encoded = body.decode("utf-8", "surrogatepass") except UnicodeError: body_encoded = repr(body) log_args.append(body_encoded) lines.append("> %s") if meta is not None: reason = HTTP_STATUS_REASONS.get(meta.status, None) if reason: lines.append("< HTTP/%s %d %s") log_args.extend((http_version, meta.status, reason)) else: lines.append("< HTTP/%s %d") log_args.extend((http_version, meta.status)) if meta.headers: for header, value in sorted(meta.headers.items()): lines.append(f"< {header.title()}: {value}") if response: try: response_decoded = response.decode("utf-8", "surrogatepass") except UnicodeError: response_decoded = repr(response) log_args.append(response_decoded) lines.append("< %s") if exception is not None: _logger.debug("\n".join(lines), log_args, exc_info=exception) else: _logger.debug("\n".join(lines), log_args) _HAS_TLS_VERSION = hasattr(ssl, "TLSVersion") _SSL_PROTOCOL_VERSION_ATTRS = ("TLSv1", "TLSv1_1", "TLSv1_2") _SSL_PROTOCOL_VERSION_DEFAULT = getattr(ssl, "OP_NO_SSLv2", 0) \| getattr( ssl, "OP_NO_SSLv3", 0 ) _SSL_PROTOCOL_VERSION_TO_OPTIONS = {} _SSL_PROTOCOL_VERSION_TO_TLS_VERSION = {} for i, _protocol_attr in enumerate(_SSL_PROTOCOL_VERSION_ATTRS): try: _protocol_value = getattr(ssl, f"PROTOCOL_{_protocol_attr}") except AttributeError: continue if _HAS_TLS_VERSION: _tls_version_value = getattr(ssl.TLSVersion, _protocol_attr) _SSL_PROTOCOL_VERSION_TO_TLS_VERSION[_protocol_value] = _tls_version_value _SSL_PROTOCOL_VERSION_TO_TLS_VERSION[_tls_version_value] = _tls_version_value # Because we're setting a minimum version we binary OR all the options together. _SSL_PROTOCOL_VERSION_TO_OPTIONS[_protocol_value] = ( _SSL_PROTOCOL_VERSION_DEFAULT \| sum( getattr(ssl, f"OP_NO_{_attr}", 0) for _attr in _SSL_PROTOCOL_VERSION_ATTRS[:i] ) ) # TLSv1.3 is unique, doesn't have a PROTOCOL_TLSvX counterpart. So we have to set it manually. if _HAS_TLS_VERSION: try: _SSL_PROTOCOL_VERSION_TO_TLS_VERSION[ssl.TLSVersion.TLSv1_3] = ( ssl.TLSVersion.TLSv1_3 ) except AttributeError: # pragma: nocover pass def ssl_context_from_node_config(node_config: NodeConfig) -> ssl.SSLContext: if node_config.ssl_context: ctx = node_config.ssl_context else: ctx = ssl.create_default_context() # Enable/disable certificate verification in these orders # to avoid 'ValueErrors' from SSLContext. We only do this # step if the user doesn't pass a preconfigured SSLContext. if node_config.verify_certs: ctx.verify_mode = ssl.CERT_REQUIRED ctx.check_hostname = not is_ipaddress(node_config.host) else: ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE # Enable logging of TLS session keys for use with Wireshark. if hasattr(ctx, "keylog_filename"): sslkeylogfile = os.environ.get("SSLKEYLOGFILE", "") if sslkeylogfile: ctx.keylog_filename = sslkeylogfile # Apply the 'ssl_version' if given, otherwise default to TLSv1.2+ ssl_version = node_config.ssl_version if ssl_version is None: if _HAS_TLS_VERSION: ssl_version = ssl.TLSVersion.TLSv1_2 else: ssl_version = ssl.PROTOCOL_TLSv1_2 try: if _HAS_TLS_VERSION: ctx.minimum_version = _SSL_PROTOCOL_VERSION_TO_TLS_VERSION[ssl_version] else: ctx.options \|= _SSL_PROTOCOL_VERSION_TO_OPTIONS[ssl_version] except KeyError: raise ValueError( f"Unsupported value for 'ssl_version': {ssl_version!r}. Must be " "either 'ssl.PROTOCOL_TLSvX' or 'ssl.TLSVersion.TLSvX'" ) from None return ctx	11.13 KB	2025-05-25 12:26
📄 _base_async.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. from typing import Optional, Union from .._models import HttpHeaders from ..client_utils import DEFAULT, DefaultType from ._base import BaseNode, NodeApiResponse class BaseAsyncNode(BaseNode): """Base class for Async HTTP node implementations""" async def perform_request( # type: ignore[override] self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: raise NotImplementedError() # pragma: nocover async def close(self) -> None: # type: ignore[override] raise NotImplementedError() # pragma: nocover	1.47 KB	2025-05-25 12:26
📄 _http_aiohttp.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import asyncio import base64 import functools import gzip import os import re import ssl import sys import warnings from typing import Optional, TypedDict, Union from .._compat import warn_stacklevel from .._exceptions import ConnectionError, ConnectionTimeout, SecurityWarning, TlsError from .._models import ApiResponseMeta, HttpHeaders, NodeConfig from ..client_utils import DEFAULT, DefaultType, client_meta_version from ._base import ( BUILTIN_EXCEPTIONS, DEFAULT_CA_CERTS, RERAISE_EXCEPTIONS, NodeApiResponse, ssl_context_from_node_config, ) from ._base_async import BaseAsyncNode try: import aiohttp import aiohttp.client_exceptions as aiohttp_exceptions _AIOHTTP_AVAILABLE = True _AIOHTTP_META_VERSION = client_meta_version(aiohttp.__version__) _version_parts = [] for _version_part in aiohttp.__version__.split(".")[:3]: try: _version_parts.append(int(re.search(r"^([0-9]+)", _version_part).group(1))) # type: ignore[union-attr] except (AttributeError, ValueError): break _AIOHTTP_SEMVER_VERSION = tuple(_version_parts) # See aio-libs/aiohttp#1769 and #5012 _AIOHTTP_FIXED_HEAD_BUG = _AIOHTTP_SEMVER_VERSION >= (3, 7, 0) class RequestKwarg(TypedDict, total=False): ssl: aiohttp.Fingerprint except ImportError: # pragma: nocover _AIOHTTP_AVAILABLE = False _AIOHTTP_META_VERSION = "" _AIOHTTP_FIXED_HEAD_BUG = False # Avoid aiohttp enabled_cleanup_closed warning: https://github.com/aio-libs/aiohttp/pull/9726 _NEEDS_CLEANUP_CLOSED_313 = (3, 13, 0) <= sys.version_info < (3, 13, 1) _NEEDS_CLEANUP_CLOSED = _NEEDS_CLEANUP_CLOSED_313 or sys.version_info < (3, 12, 7) class AiohttpHttpNode(BaseAsyncNode): """Default asynchronous node class using the ``aiohttp`` library via HTTP""" _CLIENT_META_HTTP_CLIENT = ("ai", _AIOHTTP_META_VERSION) def __init__(self, config: NodeConfig): if not _AIOHTTP_AVAILABLE: # pragma: nocover raise ValueError("You must have 'aiohttp' installed to use AiohttpHttpNode") super().__init__(config) self._ssl_assert_fingerprint = config.ssl_assert_fingerprint ssl_context: Optional[ssl.SSLContext] = None if config.scheme == "https": if config.ssl_context is not None: ssl_context = ssl_context_from_node_config(config) else: ssl_context = ssl_context_from_node_config(config) ca_certs = ( DEFAULT_CA_CERTS if config.ca_certs is None else config.ca_certs ) if config.verify_certs: if not ca_certs: raise ValueError( "Root certificates are missing for certificate " "validation. Either pass them in using the ca_certs parameter or " "install certifi to use it automatically." ) else: if config.ssl_show_warn: warnings.warn( f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure", stacklevel=warn_stacklevel(), category=SecurityWarning, ) if ca_certs is not None: if os.path.isfile(ca_certs): ssl_context.load_verify_locations(cafile=ca_certs) elif os.path.isdir(ca_certs): ssl_context.load_verify_locations(capath=ca_certs) else: raise ValueError("ca_certs parameter is not a path") # Use client_cert and client_key variables for SSL certificate configuration. if config.client_cert and not os.path.isfile(config.client_cert): raise ValueError("client_cert is not a path to a file") if config.client_key and not os.path.isfile(config.client_key): raise ValueError("client_key is not a path to a file") if config.client_cert and config.client_key: ssl_context.load_cert_chain(config.client_cert, config.client_key) elif config.client_cert: ssl_context.load_cert_chain(config.client_cert) self._loop: asyncio.AbstractEventLoop = None # type: ignore[assignment] self.session: Optional[aiohttp.ClientSession] = None # Parameters for creating an aiohttp.ClientSession later. self._connections_per_node = config.connections_per_node self._ssl_context = ssl_context async def perform_request( # type: ignore[override] self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: global _AIOHTTP_FIXED_HEAD_BUG if self.session is None: self._create_aiohttp_session() assert self.session is not None url = self.base_url + target is_head = False # There is a bug in aiohttp<3.7 that disables the re-use # of the connection in the pool when method=HEAD. # See: aio-libs/aiohttp#1769 if method == "HEAD" and not _AIOHTTP_FIXED_HEAD_BUG: method = "GET" is_head = True # total=0 means no timeout for aiohttp resolved_timeout: Optional[float] = ( self.config.request_timeout if request_timeout is DEFAULT else request_timeout ) aiohttp_timeout = aiohttp.ClientTimeout( total=resolved_timeout if resolved_timeout is not None else 0 ) request_headers = self._headers.copy() if headers: request_headers.update(headers) body_to_send: Optional[bytes] if body: if self._http_compress: body_to_send = gzip.compress(body) request_headers["content-encoding"] = "gzip" else: body_to_send = body else: body_to_send = None kwargs: RequestKwarg = {} if self._ssl_assert_fingerprint: kwargs["ssl"] = aiohttp_fingerprint(self._ssl_assert_fingerprint) try: start = self._loop.time() async with self.session.request( method, url, data=body_to_send, headers=request_headers, timeout=aiohttp_timeout, **kwargs, ) as response: if is_head: # We actually called 'GET' so throw away the data. await response.release() raw_data = b"" else: raw_data = await response.read() duration = self._loop.time() - start # We want to reraise a cancellation or recursion error. except RERAISE_EXCEPTIONS: raise except Exception as e: err: Exception if isinstance( e, (asyncio.TimeoutError, aiohttp_exceptions.ServerTimeoutError) ): err = ConnectionTimeout( "Connection timed out during request", errors=(e,) ) elif isinstance(e, (ssl.SSLError, aiohttp_exceptions.ClientSSLError)): err = TlsError(str(e), errors=(e,)) elif isinstance(e, BUILTIN_EXCEPTIONS): raise else: err = ConnectionError(str(e), errors=(e,)) self._log_request( method="HEAD" if is_head else method, target=target, headers=request_headers, body=body, exception=err, ) raise err from None meta = ApiResponseMeta( node=self.config, duration=duration, http_version="1.1", status=response.status, headers=HttpHeaders(response.headers), ) self._log_request( method="HEAD" if is_head else method, target=target, headers=request_headers, body=body, meta=meta, response=raw_data, ) return NodeApiResponse( meta, raw_data, ) async def close(self) -> None: # type: ignore[override] if self.session: await self.session.close() self.session = None def _create_aiohttp_session(self) -> None: """Creates an aiohttp.ClientSession(). This is delayed until the first call to perform_request() so that AsyncTransport has a chance to set AiohttpHttpNode.loop """ if self._loop is None: self._loop = asyncio.get_running_loop() self.session = aiohttp.ClientSession( headers=self.headers, skip_auto_headers=("accept", "accept-encoding", "user-agent"), auto_decompress=True, loop=self._loop, cookie_jar=aiohttp.DummyCookieJar(), connector=aiohttp.TCPConnector( limit_per_host=self._connections_per_node, use_dns_cache=True, enable_cleanup_closed=_NEEDS_CLEANUP_CLOSED, ssl=self._ssl_context or False, ), ) @functools.lru_cache(maxsize=64, typed=True) def aiohttp_fingerprint(ssl_assert_fingerprint: str) -> "aiohttp.Fingerprint": """Changes 'ssl_assert_fingerprint' into a configured 'aiohttp.Fingerprint' instance. Uses a cache to prevent creating tons of objects needlessly. """ return aiohttp.Fingerprint( base64.b16decode(ssl_assert_fingerprint.replace(":", ""), casefold=True) )	10.51 KB	2025-05-25 12:26
📄 _http_httpx.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import gzip import os.path import ssl import time import warnings from typing import Literal, Optional, Union from .._compat import warn_stacklevel from .._exceptions import ConnectionError, ConnectionTimeout, SecurityWarning, TlsError from .._models import ApiResponseMeta, HttpHeaders, NodeConfig from ..client_utils import DEFAULT, DefaultType, client_meta_version from ._base import ( BUILTIN_EXCEPTIONS, DEFAULT_CA_CERTS, RERAISE_EXCEPTIONS, NodeApiResponse, ssl_context_from_node_config, ) from ._base_async import BaseAsyncNode try: import httpx _HTTPX_AVAILABLE = True _HTTPX_META_VERSION = client_meta_version(httpx.__version__) except ImportError: _HTTPX_AVAILABLE = False _HTTPX_META_VERSION = "" class HttpxAsyncHttpNode(BaseAsyncNode): _CLIENT_META_HTTP_CLIENT = ("hx", _HTTPX_META_VERSION) def __init__(self, config: NodeConfig): if not _HTTPX_AVAILABLE: # pragma: nocover raise ValueError("You must have 'httpx' installed to use HttpxNode") super().__init__(config) if config.ssl_assert_fingerprint: raise ValueError( "httpx does not support certificate pinning. https://github.com/encode/httpx/issues/761" ) ssl_context: Union[ssl.SSLContext, Literal[False]] = False if config.scheme == "https": if config.ssl_context is not None: ssl_context = ssl_context_from_node_config(config) else: ssl_context = ssl_context_from_node_config(config) ca_certs = ( DEFAULT_CA_CERTS if config.ca_certs is None else config.ca_certs ) if config.verify_certs: if not ca_certs: raise ValueError( "Root certificates are missing for certificate " "validation. Either pass them in using the ca_certs parameter or " "install certifi to use it automatically." ) else: if config.ssl_show_warn: warnings.warn( f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure", stacklevel=warn_stacklevel(), category=SecurityWarning, ) if ca_certs is not None: if os.path.isfile(ca_certs): ssl_context.load_verify_locations(cafile=ca_certs) elif os.path.isdir(ca_certs): ssl_context.load_verify_locations(capath=ca_certs) else: raise ValueError("ca_certs parameter is not a path") # Use client_cert and client_key variables for SSL certificate configuration. if config.client_cert and not os.path.isfile(config.client_cert): raise ValueError("client_cert is not a path to a file") if config.client_key and not os.path.isfile(config.client_key): raise ValueError("client_key is not a path to a file") if config.client_cert and config.client_key: ssl_context.load_cert_chain(config.client_cert, config.client_key) elif config.client_cert: ssl_context.load_cert_chain(config.client_cert) self.client = httpx.AsyncClient( base_url=f"{config.scheme}://{config.host}:{config.port}", limits=httpx.Limits(max_connections=config.connections_per_node), verify=ssl_context or False, timeout=config.request_timeout, ) async def perform_request( # type: ignore[override] self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: resolved_headers = self._headers.copy() if headers: resolved_headers.update(headers) if body: if self._http_compress: resolved_body = gzip.compress(body) resolved_headers["content-encoding"] = "gzip" else: resolved_body = body else: resolved_body = None try: start = time.perf_counter() if request_timeout is DEFAULT: resp = await self.client.request( method, target, content=resolved_body, headers=dict(resolved_headers), ) else: resp = await self.client.request( method, target, content=resolved_body, headers=dict(resolved_headers), timeout=request_timeout, ) response_body = resp.read() duration = time.perf_counter() - start except RERAISE_EXCEPTIONS + BUILTIN_EXCEPTIONS: raise except Exception as e: err: Exception if isinstance(e, (TimeoutError, httpx.TimeoutException)): err = ConnectionTimeout( "Connection timed out during request", errors=(e,) ) elif isinstance(e, ssl.SSLError): err = TlsError(str(e), errors=(e,)) # Detect SSL errors for httpx v0.28.0+ # Needed until https://github.com/encode/httpx/issues/3350 is fixed elif isinstance(e, httpx.ConnectError) and e.__cause__: context = e.__cause__.__context__ if isinstance(context, ssl.SSLError): err = TlsError(str(context), errors=(e,)) else: err = ConnectionError(str(e), errors=(e,)) else: err = ConnectionError(str(e), errors=(e,)) self._log_request( method=method, target=target, headers=resolved_headers, body=body, exception=err, ) raise err from None meta = ApiResponseMeta( resp.status_code, resp.http_version, HttpHeaders(resp.headers), duration, self.config, ) self._log_request( method=method, target=target, headers=resolved_headers, body=body, meta=meta, response=response_body, ) return NodeApiResponse(meta, response_body) async def close(self) -> None: # type: ignore[override] await self.client.aclose()	7.5 KB	2025-05-25 12:26
📄 _http_requests.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import gzip import ssl import time import warnings from typing import Any, Optional, Union import urllib3 from .._compat import warn_stacklevel from .._exceptions import ConnectionError, ConnectionTimeout, SecurityWarning, TlsError from .._models import ApiResponseMeta, HttpHeaders, NodeConfig from ..client_utils import DEFAULT, DefaultType, client_meta_version from ._base import ( BUILTIN_EXCEPTIONS, RERAISE_EXCEPTIONS, BaseNode, NodeApiResponse, ssl_context_from_node_config, ) try: import requests from requests.adapters import HTTPAdapter from requests.auth import AuthBase _REQUESTS_AVAILABLE = True _REQUESTS_META_VERSION = client_meta_version(requests.__version__) # Use our custom HTTPSConnectionPool for chain cert fingerprint support. try: from ._urllib3_chain_certs import HTTPSConnectionPool except (ImportError, AttributeError): HTTPSConnectionPool = urllib3.HTTPSConnectionPool # type: ignore[assignment,misc] class _ElasticHTTPAdapter(HTTPAdapter): def __init__(self, node_config: NodeConfig, kwargs: Any) -> None: self._node_config = node_config super().__init__(kwargs) def init_poolmanager( self, connections: Any, maxsize: int, block: bool = False, pool_kwargs: Any, ) -> None: if self._node_config.scheme == "https": ssl_context = ssl_context_from_node_config(self._node_config) pool_kwargs.setdefault("ssl_context", ssl_context) # Fingerprint verification doesn't require CA certificates being loaded. # We also want to disable other verification methods as we only care # about the fingerprint of the certificates, not whether they form # a verified chain to a trust anchor. if self._node_config.ssl_assert_fingerprint: # Manually disable these in the right order on the SSLContext # so urllib3 won't think we want conflicting things. ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE pool_kwargs["assert_fingerprint"] = ( self._node_config.ssl_assert_fingerprint ) pool_kwargs["cert_reqs"] = "CERT_NONE" pool_kwargs["assert_hostname"] = False super().init_poolmanager(connections, maxsize, block=block, pool_kwargs) # type: ignore [no-untyped-call] self.poolmanager.pool_classes_by_scheme["https"] = HTTPSConnectionPool except ImportError: # pragma: nocover _REQUESTS_AVAILABLE = False _REQUESTS_META_VERSION = "" class RequestsHttpNode(BaseNode): """Synchronous node using the ``requests`` library communicating via HTTP. Supports setting :attr:`requests.Session.auth` via the :attr:`elastic_transport.NodeConfig._extras` using the ``requests.session.auth`` key. """ _CLIENT_META_HTTP_CLIENT = ("rq", _REQUESTS_META_VERSION) def __init__(self, config: NodeConfig): if not _REQUESTS_AVAILABLE: # pragma: nocover raise ValueError( "You must have 'requests' installed to use RequestsHttpNode" ) super().__init__(config) # Initialize Session so .headers works before calling super().__init__(). self.session = requests.Session() self.session.headers.clear() # Empty out all the default session headers if config.scheme == "https": # If we're using ssl_assert_fingerprint we don't want # to verify certificates the typical way. Instead we # rely on the custom ElasticHTTPAdapter and urllib3. if config.ssl_assert_fingerprint: self.session.verify = False # Otherwise we go the traditional route of verifying certs. else: if config.ca_certs: if not config.verify_certs: raise ValueError( "You cannot use 'ca_certs' when 'verify_certs=False'" ) self.session.verify = config.ca_certs else: self.session.verify = config.verify_certs if not config.ssl_show_warn: urllib3.disable_warnings() if ( config.scheme == "https" and not config.verify_certs and config.ssl_show_warn ): warnings.warn( f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure", stacklevel=warn_stacklevel(), category=SecurityWarning, ) # Requests supports setting 'session.auth' via _extras['requests.session.auth'] = ... try: requests_session_auth: Optional[AuthBase] = config._extras.pop( "requests.session.auth", None ) except AttributeError: requests_session_auth = None if requests_session_auth is not None: self.session.auth = requests_session_auth # Client certificates if config.client_cert: if config.client_key: self.session.cert = (config.client_cert, config.client_key) else: self.session.cert = config.client_cert # Create and mount custom adapter for constraining number of connections adapter = _ElasticHTTPAdapter( node_config=config, pool_connections=config.connections_per_node, pool_maxsize=config.connections_per_node, pool_block=True, ) # Preload the HTTPConnectionPool so initialization issues # are raised here instead of in perform_request() if hasattr(adapter, "get_connection_with_tls_context"): request = requests.Request(method="GET", url=self.base_url) prepared_request = self.session.prepare_request(request) adapter.get_connection_with_tls_context( prepared_request, verify=self.session.verify ) else: # elastic-transport is not vulnerable to CVE-2024-35195 because it uses # requests.Session and an SSLContext without using the verify parameter. # We should remove this branch when requiring requests 2.32 or later. adapter.get_connection(self.base_url) self.session.mount(prefix=f"{self.scheme}://", adapter=adapter) def perform_request( self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: url = self.base_url + target headers = HttpHeaders(headers or ()) request_headers = self._headers.copy() if headers: request_headers.update(headers) body_to_send: Optional[bytes] if body: if self._http_compress: body_to_send = gzip.compress(body) request_headers["content-encoding"] = "gzip" else: body_to_send = body else: body_to_send = None start = time.time() request = requests.Request( method=method, headers=request_headers, url=url, data=body_to_send ) prepared_request = self.session.prepare_request(request) send_kwargs = { "timeout": ( request_timeout if request_timeout is not DEFAULT else self.config.request_timeout ) } send_kwargs.update( self.session.merge_environment_settings( # type: ignore[arg-type] prepared_request.url, {}, None, None, None ) ) try: response = self.session.send(prepared_request, **send_kwargs) # type: ignore[arg-type] data = response.content duration = time.time() - start response_headers = HttpHeaders(response.headers) except RERAISE_EXCEPTIONS: raise except Exception as e: err: Exception if isinstance(e, requests.Timeout): err = ConnectionTimeout( "Connection timed out during request", errors=(e,) ) elif isinstance(e, (ssl.SSLError, requests.exceptions.SSLError)): err = TlsError(str(e), errors=(e,)) elif isinstance(e, BUILTIN_EXCEPTIONS): raise else: err = ConnectionError(str(e), errors=(e,)) self._log_request( method=method, target=target, headers=request_headers, body=body, exception=err, ) raise err from None meta = ApiResponseMeta( node=self.config, duration=duration, http_version="1.1", status=response.status_code, headers=response_headers, ) self._log_request( method=method, target=target, headers=request_headers, body=body, meta=meta, response=data, ) return NodeApiResponse( meta, data, ) def close(self) -> None: """ Explicitly closes connections """ self.session.close()	10.34 KB	2025-05-25 12:26
📄 _http_urllib3.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import gzip import ssl import time import warnings from typing import Any, Dict, Optional, Union try: from importlib import metadata except ImportError: import importlib_metadata as metadata # type: ignore[no-redef] import urllib3 from urllib3.exceptions import ConnectTimeoutError, NewConnectionError, ReadTimeoutError from urllib3.util.retry import Retry from .._compat import warn_stacklevel from .._exceptions import ConnectionError, ConnectionTimeout, SecurityWarning, TlsError from .._models import ApiResponseMeta, HttpHeaders, NodeConfig from ..client_utils import DEFAULT, DefaultType, client_meta_version from ._base import ( BUILTIN_EXCEPTIONS, DEFAULT_CA_CERTS, RERAISE_EXCEPTIONS, BaseNode, NodeApiResponse, ssl_context_from_node_config, ) try: from ._urllib3_chain_certs import HTTPSConnectionPool except (ImportError, AttributeError): HTTPSConnectionPool = urllib3.HTTPSConnectionPool # type: ignore[assignment,misc] class Urllib3HttpNode(BaseNode): """Default synchronous node class using the ``urllib3`` library via HTTP""" _CLIENT_META_HTTP_CLIENT = ("ur", client_meta_version(metadata.version("urllib3"))) def __init__(self, config: NodeConfig): super().__init__(config) pool_class = urllib3.HTTPConnectionPool kw: Dict[str, Any] = {} if config.scheme == "https": pool_class = HTTPSConnectionPool ssl_context = ssl_context_from_node_config(config) kw["ssl_context"] = ssl_context if config.ssl_assert_hostname and config.ssl_assert_fingerprint: raise ValueError( "Can't specify both 'ssl_assert_hostname' and 'ssl_assert_fingerprint'" ) # Fingerprint verification doesn't require CA certificates being loaded. # We also want to disable other verification methods as we only care # about the fingerprint of the certificates, not whether they form # a verified chain to a trust anchor. elif config.ssl_assert_fingerprint: # Manually disable these in the right order on the SSLContext # so urllib3 won't think we want conflicting things. ssl_context.check_hostname = False ssl_context.verify_mode = ssl.CERT_NONE kw.update( { "assert_fingerprint": config.ssl_assert_fingerprint, "assert_hostname": False, "cert_reqs": "CERT_NONE", } ) else: kw["assert_hostname"] = config.ssl_assert_hostname # Convert all sentinel values to their actual default # values if not using an SSLContext. ca_certs = ( DEFAULT_CA_CERTS if config.ca_certs is None else config.ca_certs ) if config.verify_certs: if not ca_certs: raise ValueError( "Root certificates are missing for certificate " "validation. Either pass them in using the ca_certs parameter or " "install certifi to use it automatically." ) kw.update( { "cert_reqs": "CERT_REQUIRED", "ca_certs": ca_certs, "cert_file": config.client_cert, "key_file": config.client_key, } ) else: kw["cert_reqs"] = "CERT_NONE" if config.ssl_show_warn: warnings.warn( f"Connecting to {self.base_url!r} using TLS with verify_certs=False is insecure", stacklevel=warn_stacklevel(), category=SecurityWarning, ) else: urllib3.disable_warnings() self.pool = pool_class( config.host, port=config.port, timeout=urllib3.Timeout(total=config.request_timeout), maxsize=config.connections_per_node, block=True, kw, ) def perform_request( self, method: str, target: str, body: Optional[bytes] = None, headers: Optional[HttpHeaders] = None, request_timeout: Union[DefaultType, Optional[float]] = DEFAULT, ) -> NodeApiResponse: if self.path_prefix: target = f"{self.path_prefix}{target}" start = time.time() try: kw = {} if request_timeout is not DEFAULT: kw["timeout"] = request_timeout request_headers = self._headers.copy() if headers: request_headers.update(headers) body_to_send: Optional[bytes] if body: if self._http_compress: body_to_send = gzip.compress(body) request_headers["content-encoding"] = "gzip" else: body_to_send = body else: body_to_send = None response = self.pool.urlopen( method, target, body=body_to_send, retries=Retry(False), headers=request_headers, kw, # type: ignore[arg-type] ) response_headers = HttpHeaders(response.headers) data = response.data duration = time.time() - start except RERAISE_EXCEPTIONS: raise except Exception as e: err: Exception if isinstance(e, NewConnectionError): err = ConnectionError(str(e), errors=(e,)) elif isinstance(e, (ConnectTimeoutError, ReadTimeoutError)): err = ConnectionTimeout( "Connection timed out during request", errors=(e,) ) elif isinstance(e, (ssl.SSLError, urllib3.exceptions.SSLError)): err = TlsError(str(e), errors=(e,)) elif isinstance(e, BUILTIN_EXCEPTIONS): raise else: err = ConnectionError(str(e), errors=(e,)) self._log_request( method=method, target=target, headers=request_headers, body=body, exception=err, ) raise err from e meta = ApiResponseMeta( node=self.config, duration=duration, http_version="1.1", status=response.status, headers=response_headers, ) self._log_request( method=method, target=target, headers=request_headers, body=body, meta=meta, response=data, ) return NodeApiResponse( meta, data, ) def close(self) -> None: """ Explicitly closes connection """ self.pool.close()	7.93 KB	2025-05-25 12:26
📄 _urllib3_chain_certs.py # Licensed to Elasticsearch B.V. under one or more contributor # license agreements. See the NOTICE file distributed with # this work for additional information regarding copyright # ownership. Elasticsearch B.V. licenses this file to you under # the Apache License, Version 2.0 (the "License"); you may # not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, # software distributed under the License is distributed on an # "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY # KIND, either express or implied. See the License for the # specific language governing permissions and limitations # under the License. import hashlib import sys from binascii import hexlify, unhexlify from hmac import compare_digest from typing import Any, List, Optional import _ssl # type: ignore import urllib3 import urllib3.connection from ._base import RERAISE_EXCEPTIONS if sys.version_info < (3, 10) or sys.implementation.name != "cpython": raise ImportError("Only supported on CPython 3.10+") _ENCODING_DER: int = _ssl.ENCODING_DER _HASHES_BY_LENGTH = {32: hashlib.md5, 40: hashlib.sha1, 64: hashlib.sha256} __all__ = ["HTTPSConnectionPool"] class HTTPSConnection(urllib3.connection.HTTPSConnection): def __init__(self, args: Any, kwargs: Any) -> None: self._elastic_assert_fingerprint: Optional[str] = None super().__init__(args, *kwargs) def connect(self) -> None: super().connect() # Hack to prevent a warning within HTTPSConnectionPool._validate_conn() if self._elastic_assert_fingerprint: self.is_verified = True class HTTPSConnectionPool(urllib3.HTTPSConnectionPool): ConnectionCls = HTTPSConnection """HTTPSConnectionPool implementation which supports ``assert_fingerprint`` on certificates within the chain instead of only the leaf cert using private APIs in CPython 3.10+ """ def __init__( self, args: Any, assert_fingerprint: Optional[str] = None, *kwargs: Any ) -> None: self._elastic_assert_fingerprint = ( assert_fingerprint.replace(":", "").lower() if assert_fingerprint else None ) # Complain about fingerprint length earlier than urllib3 does. if ( self._elastic_assert_fingerprint and len(self._elastic_assert_fingerprint) not in _HASHES_BY_LENGTH ): valid_lengths = "', '".join(map(str, sorted(_HASHES_BY_LENGTH.keys()))) raise ValueError( f"Fingerprint of invalid length '{len(self._elastic_assert_fingerprint)}'" f", should be one of '{valid_lengths}'" ) if self._elastic_assert_fingerprint: # Skip fingerprinting by urllib3 as we'll do it ourselves kwargs["assert_fingerprint"] = None super().__init__(args, **kwargs) def _new_conn(self) -> HTTPSConnection: """ Return a fresh :class:`urllib3.connection.HTTPSConnection`. """ conn: HTTPSConnection = super()._new_conn() # type: ignore[assignment] # Tell our custom connection if we'll assert fingerprint ourselves conn._elastic_assert_fingerprint = self._elastic_assert_fingerprint return conn def _validate_conn(self, conn: HTTPSConnection) -> None: # type: ignore[override] """ Called right before a request is made, after the socket is created. """ super(HTTPSConnectionPool, self)._validate_conn(conn) if self._elastic_assert_fingerprint: hash_func = _HASHES_BY_LENGTH[len(self._elastic_assert_fingerprint)] assert_fingerprint = unhexlify( self._elastic_assert_fingerprint.lower() .replace(":", "") .encode("ascii") ) fingerprints: List[bytes] try: if sys.version_info >= (3, 13): fingerprints = [ hash_func(cert).digest() for cert in conn.sock.get_verified_chain() # type: ignore ] else: # 'get_verified_chain()' and 'Certificate.public_bytes()' are private APIs # in CPython 3.10. They're not documented anywhere yet but seem to work # and we need them for Security on by Default so... onwards we go! # See: https://github.com/python/cpython/pull/25467 fingerprints = [ hash_func(cert.public_bytes(_ENCODING_DER)).digest() for cert in conn.sock._sslobj.get_verified_chain() # type: ignore[union-attr] ] except RERAISE_EXCEPTIONS: # pragma: nocover raise # Because these are private APIs we are super careful here # so that if anything "goes wrong" we fallback on the old behavior. except Exception: # pragma: nocover fingerprints = [] # Only add the peercert in front of the chain if it's not there for some reason. # This is to make sure old behavior of 'ssl_assert_fingerprint' still works. peercert_fingerprint = hash_func(conn.sock.getpeercert(True)).digest() # type: ignore[union-attr] if peercert_fingerprint not in fingerprints: # pragma: nocover fingerprints.insert(0, peercert_fingerprint) # If any match then that's a success! We always run them # all through though because of constant time concerns. success = False for fingerprint in fingerprints: success \|= compare_digest(fingerprint, assert_fingerprint) # Give users all the fingerprints we checked against in # order of peer -> root CA. if not success: raise urllib3.exceptions.SSLError( 'Fingerprints did not match. Expected "{0}", got "{1}".'.format( self._elastic_assert_fingerprint, '", "'.join([x.decode() for x in map(hexlify, fingerprints)]), ) ) conn.is_verified = success	6.24 KB	2025-05-25 12:26

Terminal