Snowflake SQL API の認証方法として PAT 試してみた
はじめに
Snowflake では SQL API として REST API でクエリを実行する機能が提供されています。また、この際の認証方法として2025年4月にリリースされた Programmatic access token(PAT)がサポートされています。
SQL API を使用する機会があったので、PAT での認証を試して記事としました。
サーバーへの認証
前提として本機能の概要については以下の記事でも紹介されていますので、こちらが参考になると思います。
これまでサーバーへの認証として以下の方法がサポートされていました。
- キーペア
- OAuth
上記の場合、各種キーファイルの管理や OAuth の構成が必要でしたが、PAT であればパスワードのように使用でき、以下のような機能が提供されています。
- サービスユーザーに対しても生成できる
- PAT 生成時に特定のロールに制限できる
- ローテーション/取り消しなど管理機能
キーペアの場合
例としてキーペアを使用する場合、事前に公開鍵と秘密鍵のペアを生成し、その公開鍵を Snowflake ユーザーに割り当てます。
以下はドキュメント記載のコードをベースに一部書き換えてたものですが、上記の設定後 Python などのアプリケーションコード内で JWT (JSON Web Token) を生成する仕組みが必要です。
import base64
import hashlib
import jwt
import os
from datetime import datetime, timedelta, timezone
from getpass import getpass
from dotenv import load_dotenv
from cryptography.hazmat.primitives.serialization import (
load_pem_private_key,
Encoding,
PublicFormat
)
from cryptography.hazmat.backends import default_backend
# .env ファイルを読み込む
load_dotenv()
# 環境変数から設定値を取得
ACCOUNT_IDENTIFIER = os.getenv("SNOWFLAKE_ACCOUNT")
USER_NAME = os.getenv("SNOWFLAKE_USER")
PRIVATE_KEY_FILE_PATH = os.getenv("PRIVATE_KEY_PATH")
# パスフレーズ取得(秘密鍵が暗号化されている場合)
def get_private_key_passphrase():
return getpass('Passphrase for private key: ')
# 秘密鍵の読み込み
with open(PRIVATE_KEY_FILE_PATH, 'rb') as pem_in:
pemlines = pem_in.read()
try:
private_key = load_pem_private_key(pemlines, None, default_backend())
except TypeError:
# パスフレーズが必要な場合
passphrase = get_private_key_passphrase().encode()
private_key = load_pem_private_key(pemlines, passphrase, default_backend())
# 公開鍵のバイト列を取得(DER形式)
public_key_raw = private_key.public_key().public_bytes(
encoding=Encoding.DER,
format=PublicFormat.SubjectPublicKeyInfo
)
# SHA256でフィンガープリントを計算
sha256hash = hashlib.sha256()
sha256hash.update(public_key_raw)
public_key_fp = 'SHA256:' + base64.b64encode(sha256hash.digest()).decode('utf-8')
print(f"公開鍵フィンガープリント: {public_key_fp}")
# アカウント・ユーザー情報の準備
account = ACCOUNT_IDENTIFIER.upper()
user = USER_NAME.upper()
qualified_username = f"{account}.{user}"
# JWT の有効期間と現在時刻
now = datetime.now(timezone.utc)
lifetime = timedelta(minutes=59)
# JWT ペイロードの作成
payload = {
"iss": qualified_username + '.' + public_key_fp,
"sub": qualified_username,
"iat": now,
"exp": now + lifetime
}
# JWT の生成(RS256で署名)
encoding_algorithm = "RS256"
token = jwt.encode(payload, key=private_key, algorithm=encoding_algorithm)
# バイト列なら文字列に変換(PyJWT < 2.0 対応)
if isinstance(token, bytes):
token = token.decode('utf-8')
print("\nJWT トークン(署名済み):")
print(token)
# JWT の検証(オプション)
decoded_token = jwt.decode(token, key=private_key.public_key(), algorithms=[encoding_algorithm])
print("\nJWT ペイロード(デコード済み):")
print(decoded_token)
PAT による認証
2025年4月にアクセストークンを発行し、そのアクセストークンで外部から認証ができる機能として Programmatic access token がリリースされました。
本機能の詳細は以下の記事をご参照ください。
PAT は Snowflake SQL API の認証時にも使用できます。
認証にPATを使用する場合、リクエストには以下の通り特定の HTTP ヘッダーを含める必要があります。
- Authorization ヘッダー
Authorization: Bearer <token_secret>
を設定し<token_secret>
に生成された PAT の値を指定
- X-Snowflake-Authorization-Token-Type ヘッダー
- こちらはオプション
X-Snowflake-Authorization-Token-Type: PROGRAMMATIC_ACCESS_TOKEN
を設定することも可能
その他は、PAT を使用する前提条件を満たしている必要があります。こちらについては、上記のブログ記事でも解説されていますのであわせてご参照ください。
試してみる
システムユーザーを作成
PAT を生成する検証用のシステムユーザーを作成します。
注意点として検証のため強めの権限を使用しています。
USE ROLE USERADMIN;
CREATE or replace USER api_user
DEFAULT_ROLE = SYADMIN
TYPE = SERVICE;
USE ROLE SECURITYADMIN;
GRANT ROLE SYSADMIN TO USER api_user;
ネットワークポリシーを設定
デフォルトでは PAT の生成にはネットワークポリシーの有効化が必要です。ここでは以下の手順で設定しました。
USE ROLE SECURITYADMIN;
CREATE NETWORK POLICY IF NOT EXISTS api_user_net_policy
ALLOWED_IP_LIST = ('xx.xx.xx.xx/32');
--ユーザーにネットワークポリシーを適用
ALTER USER api_user SET NETWORK_POLICY = api_user_net_policy;
PAT を生成
ALTER コマンドで PAT を生成し、控えておきます。
ALTER USER api_user ADD PROGRAMMATIC ACCESS TOKEN api_token
days_to_expiry = 365 -- アクセストークンの有効期限:最大365日
role_restriction = 'SYSADMIN';
リクエストを送信
PAT を使用するリクエストを送信します。
ここでは以下の通り環境変数に設定しました。
export SF_TOKEN="<PAT>"
リクエストを送信します。あわせてサンプルデータに対してフィルタするクエリを実行しています。
curl -X POST "https://<org>-<account>.snowflakecomputing.com/api/v2/statements" \
-H "Authorization: Bearer $SF_TOKEN" \
-H "X-Snowflake-Authorization-Token-Type: PROGRAMMATIC_ACCESS_TOKEN" \
-H "Accept: application/json" \
-H "Content-Type: application/json" \
-d '{
"statement": "select * from snowflake_sample_data.tpch_sf1.region where R_NAME = '\''AMERICA'\''",
"warehouse": "COMPUTE_WH"
}'
問題なく認証され、ここでは以下の応答でした。
クエリ結果は data プロパティに行ごとの配列として返されます。
{
"resultSetMetaData" : {
"numRows" : 1,
"format" : "jsonv2",
"partitionInfo" : [ {
"rowCount" : 1,
"uncompressedSize" : 60
} ],
"rowType" : [ {
"name" : "R_REGIONKEY",
"database" : "SNOWFLAKE_SAMPLE_DATA",
"schema" : "TPCH_SF1",
"table" : "REGION",
"nullable" : false,
"scale" : 0,
"byteLength" : null,
"length" : null,
"type" : "fixed",
"precision" : 38,
"collation" : null
}, {
"name" : "R_NAME",
"database" : "SNOWFLAKE_SAMPLE_DATA",
"schema" : "TPCH_SF1",
"table" : "REGION",
"nullable" : false,
"scale" : null,
"byteLength" : 100,
"length" : 25,
"type" : "text",
"precision" : null,
"collation" : null
}, {
"name" : "R_COMMENT",
"database" : "SNOWFLAKE_SAMPLE_DATA",
"schema" : "TPCH_SF1",
"table" : "REGION",
"nullable" : true,
"scale" : null,
"byteLength" : 608,
"length" : 152,
"type" : "text",
"precision" : null,
"collation" : null
} ]
},
"data" : [ ["1","AMERICA","hs use ironic, even requests. s"] ],
"code" : "090001",
"statementStatusUrl" : "/api/v2/statements/01bca585-0002-71b0-0002-45b20024edba?requestId=6f945777-8a33-4ead-845f-d3fa04b2c77f",
"requestId" : "6f945777-8a33-4ead-845f-d3fa04b2c77f",
"sqlState" : "00000",
"statementHandle" : "01bca585-0002-71b0-0002-45b20024edba",
"message" : "Statement executed successfully.",
"createdOn" : 1748421440715
}
ログイン履歴も確認してみます。今回のログインのに抜粋していますが、以下の通り [FIRST_AUTHENTICATION_FACTOR] に「PROGRAMMATIC_ACCESS_TOKEN」として記録されています。
>SELECT
EVENT_TYPE
,USER_NAME
,REPORTED_CLIENT_TYPE
,FIRST_AUTHENTICATION_FACTOR
FROM TABLE(information_schema.login_history())
ORDER BY event_timestamp DESC;
+------------+----------------+----------------------+-----------------------------+
| EVENT_TYPE | USER_NAME | REPORTED_CLIENT_TYPE | FIRST_AUTHENTICATION_FACTOR |
|------------+----------------+----------------------+-----------------------------|
| LOGIN | API_USER | SQL_API | PROGRAMMATIC_ACCESS_TOKEN |
+------------+----------------+----------------------+-----------------------------+
さいごに
Snowflake SQL API への認証方法として Programmatic access token(PAT)を試してみました。
こちらの内容が何かの参考になれば幸いです。