Don't send empty credentials to SR in Authorization Header

confluent_kafka/schema_registry/schema_registry_client.py

@@ -110,7 +110,7 @@ def __init__(self, conf):
                 raise ValueError("basic.auth.user.info must be in the form"
                                  " of {username}:{password}")
 
-        self.session.auth = userinfo
+        self.session.auth = userinfo if userinfo != ('', '') else None
 
         # Any leftover keys are unknown to _RestClient
         if len(conf_copy) > 0:

examples/adminapi.py

@@ -270,8 +270,9 @@ def example_list(a, args):
             else:
                 errstr = ""
 
-            print("    partition {} leader: {}, replicas: {}, isrs: {}".format(
-                p.id, p.leader, p.replicas, p.isrs, errstr))
+            print("partition {} leader: {}, replicas: {},"
+                  " isrs: {} errstr: {}".format(p.id, p.leader, p.replicas,
+                                                p.isrs, errstr))
 
 
 if __name__ == '__main__':

tests/integration/producer/test_transactions.py

@@ -41,8 +41,8 @@ def prefixed_delivery_cb(prefix):
     def delivery_err(err, msg):
         """ Reports failed message delivery to aid in troubleshooting test failures. """
         if err:
-            print("[{}]: Message delivery failed (%s [%s]): %s".format(prefix,
-                                                                       (msg.topic(), str(msg.partition()), err)))
+            print("[{}]: Message delivery failed ({} [{}]): {}".format(
+                prefix, msg.topic(), str(msg.partition()), err))
             return
 
     return delivery_err

tests/integration/schema_registry/test_api_client.py

@@ -413,6 +413,6 @@ def test_api_config_update(kafka_cluster):
     """
     sr = kafka_cluster.schema_registry()
 
-    for l in ["BACKWARD", "BACKWARD_TRANSITIVE", "FORWARD", "FORWARD_TRANSITIVE"]:
-        sr.set_compatibility(level=l)
-        assert sr.get_compatibility()['compatibilityLevel'] == l
+    for level in ["BACKWARD", "BACKWARD_TRANSITIVE", "FORWARD", "FORWARD_TRANSITIVE"]:
+        sr.set_compatibility(level=level)
+        assert sr.get_compatibility()['compatibilityLevel'] == level

tests/schema_registry/conftest.py

@@ -17,10 +17,12 @@
 #
 import os
 import re
+from base64 import b64decode
 from collections import defaultdict
 
 import pytest
 import requests_mock
+from requests_mock import create_response
 
 from confluent_kafka.schema_registry.schema_registry_client import \
     SchemaRegistryClient
@@ -68,6 +70,8 @@ class MockSchemaRegistryClient(SchemaRegistryClient):
     the only endpoint which supports this is /config which will return an
     `Invalid compatibility level` error.
 
+    To coerce Authentication errors configure credentials to
+    not match MockSchemaRegistryClient.USERINFO.
 
     Request paths to trigger exceptions:
     +--------+----------------------------------+-------+------------------------------+
@@ -130,6 +134,7 @@ class MockSchemaRegistryClient(SchemaRegistryClient):
     VERSIONS = [1, 2, 3, 4]
     SCHEMA = 'basic_schema.avsc'
     SUBJECTS = ['subject1', 'subject2']
+    USERINFO = 'mock_user:mock_password'
 
     # Counts requests handled per path by HTTP method
     # {HTTP method: { path : count}}
@@ -164,8 +169,29 @@ def __init__(self, conf):
         adapter.register_uri('POST', self.subject_versions,
                              json=self.post_subject_version_callback)
 
+        adapter.add_matcher(self._auth_matcher)
         self._rest_client.session.mount('http://', adapter)
 
+    @classmethod
+    def _auth_matcher(cls, request):
+        headers = request._request.headers
+
+        authinfo = headers.get('Authorization', None)
+        # Pass request to downstream matchers
+        if authinfo is None:
+            return None
+
+        # We only support the BASIC scheme today
+        scheme, userinfo = authinfo.split(" ")
+        if b64decode(userinfo) == cls.USERINFO:
+            return None
+
+        unauthorized = {'error_code': 401,
+                        'message': "401 Unauthorized"}
+        return create_response(request=request,
+                               status_code=401,
+                               json=unauthorized)
+
     @staticmethod
     def _load_avsc(name):
         with open(os.path.join(work_dir, '..', 'integration', 'schema_registry',

tests/schema_registry/test_api_client.py

@@ -55,6 +55,25 @@ def cmp_schema(schema1, schema2):
                 schema1.schema_type == schema2.schema_type])
 
 
+def test_basic_auth_unauthorized(mock_schema_registry, load_avsc):
+    conf = {'url': TEST_URL,
+            'basic.auth.user.info': "user:secret"}
+    sr = mock_schema_registry(conf)
+
+    with pytest.raises(SchemaRegistryError, match="401 Unauthorized"):
+        sr.get_subjects()
+
+
+def test_basic_auth_authorized(mock_schema_registry, load_avsc):
+    conf = {'url': TEST_URL,
+            'basic.auth.user.info': mock_schema_registry.USERINFO}
+    sr = mock_schema_registry(conf)
+
+    result = sr.get_subjects()
+
+    assert result == mock_schema_registry.SUBJECTS
+
+
 def test_register_schema(mock_schema_registry, load_avsc):
     conf = {'url': TEST_URL}
     sr = mock_schema_registry(conf)

tests/schema_registry/test_config.py

@@ -111,8 +111,8 @@ def test_config_auth_userinfo():
             'basic.auth.user.info': TEST_USERNAME + ':' + TEST_USER_PASSWORD}
 
     test_client = SchemaRegistryClient(conf)
-    assert test_client._rest_client.session.auth == [TEST_USERNAME,
-                                                     TEST_USER_PASSWORD]
+    assert test_client._rest_client.session.auth == (TEST_USERNAME,
+                                                     TEST_USER_PASSWORD)
 
 
 def test_config_auth_userinfo_invalid():

tools/RELEASE.md

@@ -147,12 +147,13 @@ Create a new virtualenv:
 
 Install the relevant package for your platform:
 
-    $ pip install dl-v0.11.4rc1/confluent_kafka-....whl
+    $ pip install --no-cache-dir --find-links dl-v0.11.4rc1/ confluent-kafka
 
 Verify that the package works, should print the expected Python client
 and librdkafka versions:
 
-    $ python -c 'import confluent_kafka as ck ; print "py:", ck.version(), "c:", ck.libversion()'
+    $ python -c 'import confluent_kafka as ck ; print "py: {} c: {}" \
+                 .format(ck.version(), ck.libversion())'
     py: ('0.11.4', 721920) c: ('0.11.4-RC1', 722121)
 
 ## 10. Open a release PR
@@ -203,7 +204,7 @@ In the same virtualenv as created above:
 
 
 Verify that the package works and prints the expected version:
-    $ python -c 'import confluent_kafka as ck ; print "py:", ck.version(), "c:", ck.libversion()'
+    $ python -c 'import confluent_kafka as ck ; print("py:", ck.version(), "c:", ck.libversion())'
     py: ('0.11.4', 721920) c: ('0.11.4-RC1', 722121)