はじめに

Tink という Google が開発した暗号化ライブラリと Google Cloud の Cloud KMS を活用した暗号化の後編ブログとなります。
前編 では Google の暗号化ライブラリ Tink の概要と、Cloud KMS と連携した AEAD によるエンベロープ暗号化を試しました。DEK の生成から暗号化まで Tink が全部やってくれて、Cloud KMS と連携することで暗号化キーの管理が便利になることがわかりました。

前編の AEAD にはファイル全体をメモリに読み込むという制約がありました。数MB程度なら問題ないのですが、実務で扱うデータが数GB、数TBとなるとメモリが足りなくなります。さらに、セキュリティポリシーで閉域網からインターネットに出られない環境だと、そもそも Cloud KMS に接続できず暗号化自体ができません。

後編ではそのあたりを解決するのと、結果を総まとめしていきます:

  • Streaming AEAD — 大容量ファイルをチャンク単位で暗号化する方式
  • Streaming AEAD の鍵管理の課題 — 前編の AEAD との違いと注意点
  • オフライン暗号化 — KMS に接続できない閉域環境でも暗号化する方法
  • 3方式のまとめ比較 — どの場面でどれを使うべきか

1. Streaming AEAD で大容量ファイルを暗号化する

大容量ファイルの暗号化からやっていきます。Streaming AEAD は、ファイルをセグメント単位（今回は 1MB）で少しずつ暗号化していく方式です。ファイル全体をメモリに載せないので、メモリ使用量はファイルサイズに関係なく一定で済みます。

Streaming AEAD の処理イメージはこのような形です。


※前編同様、当該ブログで利用している画像データは生成 AI により生成した画像を利用しております。

ここでは 100MB のダミーファイルを用意して、Streaming AEAD で暗号化・復号し、サイズの変化や処理速度を確認してみます。

テスト用ファイルの作成

# 100MB のダミーファイルを作成
dd if=/dev/urandom of=plaintext_large.bin bs=1M count=100

暗号化・復号スクリプトの作成（streaming_demo.py）

100MB のダミーファイルに対して、Streaming AEAD でストリーミング暗号化・復号を行い、ファイルの一致確認とサイズ比較を出力するスクリプトを組みます。以下サンプルです。

import tink
from tink import streaming_aead
import os
import time

# --- 初期化 ---
streaming_aead.register()

keyset_handle = tink.new_keyset_handle(
    streaming_aead.streaming_aead_key_templates.AES256_GCM_HKDF_1MB
)
primitive = keyset_handle.primitive(streaming_aead.StreamingAead)

INPUT_FILE = "plaintext_large.bin"
ENCRYPTED_FILE = "encrypted_large.bin"
DECRYPTED_FILE = "decrypted_large.bin"
CHUNK_SIZE = 1024 * 1024  # 1MB

# --- 暗号化 ---
print("=" * 50)
print("Streaming AEAD 暗号化デモ")
print("=" * 50)

original_size = os.path.getsize(INPUT_FILE)
print(f"\n[1] 元ファイル: {INPUT_FILE}")
print(f"    サイズ: {original_size / 1024 / 1024:.1f} MB")

print(f"\n[2] ストリーミング暗号化中...")
start_time = time.time()

with open(INPUT_FILE, "rb") as input_file:
    with open(ENCRYPTED_FILE, "wb") as output_file:
        with primitive.new_encrypting_stream(output_file, b"metadata:largefile") as enc_stream:
            bytes_processed = 0
            while True:
                chunk = input_file.read(CHUNK_SIZE)
                if not chunk:
                    break
                enc_stream.write(chunk)
                bytes_processed += len(chunk)
                # 進捗表示（10MBごと）
                if bytes_processed % (10 * 1024 * 1024) == 0:
                    print(f"    処理済み: {bytes_processed / 1024 / 1024:.0f} MB")

elapsed = time.time() - start_time
encrypted_size = os.path.getsize(ENCRYPTED_FILE)
overhead = (encrypted_size - original_size) / original_size * 100

print(f"\n[3] 暗号化完了:")
print(f"    暗号化ファイル: {ENCRYPTED_FILE}")
print(f"    暗号化後サイズ: {encrypted_size / 1024 / 1024:.1f} MB")
print(f"    オーバーヘッド:  {overhead:.4f}%")
print(f"    処理時間:       {elapsed:.2f} 秒")
print(f"    スループット:    {original_size / 1024 / 1024 / elapsed:.1f} MB/s")

# --- 復号 ---
print(f"\n[4] ストリーミング復号中...")
start_time = time.time()

