はじめに
Snowflakeはセキュリティ強化の一環として，単一要素パスワードによるサインインの廃止を段階的に進めている．公式のロールアウト計画（Planning for the deprecation of single-factor password sign-ins）によれば，MFA（多要素認証）の必須化とサービスユーザーのパスワード認証の禁止は，おおむね次の時期に強制化される見込みである（いずれも見込み時期であり，変更の可能性がある）．
- フェーズ1（2025年9月〜2026年1月見込み）: すべてのSnowsight利用者（人間ユーザー）に対するMFAの必須化
- フェーズ2（2026年5月〜7月見込み）: 新規ユーザーに対する強力な認証の必須化．新規のサービスユーザーは
TYPE = SERVICEが必須となり，パスワードを利用できなくなる - フェーズ3（2026年8月〜10月見込み）: すべての人間ユーザー・サービスユーザーに対する強力な認証の必須化．サービスユーザーはパスワード認証が利用できなくなる
このように，特にプログラムから接続するサービスユーザーでは，段階的にパスワード認証が利用できなくなる．MFA導入前は，別ユーザーのID/パスワードを用いて snowflake.connector.connect() で接続し，そのユーザーの権限で操作を行うことが可能であった．しかしながら，この強制化に伴い，プログラムからのパスワード認証に依存する従来の方法は使えなくなる．
以前，Snowflake のタスクの実行プログラムを外部ファイル化する方法2（Streamlitアプリ） において，ストアドプロシージャを編集する権限を持たないユーザーでも，Streamlitアプリを介してストアドプロシージャ内のSQLを編集できる仕組みを紹介した．しかしながら，当時の実装は snowflake.connector.connect() によるID/パスワード認証を前提としており，MFA必須化後はそのまま利用することが難しくなった．本記事は，その仕組みをMFA環境に対応させ，キーペア認証を用いて作り直したものと位置づけられる．
本記事では，既存ユーザーにキーペア認証を設定し，Streamlit in Snowflake からそのユーザーとして接続することで，ログインユーザーが持たない権限（例: CREATE PROCEDURE）を別ユーザー経由で代行するアプリの作り方について述べる．新たにユーザーを作成する必要はなく，既存ユーザーに公開鍵を追加するだけで実現できる点が特徴である．
背景
なぜ別ユーザーの権限が必要なのかについて述べる．
ユースケース
実運用では，次のような要件がしばしば生じる．
- 一般ユーザーに
CREATE PROCEDURE権限を直接付与したくない - 一方で，プロシージャの中身（ビジネスロジック）は一般ユーザーに編集させたい
- 管理者が毎回代行するのは運用負荷が高い
- 既存の管理者ユーザーの権限を借り，アプリ内で自動的に操作を代行したい
すなわち，権限そのものは絞りつつ，特定の操作だけをアプリ経由で許可したいという要求である．
MFA導入前の旧方式
MFA導入前は，別ユーザーのID/パスワードで接続する方式を使っていた．なお，前提として述べておくと，パスワードのような秘密情報をコードに直接埋め込むことは，MFA導入の有無にかかわらず推奨されない．本来は，後述するSnowflake Secretなどに格納し，コードから分離して管理すべきものである．
# 旧方式: 別ユーザーのID/パスワードで接続
import snowflake.connector
conn = snowflake.connector.connect(
user='ADMIN_USER',
password='xxxxxxxx', # 直書きは非推奨（本来はSecretに格納すべき）。かつMFA必須化後は認証自体が通らない
account='myaccount',
role='ACCOUNTADMIN'
)
すなわち，秘密情報を安全に管理すること自体は，認証方式によらず前提として求められる．その上で，本記事で扱う本質的な課題は別の点にある．MFA必須化により，パスワード認証そのものがプログラムからは利用できなくなることである．したがって，秘密情報の格納方法を変えるだけでは解決できず，認証方式自体をパスワードから別の方式へ移行する必要がある．
MFA導入後の新方式: 既存ユーザー + キーペア認証
そこで，既存ユーザーにキーペア認証を追加し，RSA秘密鍵で接続する方式を用いる．
# 新方式: 既存ユーザーにキーペア認証を追加（MFAの影響を受けない）
from snowflake.snowpark import Session
sp_session = Session.builder.configs({
"account": "myaccount",
"user": "ADMIN_USER", # ← 既存ユーザー
"private_key": private_key_bytes, # ← RSA秘密鍵
"role": "ACCOUNTADMIN",
}).create()
キーペア認証はMFAバイパスが可能なため，既存ユーザーへのプログラム接続が引き続き可能である．また，新たにユーザーを作る必要はなく，既存ユーザーに公開鍵を設定するだけで使える点も利点として挙げられる．
手順
以下では，全5ステップで構築手順を述べる．第1ステップでは既存ユーザーへのキーペア認証の設定，第2ステップではSecretへの秘密鍵の格納，第3ステップではExternal Access Integrationの設定，第4ステップではStreamlitアプリの作成，第5ステップでは動作確認について述べる．
Step 1: 既存ユーザーにRSAキーペアを設定する
1-1. RSAキーペアの生成（ローカル端末）
まず，ローカル端末でRSAキーペアを生成する．
# 秘密鍵の生成（PKCS8形式・暗号化なし）
openssl genrsa 2048 | openssl pkcs8 -topk8 -nocrypt -out sp_key.p8
# 公開鍵の生成
openssl rsa -in sp_key.p8 -pubout -out sp_key.pub
# Snowflakeに設定する公開鍵の値を取得（ヘッダー/フッターを除去）
grep -v "BEGIN\|END" sp_key.pub | tr -d '\n'
1-2. 既存ユーザーに公開鍵を設定
次に，取得した公開鍵を既存ユーザーに設定する．ユーザーを新規作成する必要はなく，既存ユーザーにそのまま設定する．
-- 既存ユーザーにRSA公開鍵を追加
-- ※ ユーザーを新規作成する必要はない。既存ユーザーにそのまま設定する
ALTER USER "ADMIN_USER" SET RSA_PUBLIC_KEY = 'MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEA...';
ここで重要な点として，ALTER USER ... SET RSA_PUBLIC_KEY は既存ユーザーの認証方法を追加するだけであり，既存のパスワード認証やMFA設定には影響しないことが挙げられる．そのため，当該ユーザーは引き続きUIからMFA付きでログインでき，同時にキーペア認証でのプログラム接続も可能になる．
1-3. 設定確認
設定が反映されたことを確認する．
-- 公開鍵のフィンガープリントが表示されることを確認
DESC USER "ADMIN_USER";
-- RSA_PUBLIC_KEY_FP の行に SHA256:... と表示されればOK
Step 2: Snowflake Secret に秘密鍵を格納
Streamlitアプリから秘密鍵を安全に参照するため，Snowflake Secretに格納する．
-- 秘密鍵を格納するSecret
CREATE OR REPLACE SECRET DEMO_DB.PUBLIC.ADMIN_PRIVATE_KEY_SECRET
TYPE = GENERIC_STRING
SECRET_STRING = '-----BEGIN PRIVATE KEY-----
MIIEvQIBADANBgkqhkiG9w0BAQEFAASC...
<sp_key.p8 ファイルの中身をそのまま貼り付け>
...CSKwJ4hQ==
-----END PRIVATE KEY-----';
なお，秘密鍵はコードにハードコードしないことが望ましい．Snowflake Secretに格納し，後述のExternal Access Integrationで参照する構成とする．
Step 3: External Access Integration の設定
Streamlit in Snowflakeから別ユーザーとしてセッションを張るには，外部アクセス許可が必要になる．
-- Network Rule: 自アカウントへの接続を許可
CREATE OR REPLACE NETWORK RULE sp_network_rule
TYPE = HOST_PORT
MODE = EGRESS
VALUE_LIST = ('<あなたのアカウント>.snowflakecomputing.com');
-- External Access Integration
CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION sp_access_integration
ALLOWED_NETWORK_RULES = (sp_network_rule)
ALLOWED_AUTHENTICATION_SECRETS = (DEMO_DB.PUBLIC.ADMIN_PRIVATE_KEY_SECRET)
ENABLED = TRUE;
Step 4: Streamlit アプリの作成
4-1. アプリコード
以下に，アプリのコードを示す．ログインユーザーのセッションでUIを構成し，別ユーザーのセッション（キーペア認証）で権限を要する操作を実行する構成である．
# Streamlit app that lets users edit stored procedures via another user (key-pair auth)
# Co-authored with CoCo
import streamlit as st
import os
import re
from snowflake.snowpark import Session
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.backends import default_backend
# --- 固定設定 ---
SP_USER = "ADMIN_USER" # 接続先の既存ユーザー
SP_ROLE = "ACCOUNTADMIN" # そのユーザーのロール
SP_WAREHOUSE = "COMPUTE_WH" # ウェアハウス
SECRET_ENV_NAME = "SVC_PRIVATE_KEY" # 環境変数名（Secrets設定のキー）
st.set_page_config(page_title="SP Editor", layout="wide")
st.title("ストアドプロシージャ エディタ")
st.caption(f"別ユーザー（{SP_USER}）の権限でプロシージャを編集")
st.divider()
# --- ログインユーザーのセッション ---
conn = st.connection("snowflake", ttl=os.getenv("SNOWFLAKE_CONNECTION_TTL"))
user_session = conn.session()
current_user = user_session.sql("SELECT CURRENT_USER()").collect()[0][0]
current_role = user_session.sql("SELECT CURRENT_ROLE()").collect()[0][0]
current_account = user_session.sql("SELECT CURRENT_ACCOUNT()").collect()[0][0]
st.info(f"ログインユーザー: **{current_user}** ({current_role}) → 代行: **{SP_USER}** ({SP_ROLE})")
# 接続モード表示
if os.environ.get(SECRET_ENV_NAME, ""):
st.success(f"接続モード: キーペア認証（別ユーザー {SP_USER} で実行）")
else:
st.warning(f"接続モード: USE ROLE フォールバック（{SP_ROLE} に切替。キーペア未設定）")
# --- 別ユーザー接続ヘルパー ---
def get_sp_session():
"""キーペア認証で別ユーザーとして接続"""
pem_key = os.environ.get(SECRET_ENV_NAME, "")
if not pem_key:
return None, "秘密鍵が未設定"
try:
private_key = serialization.load_pem_private_key(
pem_key.encode('utf-8'), password=None, backend=default_backend()
)
private_key_bytes = private_key.private_bytes(
encoding=serialization.Encoding.DER,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption()
)
sp_session = Session.builder.configs({
"account": current_account,
"user": SP_USER,
"private_key": private_key_bytes,
"role": SP_ROLE,
"warehouse": SP_WAREHOUSE,
}).create()
return sp_session, None
except Exception as e:
return None, str(e)
def execute_as_sp(sql_statements):
"""別ユーザーの権限でSQL実行（キーペア認証 or USE ROLEフォールバック）"""
pem_key = os.environ.get(SECRET_ENV_NAME, "")
if pem_key:
# キーペア認証で別ユーザーとして接続
sp_session, error = get_sp_session()
if sp_session is None:
return False, error
try:
results = []
for stmt in sql_statements:
results.append(sp_session.sql(stmt).collect())
return True, results
except Exception as e:
return False, str(e)
finally:
sp_session.close()
else:
# フォールバック: USE ROLEで権限切替（同一ユーザー内）
try:
user_session.sql(f"USE ROLE {SP_ROLE}").collect()
user_session.sql(f"USE WAREHOUSE {SP_WAREHOUSE}").collect()
results = []
for stmt in sql_statements:
results.append(user_session.sql(stmt).collect())
return True, results
except Exception as e:
return False, str(e)
finally:
user_session.sql(f"USE ROLE {current_role}").collect()
# --- DB/スキーマ選択（プルダウン） ---
db_rows = user_session.sql("SHOW DATABASES").collect()
db_names = [row.as_dict().get("name", "") for row in db_rows]
col_db, col_schema = st.columns(2)
with col_db:
target_db = st.selectbox("データベース", db_names, key="target_db")
schema_rows = user_session.sql(f'SHOW SCHEMAS IN DATABASE "{target_db}"').collect()
schema_names = [row.as_dict().get("name", "") for row in schema_rows]
with col_schema:
target_schema = st.selectbox("スキーマ", schema_names, key="target_schema")
fqn_schema = f"{target_db}.{target_schema}"
# --- プロシージャ一覧取得 ---
if st.button("プロシージャ一覧を取得", type="primary"):
success, result = execute_as_sp([f"SHOW PROCEDURES IN SCHEMA {fqn_schema}"])
if not success:
st.error(f"一覧取得失敗: {result}")
else:
procs = []
for row in result[0]:
d = row.as_dict()
procs.append({
"name": d.get("name", ""),
"arguments": d.get("arguments", ""),
"description": d.get("description", ""),
})
if procs:
st.session_state["procs"] = procs
st.session_state["fqn_schema"] = fqn_schema
else:
st.warning("プロシージャが見つかりません。")
if "procs" not in st.session_state:
st.stop()
# --- プロシージャ選択 ---
procs = st.session_state["procs"]
fqn_schema = st.session_state["fqn_schema"]
proc_display = [f"{p['name']} {p['arguments']}" for p in procs]
selected_idx = st.selectbox("編集するプロシージャ", range(len(proc_display)),
format_func=lambda i: proc_display[i])
selected_proc = procs[selected_idx]
arg_match = re.search(r'\(([^)]*)\)', selected_proc["arguments"])
arg_types_str = f"({arg_match.group(1)})" if arg_match else "()"
proc_fqn = f"{fqn_schema}.{selected_proc['name']}"
# --- ソース取得＆編集 ---
if st.button("ソースを読み込む", type="primary"):
ddl_sql = f"SELECT GET_DDL('PROCEDURE', '{proc_fqn}{arg_types_str}')"
success, result = execute_as_sp([ddl_sql])
if success and result[0]:
st.session_state["edit_source"] = result[0][0][0]
if "edit_source" in st.session_state:
st.divider()
edited = st.text_area("プロシージャのソース", value=st.session_state["edit_source"], height=450)
if st.button("保存（CREATE OR REPLACE）", type="primary"):
save_sql = re.sub(r'^CREATE\s+(OR\s+REPLACE\s+)?PROCEDURE',
'CREATE OR REPLACE PROCEDURE', edited.strip(), flags=re.IGNORECASE)
success, result = execute_as_sp([save_sql])
if success:
st.success("プロシージャを更新しました！")
del st.session_state["edit_source"]
st.rerun()
else:
st.error(f"更新失敗: {result}")
4-2. Streamlit アプリへのSecret紐付け
アプリをデプロイした後，External Access IntegrationとSecretを紐付ける．
-- Streamlitアプリに External Access Integration と Secret を紐付け
ALTER STREAMLIT <DB>.<SCHEMA>.<STREAMLIT_APP_NAME>
SET EXTERNAL_ACCESS_INTEGRATIONS = (sp_access_integration)
SECRETS = ('SVC_PRIVATE_KEY' = DEMO_DB.PUBLIC.ADMIN_PRIVATE_KEY_SECRET);
この設定により，アプリ内で os.environ['SVC_PRIVATE_KEY'] として秘密鍵を参照できるようになる．
Step 5: 動作確認
構築後の確認は，次の手順で行う．
- Streamlitアプリを開く
- 「接続モード: キーペア認証」と表示されることを確認する
- データベース・スキーマをプルダウンで選択する
- 「プロシージャ一覧を取得」をクリックする
- 編集したいプロシージャを選択し，「ソースを読み込む」を実行する
- テキストエリアで中身を編集する
- 「保存」により，別ユーザーとして
CREATE OR REPLACEが実行される
実際にどのユーザーとして実行されたかは，QUERY_HISTORY で確認できる．
-- QUERY_HISTORY でどのユーザーとして実行されたか確認
SELECT USER_NAME, QUERY_TEXT, START_TIME
FROM SNOWFLAKE.ACCOUNT_USAGE.QUERY_HISTORY
WHERE QUERY_TEXT ILIKE '%CREATE OR REPLACE PROCEDURE%'
ORDER BY START_TIME DESC
LIMIT 5;
-- → USER_NAME が "ADMIN_USER" として記録されている
USER_NAME が接続先ユーザー（ADMIN_USER）として記録されていれば，キーペア認証で別ユーザーとして動作していることを確認できる．
セキュリティ上の考慮事項
バリデーション
アプリ内で，プロシージャ本体に対して簡易バリデーションを実装できる．
# 禁止操作のチェック例
forbidden = ["DROP TABLE", "DROP DATABASE", "DROP SCHEMA", "TRUNCATE",
"CREATE USER", "ALTER USER", "GRANT ", "REVOKE "]
監査
監査の観点では，次の点が挙げられる．
- キーペア認証モードでは，
QUERY_HISTORYに接続先ユーザー名として記録される - アプリ側でログインユーザー名をCOMMENTに含めれば，誰が操作したかを追跡できる
秘密鍵の管理
秘密鍵の取り扱いについては，以下を徹底することが望ましい．
- 認証に用いる秘密情報は，パスワード・秘密鍵を問わず，コードに直書きせずSnowflake Secretに格納して管理する（これは認証方式によらず共通の前提である）
- External Access Integrationでアクセスを制御する
- 秘密鍵のローテーションは，
ALTER USER ... SET RSA_PUBLIC_KEY_2で2つ目の鍵を設定することにより，無停止で行える
MFA導入前後の比較まとめ
旧方式と新方式の違いを整理すると，次の表のとおりである．
| 項目 | 旧方式（MFA前） | 新方式（キーペア認証） |
|---|---|---|
| 認証方式 | ID/パスワード | RSAキーペア |
| MFA影響 | 動作不可 | 影響なし |
| 対象ユーザー | 既存ユーザーのPWを共有 | 既存ユーザーに公開鍵を追加 |
| 必要な変更 | なし |
ALTER USER SET RSA_PUBLIC_KEY のみ |
| 秘密情報の管理 | パスワードをSecretに格納（コード直書きは非推奨） | 秘密鍵をSnowflake Secretに格納 |
| 監査ログ | 接続先ユーザーとして記録 | 同左 + 操作者追跡可能 |
| セットアップ工数 | 低い | Secret + EAI設定が必要 |
まとめ
本記事では，MFA導入後のSnowflakeにおいて，既存の別ユーザーの権限でストアドプロシージャを管理する方法について述べた．要点は次の4つである．
- 既存ユーザーに
RSA_PUBLIC_KEYを設定する（ALTER USER1行のみ） - Snowflake Secretで秘密鍵を安全に管理する
- External Access IntegrationでStreamlitアプリからの接続を許可する
- Streamlit in SnowflakeでセルフサービスのUIを提供する
新しいユーザーを作る必要はなく，既存の管理者ユーザーにキーペア認証を追加するだけで実現できる．また，既存ユーザーのパスワード認証やMFA設定には一切影響しないため，安全に導入できると考えられる．これにより，一般ユーザーに過剰な権限を付与することなく，必要な操作だけをアプリ経由で許可する「最小権限の原則」に沿った運用が可能になる．
ソースコードは以下のリポジトリで公開している．