Merge pull request #200 from vladimir-pachnik/master

FEATURE: Add SSL client cert auth

grafana_dashboards/client/connection.py

@@ -108,3 +108,14 @@ def __init__(self, host):
     def make_request(self, uri, body=None):
         response = requests.post('{0}{1}'.format(self._host, uri), json=body, auth=HTTPKerberosAuth(), verify=False)
         return response.json()
+
+
+class SSLAuthConnection(object):
+    def __init__(self, host, cert_bundle, debug=0):
+        logger.debug('Using SSL client cert from "%s" with host=%s', cert_bundle, host)
+        self._host = host
+        self._cert = cert_bundle
+
+    def make_request(self, uri, body=None):
+        response = requests.post('{0}{1}'.format(self._host, uri), json=body, cert=self._cert)
+        return response.json()

grafana_dashboards/client/grafana.py

@@ -17,7 +17,10 @@
 import logging
 import os
 
-from grafana_dashboards.client.connection import KerberosConnection, BearerAuthConnection, BasicAuthConnection
+from grafana_dashboards.client.connection import (KerberosConnection,
+                                                  BearerAuthConnection,
+                                                  BasicAuthConnection,
+                                                  SSLAuthConnection)
 from grafana_dashboards.exporter import DashboardExporter
 
 __author__ = 'Jakub Plichta <[email protected]>'
@@ -34,11 +37,22 @@ def __init__(self, **kwargs):
         username = os.getenv('GRAFANA_USERNAME', kwargs.get('username'))
         auth_token = os.getenv('GRAFANA_TOKEN', kwargs.get('token'))
         use_kerberos = os.getenv('GRAFANA_USE_KERBEROS', kwargs.get('use_kerberos'))
+        client_crt = os.getenv('GRAFANA_SSL_CLIENT_CRT', kwargs.get('ssl_client_crt'))
 
         if use_kerberos:
             self._connection = KerberosConnection(self._host)
         elif auth_token:
             self._connection = BearerAuthConnection(auth_token, self._host)
+        elif client_crt:
+            client_key = os.getenv('GRAFANA_SSL_CLIENT_KEY', kwargs.get('ssl_client_key'))
+            derived_key_path = os.path.splitext(client_crt)[0] + '.key'
+            # pull the separate key also if not given explicitly and derived filename exists
+            if client_key or (not client_key and os.path.exists(derived_key_path)):
+                cert_bundle = (client_crt, client_key if client_key else derived_key_path)
+            # otherwise assume bundled PEM
+            else:
+                cert_bundle = client_crt
+            self._connection = SSLAuthConnection(self._host, cert_bundle)
         else:
             self._connection = BasicAuthConnection(username, password, self._host)
 

tests/grafana_dashboards/client/test_connection.py

@@ -20,7 +20,10 @@
 from mock import MagicMock, patch
 from requests_kerberos import HTTPKerberosAuth
 
-from grafana_dashboards.client.connection import KerberosConnection, BasicAuthConnection, BearerAuthConnection
+from grafana_dashboards.client.connection import (KerberosConnection,
+                                                  BearerAuthConnection,
+                                                  BasicAuthConnection,
+                                                  SSLAuthConnection)
 
 __author__ = 'Jakub Plichta <[email protected]>'
 
@@ -98,3 +101,14 @@ def test_connection_with_kerberos(post):
     capture = Capture()
     post.assert_called_with('https://host/uri', auth=capture, json={"it's": 'alive'}, verify=False)
     assert isinstance(capture.value, HTTPKerberosAuth)
+
+
+@patch('requests.post')
+def test_connection_with_sslauth(post):
+    connection = SSLAuthConnection('https://host', ('/fake/cert'))
+
+    post().json.return_value = {'hello': 'world'}
+
+    assert connection.make_request('/uri', {'it\'s': 'alive'}) == {'hello': 'world'}
+
+    post.assert_called_with('https://host/uri', json={"it's": 'alive'}, cert='/fake/cert')

tests/grafana_dashboards/client/test_grafana.py

@@ -45,3 +45,16 @@ def test_grafana_with_kerberos():
     # noinspection PyProtectedMember
     exporter._connection.make_request.assert_called_once_with('/api/dashboards/db',
                                                               body)
+
+
+def test_grafana_with_sslauth():
+    exporter = GrafanaExporter(host='host', ssl_client_crt='/file/fake')
+    exporter._connection = MagicMock()
+
+    dashboard_data = {'title': 'title', 'tags': []}
+    exporter.process_dashboard('project_name', 'dashboard_name', dashboard_data)
+
+    body = {'overwrite': True, 'dashboard': dashboard_data}
+    # noinspection PyProtectedMember
+    exporter._connection.make_request.assert_called_once_with('/api/dashboards/db',
+                                                              body)