Verify connection to Elasticsearch

Author: sethmlarson

elasticsearch/_async/transport.py

@@ -18,15 +18,19 @@
 import asyncio
 import logging
 import sys
+import warnings
 from itertools import chain
 
 from ..exceptions import (
+    AuthenticationException,
+    AuthorizationException,
     ConnectionError,
     ConnectionTimeout,
+    ElasticsearchWarning,
     SerializationError,
     TransportError,
 )
-from ..transport import Transport
+from ..transport import Transport, _verify_elasticsearch
 from .compat import get_running_loop
 from .http_aiohttp import AIOHttpConnection
 
@@ -327,6 +331,10 @@ async def perform_request(self, method, url, headers=None, params=None, body=Non
             method, headers, params, body
         )
 
+        # Before we make the actual API call we verify the Elasticsearch instance.
+        if not self._verified_elasticsearch:
+            await self._do_verify_elasticsearch(headers=headers, timeout=timeout)
+
         for attempt in range(self.max_retries + 1):
             connection = self.get_connection()
 
@@ -398,3 +406,67 @@ async def close(self):
 
         for connection in self.connection_pool.connections:
             await connection.close()
+
+    async def _do_verify_elasticsearch(self, headers, timeout):
+        """Verifies that we're connected to an Elasticsearch cluster.
+        This is done at least once before the first actual API call
+        and makes a single request to the 'GET /' API endpoint and
+        check version along with other details of the response.
+
+        If we're unable to verify we're talking to Elasticsearch
+        but we're also unable to rule it out due to a permission
+        error we instead emit an 'ElasticsearchWarning'.
+        """
+        # Product check has already been done, no need to do again.
+        if self._verified_elasticsearch:
+            return
+
+        headers = {header.lower(): value for header, value in (headers or {}).items()}
+        # We know we definitely want JSON so request it via 'accept'
+        headers.setdefault("accept", "application/json")
+
+        info_headers = {}
+        info_response = {}
+        info_error = None
+
+        for conn in chain(self.connection_pool.connections, self.seed_connections):
+            try:
+                _, info_headers, info_response = await conn.perform_request(
+                    "GET", "/", headers=headers, timeout=timeout
+                )
+
+                # Lowercase all the header names for consistency in accessing them.
+                info_headers = {
+                    header.lower(): value for header, value in info_headers.items()
+                }
+
+                info_response = self.deserializer.loads(
+                    info_response, mimetype="application/json"
+                )
+                break
+
+            # Previous versions of 7.x Elasticsearch required a specific
+            # permission so if we receive HTTP 401/403 we should warn
+            # instead of erroring out.
+            except (AuthenticationException, AuthorizationException):
+                warnings.warn(
+                    (
+                        "The client is unable to verify that the server is "
+                        "Elasticsearch due security privileges on the server side"
+                    ),
+                    ElasticsearchWarning,
+                    stacklevel=3,
+                )
+                self._verified_elasticsearch = True
+                return
+
+            # This connection didn't work, we'll try another.
+            except (ConnectionError, SerializationError):
+                if info_error is None:
+                    info_error = info_error
+
+        # Check the information we got back from the index request.
+        _verify_elasticsearch(info_headers, info_response)
+
+        # If we made it through the above call this config is verified.
+        self._verified_elasticsearch = True

elasticsearch/exceptions.py

@@ -51,6 +51,12 @@ class SerializationError(ElasticsearchException):
     """
 
 
+class NotElasticsearchError(ElasticsearchException):
+    """Error which is raised when the client detects
+    it's not connected to an Elasticsearch cluster.
+    """
+
+
 class TransportError(ElasticsearchException):
     """
     Exception raised when ES returns a non-OK (>=400) HTTP status code. Or when

elasticsearch/exceptions.pyi

@@ -20,6 +20,7 @@ from typing import Any, Dict, Union
 class ImproperlyConfigured(Exception): ...
 class ElasticsearchException(Exception): ...
 class SerializationError(ElasticsearchException): ...
+class NotElasticsearchError(ElasticsearchException): ...
 
 class TransportError(ElasticsearchException):
     @property

elasticsearch/transport.py

@@ -15,16 +15,22 @@
 #  specific language governing permissions and limitations
 #  under the License.
 
+import re
 import time
+import warnings
 from itertools import chain
 from platform import python_version
 
 from ._version import __versionstr__
 from .connection import Urllib3HttpConnection
 from .connection_pool import ConnectionPool, DummyConnectionPool, EmptyConnectionPool
 from .exceptions import (
+    AuthenticationException,
+    AuthorizationException,
     ConnectionError,
     ConnectionTimeout,
+    ElasticsearchWarning,
+    NotElasticsearchError,
     SerializationError,
     TransportError,
 )
