Complete HTTPS support for schema registry (#90)

update tox

Author: rnpridgeon

confluent_kafka/avro/__init__.py

@@ -28,11 +28,20 @@ class AvroProducer(Producer):
 
     def __init__(self, config, default_key_schema=None,
                  default_value_schema=None, schema_registry=None):
+
         schema_registry_url = config.pop("schema.registry.url", None)
+        schema_registry_ca_location = config.pop("schema.registry.ssl.ca.location", None)
+        schema_registry_certificate_location = config.pop("schema.registry.ssl.certificate.location", None)
+        schema_registry_key_location = config.pop("schema.registry.ssl.key.location", None)
+
         if schema_registry is None:
             if schema_registry_url is None:
                 raise ValueError("Missing parameter: schema.registry.url")
-            schema_registry = CachedSchemaRegistryClient(url=schema_registry_url)
+
+            schema_registry = CachedSchemaRegistryClient(url=schema_registry_url,
+                                                         ca_location=schema_registry_ca_location,
+                                                         cert_location=schema_registry_certificate_location,
+                                                         key_location=schema_registry_key_location)
         elif schema_registry_url is not None:
             raise ValueError("Cannot pass schema_registry along with schema.registry.url config")
 
@@ -92,11 +101,20 @@ class AvroConsumer(Consumer):
                         and the standard Kafka client configuration (``bootstrap.servers`` et.al).
     """
     def __init__(self, config, schema_registry=None):
+
         schema_registry_url = config.pop("schema.registry.url", None)
+        schema_registry_ca_location = config.pop("schema.registry.ssl.ca.location", None)
+        schema_registry_certificate_location = config.pop("schema.registry.ssl.certificate.location", None)
+        schema_registry_key_location = config.pop("schema.registry.ssl.key.location", None)
+
         if schema_registry is None:
             if schema_registry_url is None:
                 raise ValueError("Missing parameter: schema.registry.url")
-            schema_registry = CachedSchemaRegistryClient(url=schema_registry_url)
+
+            schema_registry = CachedSchemaRegistryClient(url=schema_registry_url,
+                                                         ca_location=schema_registry_ca_location,
+                                                         cert_location=schema_registry_certificate_location,
+                                                         key_location=schema_registry_key_location)
         elif schema_registry_url is not None:
             raise ValueError("Cannot pass schema_registry along with schema.registry.url config")
 

confluent_kafka/avro/cached_schema_registry_client.py

@@ -29,6 +29,7 @@
 from . import loads
 
 VALID_LEVELS = ['NONE', 'FULL', 'FORWARD', 'BACKWARD']
+VALID_METHODS = ['GET', 'POST', 'PUT', 'DELETE']
 
 # Common accept header sent
 ACCEPT_HDR = "application/vnd.schemaregistry.v1+json, application/vnd.schemaregistry+json, application/json"
@@ -46,7 +47,7 @@ class CachedSchemaRegistryClient(object):
     @:param: url: url to schema registry
     """
 
-    def __init__(self, url, max_schemas_per_subject=1000):
+    def __init__(self, url, max_schemas_per_subject=1000, ca_location=None, cert_location=None, key_location=None):
         """Construct a client by passing in the base URL of the schema registry server"""
 
         self.url = url.rstrip('/')
@@ -59,32 +60,41 @@ def __init__(self, url, max_schemas_per_subject=1000):
         # subj => { schema => version }
         self.subject_to_schema_versions = defaultdict(dict)
 
-    def _send_request(self, url, method='GET', body=None, headers=None):
-        if body:
-            body = json.dumps(body)
-            body = body.encode('utf8')
-        _headers = dict()
-        _headers["Accept"] = ACCEPT_HDR
+        s = requests.Session()
+        if ca_location is not None:
+            s.verify = ca_location
+        if cert_location is not None or key_location is not None:
+            if cert_location is None or key_location is None:
+                raise ValueError(
+                    "Both schema.registry.ssl.certificate.location and schema.registry.ssl.key.location must be set")
+            s.cert = (cert_location, key_location)
+
+        self._session = s
+
+    def __del__(self):
+        self.close()
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *args):
+        self.close()
+
+    def close(self):
+        self._session.close()
+
+    def _send_request(self, url, method='GET', body=None, headers={}):
+        if method not in VALID_METHODS:
+            raise ClientError("Method {} is invalid; valid methods include {}".format(method, VALID_METHODS))
+
+        _headers = {'Accept': ACCEPT_HDR}
         if body:
             _headers["Content-Length"] = str(len(body))
             _headers["Content-Type"] = "application/vnd.schemaregistry.v1+json"