with open(ENCRYPTED_FILE, "rb") as input_file:
    with open(DECRYPTED_FILE, "wb") as output_file:
        with primitive.new_decrypting_stream(input_file, b"metadata:largefile") as dec_stream:
            while True:
                chunk = dec_stream.read(CHUNK_SIZE)
                if not chunk:
                    break
                output_file.write(chunk)

elapsed = time.time() - start_time
print(f"    復号完了: {elapsed:.2f} 秒")

# --- 一致確認 ---
print(f"\n[5] ファイル比較:")
import subprocess
result = subprocess.run(["diff", INPUT_FILE, DECRYPTED_FILE], capture_output=True)
if result.returncode == 0:
    print("    -> 元ファイルと復号ファイルは完全一致!")
else:
    print("    -> 不一致が検出されました。")

# --- サイズ比較表 ---
print(f"\n[6] サイズ比較:")
print(f"    {'ファイル':<30} {'サイズ':>15}")
print(f"    {'-'*45}")
print(f"    {'元ファイル (plaintext_large.bin)':<30} {original_size:>12,} bytes")
print(f"    {'暗号化 (encrypted_large.bin)':<30} {encrypted_size:>12,} bytes")
print(f"    {'差分':<30} {encrypted_size - original_size:>12,} bytes")
print(f"    {'オーバーヘッド':<30} {overhead:>11.4f}%")

実行結果

==================================================
Streaming AEAD 暗号化デモ
==================================================

[1] 元ファイル: plaintext_large.bin
    サイズ: 100.0 MB

[2] ストリーミング暗号化中...
    処理済み: 10 MB
    処理済み: 20 MB
    処理済み: 30 MB
    処理済み: 40 MB
    処理済み: 50 MB
    処理済み: 60 MB
    処理済み: 70 MB
    処理済み: 80 MB
    処理済み: 90 MB
    処理済み: 100 MB

[3] 暗号化完了:
    暗号化ファイル: encrypted_large.bin
    暗号化後サイズ: 100.0 MB
    オーバーヘッド:  0.0016%
    処理時間:       0.09 秒
    スループット:    1153.4 MB/s

[4] ストリーミング復号中...
    復号完了: 0.09 秒

[5] ファイル比較:
    -> 元ファイルと復号ファイルは完全一致!

[6] サイズ比較:
    ファイル                                       サイズ
    ---------------------------------------------
    元ファイル (plaintext_large.bin)     104,857,600 bytes
    暗号化 (encrypted_large.bin)       104,859,256 bytes
    差分                                    1,656 bytes
    オーバーヘッド                             0.0016%

100MB のファイルを暗号化しても、サイズ増加はわずか 1,656 bytes（0.0016%）でした。暗号文自体のサイズ増加はほとんどありません。

処理速度も 1153.4 MB/s と非常に高速なんじゃないかなと思います。

なお、Streaming AEAD は前編で紹介した Cloud KMS と組み合わせて使うこともできます。今回は Streaming AEAD の仕組み自体をシンプルに見せるために、あえて KMS 連携なしのローカル鍵で動かしています。

ただ、ローカル鍵で動かしているということは、鍵の管理をどうするかという問題が出てきます。次のセクションでは、この鍵管理の課題と、KMS に接続できない環境での解決策を見ていきます。

2. Streaming AEAD の鍵管理と KMS の活用

前編のブログで試した AEAD（エンベロープ暗号化）では、DEK の生成・暗号化・保存・復号を Tink と KMS が全て自動で処理してくれました。DEK は暗号文の中に暗号化された状態で埋め込まれるため、開発者が DEK の存在を意識する必要はありませんでした。

一方、Streaming AEAD には、この「KMS と連携した全自動エンベロープ暗号化」の仕組みが用意されていません。Streaming AEAD での鍵管理の流れを整理すると以下のようになります。

  1. tink.new_keyset_handle() でメモリ上に鍵（DEK）を生成
  2. その DEK で Streaming AEAD 暗号化を実行
  3. スクリプト終了 → 鍵はどこにも保存されず消える

表現するとこんな形でしょうか。

左側に記載がありますが、今回のデモのように同じスクリプト内で暗号化と復号を両方行う分には問題ありませんが、実運用ではキーが消失し暗号化したファイルを二度と復号できなくなります。
そのため、右側の記載の通り、鍵の永続化の方式は2通りあるかなと思います。

  • KMS に接続できる環境 → Tink の keyset を KMS の鍵で暗号化してファイルに保存し、使用時に KMS で復号して読み込む。しかし、前編ブログのように DEK が暗号文に自動で同梱されるわけではなく、暗号化された keyset ファイルを独立して管理する必要がある
  • KMS に接続できない環境（閉域網、インターネット接続が制限された社内ネットワーク等）→ KMS の非対称鍵（公開鍵/秘密鍵）を活用することで解決できる

次のセクションでは、後者のオフライン環境での暗号化を実際に試していきます。

