feat: add SSL certificate validation for Druid (#9396)

* feat: add SSL certificate feature

* Address comments

* don't mutate extras

* Address comments and add polish

* Add further polish

docs/installation.rst

@@ -729,6 +729,12 @@ The native Druid connector (behind the ``DRUID_IS_ACTIVE`` feature flag)
 is slowly getting deprecated in favor of the SQLAlchemy/DBAPI connector made
 available in the ``pydruid`` library.
 
+To use a custom SSL certificate to validate HTTPS requests, the certificate
+contents can be entered in the ``Root Certificate`` field in the Database
+dialog. When using a custom certificate, ``pydruid`` will automatically use
+``https`` scheme. To disable SSL verification add the following to extras:
+``engine_params": {"connect_args": {"scheme": "https", "ssl_verify_cert": false}}``
+
 Dremio
 ------
 

setup.cfg

@@ -45,7 +45,7 @@ combine_as_imports = true
 include_trailing_comma = true
 line_length = 88
 known_first_party = superset
-known_third_party =alembic,backoff,bleach,celery,click,colorama,contextlib2,croniter,dateutil,flask,flask_appbuilder,flask_babel,flask_caching,flask_compress,flask_login,flask_migrate,flask_sqlalchemy,flask_talisman,flask_testing,flask_wtf,geohash,geopy,humanize,isodate,jinja2,markdown,markupsafe,marshmallow,msgpack,numpy,pandas,parsedatetime,pathlib2,polyline,prison,pyarrow,pyhive,pytz,retry,selenium,setuptools,simplejson,sphinx_rtd_theme,sqlalchemy,sqlalchemy_utils,sqlparse,werkzeug,wtforms,wtforms_json,yaml
+known_third_party =alembic,backoff,bleach,celery,click,colorama,contextlib2,croniter,cryptography,dateutil,flask,flask_appbuilder,flask_babel,flask_caching,flask_compress,flask_login,flask_migrate,flask_sqlalchemy,flask_talisman,flask_testing,flask_wtf,geohash,geopy,humanize,isodate,jinja2,markdown,markupsafe,marshmallow,msgpack,numpy,pandas,parsedatetime,pathlib2,polyline,prison,pyarrow,pyhive,pytz,retry,selenium,setuptools,simplejson,sphinx_rtd_theme,sqlalchemy,sqlalchemy_utils,sqlparse,werkzeug,wtforms,wtforms_json,yaml
 multi_line_output = 3
 order_by_type = false
 

superset/config.py

@@ -797,6 +797,11 @@ SQLALCHEMY_EXAMPLES_URI = None
 # Typically these should not be allowed.
 PREVENT_UNSAFE_DB_CONNECTIONS = True
 
+# Path used to store SSL certificates that are generated when using custom certs.
+# Defaults to temporary directory.
+# Example: SSL_CERT_PATH = "/certs"
+SSL_CERT_PATH: Optional[str] = None
+
 # SIP-15 should be enabled for all new Superset deployments which ensures that the time
 # range endpoints adhere to [start, end). For existing deployments admins should provide
 # a dedicated period of time to allow chart producers to update their charts before

superset/db_engine_specs/base.py

@@ -16,6 +16,8 @@
 # under the License.
 # pylint: disable=unused-argument
 import hashlib
+import json
+import logging
 import os
 import re
 from contextlib import closing
@@ -59,6 +61,8 @@ if TYPE_CHECKING:
     )
     from superset.models.core import Database  # pylint: disable=unused-import
 
+logger = logging.getLogger()
+
 
 class TimeGrain(NamedTuple):  # pylint: disable=too-few-public-methods
     name: str  # TODO: redundant field, remove