+        _headers.update(headers)
 
-        if headers:
-            for header_name in headers:
-                _headers[header_name] = headers[header_name]
-        if method == 'GET':
-            response = requests.get(url, headers=_headers)
-        elif method == 'POST':
-            response = requests.post(url, body, headers=_headers)
-        elif method == 'PUT':
-            response = requests.put(url, body, headers=_headers)
-        elif method == 'DELETE':
-            response = requests.delete(url, headers=_headers)
-        else:
-            raise ClientError("Invalid HTTP request type")
-
-        result = json.loads(response.text)
-        return (result, response.status_code)
+        response = self._session.request(method, url, headers=_headers, json=body)
+        return response.json(), response.status_code
 
     def _add_to_cache(self, cache, subject, schema, value):
         sub_cache = cache[subject]

examples/integration_test.py

@@ -45,14 +45,17 @@
 except ImportError as e:
     with_progress = False
 
+# Default test conf location
+testconf = "tests/testconf.json"
+
 # Kafka bootstrap server(s)
 bootstrap_servers = None
 
 # Confluent schema-registry
 schema_registry_url = None
 
 # Topic prefix to use
-topic = 'test'
+topic = None
 
 # API version requests are only implemented in Kafka broker >=0.10
 # but the client handles failed API version requests gracefully for older
@@ -374,6 +377,96 @@ def verify_avro():
         c.close()
 
 