3. 【オフライン版】KMS 非対称鍵で閉域環境でも暗号化する

セクション2で述べた「KMS に接続できない環境」でも暗号化したいケースに対応するのが、KMS の非対称鍵（公開鍵/秘密鍵）を使った方式です。

オフライン暗号化の仕組み

オフライン暗号化では、以下の流れで処理します。

【オフライン環境（暗号化側）】
1. DEK（データ暗号化鍵）をローカルで生成
2. KMS 非対称鍵の公開鍵で DEK を暗号化（wrap）
3. Streaming AEAD で DEK を使ってファイルを暗号化
4. 暗号化ファイル + 暗号化済みDEK を安全な経路で搬送

【オンライン環境（復号側）】
5. KMS の asymmetricDecrypt API で DEK を復号（unwrap）
6. 復号した DEK で Streaming AEAD を使ってファイルを復号

ポイントは、暗号化時は公開鍵だけあれば KMS 接続不要、復号時のみ KMS 接続が必要（秘密鍵は KMS の外に出ない）という点です。

これにより、前章で述べた「鍵の保存問題」も解決できます。DEK は KMS の非対称鍵で暗号化（wrap）された状態でファイルと一緒に保存するため、鍵管理を KMS に任せられます。

ではいきます。

非対称鍵の作成と公開鍵のエクスポート

# オフライン版で使用する非対称鍵の作成
gcloud kms keys create tink-blog-asymmetric-key \
  --location asia-northeast1 \
  --keyring tink-blog-keyring \
  --purpose asymmetric-encryption \
  --default-algorithm rsa-decrypt-oaep-2048-sha256

# 作成確認
gcloud kms keys describe tink-blog-asymmetric-key \
  --location asia-northeast1 \
  --keyring tink-blog-keyring

purpose: ASYMMETRIC_DECRYPT の出力があることを確認します。

# 公開鍵のエクスポート
gcloud kms keys versions get-public-key 1 \
  --location asia-northeast1 \
  --keyring tink-blog-keyring \
  --key tink-blog-asymmetric-key \
  --output-file public_key.pem

この public_key.pem を暗号化を行うオフライン環境側に事前に配置しておきます。

# 公開鍵の内容を確認
cat public_key.pem

# 公開鍵の詳細を確認
openssl rsa -pubin -in public_key.pem -text -noout

確認結果としては以下のようになります。

# 公開鍵内容
-----BEGIN PUBLIC KEY-----
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEArpHPwPgK4psypQeLNq3t
jxm0jMvDhHxe034UM5zFk6PiM7RBFWHoQvdEYCOL/1EPEosm22EbIS64XiOVCiCU
LpdMpXfsHZcmGkKDaZ70+xzidHQD7DhXW0VPVibckjYTFBUTT9np3nYMGOYHcrfP
nZ7i/cUozDBH/VNUtSvUOPAHmCYQT3ofrKGCjBbIX5IhQnubuQPMYPqqLwrxUbkt
xbF0qTEWFeL5lxpO5jm556/4C/LACA2B5l1bQJxjf0Pc5JAcKy9y29246ggvuF6t
AkwRGy2hquG66V74G6Ra6bqxDgWOCGiKYVFenZdH60cNX00lPb1B7csFf0S4JW+A
xQIDAQAB
-----END PUBLIC KEY-----

# 公開鍵詳細
Public-Key: (2048 bit)
Modulus:
    00:ae:91:cf:c0:f8:0a:e2:9b:32:a5:07:8b:36:ad:
    ed:8f:19:b4:8c:cb:c3:84:7c:5e:d3:7e:14:33:9c:
    c5:93:a3:e2:33:b4:41:15:61:e8:42:f7:44:60:23:
    8b:ff:51:0f:12:8b:26:db:61:1b:21:2e:b8:5e:23:
    95:0a:20:94:2e:97:4c:a5:77:ec:1d:97:26:1a:42:
    83:69:9e:f4:fb:1c:e2:74:74:03:ec:38:57:5b:45:
    4f:56:26:dc:92:36:13:14:15:13:4f:d9:e9:de:76:
    0c:18:e6:07:72:b7:cf:9d:9e:e2:fd:c5:28:cc:30:
    47:fd:53:54:b5:2b:d4:38:f0:07:98:26:10:4f:7a:
    1f:ac:a1:82:8c:16:c8:5f:92:21:42:7b:9b:b9:03:
    cc:60:fa:aa:2f:0a:f1:51:b9:2d:c5:b1:74:a9:31:
    16:15:e2:f9:97:1a:4e:e6:39:b9:e7:af:f8:0b:f2:
    c0:08:0d:81:e6:5d:5b:40:9c:63:7f:43:dc:e4:90:
    1c:2b:2f:72:db:dd:b8:ea:08:2f:b8:5e:ad:02:4c:
    11:1b:2d:a1:aa:e1:ba:e9:5e:f8:1b:a4:5a:e9:ba:
    b1:0e:05:8e:08:68:8a:61:51:5e:9d:97:47:eb:47:
    0d:5f:4d:25:3d:bd:41:ed:cb:05:7f:44:b8:25:6f:
    80:c5
