はじめに

　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 USER 1行のみ）
Snowflake Secretで秘密鍵を安全に管理する
External Access IntegrationでStreamlitアプリからの接続を許可する
Streamlit in SnowflakeでセルフサービスのUIを提供する

　新しいユーザーを作る必要はなく，既存の管理者ユーザーにキーペア認証を追加するだけで実現できる．また，既存ユーザーのパスワード認証やMFA設定には一切影響しないため，安全に導入できると考えられる．これにより，一般ユーザーに過剰な権限を付与することなく，必要な操作だけをアプリ経由で許可する「最小権限の原則」に沿った運用が可能になる．

　ソースコードは以下のリポジトリで公開している．

GitHub: mshdtksk/snowflake-sis-procedure-editor

関連記事

Snowflake のタスクの実行プログラムを外部ファイル化する方法2（Streamlitアプリ）