developersIO
Snowflake から PGP 暗号化の GCS ファイル を 出力・取り込みしてみた。

Snowflake から PGP 暗号化の GCS ファイル を 出力・取り込みしてみた。

SnowflakeGoogle Cloud Storage
2026.05.23

こんにちは、みかみです。

最近、カブトエビを育ててみたくてしかたありません。「生きた化石」という響きにロマンを感じます。
調べてみたら、野生でも田んぼなどに生息していて、食べられないこともないらしいです。とはいえ泥臭くてあまり美味しくないらしいです。
卵から育てて餌にも気を配っていれば美味しく食べられるのかどうか、気になってます。

やりたいこと

  • Snowflake から、PGP 暗号化ファイルを GCS に出力したい。
  • GCS に配置された PGP 暗号化済みファイルの復号データを、Snowflake のテーブルに取り込みたい。
  • Snowflake で PGP 暗号化・復号する方法を確認したい。

前提

Google Cloud SDK（gcloud コマンド）の実行環境は準備済みであるものとします。 本エントリでは、Cloud Shell を使用しました。

また、GCS や IAM の操作に必要な API の有効化と権限は付与済みです。
Snowflake 環境側でも、必要なユーザーの作成や権限付与は実施済みです。

なお、文中、鍵情報など一部の文字は伏字に変更しています。

本ブログの検証に使用した Snowflake は 東京リージョンの AWS プラットフォームで動いています。GCS も同じく東京リージョンのバケットを使用しています。

準備：pgpy パッケージの利用可否確認

今回、PGP 暗号化・復号処理に Python コードの UDF 作成を想定しているので、UDF から利用可能な Snowflake 管理の Anaconda に pgpy パッケージが含まれているか確認します。
Snowflake で以下の SQL を実行しました。

SELECT package_name, version, language, runtime_version
FROM MIKAMI_DB.information_schema.packages
WHERE package_name = 'pgpy'
  AND language = 'python'
ORDER BY runtime_version DESC;

check_python_package

Python 3.8〜3.12 で、pgpy パッケージが利用できることが確認できました。

準備：PGP キーペアを準備

ローカル環境で以下の Python スクリプトを実行して、PGP の秘密鍵と公開鍵を作成しました。

import pgpy
from pgpy.constants import (
    PubKeyAlgorithm, KeyFlags, HashAlgorithm,
    SymmetricKeyAlgorithm, CompressionAlgorithm
)

def generate_keypair(name: str, email: str, key_path_prefix: str = "PGP_key"):
    """RSA-4096 の PGP キーペアを生成してファイルに保存する"""

    # キーペアの生成
    key = pgpy.PGPKey.new(PubKeyAlgorithm.RSAEncryptOrSign, 4096)

    # UID（名前・メールアドレス）の追加
    uid = pgpy.PGPUID.new(name, email=email)
    key.add_uid(
        uid,
        usage={KeyFlags.Sign, KeyFlags.EncryptCommunications, KeyFlags.EncryptStorage},
        hashes=[HashAlgorithm.SHA256],
        ciphers=[SymmetricKeyAlgorithm.AES256],
        compression=[CompressionAlgorithm.ZLIB]
    )

    # 公開鍵・秘密鍵をファイルに書き出す
    pub_key_path  = f"{key_path_prefix}_public.asc"
    priv_key_path = f"{key_path_prefix}_private.asc"

    with open(pub_key_path, "w") as f:
        f.write(str(key.pubkey))

    with open(priv_key_path, "w") as f:
        f.write(str(key))

    print(f"公開鍵  : {pub_key_path}")
    print(f"秘密鍵  : {priv_key_path}")
    print("鍵ペアの生成が完了しました。")

if __name__ == "__main__":
    generate_keypair(
        name="Test User",
        email="test@example.com",
        key_path_prefix="PGP_key"
    )

作成したキーペアを、Snowflake のシークレットに登録します。

Snowflake で以下の SQL を実行して、PGP 秘密鍵・公開鍵をシークレットとして登録しました。

