fix(embedded): refuse a broken embedded key pair at save instead of 500ing every chart

UPDATING.md

@@ -39,7 +39,8 @@ assists people when migrating to a new version.
   per-user result; existing cached entries for such charts are effectively invalidated
   for guests. A malformed or incomplete `embedded_credentials` is rejected by
   `POST`/`PUT /api/v1/database/` and by `validate_parameters` rather than stored and
-  ignored while both embedding flags are enabled; asset import is unchanged. Validation
+  ignored while both embedding flags are enabled. Import refuses a key pair that
+  doesn't decode with its passphrase, as the API does. Validation
   uses the engine's credential rules and edited credentials rather than stale stored
   values, and rejects whitespace-only usernames. Disabling either flag preserves stored credentials without blocking unrelated edits. Restored async guest
   queries use the same credential as synchronous queries. Connection-mutator hooks

superset/commands/database/importers/v1/utils.py

@@ -20,18 +20,18 @@
 
 from flask import current_app as app
 
-from superset import db, security_manager
+from superset import db, is_feature_enabled, security_manager
 from superset.commands.database.exceptions import DatabaseInvalidError
 from superset.commands.database.utils import (
     add_permissions,
     import_connection_identity_changed,
     ssh_tunnel_rebind_unsafe,
 )
-from superset.commands.exceptions import ImportFailedError
-from superset.constants import PASSWORD_MASK
+from superset.commands.exceptions import ImportFailedError, ImportRefusedError
+from superset.constants import EMBEDDED_CREDENTIALS_KEY, PASSWORD_MASK
 from superset.databases.ssh_tunnel.models import SSHTunnel
-from superset.databases.utils import make_url_safe
-from superset.db_engine_specs import get_engine_spec
+from superset.databases.utils import embedded_credentials_error, make_url_safe
+from superset.db_engine_specs import BaseEngineSpec, get_engine_spec
 from superset.db_engine_specs.exceptions import SupersetDBAPIError
 from superset.exceptions import (
     OAuth2RedirectError,
@@ -79,6 +79,31 @@ def _refuse_stored_secret_reuse(existing: Database, config: dict[str, Any]) -> N
         )
 
 
