Python集成Snowflake与Okta SSO：安全认证与最佳实践

import snowflake.connector

try:
    conn = snowflake.connector.connect(
        user="your_snowflake_user",
        password="your_password", # 尝试使用Okta密码
        authenticator='https://<okta_account_name>.okta.com', # 尝试使用Okta认证URL
        account = "your_snowflake_account"
    )
    # ... 执行查询
except Exception as e:
    print(f"连接失败: {e}")

import os
import requests
import json
import time

# --- Okta配置信息 ---
# 建议从环境变量或安全的配置管理服务中加载这些敏感信息
OKTA_DOMAIN = os.getenv("OKTA_DOMAIN", "https://<your_okta_domain>.okta.com") # 替换为你的Okta域名
OKTA_CLIENT_ID = os.getenv("OKTA_CLIENT_ID", "your_okta_client_id")         # 替换为你的Okta Client ID
OKTA_CLIENT_SECRET = os.getenv("OKTA_CLIENT_SECRET", "your_okta_client_secret") # 替换为你的Okta Client Secret
OKTA_AUTH_SERVER_ID = os.getenv("OKTA_AUTH_SERVER_ID", "default") # 授权服务器ID，通常为'default'
OKTA_SCOPE = os.getenv("OKTA_SCOPE", "openid profile email snowflake_full_access") # 根据需要调整OAuth Scope

# 获取OAuth Token的URL
token_url = f"{OKTA_DOMAIN}/oauth2/{OKTA_AUTH_SERVER_ID}/v1/token"

def get_okta_access_token():
    """
    通过Okta Client Credentials Grant获取OAuth Access Token。
    """
    headers = {
        "Content-Type": "application/x-www-form-urlencoded"
    }
    data = {
        "grant_type": "client_credentials",
        "client_id": OKTA_CLIENT_ID,
        "client_secret": OKTA_CLIENT_SECRET,
        "scope": OKTA_SCOPE
    }

    try:
        print(f"尝试从 {token_url} 获取Okta OAuth Access Token...")
        response = requests.post(token_url, headers=headers, data=data)
        response.raise_for_status() # 如果请求失败，抛出HTTPError
        token_data = response.json()
        access_token = token_data.get("access_token")
        expires_in = token_data.get("expires_in") # 令牌有效期，单位秒

        if not access_token:
            raise ValueError("未能从Okta响应中获取access_token")

        print(f"成功获取Okta OAuth Access Token (有效期: {expires_in} 秒)。")
        return access_token, expires_in
    except requests.exceptions.RequestException as e:
        print(f"获取Okta OAuth Token失败: {e}")
        if e.response:
            print(f"Okta响应错误: {e.response.text}")
        return None, 0
    except ValueError as e:
        print(f"处理Okta Token响应失败: {e}")
        return None, 0

# --- 示例：获取令牌 ---
access_token, expires_in = get_okta_access_token()
if not access_token:
    print("无法获取OAuth令牌，程序退出。")
    exit(1)

import snowflake.connector
# import os # 已在上面导入

# --- Snowflake配置信息 ---
SNOWFLAKE_ACCOUNT = os.getenv("SNOWFLAKE_ACCOUNT", "your_snowflake_account_identifier") # 替换为你的Snowflake账号标识符
SNOWFLAKE_USER = os.getenv("SNOWFLAKE_USER", "your_snowflake_user_name") # 替换为Snowflake中映射的用户名

# --- 使用获取到的Access Token连接Snowflake ---
conn = None
try:
    print(f"尝试使用OAuth令牌连接Snowflake账号: {SNOWFLAKE_ACCOUNT}，用户: {SNOWFLAKE_USER}...")
    conn = snowflake.connector.connect(
        account=SNOWFLAKE_ACCOUNT,
        user=SNOWFLAKE_USER,
        authenticator='oauth', # 明确指定使用OAuth认证
        token=access_token,    # 提供OAuth访问令牌
        # warehouse='your_warehouse', # 可选：指定默认仓库
        # database='your_database',   # 可选：指定默认数据库
        # schema='your_schema'        # 可选：指定默认Schema
    )
    print("成功连接到Snowflake！")

    # 执行一个示例查询
    cursor = conn.cursor()
    cursor.execute("SELECT CURRENT_VERSION()")
    result = cursor.fetchone()
    print(f"Snowflake版本: {result[0]}")

except snowflake.connector.errors.ProgrammingError as e:
    print(f"Snowflake连接或查询失败: {e}")
    # 对于认证错误，可以检查 e.sfqid 或 e.msg 以获取更多细节
    if "AUTHENTICATION_REQUEST_FAILED" in str(e):
        print("可能原因：OAuth令牌无效、过期，或Okta与Snowflake用户映射问题。")
except Exception as e:
    print(f"发生未知错误: {e}")
finally:
    if conn:
        conn.close()
        print("Snowflake连接已关闭。")