Support insecure mode in Snowflake provider

[mik-laj]

Description

Snowflake client drivers perform OCSP checking by default when connecting to a service endpoint. In normal circumstances, we should never turn off OCSP checking, and keep the default security posture of OCSP checking always ON. Only in rare emergency situations when Airflow is unable to connect to Snowflake due to OCSP issues should we consider using this workaround temporarily till the issue is resolved. For details, see: how to: turn off OCSP checking in Snowflake client drivers

To do this, we need to add the ability to pass the insecure mode = True flag to the Snowflake client in SnowflkeHook. For now, we support a limited subset of parameters:

airflow/airflow/providers/snowflake/hooks/snowflake.py

Lines 148 to 209 in de99005

	def _get_conn_params(self) -> Dict[str, Optional[str]]:
	"""
	One method to fetch connection params as a dict
	used in get_uri() and get_connection()
	"""
	conn = self.get_connection(self.snowflake_conn_id) # type: ignore[attr-defined]
	account = conn.extra_dejson.get('extra__snowflake__account', '') or conn.extra_dejson.get(
	'account', ''
	)
	warehouse = conn.extra_dejson.get('extra__snowflake__warehouse', '') or conn.extra_dejson.get(
	'warehouse', ''
	)
	database = conn.extra_dejson.get('extra__snowflake__database', '') or conn.extra_dejson.get(
	'database', ''
	)
	region = conn.extra_dejson.get('extra__snowflake__region', '') or conn.extra_dejson.get('region', '')
	role = conn.extra_dejson.get('extra__snowflake__role', '') or conn.extra_dejson.get('role', '')
	schema = conn.schema or ''
	authenticator = conn.extra_dejson.get('authenticator', 'snowflake')
	session_parameters = conn.extra_dejson.get('session_parameters')
	conn_config = {
	"user": conn.login,
	"password": conn.password or '',
	"schema": self.schema or schema,
	"database": self.database or database,
	"account": self.account or account,
	"warehouse": self.warehouse or warehouse,
	"region": self.region or region,
	"role": self.role or role,
	"authenticator": self.authenticator or authenticator,
	"session_parameters": self.session_parameters or session_parameters,
	# application is used to track origin of the requests
	"application": os.environ.get("AIRFLOW_SNOWFLAKE_PARTNER", "AIRFLOW"),
	}
	# If private_key_file is specified in the extra json, load the contents of the file as a private
	# key and specify that in the connection configuration. The connection password then becomes the
	# passphrase for the private key. If your private key file is not encrypted (not recommended), then
	# leave the password empty.
	private_key_file = conn.extra_dejson.get('private_key_file')
	if private_key_file:
	with open(private_key_file, "rb") as key:
	passphrase = None
	if conn.password:
	passphrase = conn.password.strip().encode()
	p_key = serialization.load_pem_private_key(
	key.read(), password=passphrase, backend=default_backend()
	)
	pkb = p_key.private_bytes(
	encoding=serialization.Encoding.DER,
	format=serialization.PrivateFormat.PKCS8,
	encryption_algorithm=serialization.NoEncryption(),
	)
	conn_config['private_key'] = pkb
	conn_config.pop('password', None)
	return conn_config

Use case/motivation

As above

Related issues

No response

Are you willing to submit a PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct