Added password support for SSL private key of the cached schema registry client (#1516)

* Steps to Test:
1) Run certify.sh
2) Manually run .env.sh code
3) docker-compose up [tests/docker]
4) Wait for the docker instance to be listening for requests(roughly 30 seconds)
5) run the command -> python3 integration_test.py --avro-https testconf.json

In mac Systems the certify.sh takes the hostname and it then collides when https connection is to be made since the hostname should be mentioned in the keychain - so either change hostname command to localhost literal and then run on Mac, otherwise in Linux system it should be fine.

Author: mahajanadhitya

CHANGELOG.md

@@ -6,6 +6,7 @@
   SASL PLAIN/SCRAM credentials that will be used for subsequent (new) connections to a broker (#1511).
 - Wheels for Linux / arm64 (#1496).
 - Added support for Default num_partitions in CreateTopics Admin API.
+- Added support for password protected private key in CachedSchemaRegistryClient.
 
 ## v2.0.2
 

src/confluent_kafka/avro/cached_schema_registry_client.py

@@ -21,6 +21,8 @@
 #
 import logging
 import warnings
+import urllib3
+import json
 from collections import defaultdict
 
 from requests import Session, utils
@@ -54,6 +56,7 @@ class CachedSchemaRegistryClient(object):
     Use CachedSchemaRegistryClient(dict: config) instead.
     Existing params ca_location, cert_location and key_location will be replaced with their librdkafka equivalents:
     `ssl.ca.location`, `ssl.certificate.location` and `ssl.key.location` respectively.
+    The support for password protected private key is via the Config only using 'ssl.key.password' field.
 
     Errors communicating to the server will result in a ClientError being raised.
 
@@ -109,6 +112,9 @@ def __init__(self, url, max_schemas_per_subject=1000, ca_location=None, cert_loc
         self.url = utils.urldefragauth(self.url)
 
         self._session = s
+        key_password = conf.pop('ssl.key.password', None)
+        self._is_key_password_provided = not key_password
+        self._https_session = self._make_https_session(s.cert[0], s.cert[1], ca_path, s.auth, key_password)
 
         self.auto_register_schemas = conf.pop("auto.register.schemas", True)
 
@@ -128,6 +134,29 @@ def close(self):
         # Constructor exceptions may occur prior to _session being set.
         if hasattr(self, '_session'):
             self._session.close()
+        if hasattr(self, '_https_session'):
+            self._https_session.clear()
+
+    @staticmethod
+    def _make_https_session(cert_location, key_location, ca_certs_path, auth, key_password):
+        https_session = urllib3.PoolManager(cert_reqs='CERT_REQUIRED', ca_certs=ca_certs_path,
+                                            cert_file=cert_location, key_file=key_location, key_password=key_password)
+        https_session.auth = auth
+        return https_session
+
+    def _send_https_session_request(self, url, method, headers, body):
+        request_headers = {'Accept': ACCEPT_HDR}
+        auth = self._https_session.auth
+        if body:
+            body = json.dumps(body).encode('UTF-8')
+            request_headers["Content-Length"] = str(len(body))
+            request_headers["Content-Type"] = "application/vnd.schemaregistry.v1+json"
+        if auth[0] != '' and auth[1] != '':
+            request_headers.update(urllib3.make_headers(basic_auth=auth[0] + ":" +
+                                                        auth[1]))
+        request_headers.update(headers)
+        response = self._https_session.request(method, url, headers=request_headers, body=body)
+        return response
 
     @staticmethod
     def _configure_basic_auth(url, conf):
@@ -158,6 +187,13 @@ 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))
 
+        if url.startswith('https') and self._is_key_password_provided:
+            response = self._send_https_session_request(url, method, headers, body)
+            try:
+                return json.loads(response.data), response.status
+            except ValueError:
+                return response.content, response.status
+
         _headers = {'Accept': ACCEPT_HDR}
         if body:
             _headers["Content-Length"] = str(len(body))

tests/docker/.env.sh

@@ -13,3 +13,5 @@ export MY_SCHEMA_REGISTRY_SSL_URL_ENV=https://$(hostname -f):8082
 export MY_SCHEMA_REGISTRY_SSL_CA_LOCATION_ENV=$TLS/ca-cert
 export MY_SCHEMA_REGISTRY_SSL_CERTIFICATE_LOCATION_ENV=$TLS/client.pem
 export MY_SCHEMA_REGISTRY_SSL_KEY_LOCATION_ENV=$TLS/client.key
+export MY_SCHEMA_REGISTRY_SSL_KEY_WITH_PASSWORD_LOCATION_ENV=$TLS/client_with_password.key
+export MY_SCHEMA_REGISTRY_SSL_KEY_PASSWORD="abcdefgh"

tests/docker/bin/certify.sh

@@ -24,5 +24,6 @@ echo "Creating client cert..."
 ${PY_DOCKER_BIN}/gen-ssl-certs.sh client ${TLS}/ca-cert ${TLS}/ ${HOST} ${HOST}
 
 echo "Creating key ..."
+cp ${TLS}/client.key ${TLS}/client_with_password.key
 openssl rsa -in ${TLS}/client.key -out ${TLS}/client.key  -passin pass:${PASS}
 

tests/docker/docker-compose.yaml

@@ -21,7 +21,7 @@ services:
       KAFKA_AUTHORIZER_CLASS_NAME: kafka.security.authorizer.AclAuthorizer
       KAFKA_SUPER_USERS: "User:ANONYMOUS"
   schema-registry:
-    image: confluentinc/cp-schema-registry:7.1.0
+    image: confluentinc/cp-schema-registry
     depends_on:
     - zookeeper
     - kafka
@@ -42,7 +42,7 @@ services:
       SCHEMA_REGISTRY_SSL_TRUSTSTORE_PASSWORD: abcdefgh
       SCHEMA_REGISTRY_KAFKASTORE_CONNECTION_URL: zookeeper:2181
   schema-registry-basic-auth:
-    image: confluentinc/cp-schema-registry:7.1.0
+    image: confluentinc/cp-schema-registry
     depends_on:
       - zookeeper
       - kafka

tests/integration/integration_test.py

@@ -1254,6 +1254,16 @@ def print_usage(exitcode, reason=None):
     if 'avro-https' in modes:
         print('=' * 30, 'Verifying AVRO with HTTPS', '=' * 30)
         verify_avro_https(testconf.get('avro-https', None))
+        key_with_password_conf = testconf.get("avro-https-key-with-password", None)
+        print('=' * 30, 'Verifying AVRO with HTTPS Flow with Password',
+              'Protected Private Key of Cached-Schema-Registry-Client', '=' * 30)
+        verify_avro_https(key_with_password_conf)
+        print('Verifying Error with Wrong Password of Password Protected Private Key of Cached-Schema-Registry-Client')
+        try:
+            key_with_password_conf['schema.registry.ssl.key.password'] += '->wrongpassword'
+            verify_avro_https(key_with_password_conf)
+        except Exception:
+            print("Wrong Password Gives Error -> Successful")
 
     if 'avro-basic-auth' in modes:
         print("=" * 30, 'Verifying AVRO with Basic Auth', '=' * 30)

tests/integration/testconf.json

@@ -3,15 +3,22 @@
     "bootstrap.servers": "$MY_BOOTSTRAP_SERVER_ENV",
     "schema.registry.url": "$MY_SCHEMA_REGISTRY_URL_ENV",
     "avro-https": {
-            "schema.registry.url": "$MY_SCHEMA_REGISTRY_SSL_URL_ENV",
-            "schema.registry.ssl.ca.location": "$MY_SCHEMA_REGISTRY_SSL_CA_LOCATION_ENV",
-            "schema.registry.ssl.certificate.location": "$MY_SCHEMA_REGISTRY_SSL_CERTIFICATE_LOCATION_ENV",
-            "schema.registry.ssl.key.location": "$MY_SCHEMA_REGISTRY_SSL_KEY_LOCATION_ENV"
+        "schema.registry.url": "$MY_SCHEMA_REGISTRY_SSL_URL_ENV",
+        "schema.registry.ssl.ca.location": "$MY_SCHEMA_REGISTRY_SSL_CA_LOCATION_ENV",
+        "schema.registry.ssl.certificate.location": "$MY_SCHEMA_REGISTRY_SSL_CERTIFICATE_LOCATION_ENV",
+        "schema.registry.ssl.key.location": "$MY_SCHEMA_REGISTRY_SSL_KEY_LOCATION_ENV"
     },
     "avro-basic-auth": {
         "schema.registry.url": "http://localhost:8083",
         "schema.registry.basic.auth.user.info": "ckp_tester:test_secret",
         "sasl.username": "ckp_tester",
         "sasl.password": "test_secret"
+    },
+    "avro-https-key-with-password": {
+        "schema.registry.url": "$MY_SCHEMA_REGISTRY_SSL_URL_ENV",
+        "schema.registry.ssl.ca.location": "$MY_SCHEMA_REGISTRY_SSL_CA_LOCATION_ENV",
+        "schema.registry.ssl.certificate.location": "$MY_SCHEMA_REGISTRY_SSL_CERTIFICATE_LOCATION_ENV",
+        "schema.registry.ssl.key.location": "$MY_SCHEMA_REGISTRY_SSL_KEY_WITH_PASSWORD_LOCATION_ENV",
+        "schema.registry.ssl.key.password": "$MY_SCHEMA_REGISTRY_SSL_KEY_PASSWORD"
     }
 }