@@ -959,3 +963,21 @@ class BaseEngineSpec:  # pylint: disable=too-many-public-methods
         :param database: instance to be mutated
         """
         return None
+
+    @staticmethod
+    def get_extra_params(database: "Database") -> Dict[str, Any]:
+        """
+        Some databases require adding elements to connection parameters,
+        like passing certificates to `extra`. This can be done here.
+
+        :param database: database instance from which to extract extras
+        :raises CertificateException: If certificate is not valid/unparseable
+        """
+        extra: Dict[str, Any] = {}
+        if database.extra:
+            try:
+                extra = json.loads(database.extra)
+            except json.JSONDecodeError as e:
+                logger.error(e)
+                raise e
+        return extra

superset/db_engine_specs/druid.py

@@ -14,14 +14,20 @@
 # KIND, either express or implied.  See the License for the
 # specific language governing permissions and limitations
 # under the License.
-from typing import TYPE_CHECKING
+import json
+import logging
+from typing import Any, Dict, TYPE_CHECKING
 
 from superset.db_engine_specs.base import BaseEngineSpec
+from superset.utils import core as utils
 
 if TYPE_CHECKING:
     from superset.connectors.sqla.models import (  # pylint: disable=unused-import
         TableColumn,
     )
+    from superset.models.core import Database  # pylint: disable=unused-import
+
+logger = logging.getLogger()
 
 
 class DruidEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
@@ -47,3 +53,27 @@ class DruidEngineSpec(BaseEngineSpec):  # pylint: disable=abstract-method
     def alter_new_orm_column(cls, orm_col: "TableColumn") -> None:
         if orm_col.column_name == "__time":
             orm_col.is_dttm = True
+
+    @staticmethod
+    def get_extra_params(database: "Database") -> Dict[str, Any]:
+        """
+        For Druid, the path to a SSL certificate is placed in `connect_args`.
+
+        :param database: database instance from which to extract extras
+        :raises CertificateException: If certificate is not valid/unparseable
+        """
+        try:
+            extra = json.loads(database.extra or "{}")
+        except json.JSONDecodeError as e:
+            logger.error(e)
+            raise e
+
+        if database.server_cert:
+            engine_params = extra.get("engine_params", {})
+            connect_args = engine_params.get("connect_args", {})
+            connect_args["scheme"] = "https"
+            path = utils.create_ssl_cert_file(database.server_cert)
+            connect_args["ssl_verify_cert"] = path
+            engine_params["connect_args"] = connect_args
+            extra["engine_params"] = engine_params
+        return extra

superset/exceptions.py

@@ -60,5 +60,9 @@ class SpatialException(SupersetException):
     pass
 
 
+class CertificateException(SupersetException):
+    pass
+
+
 class DatabaseNotFound(SupersetException):
     status = 400

superset/migrations/versions/b5998378c225_add_certificate_to_dbs.py

@@ -0,0 +1,47 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+"""add certificate to dbs
+
+Revision ID: b5998378c225
+Revises: 72428d1ea401
+Create Date: 2020-03-25 10:49:10.883065
+
+"""
+
+# revision identifiers, used by Alembic.
+revision = "b5998378c225"
+down_revision = "72428d1ea401"
+
+from typing import Dict
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects.postgresql.base import PGDialect
+from sqlalchemy_utils import EncryptedType
+
+
+def upgrade():
+    kwargs: Dict[str, str] = {}
+    bind = op.get_bind()
+    op.add_column(
+        "dbs",
+        sa.Column("server_cert", EncryptedType(sa.Text()), nullable=True, **kwargs),
+    )
+
+
+def downgrade():
+    op.drop_column("dbs", "server_cert")

superset/models/core.py

@@ -139,6 +139,7 @@ class Database(
     encrypted_extra = Column(EncryptedType(Text, config["SECRET_KEY"]), nullable=True)
     perm = Column(String(1000))
     impersonate_user = Column(Boolean, default=False)
+    server_cert = Column(EncryptedType(Text, config["SECRET_KEY"]), nullable=True)
     export_fields = [
         "database_name",
         "sqlalchemy_uri",
@@ -309,6 +310,7 @@ class Database(
         )
         if configuration:
             connect_args["configuration"] = configuration
+        if connect_args:
             params["connect_args"] = connect_args
 
         params.update(self.get_encrypted_extra())
@@ -555,14 +557,7 @@ class Database(
         return self.db_engine_spec.get_time_grains()
 
     def get_extra(self) -> Dict[str, Any]:
-        extra: Dict[str, Any] = {}
-        if self.extra:
-            try:
-                extra = json.loads(self.extra)
-            except json.JSONDecodeError as e:
-                logger.error(e)
-                raise e
-        return extra
+        return self.db_engine_spec.get_extra_params(self)
 
     def get_encrypted_extra(self):
         encrypted_extra = {}

superset/templates/superset/models/database/add.html

@@ -24,4 +24,5 @@
   {{ macros.testconn() }}
   {{ macros.expand_extra_textarea() }}
   {{ macros.expand_encrypted_extra_textarea() }}
+  {{ macros.expand_server_cert_textarea() }}
 {% endblock %}

superset/templates/superset/models/database/edit.html

@@ -24,4 +24,5 @@
   {{ macros.testconn() }}
   {{ macros.expand_extra_textarea() }}
   {{ macros.expand_encrypted_extra_textarea() }}
+  {{ macros.expand_server_cert_textarea() }}
 {% endblock %}

superset/templates/superset/models/database/macros.html

@@ -43,6 +43,7 @@
           impersonate_user: $('#impersonate_user').is(':checked'),
           extras: extra ? JSON.parse(extra) : {},
           encrypted_extra: encryptedExtra ? JSON.parse(encryptedExtra) : {},
+          server_cert: $("#server_cert").val(),
         })
       } catch(parse_error){
         alert("Malformed JSON in the extras field: " + parse_error);
@@ -81,3 +82,9 @@
     $('#encrypted_extra').attr('rows', '5');
   </script>
 {% endmacro %}
+
+{% macro expand_server_cert_textarea() %}
+  <script>
+    $('#server_cert').attr('rows', '5');
+  </script>
+{% endmacro %}

superset/utils/core.py

@@ -19,12 +19,13 @@
 import decimal
 import errno
 import functools
+import hashlib
 import json
 import logging
 import os
-import re
 import signal
 import smtplib
+import tempfile
 import traceback
 import uuid
 import zlib
@@ -45,6 +46,9 @@ import numpy as np
 import pandas as pd
 import parsedatetime
 import sqlalchemy as sa
+from cryptography import x509
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.backends.openssl.x509 import _Certificate
 from dateutil.parser import parse
 from dateutil.relativedelta import relativedelta
 from flask import current_app, flash, Flask, g, Markup, render_template
@@ -56,7 +60,11 @@ from sqlalchemy.dialects.mysql import MEDIUMTEXT
 from sqlalchemy.sql.type_api import Variant
 from sqlalchemy.types import TEXT, TypeDecorator
 
-from superset.exceptions import SupersetException, SupersetTimeoutException
+from superset.exceptions import (
+    CertificateException,
+    SupersetException,
+    SupersetTimeoutException,
+)
 from superset.utils.dates import datetime_to_epoch, EPOCH
 
 try:
@@ -1163,6 +1171,46 @@ def get_username() -> Optional[str]:
         return None
 
 
+def parse_ssl_cert(certificate: str) -> _Certificate:
+    """
+    Parses the contents of a certificate and returns a valid certificate object
+    if valid.
+
+    :param certificate: Contents of certificate file
+    :return: Valid certificate instance
+    :raises CertificateException: If certificate is not valid/unparseable
+    """
+    try:
+        return x509.load_pem_x509_certificate(
+            certificate.encode("utf-8"), default_backend()
+        )
+    except ValueError as e:
+        raise CertificateException("Invalid certificate")
+
+
+def create_ssl_cert_file(certificate: str) -> str:
+    """
+    This creates a certificate file that can be used to validate HTTPS
+    sessions. A certificate is only written to disk once; on subsequent calls,
+    only the path of the existing certificate is returned.
+
+    :param certificate: The contents of the certificate
+    :return: The path to the certificate file
+    :raises CertificateException: If certificate is not valid/unparseable
+    """
+    filename = f"{hashlib.md5(certificate.encode('utf-8')).hexdigest()}.crt"
+    cert_dir = current_app.config["SSL_CERT_PATH"]
+    path = cert_dir if cert_dir else tempfile.gettempdir()
+    path = os.path.join(path, filename)
+    if not os.path.exists(path):
+        # Validate certificate prior to persisting to temporary directory
+        parse_ssl_cert(certificate)
+        cert_file = open(path, "w")
+        cert_file.write(certificate)
+        cert_file.close()
+    return path
+
+
 def MediumText() -> Variant:
     return Text().with_variant(MEDIUMTEXT(), "mysql")
 

superset/views/core.py

@@ -67,6 +67,7 @@ from superset.connectors.connector_registry import ConnectorRegistry
 from superset.connectors.sqla.models import AnnotationDatasource
 from superset.constants import RouteMethod
 from superset.exceptions import (
+    CertificateException,
     DatabaseNotFound,
     SupersetException,
     SupersetSecurityException,
@@ -1353,6 +1354,7 @@ class Superset(BaseSupersetView):
             # this is the database instance that will be tested
             database = models.Database(
                 # extras is sent as json, but required to be a string in the Database model
+                server_cert=request.json.get("server_cert"),
                 extra=json.dumps(request.json.get("extras", {})),
                 impersonate_user=request.json.get("impersonate_user"),
                 encrypted_extra=json.dumps(request.json.get("encrypted_extra", {})),
@@ -1366,6 +1368,17 @@ class Superset(BaseSupersetView):
             with closing(engine.connect()) as conn:
                 conn.scalar(select([1]))
                 return json_success('"OK"')
+        except CertificateException as e:
+            logger.info("Invalid certificate %s", e)
+            return json_error_response(
+                _(
+                    "Invalid certificate. "
+                    "Please make sure the certificate begins with\n"
+                    "-----BEGIN CERTIFICATE-----\n"
+                    "and ends with \n"
+                    "-----END CERTIFICATE-----"
+                )
+            )
         except NoSuchModuleError as e:
             logger.info("Invalid driver %s", e)
             driver_name = make_url(uri).drivername

superset/views/database/mixins.py

@@ -21,7 +21,7 @@ from flask_babel import lazy_gettext as _
 from sqlalchemy import MetaData
 
 from superset import app, security_manager
-from superset.exceptions import SupersetException
+from superset.exceptions import CertificateException, SupersetException
 from superset.security.analytics_db_safety import check_sqlalchemy_uri
 from superset.utils import core as utils
 from superset.views.database.filters import DatabaseFilter
@@ -65,6 +65,7 @@ class DatabaseMixin:
         "allow_multi_schema_metadata_fetch",
         "extra",
         "encrypted_extra",
+        "server_cert",
     ]
     search_exclude_columns = (
         "password",
@@ -74,6 +75,7 @@ class DatabaseMixin:
         "queries",
         "saved_queries",
         "encrypted_extra",
+        "server_cert",
     )
     edit_columns = add_columns
     show_columns = [
@@ -149,6 +151,11 @@ class DatabaseMixin:
             "syntax normally used by SQLAlchemy.",
             True,
         ),
+        "server_cert": utils.markdown(
+            "Optional CA_BUNDLE contents to validate HTTPS requests. Only available "
+            "on certain database engines.",
+            True,
+        ),
         "impersonate_user": _(
             "If Presto, all the queries in SQL Lab are going to be executed as the "
             "currently logged on user who must have permission to run them.<br/>"
@@ -183,6 +190,7 @@ class DatabaseMixin:
         "cache_timeout": _("Chart Cache Timeout"),
         "extra": _("Extra"),
         "encrypted_extra": _("Secure Extra"),
+        "server_cert": _("Root certificate"),
         "allow_run_async": _("Asynchronous Query Execution"),
         "impersonate_user": _("Impersonate the logged on user"),
         "allow_csv_upload": _("Allow Csv Upload"),
@@ -196,6 +204,10 @@ class DatabaseMixin:
             check_sqlalchemy_uri(database.sqlalchemy_uri)
         self.check_extra(database)
         self.check_encrypted_extra(database)
+        utils.parse_ssl_cert(database.server_cert)
+        database.server_cert = (
+            database.server_cert.strip() if database.server_cert else ""
+        )
         database.set_sqlalchemy_uri(database.sqlalchemy_uri)
         security_manager.add_permission_view_menu("database_access", database.perm)
         # adding a new database we always want to force refresh schema list
@@ -224,17 +236,24 @@ class DatabaseMixin:
         # this will check whether json.loads(extra) can succeed
         try:
             extra = database.get_extra()
+        except CertificateException:
+            raise Exception(_("Invalid certificate"))
         except Exception as e:
-            raise Exception("Extra field cannot be decoded by JSON. {}".format(str(e)))
+            raise Exception(
+                _("Extra field cannot be decoded by JSON. %{msg}s", msg=str(e))
+            )
 
         # this will check whether 'metadata_params' is configured correctly
         metadata_signature = inspect.signature(MetaData)
         for key in extra.get("metadata_params", {}):
             if key not in metadata_signature.parameters:
                 raise Exception(
-                    "The metadata_params in Extra field "
-                    "is not configured correctly. The key "
-                    "{} is invalid.".format(key)
+                    _(
+                        "The metadata_params in Extra field "
+                        "is not configured correctly. The key "
+                        "%{key}s is invalid.",
+                        key=key,
+                    )
                 )
 
     def check_encrypted_extra(self, database):  # pylint: disable=no-self-use
@@ -242,4 +261,6 @@ class DatabaseMixin:
         try:
             database.get_encrypted_extra()
         except Exception as e:
-            raise Exception(f"Secure Extra field cannot be decoded as JSON. {str(e)}")
+            raise Exception(
+                _("Extra field cannot be decoded by JSON. %{msg}s", msg=str(e))
+            )

tests/fixtures/certificates.py

@@ -0,0 +1,38 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements.  See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership.  The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License.  You may obtain a copy of the License at
+#
+#   http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied.  See the License for the
+# specific language governing permissions and limitations
+# under the License.
+ssl_certificate = """-----BEGIN CERTIFICATE-----
+MIIDnDCCAoQCCQCrdpcNPCA/eDANBgkqhkiG9w0BAQsFADCBjzELMAkGA1UEBhMC
+VVMxEzARBgNVBAgMCkNhbGlmb3JuaWExEjAQBgNVBAcMCVNhbiBNYXRlbzEPMA0G
+A1UECgwGUHJlc2V0MRMwEQYDVQQLDApTa3Vua3dvcmtzMRIwEAYDVQQDDAlwcmVz
+ZXQuaW8xHTAbBgkqhkiG9w0BCQEWDmluZm9AcHJlc2V0LmlvMB4XDTIwMDMyNjEw
+NTE1NFoXDTQwMDMyNjEwNTE1NFowgY8xCzAJBgNVBAYTAlVTMRMwEQYDVQQIDApD
+YWxpZm9ybmlhMRIwEAYDVQQHDAlTYW4gTWF0ZW8xDzANBgNVBAoMBlByZXNldDET
+MBEGA1UECwwKU2t1bmt3b3JrczESMBAGA1UEAwwJcHJlc2V0LmlvMR0wGwYJKoZI
+hvcNAQkBFg5pbmZvQHByZXNldC5pbzCCASIwDQYJKoZIhvcNAQEBBQADggEPADCC
+AQoCggEBAKNHQZcu2L/6HvZfzy4Hnm3POeztfO+NJ7OzppAcNlLbTAatUk1YoDbJ
+5m5GUW8m7pVEHb76UL6Xxei9MoMVvHGuXqQeZZnNd+DySW/227wkOPYOCVSuDsWD
+1EReG+pv/z8CDhdwmMTkDTZUDr0BUR/yc8qTCPdZoalj2muDl+k2J3LSCkelx4U/
+2iYhoUQD+lzFS3k7ohAfaGc2aZOlwTITopXHSFfuZ7j9muBOYtU7NgpnCl6WgxYP
+1+4ddBIauPTBY2gWfZC2FeOfYEqfsUUXRsw1ehEQf4uxxTKNJTfTuVbdgrTYx5QQ
+jrM88WvWdyVnIM7u7/x9bawfGX/b/F0CAwEAATANBgkqhkiG9w0BAQsFAAOCAQEA
+XYLLk3T5RWIagNa3DPrMI+SjRm4PAI/RsijtBV+9hrkCXOQ1mvlo/ORniaiemHvF
+Kh6u6MTl014+f6Ytg/tx/OzuK2ffo9x44ZV/yqkbSmKD1pGftYNqCnBCN0uo1Gzb
+HZ+bTozo+9raFN7OGPgbdBmpQT2c+LG5n+7REobHFb7VLeY2/7BKtxNBRXfIxn4X
++MIhpASwLH5X64a1f9LyuPNMyUvKgzDe7jRdX1JZ7uw/1T//OHGQth0jLiapa6FZ
+GwgYUaruSZH51ZtxrJSXKSNBA7asPSBbyOmGptLsw2GTAsoBd5sUR4+hbuVo+1ai
+XeA3AKTX/OdYWJvr5YIgeQ==
+-----END CERTIFICATE-----"""

tests/utils_tests.py

@@ -19,6 +19,8 @@ import unittest
 import uuid
 from datetime import date, datetime, time, timedelta
 from decimal import Decimal
+import hashlib
+import os
 from unittest.mock import Mock, patch
 
 import numpy
@@ -28,12 +30,13 @@ from sqlalchemy.exc import ArgumentError
 
 import tests.test_app
 from superset import app, db, security_manager
-from superset.exceptions import SupersetException
+from superset.exceptions import CertificateException, SupersetException
 from superset.models.core import Database
 from superset.utils.cache_manager import CacheManager
 from superset.utils.core import (
     base_json_conv,
     convert_legacy_filters_into_adhoc,
+    create_ssl_cert_file,
     datetime_f,
     format_timedelta,
     get_iterable,
@@ -46,6 +49,7 @@ from superset.utils.core import (
     memoized,
     merge_extra_filters,
     merge_request_params,
+    parse_ssl_cert,
     parse_human_timedelta,
     parse_js_uri_path_item,
     parse_past_timedelta,
@@ -59,6 +63,8 @@ from superset.views.utils import get_time_range_endpoints
 from superset.views.utils import build_extra_filters
 from tests.base_tests import SupersetTestCase
 
+from .fixtures.certificates import ssl_certificate
+
 
 def mock_parse_human_datetime(s):
     if s == "now":
@@ -1221,3 +1227,14 @@ class UtilsTestCase(SupersetTestCase):
         )
         expected = []
         self.assertEqual(extra_filters, expected)
+
+    def test_ssl_certificate_parse(self):
+        parsed_certificate = parse_ssl_cert(ssl_certificate)
+        self.assertEqual(parsed_certificate.serial_number, 12355228710836649848)
+        self.assertRaises(CertificateException, parse_ssl_cert, "abc" + ssl_certificate)
+
+    def test_ssl_certificate_file_creation(self):
+        path = create_ssl_cert_file(ssl_certificate)
+        expected_filename = hashlib.md5(ssl_certificate.encode("utf-8")).hexdigest()
+        self.assertIn(expected_filename, path)
+        self.assertTrue(os.path.exists(path))