CREATE OR REPLACE SECRET MIKAMI_DB.PUBLIC.PGP_PUBLIC_KEY
  TYPE = GENERIC_STRING
  SECRET_STRING = '-----BEGIN PGP PUBLIC KEY BLOCK-----
（PGP_key_public.asc の内容をペースト）
-----END PGP PUBLIC KEY BLOCK-----';

CREATE OR REPLACE SECRET MIKAMI_DB.PUBLIC.PGP_PRIVATE_KEY
  TYPE = GENERIC_STRING
  SECRET_STRING = '-----BEGIN PGP PRIVATE KEY BLOCK-----
（PGP_key_private.asc の内容をペースト）
-----END PGP PRIVATE KEY BLOCK-----';

それぞれのキーファイルの内容は、以下のコマンドでクリップボードにコピーできます。

cat PGP_key_public.asc | pbcopy
cat PGP_key_private.asc | pbcopy

SHOW SECRETS IN SCHEMA MIKAMI_DB.PUBLIC;

で、シークレットが正常に作成されたことが確認できました。

show_secret

準備：暗号化・復号 UDF 作成

PGP 暗号化・復号 UDF を作成します。

まずは、UDF からシークレットにアクセスするためのネットワークルールと外部アクセス統合を作成しておきます。
今回、UDF 内で get_generic_secret_string() を使ってシークレット情報を取得しますが、この get_generic_secret_string() を使う場合、UDF 作成時に外部アクセス統合（EXTERNAL_ACCESS_INTEGRATIONS）を指定する必要があります。

https://docs.snowflake.com/ja/developer-guide/external-network-access/secret-api-reference#python-api-for-secret-access

また、外部アクセス統合の作成時にネットワークルール（ALLOWED_NETWORK_RULES）の指定が必要なため、ネットワークルールも作成する必要があります。
実際に外部ネットワークへアクセスするわけではないので、ネットワークルールの VALUE_LIST はダミー値（example.com）で作成します。

Snowflake で以下の SQL を実行しました。

CREATE OR REPLACE NETWORK RULE MIKAMI_DB.PUBLIC.PGP_DUMMY_NETWORK_RULE
  TYPE = HOST_PORT
  VALUE_LIST = ('example.com')
  MODE = EGRESS;

CREATE OR REPLACE EXTERNAL ACCESS INTEGRATION PGP_SECRETS_INTEGRATION
  ALLOWED_NETWORK_RULES = (MIKAMI_DB.PUBLIC.PGP_DUMMY_NETWORK_RULE)
  ALLOWED_AUTHENTICATION_SECRETS = (
    MIKAMI_DB.PUBLIC.PGP_PUBLIC_KEY,
    MIKAMI_DB.PUBLIC.PGP_PRIVATE_KEY
  )
  ENABLED = TRUE;

PGP 暗号化 UDF を作成します。

CREATE OR REPLACE FUNCTION MIKAMI_DB.PUBLIC.PGP_ENCRYPT(plaintext STRING)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('pgpy')
EXTERNAL_ACCESS_INTEGRATIONS = (PGP_SECRETS_INTEGRATION)
SECRETS = ('public_key' = MIKAMI_DB.PUBLIC.PGP_PUBLIC_KEY)
HANDLER = 'encrypt'
AS $$
import pgpy
import _snowflake

def encrypt(plaintext: str) -> str:
    # シークレットから公開鍵を取得
    public_key_str = _snowflake.get_generic_secret_string('public_key')
    pub_key, _ = pgpy.PGPKey.from_blob(public_key_str)

    # メッセージの作成と暗号化
    message   = pgpy.PGPMessage.new(plaintext.encode('utf-8'))
    encrypted = pub_key.encrypt(message)

    # ASCII Armor 形式の文字列として返す
    return str(encrypted)
$$;

正常に動くか、以下の SQL で確認してみます。

SELECT MIKAMI_DB.PUBLIC.PGP_ENCRYPT('Hello, Snowflake!') AS encrypted_text;