+def verify_avro_https():
+    from confluent_kafka import avro
+    avsc_dir = os.path.join(os.path.dirname(__file__), os.pardir, 'tests', 'avro')
+
+    # Producer config
+    conf = {'bootstrap.servers': bootstrap_servers,
+            'error_cb': error_cb,
+            'api.version.request': api_version_request}
+
+    conf.update(testconf.get('schema_registry_https', {}))
+
+    p = avro.AvroProducer(conf)
+
+    prim_float = avro.load(os.path.join(avsc_dir, "primitive_float.avsc"))
+    prim_string = avro.load(os.path.join(avsc_dir, "primitive_string.avsc"))
+    basic = avro.load(os.path.join(avsc_dir, "basic_schema.avsc"))
+    str_value = 'abc'
+    float_value = 32.0
+
+    combinations = [
+        dict(key=float_value, key_schema=prim_float),
+        dict(value=float_value, value_schema=prim_float),
+        dict(key={'name': 'abc'}, key_schema=basic),
+        dict(value={'name': 'abc'}, value_schema=basic),
+        dict(value={'name': 'abc'}, value_schema=basic, key=float_value, key_schema=prim_float),
+        dict(value={'name': 'abc'}, value_schema=basic, key=str_value, key_schema=prim_string),
+        dict(value=float_value, value_schema=prim_float, key={'name': 'abc'}, key_schema=basic),
+        dict(value=float_value, value_schema=prim_float, key=str_value, key_schema=prim_string),
+        dict(value=str_value, value_schema=prim_string, key={'name': 'abc'}, key_schema=basic),
+        dict(value=str_value, value_schema=prim_string, key=float_value, key_schema=prim_float),
+        # Verify identity check allows Falsy object values(e.g., 0, empty string) to be handled properly (issue #342)
+        dict(value='', value_schema=prim_string, key=0.0, key_schema=prim_float),
+        dict(value=0.0, value_schema=prim_float, key='', key_schema=prim_string),
+    ]
+
+    for i, combo in enumerate(combinations):
+        combo['topic'] = str(uuid.uuid4())
+        combo['headers'] = [('index', str(i))]
+        p.produce(**combo)
+    p.flush()
+
+    conf = {'bootstrap.servers': bootstrap_servers,
+            'group.id': generate_group_id(),
+            'session.timeout.ms': 6000,
+            'enable.auto.commit': False,
+            'api.version.request': api_version_request,
+            'on_commit': print_commit_result,
+            'error_cb': error_cb,
+            'default.topic.config': {
+                'auto.offset.reset': 'earliest'
+            }}
+
+    conf.update(testconf.get('schema_registry_https', {}))
+
+    c = avro.AvroConsumer(conf)
+    c.subscribe([(t['topic']) for t in combinations])
+
+    msgcount = 0
+    while msgcount < len(combinations):
+        msg = c.poll(0)
+
+        if msg is None or msg.error():
+            continue
+
+        tstype, timestamp = msg.timestamp()
+        print('%s[%d]@%d: key=%s, value=%s, tstype=%d, timestamp=%s' %
+              (msg.topic(), msg.partition(), msg.offset(),
+               msg.key(), msg.value(), tstype, timestamp))
+
+        # omit empty Avro fields from payload for comparison
+        record_key = msg.key()
+        record_value = msg.value()
+        index = int(dict(msg.headers())['index'])
+
+        if isinstance(msg.key(), dict):
+            record_key = {k: v for k, v in msg.key().items() if v is not None}
+
+        if isinstance(msg.value(), dict):
+            record_value = {k: v for k, v in msg.value().items() if v is not None}
+
+        assert combinations[index].get('key') == record_key
+        assert combinations[index].get('value') == record_value
+
+        c.commit()
+        msgcount += 1
+
+    # Close consumer
+    c.close()
+
+
 def verify_producer_performance(with_dr_cb=True):
     """ Time how long it takes to produce and delivery X messages """
     conf = {'bootstrap.servers': bootstrap_servers,
@@ -1125,7 +1218,7 @@ def verify_config(expconfig, configs):
 
 # Exclude throttle since from default list
 default_modes = ['consumer', 'producer', 'avro', 'performance', 'admin']
-all_modes = default_modes + ['throttle', 'none']
+all_modes = default_modes + ['throttle', 'avro-https', 'none']
 """All test modes"""
 
 
@@ -1140,6 +1233,21 @@ def print_usage(exitcode, reason=None):
     sys.exit(exitcode)
 
 
+def generate_group_id():
+    return str(uuid.uuid1())
+
+
+def resolve_envs(_conf):
+    """Resolve environment variables"""
+
+    for k, v in _conf.items():
+        if isinstance(v, dict):
+            resolve_envs(v)
+
+        if str(v).startswith('$'):
+            _conf[k] = os.getenv(v[1:])
+
+
 if __name__ == '__main__':
     """Run test suites"""
 
@@ -1152,22 +1260,30 @@ def print_usage(exitcode, reason=None):
     # Parse options
     while len(sys.argv) > 1 and sys.argv[1].startswith('--'):
         opt = sys.argv.pop(1)[2:]
+
+        if opt == 'conf':
+            testconf = sys.argv.pop(1)
+            continue
+
         if opt not in all_modes:
             print_usage(1, 'unknown option --' + opt)
         modes.append(opt)
 
-    if len(sys.argv) > 1:
-        bootstrap_servers = sys.argv[1]
-        if len(sys.argv) > 2:
-            topic = sys.argv[2]
-        if len(sys.argv) > 3:
-            schema_registry_url = sys.argv[3]
-    else:
-        print_usage(1)
+    with open(testconf) as f:
+        testconf = json.load(f)
+        resolve_envs(testconf)
+
+    bootstrap_servers = testconf.get('bootstrap.servers', None)
+    topic = testconf.get('topic', None)
+    schema_registry_url = testconf.get('schema.registry.url', None)
 
     if len(modes) == 0:
         modes = default_modes
 
+    if bootstrap_servers is None or topic is None:
+        print_usage(1, "Properties bootstrap.servers and topic must be set. "
+                       "Use tests/testconf-example.json as a template when creating a new conf file.")
+
     print('Using confluent_kafka module version %s (0x%x)' % confluent_kafka.version())
     print('Using librdkafka version %s (0x%x)' % confluent_kafka.libversion())
     print('Testing: %s' % modes)
@@ -1213,6 +1329,10 @@ def print_usage(exitcode, reason=None):
         print('=' * 30, 'Verifying AVRO', '=' * 30)
         verify_avro()
 
+    if 'avro-https' in modes:
+        print('=' * 30, 'Verifying AVRO with HTTPS', '=' * 30)
+        verify_avro_https()
+
     if 'admin' in modes:
         print('=' * 30, 'Verifying Admin API', '=' * 30)
         verify_admin()

tests/README.md

@@ -10,30 +10,36 @@ From top-level directory run:
 Integration tests
 =================
 
-**WARNING**: These tests require an active Kafka cluster and will create new topics.
+**NOTE**: Integration tests require an existing Kafka cluster and a `testconf.json` configuration file. Any value provided
+in `testconf.json` prefixed with '$' will be treated as an environment variable and automatically resolved.
 
+At a minimum you must specify `bootstrap.servers` and `topic` within `testconf.json`. Please reference [tests/testconf-example.json](tests/testconf-example.json) for formatting.
 
-To run all of the integration test `modes` uncomment the following line from `tox.ini` and add the addresses to your Kafka and Confluent Schema Registry instances.
+**WARNING**: These tests will create new topics and consumer groups.
 
-    #python examples/integration_test.py <bootstrap-servers> confluent-kafka-testing [<schema-registry-url>]
+To run all of the integration test `modes` uncomment the following line from `tox.ini` and provide the location to your `testconf.json`
+
+    #python examples/integration_test.py --conf <testconf.json>
 
 You can also run the integration tests outside of `tox` by running this command from the source root directory
 
-    examples/integration_test.py <kafka-broker> [<test-topic>] [<schema-registry>]
+    python examples/integration_test.py --conf <testconf.json>
 
 To run individual integration test `modes` use the following syntax
 
-     examples/integration_test.py --<test mode>  <kafka-broker> [<test-topic>] [<schema-registry>]
+    python examples/integration_test.py --<test mode> --conf <testconf.json>
 
 For example:
 
-    examples/integration_test.py --producer <kafka-broker> [<test-topic>]
+    python examples/integration_test.py --producer --conf testconf.json
 
 To get a list of modes you can run the integration test manually with the `--help` flag
 
-    examples/integration_tests.py --help
+    python examples/integration_tests.py --help
 
 
+Throttle Callback test
+======================
 The throttle_cb integration test requires an additional step and as such is not included in the default test modes.
 In order to execute the throttle_cb test you must first set a throttle for the client 'throttled_client' with the command below:
 
@@ -43,11 +49,26 @@ In order to execute the throttle_cb test you must first set a throttle for the c
 
 Once the throttle has been set you can proceed with the following command:
 
-    examples/integration_test.py --throttle <kafka-broker> [<test-topic>]
+    python examples/integration_test.py --throttle --conf testconf.json
 
 
 To remove the throttle you can execute the following
 
     kafka-configs  --zookeeper <zookeeper host>:<zookeeper port> \
         --alter --delete-config 'request_percentage' \
         --entity-name throttled_client --entity-type clients
+
+
+HTTPS Schema Registry test
+==========================
+
+HTTPS tests require access to a Schema Registry instance configured to with at least one HTTPS listener.
+
+For instructions on how to configure the Schema Registry please see the Confluent documentation:
+
+[Schema Registry documentation](https://docs.confluent.io/current/schema-registry/docs/security.html#configuring-the-rest-api-for-http-or-https)
+
+If client authentication has been enabled you will need to provide both the client certificate, `schema.registry.ssl.certificate.location`,
+and the client's private key, `schema.registry.ssl.key.location`
+
+    python examples/integration_test.py --avro-https --conf testconf.json