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 = '"'; 

id,sensitive_column
1,user_secret_data_001
2,confidential_info_alpha
3,my_password_123

-- 暗号化時のパスフレーズを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')
$$;

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;

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;