udf_pgp_encrypt

パラメータで渡した文字列が暗号化されていることが確認できました。

続いて、PGP 復号 UDF を作成します。

CREATE OR REPLACE FUNCTION MIKAMI_DB.PUBLIC.PGP_DECRYPT(encrypted_text STRING)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = '3.9'
PACKAGES = ('pgpy')
EXTERNAL_ACCESS_INTEGRATIONS = (PGP_SECRETS_INTEGRATION)
SECRETS = ('private_key' = MIKAMI_DB.PUBLIC.PGP_PRIVATE_KEY)
HANDLER = 'decrypt'
AS $$
import pgpy
import _snowflake

def decrypt(encrypted_text: str) -> str:
    # シークレットから秘密鍵を取得
    private_key_str = _snowflake.get_generic_secret_string('private_key')
    priv_key, _ = pgpy.PGPKey.from_blob(private_key_str)

    # 暗号化メッセージの読み込みと復号
    encrypted_message = pgpy.PGPMessage.from_blob(encrypted_text)
    decrypted         = priv_key.decrypt(encrypted_message)
    plaintext         = decrypted.message

    # bytes / bytearray / str の3パターンに対応
    if isinstance(plaintext, (bytes, bytearray)):
        return plaintext.decode('utf-8')
    return plaintext
$$;

以下の SQL で動作確認します。

SELECT MIKAMI_DB.PUBLIC.PGP_DECRYPT(
  MIKAMI_DB.PUBLIC.PGP_ENCRYPT('Hello, Snowflake!')
) AS decrypted_text;

こちらも、期待通り復号処理が正常に実行されたことが確認できました。

udf_pgp_decrypt

PGP 暗号化ファイルを GCS に出力

Snowflake の以下の DOGS テーブルのデータを、PGP 暗号化して GCS に CSV ファイルとして出力します。

sf_data_dogs

GCS アクセスのためのストレージ統合と外部ステージは作成済みのものを使用します。
新規作成が必要な場合は、以下をご参照ください。

https://dev.classmethod.jp/articles/snowflake-gcs-as-external-stage/

以下の　SQL で、出力先にはまだファイルがないことを確認しておきます。

LIST @MIKAMI_DB.PUBLIC.MIKAMI_GCS_TEST_STAGE/pgp/;

list_gcs_before

以下の SQL を実行して、PGP 暗号化したファイルを GCS に出力します。

COPY INTO @MIKAMI_DB.PUBLIC.MIKAMI_GCS_TEST_STAGE/pgp/dogs_row_encrypted.csv.pgp
FROM (
  SELECT
    MIKAMI_DB.PUBLIC.PGP_ENCRYPT(
      OBJECT_CONSTRUCT('id', id, 'name', name)::STRING
    ) AS row_encrypted
  FROM MIKAMI_DB.PUBLIC.DOGS
)
FILE_FORMAT = (TYPE = 'CSV' COMPRESSION = NONE)
SINGLE    = TRUE
OVERWRITE = TRUE;

再度、出力先を確認してみると

LIST @MIKAMI_DB.PUBLIC.MIKAMI_GCS_TEST_STAGE/pgp/;

list_gcs_after

ファイルが出力されたことが確認できました。

実際のファイルの中身も確認してみます。

