feat: add snowflake keypair authentication

requirements/testing.txt

@@ -136,7 +136,10 @@ websocket-client==1.2.0
     # via docker
 wrapt==1.12.1
     # via astroid
-
+snowflake-connector-python==2.7.9
+    # via snowflake
+snowflake-sqlalchemy==1.2.4
+    # via snowflake
 # The following packages are considered to be unsafe in a requirements file:
 # pip
 # setuptools

superset/db_engine_specs/snowflake.py

@@ -16,6 +16,7 @@
 # under the License.
 import json
 import re
+import logging
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Pattern, Tuple, TYPE_CHECKING
 from urllib import parse
@@ -26,6 +27,8 @@
 from marshmallow import fields, Schema
 from sqlalchemy.engine.url import URL
 from typing_extensions import TypedDict
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import serialization
 
 from superset.databases.utils import make_url_safe
 from superset.db_engine_specs.postgres import PostgresBaseEngineSpec
@@ -46,6 +49,8 @@
     "unexpected '(?P<syntax_error>.+?)'."
 )
 
+logger = logging.getLogger(__name__)
+
 
 class SnowflakeParametersSchema(Schema):
     username = fields.Str(required=True)
@@ -279,3 +284,33 @@ def parameters_json_schema(cls) -> Any:
 
         spec.components.schema(cls.__name__, schema=cls.parameters_schema)
         return spec.to_dict()["components"]["schemas"][cls.__name__]
+
+    @staticmethod
+    def update_params_from_encrypted_extra(
+        database: "Database",
+        params: Dict[str, Any],
+    ) -> None:
+        if not database.encrypted_extra:
+            return
+        try:
+            encrypted_extra = json.loads(database.encrypted_extra)
+            auth_method = encrypted_extra.pop("auth_method", None)
+            auth_params = encrypted_extra.pop("auth_params", {})
+            if not auth_method:
+                return
+            connect_args = params.setdefault("connect_args", {})
+            if auth_method == "keypair":
+                with open(auth_params['privatekey_path'], "rb") as key:
+                    p_key = serialization.load_pem_private_key(
+                        key.read(),
+                        password=auth_params['privatekey_pass'].encode(),
+                        backend=default_backend()
+                    )
+                pkb = p_key.private_bytes(
+                    encoding=serialization.Encoding.DER,
+                    format=serialization.PrivateFormat.PKCS8,
+                    encryption_algorithm=serialization.NoEncryption())
+                connect_args["private_key"] = pkb
+        except json.JSONDecodeError as ex:
+            logger.error(ex, exc_info=True)
+            raise ex