Need to add Snowflake connection using DVT command with RSA key.

[shubhamTeli]

i want setup a connection of Snowflake DB using RSA key as don't want to use password for security purpose.

Can anyone suggest how can i setup the snowflake connection?
As i did not find the RSA key option in DVT connection guide.

[shubhamTeli]

@helensilva14 Any update on above question?

[sundar-mudupalli-work]

Shubham,

DVT uses ibis and ibis may provide a way to connect with a private key. The documentation is here. Assuming you are comfortable using Python, pandas and installing the ibis library, could you test if you can connect to snowflake using ibis. The python code would be something like this

import ibis
con=ibis.snowflake.connect( ....
     - as described in the link -
)
results=con.sql('some sql query').execute()
print(results)

If you are successful with this, we may be able some changes to DVT to allow authenticating to Snowflake with private keys. If you are a GCP customer or working with a GCP customer, can you ask your GCP account rep to reach out to me internally - this will help us prioritize your request.

Thank you.

Sundar Mudupalli

[shubhamTeli]

Thanks @sundar-mudupalli-work for reply.

Tried to connect Snowflake using ibis with below option from documentation https://ibis-project.org/backends/snowflake#authenticating-with-key-pair-authentication

However, getting below error:

Please find below code which i have tried.

import ibis
from ibis import snowflake
import os

private_key_path = r'rsa_key.der'

with open(private_key_path, 'rb') as key:
    p_key = key.read()

print(type(p_key))

connection_details = {
    'user': '<my_user>',
    'account': '<my_account>',
    'database': '<my_database>/<my_schema>',
    'warehouse': '<my_warehouse>',
    'private_key': p_key,  # Directly pass the private key as bytes
    'password': ''  # Leave this empty if using private key authentication
}

try:
    con = snowflake.connect(**connection_details)
    print("Connection successful")
    table = con.table("<my_table>")
    result = table.limit(3).execute()
    print(table.schema)
    print(result)
    print(type(result))

except Exception as e:
    print("Error connecting to Snowflake:", e)

Can you please help here?

[sundar-mudupalli-work]

Hi,

You may need to provide the RSAPrivateKey, this example from the doc seems to create an object of the type cryptography.hazmat.bindings._rust.openssl.rsa.RSAPrivateKey.

Try that out.

Thanks.

Sundar Mudupalli

[shubhamTeli]

hi @sundar-mudupalli-work

i tried the option which you suggested. However still getting below error.

Please find below code which i have tried.

import ibis
from ibis import snowflake

from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.backends import default_backend

# Load the private key from a file
private_key_path = 'rsa_key.pem'
with open(private_key_path, 'rb') as key_file:
    private_key_data = key_file.read()

# Convert the private key data to an RSAPrivateKey object
p_key = serialization.load_pem_private_key(
    private_key_data,
    password=None,  # If your key is encrypted, provide the password here
    backend=default_backend()
)

print(type(p_key))

# Define your Snowflake connection parameters
connection_details = {
    'user': '<my_user>',
    'account': '<my_account>',
    'database': '<my_database>/<my_schema>',
    'warehouse': '<my_warehouse>',
    'private_key': p_key,  # Directly pass the private key as RSAPrivateKey
    'password': ''  # Leave this empty if using private key authentication
}

try:
    # Create a connection to Snowflake
    con = snowflake.connect(**connection_details)
    print("Connection successful")
    table = con.table("adh_com_globe_dh_ds_customerdataset_vw")
    result = table.limit(3).execute()
    print(table.schema)
    print(result)
    print(type(result))

except Exception as e:
    print("Error connecting to Snowflake:", e)