$ gcloud storage cat gs://test-mikami/pgp/dogs_row_encrypted.csv.pgp | head -10
-----BEGIN PGP MESSAGE-----\
\
wcFMA2USQwCKCKhMAQ//TydM2Gka5DrnU/7JpZBARae7PJYi0rRjgiXWIHF6bSgE\
nFX4ulIXe9BcwSFOVfS7u3GOuBUADyaOCV3f0pnZTRf1nRCy5EaXV/w9l5xeQS7l\
zcCHTmJfcWpVIlCPVZbkVh+NItjwAfKGl7ouHOIr1+2nB488/sk7110AUNE+8LbV\
USl3JdIMk0SIwHdFoDs2FtNbyNk411bjAhHL/oysyf2vHOgwzNYXaOs6eWNQfz/+\
iwHgXDsPeO1d+NT0fcGEPAm+Fjk3slLlDMMPHqNwX9Fe7wqC1oGiPmqix/sEnlOz\
irC34bQ92yPZKCMQMkigf4RRGegtncZFY3CoA19DJ+p5MZENQv5vof520Hhk7L+M\
AIwF/uZDlIPYkMrIXHXEJFN3xilEPYcU1tf1bA9PDbmUBzs7RR9ENEqV1GkPRg1e\
vY5ZiNtvzNxKtl+buurQgi4dJx1m6uPvjg+fO6Pd/YkGSFkCT9VNXi4SDKHwt88+\

想定通り、暗号化されていることが確認できました。

PGP 暗号化されたファイルを Snowflake に取り込み

先ほど出力した PGP 暗号化済みの CSV ファイルを、新しい DOGS_DECRYPT テーブルに取り込みます。

-- ファイルフォーマット作成
CREATE OR REPLACE FILE FORMAT MIKAMI_DB.PUBLIC.PGP_CSV_FORMAT
  TYPE = 'CSV'
  FIELD_OPTIONALLY_ENCLOSED_BY = '"';

-- テーブル作成・取り込み
CREATE OR REPLACE TABLE MIKAMI_DB.PUBLIC.DOGS_DECRYPT AS
SELECT
  parsed_row:id::INT      AS id,
  parsed_row:name::STRING AS name
FROM (
  SELECT
    PARSE_JSON(
      MIKAMI_DB.PUBLIC.PGP_DECRYPT($1)
    ) AS parsed_row
  FROM @MIKAMI_DB.PUBLIC.MIKAMI_GCS_TEST_STAGE/pgp/dogs_row_encrypted.csv.pgp
  (FILE_FORMAT => MIKAMI_DB.PUBLIC.PGP_CSV_FORMAT)
);

テーブルデータを確認してみます。

load_decript_data

想定通り、復号されたデータが DOGS_DECRYPT テーブルに格納できました。

念のためバイト単位でデータ出力元テーブルと比較して、目視では見落としがちな改行やスペースなどのデータ差異も発生していないか確認しておきます。

SELECT 'DOGS'         AS table_name, HASH_AGG(id, name) AS hash FROM MIKAMI_DB.PUBLIC.DOGS
UNION ALL
SELECT 'DOGS_DECRYPT' AS table_name, HASH_AGG(id, name) AS hash FROM MIKAMI_DB.PUBLIC.DOGS_DECRYPT;

diff_table_data

HASH 値が一致しているので、暗号化前・復号後のデータに差異がないことが確認できました。

まとめ（所感）

Snowflake の UDF を使って、PGP 暗号ファイルを出力・取り込みできることが確認できました。

今回は PGP 暗号化・復号を検証しましたが、Snowflake 管理の Anaconda に含まれる Python パッケージであれば UDF から利用できるので、cryptography パッケージを使って他の AES・RSA アルゴリズムによるデータの暗号化・復号も可能です。

UDF を活用することで、Python の豊富なライブラリをデータ処理に組み込むことができ、Snowflake 上でのデータ処理の可能性がさらに広がるのではないかと思います。

参考

Snowflakeの導入支援はクラスメソッドに！

クラスメソッドでは Snowflake の導入を支援しております。
製品の詳細や支援の内容についてお気軽にお問い合わせください。

Snowflakeの詳細を見る

この記事をシェアする

関連記事

GCS ファイルを Snowflake on AWS に Snowpipe で取り込んでみた。
みかみ
2026.05.23
Snowflakeでデータロード時にハッシュ化・暗号化をしてみた
さがら
2025.05.09
Snowflakeとdbtの接続設定でキーペア認証とProgrammatic access tokensを試してみた
かわばた
2025.08.05
Snowflake の type=service なユーザーの key-pair で dbt Cloud からの接続設定をしてみた
とーかみ
2025.03.11