+def _refuse_unreadable_embedded_key(
+    config: dict[str, Any], engine_spec: type[BaseEngineSpec]
+) -> None:
+    """
+    Refuse an imported embedded credential the engine can't use, as the API does.
+
+    ``config["encrypted_extra"]`` has its masked secrets already revealed, so a
+    key pair is decoded with the passphrase it will actually be stored with.
+    """
+    if not (
+        is_feature_enabled("EMBEDDED_CREDENTIAL_FALLBACK")
+        and is_feature_enabled("EMBEDDED_SUPERSET")
+    ):
+        return
+    payload = json.loads(config["encrypted_extra"])
+    if not isinstance(payload, dict):
+        return
+    credentials = payload.get(EMBEDDED_CREDENTIALS_KEY)
+    if message := embedded_credentials_error(engine_spec, credentials, None):
+        raise ImportRefusedError(
+            f"Embedded guest credentials of database '{config['database_name']}' "
+            f"can't be used. {message}"
+        )
+
+
 def _sync_permissions_best_effort(database: Database) -> None:
     """
     Sync catalog/schema permissions for ``database``, tolerating a transient
@@ -191,6 +216,7 @@ def import_database(  # noqa: C901
                 "Must provide value for masked_encrypted_extra field: "
                 + ", ".join(still_masked)
             )
+        _refuse_unreadable_embedded_key(config, engine_spec)
 
     ssh_tunnel_config = config.pop("ssh_tunnel", None)
 

superset/commands/database/validate.py

@@ -33,7 +33,11 @@
 )
 from superset.constants import EMBEDDED_CREDENTIALS_KEY
 from superset.daos.database import DatabaseDAO
-from superset.databases.utils import embedded_credentials_incomplete, make_url_safe
+from superset.databases.utils import (
+    embedded_credentials_error,
+    embedded_credentials_incomplete,
+    make_url_safe,
+)
 from superset.db_engine_specs import get_engine_spec
 from superset.errors import ErrorLevel, SupersetError, SupersetErrorType
 from superset.extensions import event_logger
@@ -282,6 +286,17 @@ def _get_embedded_credentials_error(self) -> Optional[SupersetError]:
         if not embedded_credentials_incomplete(
             credentials, engine_spec.validate_embedded_credentials
         ):
+            if message := embedded_credentials_error(
+                engine_spec,
+                credentials,
+                self._model.encrypted_extra if self._model else None,
+            ):
+                return SupersetError(
+                    message=message,
+                    error_type=SupersetErrorType.INVALID_PAYLOAD_SCHEMA_ERROR,
+                    level=ErrorLevel.ERROR,
+                    extra={"invalid": [EMBEDDED_CREDENTIALS_KEY]},
+                )
             return None
 
         return SupersetError(

superset/commands/exceptions.py

@@ -105,6 +105,12 @@ class ImportFailedError(CommandException):
     message = "Import failed for an unknown reason"
 
 
+class ImportRefusedError(ImportFailedError):
+    """An import refused because of what the bundle contains, not a server fault."""
+
+    status = 400
+
+
 class DatasourceTypeInvalidError(ValidationError):
     status = 422
 

superset/databases/schemas.py

@@ -53,7 +53,11 @@
     EncryptedDict,  # noqa: F401
     EncryptedString,  # noqa: F401
 )
-from superset.databases.utils import embedded_credentials_incomplete, make_url_safe
+from superset.databases.utils import (
+    embedded_credentials_error,
+    embedded_credentials_incomplete,
+    make_url_safe,
+)
 from superset.db_engine_specs import get_engine_spec
 from superset.db_engine_specs.base import BaseEngineSpec
 from superset.exceptions import CertificateException, SupersetSecurityException
@@ -488,6 +492,10 @@ def validate_embedded_credentials(
                 ),
                 field_name="masked_encrypted_extra",
             )
+        if message := embedded_credentials_error(
+            engine_spec, credentials, stored.encrypted_extra if stored else None
+        ):
+            raise ValidationError(message, field_name="masked_encrypted_extra")
 
         # A credential only applies to a per-user OAuth2 connection with
         # impersonation on (`Database.embedded_fallback_applies`). Anywhere else

superset/databases/utils.py

@@ -24,6 +24,7 @@
 from sqlalchemy.exc import NoSuchTableError
 
 from superset.commands.database.exceptions import DatabaseInvalidError
+from superset.constants import EMBEDDED_CREDENTIALS_KEY, PASSWORD_MASK
 from superset.sql.parse import Table
 from superset.utils import json
 
@@ -33,6 +34,7 @@
         TableMetadataForeignKeysIndexesResponse,
         TableMetadataResponse,
     )
+    from superset.db_engine_specs.base import BaseEngineSpec
 
 
 def get_foreign_keys_metadata(
@@ -245,3 +247,33 @@ def embedded_credentials_incomplete(
         return False
 
     return validate_credentials(value) is None
+
+
+def embedded_credentials_error(
+    engine_spec: type[BaseEngineSpec],
+    credentials: Any,
+    stored_encrypted_extra: str | None,
+) -> str | None:
+    """
+    Why an authored embedded credential cannot be used, or ``None``.
+
+    Complements :func:`embedded_credentials_incomplete`, which only checks that the
+    fields are filled in, by asking the engine whether their values work (see
+    ``BaseEngineSpec.get_embedded_credentials_error``). An edit sends unchanged
+    secrets masked, so they are first restored from ``stored_encrypted_extra``; a
+    secret that is still masked has nothing to restore it from and is not checked.
+    """
+    if not isinstance(credentials, dict):
+        return None
+    unmasked = engine_spec.unmask_encrypted_extra(
+        stored_encrypted_extra,
+        json.dumps({EMBEDDED_CREDENTIALS_KEY: credentials}),
+    )
+    try:
+        credentials = json.loads(unmasked or "{}").get(EMBEDDED_CREDENTIALS_KEY)
+    except (AttributeError, json.JSONDecodeError):
+        return None
+    validated = engine_spec.validate_embedded_credentials(credentials)
+    if validated is None or PASSWORD_MASK in validated.values():
+        return None
+    return engine_spec.get_embedded_credentials_error(validated)

superset/db_engine_specs/base.py

@@ -2651,6 +2651,23 @@ def validate_embedded_credentials(cls, credentials: Any) -> dict[str, str] | Non
         """
         return validated_embedded_credentials(credentials)
 