@@ -198,6 +204,10 @@ def __init__(
         if http_client_meta:
             self._client_meta += (http_client_meta,)
 
+        # Flag which is set after verifying that we're
+        # connected to Elasticsearch.
+        self._verified_elasticsearch = False
+
     def add_connection(self, host):
         """
         Create a new :class:`~elasticsearch.Connection` instance and add it to the pool.
@@ -380,6 +390,9 @@ def perform_request(self, method, url, headers=None, params=None, body=None):
             method, headers, params, body
         )
 
+        # Before we make the actual API call we verify the Elasticsearch instance.
+        self._do_verify_elasticsearch(headers=headers, timeout=timeout)
+
         for attempt in range(self.max_retries + 1):
             connection = self.get_connection()
 
@@ -488,3 +501,115 @@ def _resolve_request_args(self, method, headers, params, body):
             )
 
         return method, headers, params, body, ignore, timeout
+
+    def _do_verify_elasticsearch(self, headers, timeout):
+        """Verifies that we're connected to an Elasticsearch cluster.
+        This is done at least once before the first actual API call
+        and makes a single request to the 'GET /' API endpoint to
+        check the version along with other details of the response.
+
+        If we're unable to verify we're talking to Elasticsearch
+        but we're also unable to rule it out due to a permission
+        error we instead emit an 'ElasticsearchWarning'.
+        """
+        # Product check has already been done, no need to do again.
+        if self._verified_elasticsearch:
+            return
+
+        headers = {header.lower(): value for header, value in (headers or {}).items()}
+        # We know we definitely want JSON so request it via 'accept'
+        headers.setdefault("accept", "application/json")
+
+        info_headers = {}
+        info_response = {}
+
+        for conn in chain(self.connection_pool.connections, self.seed_connections):
+            try:
+                _, info_headers, info_response = conn.perform_request(
+                    "GET", "/", headers=headers, timeout=timeout
+                )
+
+                # Lowercase all the header names for consistency in accessing them.
+                info_headers = {
+                    header.lower(): value for header, value in info_headers.items()
+                }
+
+                info_response = self.deserializer.loads(
+                    info_response, mimetype="application/json"
+                )
+                break
+
+            # Previous versions of 7.x Elasticsearch required a specific
+            # permission so if we receive HTTP 401/403 we should warn
+            # instead of erroring out.
+            except (AuthenticationException, AuthorizationException):
+                warnings.warn(
+                    (
+                        "The client is unable to verify that the server is "
+                        "Elasticsearch due security privileges on the server side"
+                    ),
+                    ElasticsearchWarning,
+                    stacklevel=3,
+                )
+                self._verified_elasticsearch = True
+                return
+
+            # This connection didn't work, we'll try another.
+            except (ConnectionError, SerializationError):
+                pass
+
+        # Check the information we got back from the index request.
+        _verify_elasticsearch(info_headers, info_response)
+
+        # If we made it through the above call this config is verified.
+        self._verified_elasticsearch = True
+
+
+def _verify_elasticsearch(headers, response):
+    """Verifies that the server we're talking to is Elasticsearch.
+    Does this by checking HTTP headers and the deserialized
+    response to the 'info' API.
+
+    If there's a problem this function raises 'NotElasticsearchError'
+    otherwise doesn't do anything.
+    """
+    try:
+        version = response.get("version", {})
+        version_number = tuple(
+            int(x) if x is not None else 999
+            for x in re.search(
+                r"^([0-9]+)\.([0-9]+)(?:\.([0-9]+))?", version["number"]
+            ).groups()
+        )
+    except (KeyError, TypeError, ValueError, AttributeError):
+        # No valid 'version.number' field, effectively 0.0.0
+        version = {}
+        version_number = (0, 0, 0)
+
+    # Check all of the fields and headers for missing/valid values.
+    try:
+        bad_tagline = response.get("tagline", None) != "You Know, for Search"
+        bad_build_flavor = version.get("build_flavor", None) != "default"
+        bad_product_header = headers.get("x-elastic-product", None) != "Elasticsearch"
+    except (AttributeError, TypeError):
+        bad_tagline = True
+        bad_build_flavor = True
+        bad_product_header = True
+
+    if (
+        # No version or version less than 6.x
+        version_number < (6, 0, 0)
+        # 6.x and there's a bad 'tagline'
+        or ((6, 0, 0) <= version_number < (7, 0, 0) and bad_tagline)
+        # 7.0-7.13 and there's a bad 'tagline' or 'build_flavor'
+        or (
+            (7, 0, 0) <= version_number < (7, 14, 0)
+            and (bad_tagline or bad_build_flavor)
+        )
+        # 7.14+ and there's a bad 'X-Elastic-Product' HTTP header
+        or ((7, 14, 0) <= version_number and bad_product_header)
+    ):
+        raise NotElasticsearchError(
+            "The client noticed that the server is not Elasticsearch "
+            "and we do not support this unknown product"
+        )