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连接已关闭。")