FrancescoCastaldi commented on code in PR #45186:
URL: https://github.com/apache/superset/pull/45186#discussion_r4272918622
##########
tests/unit_tests/db_engine_specs/test_snowflake.py:
##########
@@ -854,3 +854,55 @@ def
test_snowflake_oauth2_exception_catches_refresh_token_error() -> None:
"OAuth2TokenRefreshError must be caught by "
"SnowflakeEngineSpec.oauth2_exception"
)
+
+
+def test_snowflake_parameters_json_schema() -> None:
+ """
+ Verify parameters_json_schema constructs OpenAPI schema via
+ APISpec and MarshmallowPlugin.
+ """
+ from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+ schema = SnowflakeEngineSpec.parameters_json_schema()
+ assert isinstance(schema, dict)
+ assert "properties" in schema
+ assert "account" in schema["properties"]
+
+
+def test_snowflake_update_params_from_encrypted_extra_keypair() -> None:
+ """
+ Verify update_params_from_encrypted_extra parses PEM private key and sets
DER bytes.
+ Protects against missing cryptography.hazmat imports.
+ """
+ from cryptography.hazmat.primitives import serialization
+ from cryptography.hazmat.primitives.asymmetric import rsa
+
+ from superset.db_engine_specs.snowflake import SnowflakeEngineSpec
+
+ private_key = rsa.generate_private_key(
+ public_exponent=65537,
+ key_size=2048,
+ )
+ pem_bytes = private_key.private_bytes(
+ encoding=serialization.Encoding.PEM,
+ format=serialization.PrivateFormat.PKCS8,
+ encryption_algorithm=serialization.NoEncryption(),
+ )
+
+ mock_db = mock.MagicMock()
+ mock_db.encrypted_extra = json.dumps(
+ {
+ "auth_method": "keypair",
+ "auth_params": {
+ "privatekey_body": pem_bytes.decode("utf-8"),
+ },
+ }
+ )
+
+ params: dict[str, Any] = {}
+ SnowflakeEngineSpec.update_params_from_encrypted_extra(mock_db, params)
+
+ assert "connect_args" in params
+ assert "private_key" in params["connect_args"]
+ assert isinstance(params["connect_args"]["private_key"], bytes)
Review Comment:
Good catch. Generic `isinstance(bytes)` and length checks were insufficient
to guard against PEM regression.
### Direct Resolution
In commit `af547897`, we replaced the generic byte checks in
`test_snowflake_update_params_from_encrypted_extra_keypair` with rigorous DER
verification:
1. **Strict DER Byte Equality**: We assert that
`params["connect_args"]["private_key"]` exactly matches
`private_key.private_bytes(encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption())`.
2. **Explicit Anti-PEM Guard**: We assert `not
params["connect_args"]["private_key"].startswith(b"-----BEGIN")`, immediately
failing if PEM text/bytes are produced.
3. **Round-Trip DER Deserialization**: We deserialize
`params["connect_args"]["private_key"]` using
`serialization.load_der_private_key(..., password=None)` and assert that the
resulting private key numbers match the original RSA key. If PEM bytes were
supplied, `load_der_private_key` raises a `ValueError`.
### Anticipated Edge Case and Preemptive Hardening
We analyzed related keypair execution paths:
- **Passphrase-Protected Keys**: Added
`test_snowflake_update_params_from_encrypted_extra_keypair_passphrase` to
verify that when `auth_params` contains a passphrase-encrypted PEM key
(`privatekey_pass`), `update_params_from_encrypted_extra` successfully decrypts
it and outputs unencrypted DER-encoded PKCS8 bytes without requiring runtime
credentials.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