+    @classmethod
+    def get_embedded_credentials_error(
+        cls,
+        credentials: dict[str, str],  # pylint: disable=unused-argument
+    ) -> str | None:
+        """
+        Explain why a complete embedded fallback credential cannot be used, or
+        return ``None``.
+
+        ``validate_embedded_credentials`` only checks that the fields are filled in;
+        this checks that what they hold is usable, e.g. that a private key decodes
+        with its passphrase. Called when a credential is saved, never at query time,
+        and with unmasked secrets. The message is shown to the administrator saving
+        the connection, so it must never quote the credential's values.
+        """
+        return None
+
     @classmethod
     def apply_embedded_credentials(
         cls,

superset/db_engine_specs/snowflake.py

@@ -24,6 +24,7 @@
 from urllib import parse
 
 import sqlalchemy as sa
+from cryptography.exceptions import UnsupportedAlgorithm
 from cryptography.hazmat.backends import default_backend
 from cryptography.hazmat.primitives import serialization
 from flask import current_app as app, has_request_context
@@ -470,6 +471,28 @@ def apply_embedded_credentials(
 
         return url, engine_kwargs
 
+    @classmethod
+    def get_embedded_credentials_error(cls, credentials: dict[str, str]) -> str | None:
+        """
+        Refuse a key pair whose key does not decode with its passphrase.
+
+        Decoding happens before any connection exists, so a bad key would otherwise
+        save cleanly and only fail once an embedded viewer opens a chart.
+        """
+        if credentials.get("auth_method") != "keypair":
+            return None
+        try:
+            cls._load_private_key(credentials)
+        except (TypeError, ValueError, UnsupportedAlgorithm):
+            # The exception text is not passed on: keep key material out of the
+            # response however the parser chooses to describe it.
+            return __(
+                "The embedded guest private key could not be read. Provide a PEM "
+                "private key and its correct passphrase (leave the passphrase "
+                "empty for an unencrypted key)."
+            )
+        return None
+
     @classmethod
     def get_oauth2_authorization_uri(
         cls,

superset/models/core.py

@@ -840,14 +840,36 @@ def _get_sqla_engine(  # pylint: disable=too-many-locals  # noqa: C901
             # password on the URL, a PEM key in connect_args. Both the URL and the
             # engine kwargs participate in the cache key below, keeping guest and
             # OAuth engines separate even when they use the same database username.
-            sqlalchemy_url, engine_kwargs = (
-                self.db_engine_spec.apply_embedded_credentials(
-                    self,
-                    embedded_credentials,
-                    sqlalchemy_url,
-                    engine_kwargs,
+            try:
+                sqlalchemy_url, engine_kwargs = (
+                    self.db_engine_spec.apply_embedded_credentials(
+                        self,
+                        embedded_credentials,
+                        sqlalchemy_url,
+                        engine_kwargs,
+                    )
                 )
-            )
+            except Exception as ex:
+                # Applying it can itself reject the credential -- a key pair is
+                # decoded here, before any engine (and its `handle_error`
+                # listener) exists. Log it and refuse it as a sign-in the database
+                # rejected would be, rather than letting the decoding error
+                # surface as a server fault (sc-130269). The decoding error is
+                # dropped from the chain: the line above already names its class,
+                # and the error handler would log its traceback with the refusal.
+                _log_embedded_rejection(self.id, self.database_name, ex)
+                raise SupersetErrorException(
+                    SupersetError(
+                        message=__(
+                            "The credential configured for embedded viewers of this "
+                            "connection could not be used. An administrator needs "
+                            "to update it."
+                        ),
+                        error_type=SupersetErrorType.CONNECTION_ACCESS_DENIED_ERROR,
+                        level=ErrorLevel.ERROR,
+                    ),
+                    status=400,
+                ) from None
             # Hooks must receive the selected connection identity, not the
             # username claim supplied by the guest token.
             effective_username = sqlalchemy_url.username

superset/translations/messages.pot

@@ -14305,6 +14305,9 @@ msgstr ""
 msgid "The country code standard that Superset should expect to find in the [country] column"
 msgstr ""
 
+msgid "The credential configured for embedded viewers of this connection could not be used. An administrator needs to update it."
+msgstr ""
+
 msgid "The current version could not be identified. Restore is temporarily unavailable."
 msgstr ""
 
@@ -14406,6 +14409,9 @@ msgstr ""
 msgid "The duration of time in seconds before the cache is invalidated. Set to -1 to bypass the cache."
 msgstr ""
 
+msgid "The embedded guest private key could not be read. Provide a PEM private key and its correct passphrase (leave the passphrase empty for an unencrypted key)."
+msgstr ""
+
 msgid "The encoding format of the lines"
 msgstr ""
 

tests/unit_tests/commands/databases/validate_test.py

@@ -967,6 +967,27 @@ def test_incomplete_embedded_credentials_from_the_api(mocker: MockerFixture) ->
     assert excinfo.value.errors[0].extra["invalid"] == ["embedded_credentials"]
 
 
+def _pem(passphrase: str) -> str:
+    """A freshly generated RSA private key, PEM-encoded under ``passphrase``."""
+    from cryptography.hazmat.primitives import serialization
+    from cryptography.hazmat.primitives.asymmetric import rsa
+
+    encryption = (
+        serialization.BestAvailableEncryption(passphrase.encode())
+        if passphrase
+        else serialization.NoEncryption()
+    )
+    return (
+        rsa.generate_private_key(public_exponent=65537, key_size=2048)
+        .private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            encryption,
+        )
+        .decode()
+    )
+
+
 @pytest.mark.parametrize(
     "credentials",
     [
@@ -976,7 +997,7 @@ def test_incomplete_embedded_credentials_from_the_api(mocker: MockerFixture) ->
         {
             "username": "embedded_svc",
             "auth_method": "keypair",
-            "privatekey_body": "-----BEGIN PRIVATE KEY-----",
+            "privatekey_body": _pem(""),
         },
         {
             "username": "embedded_svc",
@@ -1123,3 +1144,42 @@ def validate_embedded_credentials(
     else:
         with pytest.raises(InvalidParametersError):
             command.run()
+
+
+@pytest.mark.parametrize(
+    "privatekey_body, privatekey_pass",
+    [
+        pytest.param(_pem("key passphrase"), "wrong passphrase", id="wrong-passphrase"),
+        pytest.param("-----BEGIN PRIVATE KEY-----", "", id="not-a-pem-key"),
+    ],
+)
+@with_feature_flags(EMBEDDED_CREDENTIAL_FALLBACK=True, EMBEDDED_SUPERSET=True)
+def test_an_unreadable_embedded_key_is_reported_on_its_field(
+    mocker: MockerFixture, privatekey_body: str, privatekey_pass: str
+) -> None:
+    """
+    The dialog validates before saving, so an unreadable key is reported on the
+    embedded credential field there too, without quoting the key (sc-130269).
+    """
+    mocker.patch("superset.commands.database.validate.DatabaseDAO")
+
+    properties = {
+        "engine": "snowflake",
+        "parameters": {
+            "embedded_credentials": {
+                "username": "embedded_svc",
+                "auth_method": "keypair",
+                "privatekey_body": privatekey_body,
+                "privatekey_pass": privatekey_pass,
+            }
+        },
+    }
+    with pytest.raises(InvalidParametersError) as excinfo:
+        ValidateDatabaseParametersCommand(properties).run()
+
+    error = excinfo.value.errors[0]
+    assert error.extra["invalid"] == ["embedded_credentials"]
+    assert "private key could not be read" in error.message
+    for secret in (privatekey_body, privatekey_pass):
+        if secret:
+            assert secret not in error.message