Mutual TLS authentication (#702)

* Add support for providing a client certificate for mutual TLS authentication.

* Be more explicit in documentation on valid values for the  parameter

Author: Lloyd Wallis

influxdb/client.py

@@ -61,6 +61,13 @@ class InfluxDBClient(object):
     :type proxies: dict
     :param path: path of InfluxDB on the server to connect, defaults to ''
     :type path: str
+    :param cert: Path to client certificate information to use for mutual TLS
+        authentication. You can specify a local cert to use
+        as a single file containing the private key and the certificate, or as
+        a tuple of both files’ paths, defaults to None
+    :type cert: str
+
+    :raises ValueError: if cert is provided but ssl is disabled (set to False)
     """
 
     def __init__(self,
@@ -78,6 +85,7 @@ def __init__(self,
                  proxies=None,
                  pool_size=10,
                  path='',
+                 cert=None,
                  ):
         """Construct a new InfluxDBClient object."""
         self.__host = host
@@ -120,6 +128,14 @@ def __init__(self,
         else:
             self._proxies = proxies
 
+        if cert:
+            if not ssl:
+                raise ValueError(
+                    "Client certificate provided but ssl is disabled."
+                )
+            else:
+                self._session.cert = cert
+
         self.__baseurl = "{0}://{1}:{2}{3}".format(
             self._scheme,
             self._host,

influxdb/tests/client_test.py

@@ -149,6 +149,14 @@ def test_dsn(self):
                                       **{'ssl': False})
         self.assertEqual('http://my.host.fr:1886', cli._baseurl)
 
+    def test_cert(self):
+        """Test mutual TLS authentication for TestInfluxDBClient object."""
+        cli = InfluxDBClient(ssl=True, cert='/etc/pki/tls/private/dummy.crt')
+        self.assertEqual(cli._session.cert, '/etc/pki/tls/private/dummy.crt')
+
+        with self.assertRaises(ValueError):
+            cli = InfluxDBClient(cert='/etc/pki/tls/private/dummy.crt')
+
     def test_switch_database(self):
         """Test switch database in TestInfluxDBClient object."""
         cli = InfluxDBClient('host', 8086, 'username', 'password', 'database')