Add check for SSL certificate and add form validators

superset/exceptions.py

@@ -16,6 +16,8 @@
 # under the License.
 from typing import Optional
 
+from flask_babel import gettext as _
+
 
 class SupersetException(Exception):
     status = 500
@@ -61,7 +63,7 @@ class SpatialException(SupersetException):
 
 
 class CertificateException(SupersetException):
-    pass
+    message = _("Invalid certificate")
 
 
 class DatabaseNotFound(SupersetException):

superset/views/core.py

@@ -1369,16 +1369,8 @@ def testconn(self):
                 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-----"
-                )
-            )
+            logger.info(e.message)
+            return json_error_response(e.message)
         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 sqlalchemy import MetaData
 
 from superset import app, security_manager
-from superset.exceptions import CertificateException, SupersetException
+from superset.exceptions import 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
@@ -204,10 +204,8 @@ def _pre_add_update(self, database):
             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 ""
-        )
+        if database.server_cert:
+            utils.parse_ssl_cert(database.server_cert)
         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
@@ -236,8 +234,6 @@ def check_extra(self, database):  # pylint: disable=no-self-use
         # 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. %{msg}s", msg=str(e))

superset/views/database/views.py

@@ -29,6 +29,7 @@
 from superset import app, db
 from superset.connectors.sqla.models import SqlaTable
 from superset.constants import RouteMethod
+from superset.exceptions import CertificateException
 from superset.utils import core as utils
 from superset.views.base import DeleteMixin, SupersetModelView, YamlExportMixin
 
@@ -50,6 +51,17 @@ def sqlalchemy_uri_form_validator(_, field: StringField) -> None:
     sqlalchemy_uri_validator(field.data, exception=ValidationError)
 
 
+def certificate_form_validator(_, field: StringField) -> None:
+    """
+        Check if user has submitted a valid SSL certificate
+    """
+    if field.data:
+        try:
+            utils.parse_ssl_cert(field.data)
+        except CertificateException as ex:
+            raise ValidationError(ex.message)
+
+
 def upload_stream_write(form_file_field: "FileStorage", path: str):
     chunk_size = app.config["UPLOAD_CHUNK_SIZE"]
     with open(path, "bw") as file_description:
@@ -68,7 +80,10 @@ class DatabaseView(
 
     add_template = "superset/models/database/add.html"
     edit_template = "superset/models/database/edit.html"
-    validators_columns = {"sqlalchemy_uri": [sqlalchemy_uri_form_validator]}
+    validators_columns = {
+        "sqlalchemy_uri": [sqlalchemy_uri_form_validator],
+        "server_cert": [certificate_form_validator],
+    }
 
     yaml_dict_key = "databases"