Exponent: 65537 (0x10001)

テスト用ファイルの作成

# テスト用の機密データファイル
echo "ProjectID: PRJ-001, DataID: D-20260401, Status: Pending" > secret_data.txt

# 大きめのテストファイル（50MB）
dd if=/dev/urandom of=secret_data_large.bin bs=1M count=50

オフライン暗号化スクリプトの作成（offline_encrypt.py）

KMS に接続せず、事前に配布した公開鍵のみを使って、ファイルの暗号化と DEK の wrap を行うスクリプトを組みます。暗号化されたファイルと、復号に必要なメタデータ（wrap 済み DEK 等）を出力します。以下サンプルです。

"""
オフライン暗号化スクリプト
- KMS への接続は一切不要
- 公開鍵（public_key.pem）のみ使用
"""
import os
import json
import time
import tink
from tink import streaming_aead
from cryptography.hazmat.primitives import serialization, hashes
from cryptography.hazmat.primitives.asymmetric import padding

streaming_aead.register()

def offline_encrypt(input_file, public_key_file):
    """オフライン環境でファイルを暗号化する"""

    output_enc = input_file + ".enc"
    output_meta = input_file + ".meta.json"

    print("=" * 60)
    print("オフライン暗号化（KMS 接続なし）")
    print("=" * 60)

    # --- Step A: Tink keyset（DEK）をローカルで生成 ---
    keyset_handle = tink.new_keyset_handle(
        streaming_aead.streaming_aead_key_templates.AES256_GCM_HKDF_1MB
    )

    # keyset をバイト列にシリアライズ（後で公開鍵で wrap するため）
    keyset_bytes = tink.proto_keyset_format.serialize(
        keyset_handle, tink._secret_key_access.TOKEN
    )

    print(f"\n[A] Tink keyset（DEK）を生成しました")
    print(f"    keyset サイズ: {len(keyset_bytes)} bytes")

    # --- Step B: 公開鍵で keyset を暗号化（wrap）---
    with open(public_key_file, "rb") as f:
        public_key = serialization.load_pem_public_key(f.read())

    wrapped_keyset = public_key.encrypt(
        keyset_bytes,
        padding.OAEP(
            mgf=padding.MGF1(algorithm=hashes.SHA256()),
            algorithm=hashes.SHA256(),
            label=None
        )
    )

    print(f"\n[B] 公開鍵で keyset を暗号化（wrap）しました")
    print(f"    公開鍵ファイル: {public_key_file}")
    print(f"    暗号化済み keyset サイズ: {len(wrapped_keyset)} bytes")
    print(f"    ※ この暗号化には KMS 接続は不要です")

    # --- Step C: 同じ keyset で Streaming AEAD 暗号化 ---
    primitive = keyset_handle.primitive(streaming_aead.StreamingAead)

    original_size = os.path.getsize(input_file)
    print(f"\n[C] Streaming AEAD で暗号化中...")
    print(f"    入力ファイル: {input_file} ({original_size:,} bytes)")

    start_time = time.time()
    with open(input_file, "rb") as inf:
        with open(output_enc, "wb") as outf:
            with primitive.new_encrypting_stream(outf, b"offline-encryption") as enc_stream:
                bytes_processed = 0
                while True:
                    chunk = inf.read(1024 * 1024)  # 1MB チャンク
                    if not chunk:
                        break
                    enc_stream.write(chunk)
                    bytes_processed += len(chunk)
                    if bytes_processed % (10 * 1024 * 1024) == 0:
                        print(f"    処理済み: {bytes_processed / 1024 / 1024:.0f} MB")

    elapsed = time.time() - start_time
    encrypted_size = os.path.getsize(output_enc)

    print(f"    暗号化完了! ({elapsed:.2f} 秒)")

    # --- Step D: メタデータを保存 ---
    metadata = {
        "original_filename": os.path.basename(input_file),
        "original_size": original_size,
        "encrypted_size": encrypted_size,
        "wrapped_keyset_hex": wrapped_keyset.hex(),
        "kms_key_version": "projects/YOUR_PROJECT/locations/asia-northeast1/keyRings/tink-blog-keyring/cryptoKeys/tink-blog-asymmetric-key/cryptoKeyVersions/1",
        "algorithm": "AES256_GCM_HKDF_1MB",
        "wrap_algorithm": "RSA_DECRYPT_OAEP_2048_SHA256",
        "encrypted_at": time.strftime("%Y-%m-%dT%H:%M:%S%z")
    }

    with open(output_meta, "w") as f:
        json.dump(metadata, f, indent=2, ensure_ascii=False)

    # --- 結果表示 ---
    overhead = (encrypted_size - original_size) / original_size * 100

    print(f"\n{'=' * 60}")
    print(f"暗号化結果サマリ")
    print(f"{'=' * 60}")
    print(f"  元ファイル:       {input_file} ({original_size:,} bytes)")
    print(f"  暗号化ファイル:   {output_enc} ({encrypted_size:,} bytes)")
    print(f"  メタデータ:       {output_meta}")
    print(f"  オーバーヘッド:   {overhead:.4f}%")
    print(f"  処理時間:         {elapsed:.2f} 秒")
    if elapsed > 0:
        print(f"  スループット:     {original_size / 1024 / 1024 / elapsed:.1f} MB/s")
    print(f"")
    print(f"  搬送対象ファイル:")
    print(f"    1. {output_enc}      （暗号化されたデータ）")
    print(f"    2. {output_meta}  （復号に必要なメタデータ）")
    print(f"")
    print(f"  ※ 秘密鍵は KMS 内に封じ込められており、")
    print(f"    このスクリプトでは一切使用していません。")

