Snowflakeでデータロード時にハッシュ化・暗号化をしてみた
さがらです。
Snowflakeでデータロード時にハッシュ化・暗号化をしてみたので、その内容をまとめてみます。
やってみた
テーブル・内部ステージ・ファイルフォーマットの作成
以下のクエリを実行して、検証用のテーブル・内部ステージ・ファイルフォーマットを作成します。
-- スキーマの作成
use database sagara_test_db;
create schema dataload_with_hash_encrypt;
use schema dataload_with_hash_encrypt;
-- ロード先テーブルの作成
create or replace table sensitive_data_transformed (
id int,
original_value varchar,
hashed_value_sha2 varchar, -- ハッシュ化された値を格納する列
encrypted_value binary, -- 暗号化された値を格納する列 (encrypt関数の結果はbinary型)
load_timestamp timestamp_ltz default current_timestamp()
);
-- 内部ステージの作成 (ファイルアップロード用)
create or replace stage my_internal_stage;
-- ファイルフォーマットの作成 (csv形式を想定)
create or replace file format my_csv_format
type = csv
field_delimiter = ','
skip_header = 1
empty_field_as_null = true
field_optionally_enclosed_by = '"';
内部ステージにファイルをアップロード
以下のCSVファイルを、作成した内部ステージにアップロードします。
id,sensitive_column
1,user_secret_data_001
2,confidential_info_alpha
3,my_password_123
暗号化用のパスフレーズを格納するSecret・Secretを参照するUDF及び必要なオブジェクト
以下のクエリを実行して、暗号化用のパスフレーズを格納するSecret・Secretを参照するUDF及び必要なオブジェクトを作成します。
network ruleとexternal access integrationを作成して、これらをUDFの定義時に参照するのがポイントです。
-- 暗号化時のパスフレーズをsecretで保存
create or replace secret my_encryption_passphrase_secret
type = generic_string
secret_string = 'your-passphrase';
-- network rule の作成
create or replace network rule dummy_secret_access_rule
mode = egress
type = host_port
value_list = ('example.com');
-- external access integration の作成 (account-level object)
use role accountadmin;
create or replace external access integration sagara_encrypt_udf_secret_access_integration
allowed_network_rules = (sagara_test_db.dataload_with_hash_encrypt.dummy_secret_access_rule)
allowed_authentication_secrets = (sagara_test_db.dataload_with_hash_encrypt.my_encryption_passphrase_secret)
enabled = true;
-- python udf の作成
use role accountadmin;
create or replace function get_secret_passphrase_py(secret_name string)
returns string
language python
runtime_version = '3.11'
handler = 'get_passphrase'
-- 上で作成した integration を参照
external_access_integrations = (sagara_encrypt_udf_secret_access_integration)
-- secret はカレントスキーマにあるものを参照し、エイリアス 'secret_alias' を使用
secrets = ('secret_alias' = sagara_test_db.dataload_with_hash_encrypt.my_encryption_passphrase_secret)
as
$$
import _snowflake
def get_passphrase(secret_identifier: str) -> str:
# udfのsecrets句で指定したエイリアス名 ('secret_alias') を使用してsecretを取得
return _snowflake.get_generic_secret_string('secret_alias')
$$;
このクエリの作成にあたっては、以下の記事を非常に参考にさせて頂きました。
ハッシュ化・暗号化を行いながらデータロード
以下のクエリを実行して、ハッシュ化・暗号化を行いながらデータロードします。
- ハッシュ化:SHA2関数を利用してハッシュ化
- 暗号化:ENCRYPT関数を利用して暗号化。第2引数にはパスフレーズを入れるが、先ほど作成したUDFを用いてSecretの値を取得してパスフレーズとしている
copy into sensitive_data_transformed (id, original_value, hashed_value_sha2, encrypted_value)
from (
select
$1::int,
$2::varchar,
sha2($2::varchar, 256),
encrypt($2::varchar, get_secret_passphrase_py('my_encryption_passphrase_secret'))
from @my_internal_stage/sample_sensitive_data.csv
(file_format => 'my_csv_format')
);
実際にロードしたデータをクエリしてみると、ハッシュ化・暗号化されていることがわかります。
select * from sensitive_data_transformed;
復号化を行うクエリ
encrypted_value
列については暗号化をしていたため、復号化が可能です。
DECRYPT関数で復号化が可能ですが、この関数の戻り値はBINARY型のため、明示的にVARCHAR型にする必要があります。
select
id,
original_value,
-- decryptの結果をto_varcharで文字列に変換
to_varchar(decrypt(encrypted_value, get_secret_passphrase_py('any_string_as_placeholder')), 'utf-8') as decrypted_value
from sensitive_data_transformed;
最後に
Snowflakeでデータロード時にハッシュ化・暗号化をしてみました。
ENCRYPT関数・DECRYPT関数で暗号化復号化はできますが、その時にSecretを用いてパスフレーズを管理しようとすると、今回のようにUDFを介さないと行えないのが少し大変でしたね…(@roki18dさんのブログに本当に助けられました…)
データロード時からハッシュ化・暗号化したい場合にこの記事が参考になると嬉しいです。