Object Storage Service:クライアントサイド暗号化 (Python SDK V2)

次のサンプルコードは、単純なオブジェクトのアップロードとダウンロードに RSA マスターキーを使用する方法を示しています：

import argparse
import alibabacloud_oss_v2 as oss
import alibabacloud_oss_v2.crypto
from alibabacloud_oss_v2.encryption_client import EncryptionClient, EncryptionMultiPartContext

# ユーザー入力を受け取るためのコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="encryption put object sample")

# バケットが配置されているリージョンを指定する --region コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# バケットの名前を指定する --bucket コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 他のサービスが OSS にアクセスするために使用するドメイン名を指定する --endpoint コマンドライン引数を追加します。この引数はオプションです。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# オブジェクトの名前を指定する --key コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--key', help='The name of the object.', required=True)

# 暗号化と復号のための RSA 公開鍵と秘密鍵を定義します。
RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0G6mse2QsIgz3******GBcom6kEF6MmR1EKixaQIDAQAB
-----END PUBLIC KEY-----"""

RSA_PRIVATE_KEY = """-----BEGIN PRIVATE KEY-----
MIICdQIBADANBgk******ItewfwXIL1Mqz53lO/gK+q6TR92gGc+4ajL
-----END PRIVATE KEY-----"""

def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 環境変数から認証情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # SDK のデフォルト構成を読み込みます。
    cfg = oss.config.load_default()

    # 認証情報プロバイダーを設定します。
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが提供されている場合は、構成で設定します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 構成オブジェクトを使用して OSS クライアントを初期化します。
    client = oss.Client(cfg)

    # 暗号化と復号のために MasterRsaCipher オブジェクトを初期化します。
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},
        public_key=RSA_PUBLIC_KEY,  # 暗号化用の RSA 公開鍵。
        private_key=RSA_PRIVATE_KEY  # 復号用の RSA 秘密鍵。
    )

    # 暗号化クライアントを初期化します。
    encryption_client = oss.EncryptionClient(client, mc)

    # アップロードするデータを定義します。
    data = b'hello world'

    # 暗号化クライアントの put_object メソッドを呼び出して、暗号化されたオブジェクトをアップロードします。
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 宛先バケットの名前。
            key=args.key,        # オブジェクトの名前。
            body=data,           # アップロードするデータ。
        )
    )

    # 操作に関するステータスコードとその他の情報を出力します。
    print(f'status code: {result.status_code}, '  # HTTP ステータスコード。リクエストが成功したかどうかを示します。
          f'request id: {result.request_id}, '    # リクエスト ID。ログの追跡とデバッグに使用されます。
          f'content md5: {result.content_md5}, '  # 返されたオブジェクトコンテンツの MD5 チェックサム。
          f'etag: {result.etag}, '               # 返されたオブジェクトの ETag。
          f'hash crc64: {result.hash_crc64}, '   # 返されたオブジェクトの CRC-64 チェックサム。
          f'version id: {result.version_id}')    # バージョン管理が有効な場合のオブジェクトのバージョン ID。


    # 暗号化クライアントの get_object メソッドを呼び出して、暗号化されたオブジェクトのコンテンツを取得します。
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 宛先バケットの名前。
            key=args.key,        # 宛先オブジェクトの名前 (ファイルパス)。
        )
    )

    # 操作に関する情報を出力します。
    print(f'status code: {result.status_code}, '  # HTTP ステータスコード。リクエストが成功したかどうかを示します。
          f'request id: {result.request_id}, '   # リクエスト ID。ログの追跡とデバッグに使用されます。
          f'content md5: {result.content_md5}, '  # オブジェクトコンテンツの MD5 チェックサム。
          f'etag: {result.etag}, '               # オブジェクトの ETag。
          f'hash crc64: {result.hash_crc64}, '   # オブジェクトの CRC-64 チェックサム。
          f'version id: {result.version_id}')    # バージョン管理が有効な場合のオブジェクトのバージョン ID。


if __name__ == "__main__":
    # プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
    main()

次のサンプルコードは、マルチパートオブジェクトのアップロードに RSA マスターキーを使用する方法を示しています：

import argparse
import alibabacloud_oss_v2 as oss
import os
import alibabacloud_oss_v2.crypto
from alibabacloud_oss_v2.encryption_client import EncryptionClient, EncryptionMultiPartContext

# ユーザー入力を受け取るためのコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="encryption put object sample")

# バケットが配置されているリージョンを指定する --region コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# バケットの名前を指定する --bucket コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 他のサービスが OSS にアクセスするために使用するドメイン名を指定する --endpoint コマンドライン引数を追加します。この引数はオプションです。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# オブジェクトの名前 (ファイルパス) を指定する --key コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--key', help='The name of the object.', required=True)

# 暗号化と復号のための RSA 公開鍵と秘密鍵を定義します。
RSA_PUBLIC_KEY = """-----BEGIN PUBLIC KEY-----
MIGfMA0G6mse2QsIgz3******GBcom6kEF6MmR1EKixaQIDAQAB
-----END PUBLIC KEY-----"""

RSA_PRIVATE_KEY = """-----BEGIN PRIVATE KEY-----
MIICdQIBADANBgk******ItewfwXIL1Mqz53lO/gK+q6TR92gGc+4ajL
-----END PRIVATE KEY-----"""

def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 環境変数から認証情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # SDK のデフォルト構成を読み込みます。
    cfg = oss.config.load_default()

    # 認証情報プロバイダーを設定します。
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが提供されている場合は、構成で設定します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 構成オブジェクトを使用して OSS クライアントを初期化します。
    client = oss.Client(cfg)

    # 暗号化と復号のためにマスターキーサイファ (MasterRsaCipher) を初期化します。
    mc = oss.crypto.MasterRsaCipher(
        mat_desc={"tag": "value"},# マスターキーの説明を作成します。説明は作成後に変更できません。マスターキーの説明はマスターキーと 1 対 1 で対応します。
        public_key=RSA_PUBLIC_KEY,  # 暗号化用の RSA 公開鍵。
        private_key=RSA_PRIVATE_KEY  # 復号用の RSA 秘密鍵。
    )

    # 暗号化クライアントを作成します。
    encryption_client = oss.EncryptionClient(client, mc)

    # マルチパートアップロードの各フラグメントのサイズをバイト単位で定義します。ここでは 100 KB に設定されています。
    part_size = 100 * 1024

    # ローカルファイルのサイズをバイト単位で取得します。
    data_size = os.path.getsize("/local/dir/example")  # ローカルファイルの実際のパスに置き換えてください。

    # マルチパートアップロードタスクを初期化し、アップロードタスクの初期情報を返します。
    result = encryption_client.initiate_multipart_upload(
        oss.InitiateMultipartUploadRequest(
            bucket="example_bucket",  # 宛先バケットの名前。
            key="example_key",        # 宛先オブジェクトの名前 (ファイルパス)。
            cse_part_size=part_size,  # 各フラグメントのサイズ。
            cse_data_size=data_size   # ファイルの合計サイズ。
        )
    )

    # マルチパートアップロードタスクの初期化結果を出力します。
    print(vars(result))

    # フラグメント番号とフラグメントのリストを初期化します。
    part_number = 1
    upload_parts = []

    # ローカルファイルを開き、フラグメントサイズに基づいてコンテンツをフラグメントごとに読み取ります。
    with open("/local/dir/example", 'rb') as f:  # ローカルファイルの実際のパスに置き換えてください。
        for start in range(0, data_size, part_size):  # フラグメントサイズごとにファイルコンテンツを反復処理します。
            n = part_size  # 現在のフラグメントのサイズ。
            if start + n > data_size:  # 最後のフラグメントがフラグメントサイズより小さい場合は、サイズを調整します。
                n = data_size - start

            # SectionReader を使用して、ファイルの現在のフラグメントのコンテンツを読み取ります。
            reader = oss.io_utils.SectionReader(
                oss.io_utils.ReadAtReader(f),  # ファイルをランダム読み取りをサポートするオブジェクトとしてラップします。
                start,  # 現在のフラグメントの開始位置。
                n       # 現在のフラグメントのサイズ。
            )

            # 現在のフラグメントをアップロードします。
            up_result = encryption_client.upload_part(
                oss.UploadPartRequest(
                    bucket="example_bucket",  # 宛先バケットの名前。
                    key="example_key",        # 宛先オブジェクトの名前 (ファイルパス)。
                    upload_id=result.upload_id,  # マルチパートアップロードタスクの一意の識別子。
                    part_number=part_number,  # 現在のフラグメントの番号。
                    cse_multipart_context=result.cse_multipart_context,  # 暗号化コンテキスト情報。
                    body=reader  # 現在のフラグメントのデータコンテンツ。
                )
            )

            # フラグメントのアップロード結果を出力します。
            print(vars(result))

            # 現在のフラグメントの番号と ETag をフラグメントのリストに追加します。
            upload_parts.append(
                oss.UploadPart(
                    part_number=part_number,  # 現在のフラグメントの番号。
                    etag=up_result.etag       # 現在のフラグメントがアップロードされた後の ETag 値。
                )
            )

            # フラグメント番号を更新します。
            part_number += 1

    # フラグメント番号でフラグメントのリストをソートします。
    parts = sorted(upload_parts, key=lambda p: p.part_number)

    # マルチパートアップロードタスクを完了します。これにより、すべてのフラグメントがマージされて最終的なオブジェクトが生成されます。
    result = encryption_client.complete_multipart_upload(
        oss.CompleteMultipartUploadRequest(
            bucket="example_bucket",  # 宛先バケットの名前。
            key="example_key",        # 宛先オブジェクトの名前 (ファイルパス)。
            upload_id=result.upload_id,  # マルチパートアップロードタスクの一意の識別子。
            complete_multipart_upload=oss.CompleteMultipartUpload(
                parts=parts  # ソートされたフラグメントのリスト。
            )
        )
    )

    # マルチパートアップロードタスクの完了結果を出力します。
    print(vars(result))


if __name__ == "__main__":
    # プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
    main()

SDK が提供するデフォルトの RSA 実装が要件を満たさない場合は、マスターキーに対して独自の暗号化および復号動作を実装できます。次のサンプルコードでは、Alibaba Cloud KMS を例として使用し、単純なオブジェクトのアップロードとダウンロードにカスタムマスターキーを使用する方法を示します。

import argparse
import base64
import json
from aliyunsdkkms.request.v20160120.DecryptRequest import DecryptRequest
from aliyunsdkkms.request.v20160120.EncryptRequest import EncryptRequest
from aliyunsdkcore.client import AcsClient
from typing import Optional, Dict
import alibabacloud_oss_v2 as oss

# ユーザー入力を受け取るためのコマンドライン引数パーサーを作成します。
parser = argparse.ArgumentParser(description="encryption kms sample")

# バケットが配置されているリージョンを指定する --region コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# バケットの名前を指定する --bucket コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 他のサービスが OSS にアクセスするために使用するドメイン名を指定する --endpoint コマンドライン引数を追加します。この引数はオプションです。
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# オブジェクトの名前 (ファイルパス) を指定する --key コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--key', help='The name of the object.', required=True)

# ユーザーのカスタマーマスターキー (CMK) ID を指定する --kms_id コマンドライン引数を追加します。この引数は必須です。
parser.add_argument('--kms_id', help='The id of the your CMK ID.', required=True)


# oss.crypto.MasterCipher を継承するカスタムマスターキーサイファクラス。
class MasterKmsCipher(oss.crypto.MasterCipher):

    def __init__(
        self,
        mat_desc: Optional[Dict] = None,
        kms_client: Optional[AcsClient] = None,
        kms_id: Optional[str] = None,
    ):
        self.kms_client = kms_client
        self.kms_id = kms_id
        self._mat_desc = None

        # マスターキーの説明が提供されている場合は、それを JSON 文字列にシリアル化します。
        if mat_desc is not None and len(mat_desc.items()) > 0:
            self._mat_desc = json.dumps(mat_desc)

    def get_wrap_algorithm(self) -> str:
        # ラップアルゴリズムの名前を返します。これは 'KMS/ALICLOUD' に固定されています。
        return 'KMS/ALICLOUD'

    def get_mat_desc(self) -> str:
        return self._mat_desc or ''

    def encrypt(self, data: bytes) -> bytes:
        """
        KMS サービスを使用してデータを暗号化します。
        :param data: 暗号化する生データ (バイト単位)。
        :return: 暗号化されたデータ (バイト単位)。
        """
        # 生データを Base64 形式でエンコードします。
        base64_crypto = base64.b64encode(data)

        # 暗号化リクエストオブジェクトを構築します。
        request = EncryptRequest()
        request.set_KeyId(self.kms_id)  # CMK ID を設定します。
        request.set_Plaintext(base64_crypto)  # 暗号化する Base64 エンコードされたデータを設定します。

        # KMS クライアントを呼び出して暗号化操作を実行し、応答を取得します。
        response = self.kms_client.do_action_with_exception(request)

        # 応答から暗号化されたデータフィールドを解析し、バイトにデコードします。
        return base64.b64decode(json.loads(response).get('CiphertextBlob'))

    def decrypt(self, data: bytes) -> bytes:
        """
        KMS サービスを使用してデータを復号します。
        :param data: 暗号化されたデータ (バイト単位)。
        :return: 復号された生データ (バイト単位)。
        """
        # 暗号化されたデータを Base64 形式でエンコードします。
        base64_crypto = base64.b64encode(data)

        # 復号リクエストオブジェクトを構築します。
        request = DecryptRequest()
        request.set_CiphertextBlob(base64_crypto)  # 暗号化されたデータを設定します。

        # KMS クライアントを呼び出して復号操作を実行し、応答を取得します。
        response = self.kms_client.do_action_with_exception(request)

        # 応答から平文フィールドを解析し、バイトにデコードします。
        return base64.b64decode(json.loads(response).get('Plaintext'))


def main():
    # コマンドライン引数を解析します。
    args = parser.parse_args()

    # 環境変数から認証情報 (AccessKey ID と AccessKey Secret) を読み込みます。
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # SDK のデフォルト構成を読み込みます。
    cfg = oss.config.load_default()

    # 認証情報プロバイダーを設定します。
    cfg.credentials_provider = credentials_provider

    # バケットが配置されているリージョンを設定します。
    cfg.region = args.region

    # カスタムエンドポイントが提供されている場合は、構成で設定します。
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 構成オブジェクトを使用して OSS クライアントを初期化します。
    client = oss.Client(cfg)

    # KMS サービスと対話するために KMS クライアントを初期化します。
    kms_client = AcsClient(
        ak=credentials_provider._credentials.access_key_id,  # 認証情報プロバイダーから AccessKey ID を取得します。
        secret=credentials_provider._credentials.access_key_secret,  # 認証情報プロバイダーから AccessKey Secret を取得します。
        region_id=args.region  # リージョンを指定します。
    )

    # 暗号化および復号操作のためにマスターキーサイファ (MasterKmsCipher) を初期化します。
    mc = MasterKmsCipher(
        mat_desc={"desc": "your master encrypt key material describe information"},  # マスターキーの説明。
        kms_client=kms_client,  # KMS クライアントインスタンス。
        kms_id=args.kms_id  # ユーザーの CMK ID。
    )

    # 暗号化クライアントを作成します。
    encryption_client = oss.EncryptionClient(client, mc)

    # アップロードするデータを定義します。
    data = b'hello world'

    # 暗号化クライアントの put_object メソッドを呼び出して、暗号化されたオブジェクトをアップロードします。
    result = encryption_client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 宛先バケットの名前。
            key=args.key,        # オブジェクトの名前 (ファイルパス)。
            body=data,           # アップロードするデータ。
        )
    )

    # 暗号化されたオブジェクトのアップロード結果を出力します。
    print(vars(result))

    # 暗号化クライアントの get_object メソッドを呼び出して、暗号化されたオブジェクトのコンテンツを取得します。
    result = encryption_client.get_object(
        oss.GetObjectRequest(
            bucket=args.bucket,  # 宛先バケットの名前。
            key=args.key,        # オブジェクトの名前 (ファイルパス)。
        )
    )

    # 暗号化されたオブジェクトの取得結果を出力します。
    print(vars(result))

    # 復号されたオブジェクトのコンテンツを出力します。
    print(result.body.read())


if __name__ == "__main__":
    # プログラムのエントリポイント。main 関数を呼び出してロジックを実行します。
    main()