# --- 実行 ---
if __name__ == "__main__":
    # 小さいファイル
    print("\n" + "#" * 60)
    print("# テスト1: 小さいファイル")
    print("#" * 60)
    offline_encrypt("secret_data.txt", "public_key.pem")

    # 大きいファイル
    print("\n" + "#" * 60)
    print("# テスト2: 大きいファイル (50MB)")
    print("#" * 60)
    offline_encrypt("secret_data_large.bin", "public_key.pem")

python offline_encrypt.py

実行結果

実行結果としては以下のようにでるはずです。
小さいファイル（56 bytes）と大きいファイル（50MB）の2パターンで実行して、KMS 接続なしで暗号化できること、オーバーヘッドがどの程度かが確認できるかと思います。

############################################################
# テスト1: 小さいファイル
############################################################
============================================================
オフライン暗号化（KMS 接続なし）
============================================================

[A] Tink keyset（DEK）を生成しました
    keyset サイズ: 129 bytes

[B] 公開鍵で keyset を暗号化（wrap）しました
    公開鍵ファイル: public_key.pem
    暗号化済み keyset サイズ: 256 bytes
    ※ この暗号化には KMS 接続は不要です

[C] Streaming AEAD で暗号化中...
    入力ファイル: secret_data.txt (56 bytes)
    暗号化完了! (0.00 秒)

============================================================
暗号化結果サマリ
============================================================
  元ファイル:       secret_data.txt (56 bytes)
  暗号化ファイル:   secret_data.txt.enc (112 bytes)
  メタデータ:       secret_data.txt.meta.json
  オーバーヘッド:   100.0000%
  処理時間:         0.00 秒
  スループット:     0.0 MB/s

  搬送対象ファイル:
    1. secret_data.txt.enc      （暗号化されたデータ）
    2. secret_data.txt.meta.json  （復号に必要なメタデータ）

  ※ 秘密鍵は KMS 内に封じ込められており、
    このスクリプトでは一切使用していません。

############################################################
# テスト2: 大きいファイル (50MB)
############################################################
============================================================
オフライン暗号化（KMS 接続なし）
============================================================

[A] Tink keyset（DEK）を生成しました
    keyset サイズ: 131 bytes

[B] 公開鍵で keyset を暗号化（wrap）しました
    公開鍵ファイル: public_key.pem
    暗号化済み keyset サイズ: 256 bytes
    ※ この暗号化には KMS 接続は不要です

[C] Streaming AEAD で暗号化中...
    入力ファイル: secret_data_large.bin (52,428,800 bytes)
    処理済み: 10 MB
    処理済み: 20 MB
    処理済み: 30 MB
    処理済み: 40 MB
    処理済み: 50 MB
    暗号化完了! (0.06 秒)

============================================================
暗号化結果サマリ
============================================================
  元ファイル:       secret_data_large.bin (52,428,800 bytes)
  暗号化ファイル:   secret_data_large.bin.enc (52,429,656 bytes)
  メタデータ:       secret_data_large.bin.meta.json
  オーバーヘッド:   0.0016%
  処理時間:         0.06 秒
  スループット:     799.9 MB/s

  搬送対象ファイル:
    1. secret_data_large.bin.enc      （暗号化されたデータ）
    2. secret_data_large.bin.meta.json  （復号に必要なメタデータ）

  ※ 秘密鍵は KMS 内に封じ込められており、
    このスクリプトでは一切使用していません。

