1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381
|
# 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 dataclasses
import enum
import re
import ssl
from dataclasses import dataclass, field
from typing import (
TYPE_CHECKING,
Any,
Collection,
Dict,
Iterator,
KeysView,
Mapping,
MutableMapping,
Optional,
Tuple,
TypeVar,
Union,
ValuesView,
)
if TYPE_CHECKING:
from typing import Final
class DefaultType(enum.Enum):
"""
Sentinel used as a default value when ``None`` has special meaning like timeouts.
The only comparisons that are supported for this type are ``is``.
"""
value = 0
def __repr__(self) -> str:
return "<DEFAULT>"
def __str__(self) -> str:
return "<DEFAULT>"
DEFAULT: "Final[DefaultType]" = DefaultType.value
T = TypeVar("T")
_TYPE_SSL_VERSION = Union[int, ssl.TLSVersion]
class HttpHeaders(MutableMapping[str, str]):
"""HTTP headers
Behaves like a Python dictionary. Can be used like this::
headers = HttpHeaders()
headers["foo"] = "bar"
headers["foo"] = "baz"
print(headers["foo"]) # prints "baz"
"""
__slots__ = ("_internal", "_frozen")
def __init__(
self,
initial: Optional[Union[Mapping[str, str], Collection[Tuple[str, str]]]] = None,
) -> None:
self._internal = {}
self._frozen = False
if initial:
for key, val in dict(initial).items():
self._internal[self._normalize_key(key)] = (key, val)
def __setitem__(self, key: str, value: str) -> None:
if self._frozen:
raise ValueError("Can't modify headers that have been frozen")
self._internal[self._normalize_key(key)] = (key, value)
def __getitem__(self, item: str) -> str:
return self._internal[self._normalize_key(item)][1]
def __delitem__(self, key: str) -> None:
if self._frozen:
raise ValueError("Can't modify headers that have been frozen")
del self._internal[self._normalize_key(key)]
def __eq__(self, other: object) -> bool:
if not isinstance(other, Mapping):
return NotImplemented
if not isinstance(other, HttpHeaders):
other = HttpHeaders(other)
return {k: v for k, (_, v) in self._internal.items()} == {
k: v for k, (_, v) in other._internal.items()
}
def __ne__(self, other: object) -> bool:
if not isinstance(other, Mapping):
return NotImplemented
return not self == other
def __iter__(self) -> Iterator[str]:
return iter(self.keys())
def __len__(self) -> int:
return len(self._internal)
def __bool__(self) -> bool:
return bool(self._internal)
def __contains__(self, item: object) -> bool:
return isinstance(item, str) and self._normalize_key(item) in self._internal
def __repr__(self) -> str:
return repr(self._dict_hide_auth())
def __str__(self) -> str:
return str(self._dict_hide_auth())
def __hash__(self) -> int:
if not self._frozen:
raise ValueError("Can't calculate the hash of headers that aren't frozen")
return hash(tuple((k, v) for k, (_, v) in sorted(self._internal.items())))
def get(self, key: str, default: Optional[str] = None) -> Optional[str]: # type: ignore[override]
return self._internal.get(self._normalize_key(key), (None, default))[1]
def keys(self) -> KeysView[str]:
return self._internal.keys()
def values(self) -> ValuesView[str]:
return {"": v for _, v in self._internal.values()}.values()
def items(self) -> Collection[Tuple[str, str]]: # type: ignore[override]
return [(key, val) for _, (key, val) in self._internal.items()]
def freeze(self) -> "HttpHeaders":
"""Freezes the current set of headers so they can be used in hashes.
Returns the same instance, doesn't make a copy.
"""
self._frozen = True
return self
@property
def frozen(self) -> bool:
return self._frozen
def copy(self) -> "HttpHeaders":
return HttpHeaders(self.items())
def _normalize_key(self, key: str) -> str:
try:
return key.lower()
except AttributeError:
return key
def _dict_hide_auth(self) -> Dict[str, str]:
def hide_auth(val: str) -> str:
# Hides only the authentication value, not the method.
match = re.match(r"^(ApiKey|Basic|Bearer) ", val)
if match:
return f"{match.group(1)} <hidden>"
return "<hidden>"
return {
key: hide_auth(val) if key.lower() == "authorization" else val
for key, val in self.items()
}
@dataclass
class ApiResponseMeta:
"""Metadata that is returned from Transport.perform_request()
:ivar int status: HTTP status code
:ivar str http_version: HTTP version being used
:ivar HttpHeaders headers: HTTP headers
:ivar float duration: Number of seconds from start of request to start of response
:ivar NodeConfig node: Node which handled the request
:ivar typing.Optional[str] mimetype: Mimetype to be used by the serializer to decode the raw response bytes.
"""
status: int
http_version: str
headers: HttpHeaders
duration: float
node: "NodeConfig"
@property
def mimetype(self) -> Optional[str]:
try:
content_type = self.headers["content-type"]
return content_type.partition(";")[0] or None
except KeyError:
return None
def _empty_frozen_http_headers() -> HttpHeaders:
"""Used for the 'default_factory' of the 'NodeConfig.headers'"""
return HttpHeaders().freeze()
@dataclass(repr=True)
class NodeConfig:
"""Configuration options available for every node."""
#: Protocol in use to connect to the node
scheme: str
#: IP address or hostname to connect to
host: str
#: IP port to connect to
port: int
#: Prefix to add to the path of every request
path_prefix: str = ""
#: Default HTTP headers to add to every request
headers: Union[HttpHeaders, Mapping[str, str]] = field(
default_factory=_empty_frozen_http_headers
)
#: Number of concurrent connections that are
#: able to be open at one time for this node.
#: Having multiple connections per node allows
#: for higher concurrency of requests.
connections_per_node: int = 10
#: Number of seconds to wait before a request should timeout.
request_timeout: Optional[float] = 10.0
#: Set to ``True`` to enable HTTP compression
#: of request and response bodies via gzip.
http_compress: Optional[bool] = False
#: Set to ``True`` to verify the node's TLS certificate against 'ca_certs'
#: Setting to ``False`` will disable verifying the node's certificate.
verify_certs: Optional[bool] = True
#: Path to a CA bundle or directory containing bundles. By default
#: If the ``certifi`` package is installed and ``verify_certs`` is
#: set to ``True`` this value will be set to ``certifi.where()``.
ca_certs: Optional[str] = None
#: Path to a client certificate for TLS client authentication.
client_cert: Optional[str] = None
#: Path to a client private key for TLS client authentication.
client_key: Optional[str] = None
#: Hostname or IP address to verify on the node's certificate.
#: This is useful if the certificate contains a different value
#: than the one supplied in ``host``. An example of this situation
#: is connecting to an IP address instead of a hostname.
#: Set to ``False`` to disable certificate hostname verification.
ssl_assert_hostname: Optional[str] = None
#: SHA-256 fingerprint of the node's certificate. If this value is
#: given then root-of-trust verification isn't done and only the
#: node's certificate fingerprint is verified.
#:
#: On CPython 3.10+ this also verifies if any certificate in the
#: chain including the Root CA matches this fingerprint. However
#: because this requires using private APIs support for this is
#: **experimental**.
ssl_assert_fingerprint: Optional[str] = None
#: Minimum TLS version to use to connect to the node. Can be either
#: :class:`ssl.TLSVersion` or one of the deprecated
#: ``ssl.PROTOCOL_TLSvX`` instances.
ssl_version: Optional[_TYPE_SSL_VERSION] = None
#: Pre-configured :class:`ssl.SSLContext` object. If this value
#: is given then no other TLS options (besides ``ssl_assert_fingerprint``)
#: can be set on the :class:`elastic_transport.NodeConfig`.
ssl_context: Optional[ssl.SSLContext] = field(default=None, hash=False)
#: Set to ``False`` to disable the :class:`elastic_transport.SecurityWarning`
#: issued when using ``verify_certs=False``.
ssl_show_warn: bool = True
#: Extras that can be set to anything, typically used
#: for annotating this node with additional information for
#: future decisions like sniffing, instance roles, etc.
#: Third-party keys should start with an underscore and prefix.
_extras: Dict[str, Any] = field(default_factory=dict, hash=False)
def replace(self, **kwargs: Any) -> "NodeConfig":
if not kwargs:
return self
return dataclasses.replace(self, **kwargs)
def __post_init__(self) -> None:
if not isinstance(self.headers, HttpHeaders) or not self.headers.frozen:
self.headers = HttpHeaders(self.headers).freeze()
if self.scheme != self.scheme.lower():
raise ValueError("'scheme' must be lowercase")
if "[" in self.host or "]" in self.host:
raise ValueError("'host' must not have square braces")
if self.port < 0:
raise ValueError("'port' must be a positive integer")
if self.connections_per_node <= 0:
raise ValueError("'connections_per_node' must be a positive integer")
if self.path_prefix:
self.path_prefix = (
("/" + self.path_prefix.strip("/")) if self.path_prefix else ""
)
tls_options = [
"ca_certs",
"client_cert",
"client_key",
"ssl_assert_hostname",
"ssl_assert_fingerprint",
"ssl_context",
]
# Disallow setting TLS options on non-HTTPS connections.
if self.scheme != "https":
if any(getattr(self, attr) is not None for attr in tls_options):
raise ValueError("TLS options require scheme to be 'https'")
elif self.scheme == "https":
# It's not valid to set 'ssl_context' and any other
# TLS option, the SSLContext object must be configured
# the way the user wants already.
def tls_option_filter(attr: object) -> bool:
return (
isinstance(attr, str)
and attr not in ("ssl_context", "ssl_assert_fingerprint")
and getattr(self, attr) is not None
)
if self.ssl_context is not None and any(
filter(
tls_option_filter,
tls_options,
)
):
raise ValueError(
"The 'ssl_context' option can't be combined with other TLS options"
)
def __eq__(self, other: object) -> bool:
if not isinstance(other, NodeConfig):
return NotImplemented
return (
self.scheme == other.scheme
and self.host == other.host
and self.port == other.port
and self.path_prefix == other.path_prefix
)
def __ne__(self, other: object) -> bool:
if not isinstance(other, NodeConfig):
return NotImplemented
return not self == other
def __hash__(self) -> int:
return hash(
(
self.scheme,
self.host,
self.port,
self.path_prefix,
)
)
@dataclass()
class SniffOptions:
"""Options which are passed to Transport.sniff_callback"""
is_initial_sniff: bool
sniff_timeout: Optional[float]
|