#!/usr/bin/env -S uv run --script # /// script # requires-python = ">=3.10" # dependencies = ["snowflake-connector-python", "cryptography"] # /// """ Test Snowflake Key-Pair Authentication Verifies that a private key + passphrase can authenticate against a Snowflake account. Optionally verifies which key slot (RSA_PUBLIC_KEY or RSA_PUBLIC_KEY_2) the key matches, which is useful during key-pair rotation. Usage: # Basic auth test SNOWFLAKE_ACCOUNT="" \ SNOWFLAKE_USER="" \ SNOWFLAKE_KEY_PATH="" \ SNOWFLAKE_KEY_PASSPHRASE="op:////passphrase" \ uv run test_keypair_auth.py # With fingerprint verification (checks which key slot matches) SNOWFLAKE_ACCOUNT="" \ SNOWFLAKE_USER="" \ SNOWFLAKE_KEY_PATH="" \ SNOWFLAKE_KEY_PASSPHRASE="op:////passphrase" \ SNOWFLAKE_VERIFY_KEY_SLOT="2" \ uv run test_keypair_auth.py Environment variables: SNOWFLAKE_ACCOUNT - Snowflake account identifier (required) SNOWFLAKE_USER - Snowflake username (required) SNOWFLAKE_KEY_PATH - Path to private key .p8 file (required) SNOWFLAKE_KEY_PASSPHRASE - Passphrase for encrypted key (optional, empty for unencrypted). Pass a 1Password secret reference (op:////) and the script resolves it via `op read` at runtime — the recommended form, since a literal passphrase in the command line or environment can leak into logs and session context. A literal passphrase value is also accepted. SNOWFLAKE_VERIFY_KEY_SLOT - If set to "1" or "2", verifies the key matches that specific slot (RSA_PUBLIC_KEY or RSA_PUBLIC_KEY_2). Useful during rotation to confirm the new key was deployed to the correct slot. """ import hashlib import os import subprocess import sys from cryptography.hazmat.backends import default_backend from cryptography.hazmat.primitives import serialization import snowflake.connector def compute_public_key_fingerprint(private_key) -> str: """Compute the SHA-256 fingerprint of the public key derived from a private key. This matches the fingerprint format Snowflake stores in RSA_PUBLIC_KEY_FP and RSA_PUBLIC_KEY_2_FP (without the 'SHA256:' prefix). """ public_key_der = private_key.public_key().public_bytes( encoding=serialization.Encoding.DER, format=serialization.PublicFormat.SubjectPublicKeyInfo, ) digest = hashlib.sha256(public_key_der).digest() import base64 return base64.b64encode(digest).decode("ascii") def get_snowflake_key_fingerprints(cursor, username: str) -> dict[str, str]: """Query Snowflake for the user's public key fingerprints.""" cursor.execute(f"DESC USER {username}") rows = cursor.fetchall() fingerprints = {} for row in rows: prop = row[0] # property name val = row[1] # property value if prop == "RSA_PUBLIC_KEY_FP" and val: # Strip 'SHA256:' prefix if present fingerprints["1"] = val.replace("SHA256:", "") elif prop == "RSA_PUBLIC_KEY_2_FP" and val: fingerprints["2"] = val.replace("SHA256:", "") return fingerprints def resolve_passphrase(value: str) -> str | None: """Resolve the passphrase, reading op:// secret references via the 1Password CLI. Resolving inside the script keeps the secret out of the shell command string and tool output — never resolve it with `$(op read ...)` on the command line. Returns None if resolution fails (an error has been printed). """ if not value.startswith("op://"): return value try: result = subprocess.run( ["op", "read", "--no-newline", value], capture_output=True, text=True, ) except FileNotFoundError: print("FAILED: `op` CLI not found. Install the 1Password CLI to resolve op:// references.") return None if result.returncode != 0: print(f"FAILED: could not resolve passphrase reference {value}") print(f" op read error: {result.stderr.strip()}") print(" Check that the op:// reference is correct and `op` is signed in (`op whoami`).") return None return result.stdout def main() -> int: # Read config from environment account = os.environ.get("SNOWFLAKE_ACCOUNT", "") user = os.environ.get("SNOWFLAKE_USER", "") key_path = os.environ.get("SNOWFLAKE_KEY_PATH", "") passphrase_str = os.environ.get("SNOWFLAKE_KEY_PASSPHRASE", "") verify_slot = os.environ.get("SNOWFLAKE_VERIFY_KEY_SLOT", "") if not account or not user or not key_path: print("Error: SNOWFLAKE_ACCOUNT, SNOWFLAKE_USER, and SNOWFLAKE_KEY_PATH must be set.") return 1 if verify_slot and verify_slot not in ("1", "2"): print("Error: SNOWFLAKE_VERIFY_KEY_SLOT must be '1' or '2' if set.") return 1 passphrase_str = resolve_passphrase(passphrase_str) if passphrase_str is None: return 1 passphrase = passphrase_str.encode() if passphrase_str else None print("Snowflake Key-Pair Auth Test") print("============================") print(f" Account: {account}") print(f" User: {user}") print(f" Key file: {key_path}") print(f" Encrypted: {'yes' if passphrase else 'no'}") if verify_slot: print(f" Verify slot: RSA_PUBLIC_KEY{'_2' if verify_slot == '2' else ''}") print() # Load and decrypt private key try: with open(key_path, "rb") as f: private_key = serialization.load_pem_private_key( f.read(), password=passphrase, backend=default_backend() ) except FileNotFoundError: print(f"FAILED: Key file not found: {key_path}") return 1 except ValueError as e: print(f"FAILED: Could not decrypt private key (wrong passphrase?): {e}") return 1 # Convert to DER for the Snowflake connector private_key_bytes = private_key.private_bytes( encoding=serialization.Encoding.DER, format=serialization.PrivateFormat.PKCS8, encryption_algorithm=serialization.NoEncryption(), ) # Connect to Snowflake try: conn = snowflake.connector.connect( account=account, user=user, private_key=private_key_bytes, ) cursor = conn.cursor() cursor.execute("SELECT CURRENT_USER(), CURRENT_ROLE(), CURRENT_ACCOUNT()") row = cursor.fetchone() print("Connection successful!") print(f" User: {row[0]}") print(f" Role: {row[1]}") print(f" Account: {row[2]}") except Exception as e: print(f"Connection FAILED: {e}") return 1 # Fingerprint verification (optional) if verify_slot: print() print("Fingerprint Verification") print("------------------------") local_fp = compute_public_key_fingerprint(private_key) print(f" Local key fingerprint: {local_fp}") try: sf_fingerprints = get_snowflake_key_fingerprints(cursor, user) except Exception as e: print(f" WARNING: Could not query fingerprints (need SECURITYADMIN or OWNERSHIP): {e}") print(" Skipping fingerprint verification.") cursor.close() conn.close() return 0 for slot, fp in sf_fingerprints.items(): slot_name = f"RSA_PUBLIC_KEY{'_2' if slot == '2' else ''}" match = "MATCH" if fp == local_fp else "no match" print(f" {slot_name}_FP: {fp} ({match})") if not sf_fingerprints: print(" No public key fingerprints found on user.") target_fp = sf_fingerprints.get(verify_slot) if target_fp and target_fp == local_fp: slot_name = f"RSA_PUBLIC_KEY{'_2' if verify_slot == '2' else ''}" print(f"\n Verified: key matches {slot_name}") elif target_fp: slot_name = f"RSA_PUBLIC_KEY{'_2' if verify_slot == '2' else ''}" print(f"\n FAILED: key does NOT match {slot_name}") cursor.close() conn.close() return 1 else: slot_name = f"RSA_PUBLIC_KEY{'_2' if verify_slot == '2' else ''}" print(f"\n FAILED: {slot_name} is not set on this user") cursor.close() conn.close() return 1 cursor.close() conn.close() return 0 if __name__ == "__main__": sys.exit(main())