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