ついでに、暗号化前後のファイル内容、サイズを見てみます。

# 暗号化前後の比較
=== 元ファイル（読める）===
$ cat secret_data.txt
ProjectID: PRJ-001, DataID: D-20260401, Status: Pending

=== 暗号化ファイル（読めない）===
$ hexdump -C -n 64 secret_data.txt.enc
00000000  28 ee fa 15 04 f2 a2 86  e8 a1 81 6d 2e 42 69 74  |(..........m.Bit|
00000010  cb e7 4b c0 05 96 a2 89  eb 08 aa 36 48 01 d1 0f  |..K........6H...|
00000020  64 92 c6 eb 08 c8 89 e4  e5 09 6a 7a 7b 2b 76 71  |d.........jz{+vq|
00000030  f0 e4 21 b3 08 fa 7c ac  27 94 b4 43 02 b7 f1 5f  |..!...|.'..C..._|
00000040


=== メタデータ（復号に必要な情報）===
$ cat secret_data.txt.meta.json | python3 -m json.tool
{
    "original_filename": "secret_data.txt",
    "original_size": 56,
    "encrypted_size": 112,
    "wrapped_keyset_hex": "06b4771e53714ddc511021b729013a213bc46bc1a37d47ab92fd4a318155bd57cb45453ef0cbc97d1e1048ec26df03f3c1bc31ab7cb07542854f785f869b8aaebc81eafea7e224e422683ccf7142d861dcc5b185ccafeac8d59e166956db25ddad755759ea3dc0d368f0f02c69a6b6b73112cc8b3363b3b1b67eacdc83b64b675c8a260144584a4f7964f9ef4a1e82edc4a34d7925f3103bd9f694975b03713e59cfb070a20c108e84a7b6fd5905c33e53d6d83273b38cc3aaa16036c33571d2488a7f8cba7fc5500f0bcc0a8ed8a0cef40829c8ca41105503700722983b2ffec02a24b438bd1408600f649636ef46adb4d5f0dee2c0951e97493c9e9453a4c3",
    "kms_key_version": "projects/YOUR_PROJECT/locations/asia-northeast1/keyRings/tink-blog-keyring/cryptoKeys/tink-blog-asymmetric-key/cryptoKeyVersions/1",
    "algorithm": "AES256_GCM_HKDF_1MB",
    "wrap_algorithm": "RSA_DECRYPT_OAEP_2048_SHA256",
    "encrypted_at": "2026-04-07T16:23:49+0900"
}

=== ファイルサイズ比較 ===
$ ls -lh secret_data.txt secret_data.txt.enc secret_data.txt.meta.json
-rw-r--r--@ 1 user  staff    56B Apr  7 09:52 secret_data.txt
-rw-r--r--@ 1 user  staff   112B Apr  7 16:23 secret_data.txt.enc
-rw-r--r--@ 1 user  staff   954B Apr  7 16:23 secret_data.txt.meta.json

$ ls -lh secret_data_large.bin secret_data_large.bin.enc secret_data_large.bin.meta.json
-rw-r--r--@ 1 user  staff    50M Apr  7 09:52 secret_data_large.bin
-rw-r--r--@ 1 user  staff    50M Apr  7 16:23 secret_data_large.bin.enc
-rw-r--r--@ 1 user  staff   971B Apr  7 16:23 secret_data_large.bin.meta.json

ファイル内容が閲覧できること、今回の暗号化に伴い、ディスクサイズにはわずかな増加があることがわかるかと思います。

次に復号化処理やっていきます。

オンライン復号スクリプト（online_decrypt.py）

オフライン暗号化で生成された暗号化ファイルとメタデータを受け取り、KMS の asymmetricDecrypt API で DEK を復号（unwrap）した上で、Streaming AEAD でファイルを復号するスクリプトを組みます。

以下サンプルです。

"""
オンライン復号スクリプト
- KMS への接続が必要（asymmetricDecrypt で keyset を復号）
- 秘密鍵は KMS 内から出ない
"""
import os
import json
import time
import tink
from tink import streaming_aead
from google.cloud import kms

streaming_aead.register()

def online_decrypt(encrypted_file, meta_file):
    """オンライン環境でファイルを復号する"""

    output_file = encrypted_file.replace(".enc", ".decrypted")
    if output_file == encrypted_file:
        output_file = encrypted_file + ".decrypted"

    print("=" * 60)
    print("オンライン復号（KMS 接続あり）")
    print("=" * 60)

    # --- Step A: メタデータの読み込み ---
    with open(meta_file, "r") as f:
        metadata = json.load(f)

    wrapped_keyset = bytes.fromhex(metadata["wrapped_keyset_hex"])
    key_version = metadata["kms_key_version"]

    print(f"\n[A] メタデータを読み込みました")
    print(f"    元ファイル名: {metadata['original_filename']}")
    print(f"    元サイズ: {metadata['original_size']:,} bytes")
    print(f"    暗号化日時: {metadata['encrypted_at']}")
    print(f"    KMS 鍵バージョン: {key_version}")

    # --- Step B: KMS で keyset を復号（unwrap）---
    print(f"\n[B] KMS asymmetricDecrypt で keyset を復号中...")
    print(f"    ※ ここで初めて KMS に接続します")

    kms_client = kms.KeyManagementServiceClient()

    response = kms_client.asymmetric_decrypt(
        request={
            "name": key_version,
            "ciphertext": wrapped_keyset,
        }
    )
    keyset_bytes = response.plaintext

    print(f"    keyset の復号に成功しました ({len(keyset_bytes)} bytes)")
    print(f"    ※ 秘密鍵は KMS の外に出ていません")

    # --- Step C: 復号した keyset で Streaming AEAD 復号 ---
    keyset_handle = tink.proto_keyset_format.parse(
        keyset_bytes, tink._secret_key_access.TOKEN
    )
    primitive = keyset_handle.primitive(streaming_aead.StreamingAead)

    encrypted_size = os.path.getsize(encrypted_file)
    print(f"\n[C] Streaming AEAD で復号中...")
    print(f"    入力ファイル: {encrypted_file} ({encrypted_size:,} bytes)")

    start_time = time.time()
    with open(encrypted_file, "rb") as inf:
        with open(output_file, "wb") as outf:
            with primitive.new_decrypting_stream(inf, b"offline-encryption") as dec_stream:
                while True:
                    chunk = dec_stream.read(1024 * 1024)
                    if not chunk:
                        break
                    outf.write(chunk)

    elapsed = time.time() - start_time
    decrypted_size = os.path.getsize(output_file)

    # --- 結果表示 ---
    print(f"\n{'=' * 60}")
    print(f"復号結果サマリ")
    print(f"{'=' * 60}")
    print(f"  暗号化ファイル:   {encrypted_file}")
    print(f"  復号ファイル:     {output_file}")
    print(f"  復号後サイズ:     {decrypted_size:,} bytes")
    print(f"  元サイズとの一致: {'一致' if decrypted_size == metadata['original_size'] else '不一致!'}")
    print(f"  処理時間:         {elapsed:.2f} 秒")

    return output_file

# --- 実行 ---
if __name__ == "__main__":
    # 小さいファイル
    print("\n" + "#" * 60)
    print("# テスト1: 小さいファイルの復号")
    print("#" * 60)
    output1 = online_decrypt("secret_data.txt.enc", "secret_data.txt.meta.json")

    print(f"\n[検証] 復号後の内容:")
    with open(output1, "r") as f:
        print(f"    {f.read()}")

    # 大きいファイル
    print("\n" + "#" * 60)
    print("# テスト2: 大きいファイルの復号")
    print("#" * 60)
    output2 = online_decrypt("secret_data_large.bin.enc", "secret_data_large.bin.meta.json")

    # diff で確認
    import subprocess
    print(f"\n[検証] 元ファイルとの比較:")
    result = subprocess.run(["diff", "secret_data_large.bin", output2], capture_output=True)
    if result.returncode == 0:
        print("    -> 完全一致! オフライン暗号化 → オンライン復号が正常に動作しています。")
    else:
        print("    -> 不一致が検出されました。")

実行結果

実行結果として以下のように出るはずです。
ファイルが復号できること、暗号化前と暗号化及び復号化後の内容が一致していることが、結果として現れています。
当たり前ですが、暗号化を経由しても同じデータとなることが確認できました。

############################################################
# テスト1: 小さいファイルの復号
############################################################
============================================================
オンライン復号（KMS 接続あり）
============================================================

[A] メタデータを読み込みました
    元ファイル名: secret_data.txt
    元サイズ: 56 bytes
    暗号化日時: 2026-04-07T16:23:49+0900
    KMS 鍵バージョン: projects/YOUR_PROJECT/locations/asia-northeast1/keyRings/tink-blog-keyring/cryptoKeys/tink-blog-asymmetric-key/cryptoKeyVersions/1

[B] KMS asymmetricDecrypt で keyset を復号中...
    ※ ここで初めて KMS に接続します
    keyset の復号に成功しました (129 bytes)
    ※ 秘密鍵は KMS の外に出ていません

[C] Streaming AEAD で復号中...
    入力ファイル: secret_data.txt.enc (112 bytes)

============================================================
復号結果サマリ
============================================================
  暗号化ファイル:   secret_data.txt.enc
  復号ファイル:     secret_data.txt.decrypted
  復号後サイズ:     56 bytes
  元サイズとの一致: 一致
  処理時間:         0.00 秒

[検証] 復号後の内容:
    ProjectID: PRJ-001, DataID: D-20260401, Status: Pending


############################################################
# テスト2: 大きいファイルの復号
############################################################
============================================================
オンライン復号（KMS 接続あり）
============================================================

[A] メタデータを読み込みました
    元ファイル名: secret_data_large.bin
    元サイズ: 52,428,800 bytes
    暗号化日時: 2026-04-07T16:23:49+0900
    KMS 鍵バージョン: projects/YOUR_PROJECT/locations/asia-northeast1/keyRings/tink-blog-keyring/cryptoKeys/tink-blog-asymmetric-key/cryptoKeyVersions/1

[B] KMS asymmetricDecrypt で keyset を復号中...
    ※ ここで初めて KMS に接続します
    keyset の復号に成功しました (131 bytes)
    ※ 秘密鍵は KMS の外に出ていません

[C] Streaming AEAD で復号中...
    入力ファイル: secret_data_large.bin.enc (52,429,656 bytes)

============================================================
復号結果サマリ
============================================================
  暗号化ファイル:   secret_data_large.bin.enc
  復号ファイル:     secret_data_large.bin.decrypted
  復号後サイズ:     52,428,800 bytes
  元サイズとの一致: 一致
  処理時間:         0.65 秒

[検証] 元ファイルとの比較:
    -> 完全一致! オフライン暗号化 → オンライン復号が正常に動作しています。

4. 3方式のまとめ比較

前編・後編で扱った3つの方式を振り返ります。

項目 AEAD エンベロープ暗号化（前編） Streaming AEAD + KMS keyset 保存 ※ オフライン版: 非対称鍵（本記事）
暗号化方式 AEAD（ファイル全体をメモリに読み込み） Streaming AEAD（チャンク単位） Streaming AEAD（チャンク単位）
大容量ファイル 不向き（メモリに全体を載せる） 向いている（メモリ使用量一定） 向いている（メモリ使用量一定）
KMS 接続 暗号化・復号ともに必要 暗号化・復号ともに必要 暗号化時は不要、復号時のみ必要
鍵の種類 対称鍵（KEK） 対称鍵（KEK） 非対称鍵（公開鍵/秘密鍵）
DEK の管理 Tink + KMS が全自動（暗号文に同梱） 自分で管理（暗号化した keyset ファイルを保存） 自分で管理（公開鍵で wrap した DEK を保存）
実装の複雑さ シンプル（Tink が自動処理） やや複雑（keyset の保存・読み込みが必要） やや複雑（DEK の生成・wrap を自前実装）
ユースケース クラウド接続環境で小〜中規模ファイル クラウド接続環境で大容量ファイル 閉域網・オフライン環境
セキュリティ 秘密鍵は KMS 内 秘密鍵は KMS 内 秘密鍵は KMS 内（同等）

※ 本記事の Streaming AEAD デモではローカル鍵で実施しています。KMS keyset 保存は Tink の機能として対応しており、KMS 接続可能な環境であれば利用できます。

前編の AEAD エンベロープ暗号化は DEK の管理を全て Tink と KMS に任せられる手軽さが魅力ですが、大容量ファイルには向きません。Streaming AEAD を使えば大容量ファイルに対応できますが、DEK（keyset）を自分で管理する必要が出てきます。さらに暗号化時に KMS に接続できない環境では、非対称鍵の公開鍵で DEK を wrap するオフライン版が有効です。いずれの方式でも、秘密鍵が KMS の外に出ないという点は共通しており、Google の Tink と Google Cloud の両方が上手に連携できていることによるメリットがでています。

まとめ

後編では、Streaming AEAD による大容量ファイルの暗号化と、KMS 非接続環境でのオフライン暗号化を試しました。

特に印象的だったのは以下の点です:

  • Streaming AEAD はサイズ増加がほぼゼロで処理速度も速いので、もっと大容量のファイルでも試してみたいなと思いました
  • オフライン暗号化で鍵管理の課題を解決 — KMS の非対称鍵の公開鍵で DEK を wrap すれば、クラウド非接続環境でも暗号化でき且つ、Google Cloud の KMS の力も発揮できる
  • 秘密鍵が KMS の外に出ない — Google Cloud の KMS の力がここでも発揮できる(2回目)

今回はブログ用の簡易検証なので省略しましたが、実務で使う場合は鍵ローテーション（非対称鍵は自動ローテーションできないので手動運用が必要）や IAM による復号権限の厳密な管理も重要になってきます。運用上の複雑さはまだありますが、その前段の暗号化における手法として参考になれば幸いです。

参考