make host url parsing more resilient

Reject invalid combinations of url scheme and connection_class

elasticsearch/client/__init__.py

@@ -29,7 +29,7 @@ def _normalize_hosts(hosts):
 
     out = []
     # normalize hosts to dicts
-    for i, host in enumerate(hosts):
+    for host in hosts:
         if isinstance(host, string_types):
             if '://' not in host:
                 host = "//%s" % host
@@ -43,11 +43,10 @@ def _normalize_hosts(hosts):
             if parsed_url.scheme == "https":
                 h['port'] = parsed_url.port or 443
                 h['use_ssl'] = True
-            elif parsed_url.scheme == "http":
-                logger.warning(
-                    "List of nodes should not include schema information (http://): %r.",
-                    host
-                )
+                h['scheme'] = 'http'
+
+            elif parsed_url.scheme:
+                h['scheme'] = parsed_url.scheme
 
             if parsed_url.username or parsed_url.password:
                 h['http_auth'] = '%s:%s' % (parsed_url.username, parsed_url.password)

elasticsearch/transport.py

@@ -5,7 +5,7 @@ from .connection import Urllib3HttpConnection
 from .connection_pool import ConnectionPool
 from .serializer import JSONSerializer, Deserializer, DEFAULT_SERIALIZERS
 from .exceptions import ConnectionError, TransportError, SerializationError, \
-                        ConnectionTimeout
+                        ConnectionTimeout, ImproperlyConfigured
 
 # get ip/port from "inet[wind/127.0.0.1:9200]"
 ADDRESS_RE = re.compile(r'/(?P<host>[\.:0-9a-f]*):(?P<port>[0-9]+)\]?$')
@@ -153,6 +153,12 @@ class Transport(object):
             # previously unseen params, create new connection
             kwargs = self.kwargs.copy()
             kwargs.update(host)
+
+            if 'scheme' in host and host['scheme'] != self.connection_class.transport_schema:
+                raise ImproperlyConfigured(
+                    'Scheme specified in connection (%s) is not the same as the connection class (%s) specifies (%s).' % (
+                        host['scheme'], self.connection_class.__name__, self.connection_class.transport_schema
+                ))
             return self.connection_class(**kwargs)
         connections = map(_create_connection, hosts)
 

test_elasticsearch/test_client/__init__.py

@@ -1,7 +1,5 @@
 from __future__ import unicode_literals
 
-from mock import patch
-
 from elasticsearch.client import _normalize_hosts, Elasticsearch
 
 from ..test_cases import TestCase, ElasticsearchTestCase
@@ -21,23 +19,16 @@ class TestNormalizeHosts(TestCase):
 
     def test_strings_are_parsed_for_scheme(self):
         self.assertEquals(
-            [{"host": "elasticsearch.org", "port": 42, "use_ssl": True},
-             {"host": "elasticsearch.com", "http_auth": "user:secret", "use_ssl": True, "port": 443}],
+            [
+                {"host": "elasticsearch.org", "port": 42, "use_ssl": True, 'scheme': 'http'},
+                {"host": "elasticsearch.com", "http_auth": "user:secret", "use_ssl": True, "port": 443, 'scheme': 'http'}
+            ],
             _normalize_hosts(["https://elasticsearch.org:42", "https://user:[email protected]"])
         )
 
     def test_dicts_are_left_unchanged(self):
         self.assertEquals([{"host": "local", "extra": 123}], _normalize_hosts([{"host": "local", "extra": 123}]))
 
-    @patch('elasticsearch.client.logger')
-    def test_schema_is_stripped_out(self, logger):
-        self.assertEquals(
-            [{"host": "elasticsearch.org", "port": 9200}],
-            _normalize_hosts(["http://elasticsearch.org:9200/"])
-        )
-        # schema triggers a warning
-        self.assertEquals(1, logger.warning.call_count)
-
     def test_single_string_is_wrapped_in_list(self):
         self.assertEquals(
             [{"host": "elasticsearch.org"}],

test_elasticsearch/test_transport.py

@@ -3,8 +3,8 @@ from __future__ import unicode_literals
 import time
 
 from elasticsearch.transport import Transport, get_host_info
-from elasticsearch.connection import Connection
-from elasticsearch.exceptions import ConnectionError
+from elasticsearch.connection import Connection, ThriftConnection
+from elasticsearch.exceptions import ConnectionError, ImproperlyConfigured
 
 from .test_cases import TestCase
 
@@ -52,6 +52,10 @@ class TestHostsInfoCallback(TestCase):
 
 
 class TestTransport(TestCase):
+    def test_host_with_scheme_different_from_connection_fails(self):
+        self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'thrift'}])
+        self.assertRaises(ImproperlyConfigured, Transport, [{'host': 'localhost', 'scheme': 'http'}], connection_class=ThriftConnection)
+
     def test_request_timeout_extracted_from_params_and_passed(self):
         t = Transport([{}], connection_class=DummyConnection)