通過Python設定用戶端加密實現上傳至OSS前已在本地完成資料加密 - Object Storage Service

OSS用戶端加密是在資料上傳至OSS之前，由使用者在本地對資料進行加密處理，確保只有密鑰持有人才能解密資料，增強資料在傳輸和預存程序中的安全性。

注意事項

本文範例程式碼以華東1（杭州）的地區IDcn-hangzhou為例，預設使用外網Endpoint，如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS地區和訪問網域名稱。
使用用戶端加密功能時，您需要對主要金鑰的完整性和正確性負責。
在對加密資料進行複製或者遷移時，您需要對加密中繼資料的完整性和正確性負責。

方法定義

對於主要金鑰的使用，Python SDK V2目前支援如下兩種方式：

使用使用者自主管理的主要金鑰（RSA）
SDK提供了RSA的預設實現，當主要金鑰資訊由使用者提供時，使用者需要將主要金鑰的公開金鑰、私密金鑰資訊作為參數傳遞給SDK。
使用使用者自訂的主要金鑰
當RSA主要金鑰方式無法滿足需求時，使用者可以自己實現主要金鑰的加解密行為。

使用以上兩種加密方式能夠有效地避免資料泄漏，保護用戶端資料安全。即使資料泄漏，其他人也無法解密得到未經處理資料。

重要

如果您需要瞭解OSS用戶端加密實現的原理，請參考OSS使用者指南中的用戶端加密。

使用用戶端加密，首先您需要執行個體化加密用戶端，然後調用其提供的介面進行操作。您的對象將作為請求的一部分自動加密和解密。

class EncryptionClient:
  ...

def __init__(self,client: Client, master_cipher: MasterCipher, decrypt_master_ciphers: Optional[List[MasterCipher]] = None)

請求參數列表

參數名	類型	說明
client	*Client	非加密用戶端執行個體
master_cipher	MasterCipher	主要金鑰執行個體，用於加密和解密資料密鑰
decrypt_master_ciphers	List[MasterCipher]	主要金鑰執行個體，用於解密資料密鑰

EncryptionClient介面列舉如下：

基礎介面名	說明
get_object_meta	擷取對象的部分元資訊
head_object	擷取對象的部元資訊
get_object	下載對象，並自動解密
put_object	上傳對象，並自動加密
initiate_multipart_upload	初始化一個分區上傳事件和分區加密上下文（EncryptionMultiPartContext）
upload_part	初始化一個分區上傳事件, 調用該介面上傳分區資料，並自動加密。調用該介面時，需要設定分區加密上下文
complete_multipart_upload	在將所有分區資料上傳完成後，調用該介面合并成一個檔案
abort_multipart_upload	取消分區上傳事件,並刪除對應的分區資料
list_parts	列舉指定上傳事件所屬的所有已經上傳成功分區

使用RSA主要金鑰

使用主要金鑰RSA簡單上傳和下載Object

使用主要金鑰RSA簡單上傳和下載Object範例程式碼如下：

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)

# 添加命令列參數 --endpoint，表示其他服務訪問 OSS 時使用的網域名稱，可選項
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()

    # 從環境變數中載入憑證資訊（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 載入 SDK 的預設配置
    cfg = oss.config.load_default()

    # 設定憑證提供者
    cfg.credentials_provider = credentials_provider

    # 設定儲存空間所在的地區
    cfg.region = args.region

    # 如果使用者提供了自訂的 endpoint，則設定到配置中
    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}, '   # 返回的對象的 CRC64 校正值
          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}, '   # 對象內容的 CRC64 校正值
          f'version id: {result.version_id}')    # 對象的版本 ID（如果啟用了版本控制）


if __name__ == "__main__":
    # 程式入口，調用 main 函數執行邏輯
    main()

使用主要金鑰RSA分區上傳Object

使用主要金鑰RSA分區上傳Object範例程式碼如下：

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)

# 添加命令列參數 --endpoint，表示其他服務訪問 OSS 時使用的網域名稱，可選項
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()

    # 從環境變數中載入憑證資訊（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 載入 SDK 的預設配置
    cfg = oss.config.load_default()

    # 設定憑證提供者
    cfg.credentials_provider = credentials_provider

    # 設定儲存空間所在的地區
    cfg.region = args.region

    # 如果使用者提供了自訂的 endpoint，則設定到配置中
    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)

    # 定義分區上傳的每個分區大小（單位：位元組），這裡設定為 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()

使用自訂主要金鑰

使用自訂主要金鑰簡單上傳和下載Object

SDK提供了RSA預設實現，當這個方式不滿足使用者的需求時，使用者可以自己實現主要金鑰的加解密行為。以下範例程式碼以阿里雲KMS為例，示範如何自訂主要金鑰加解密進行簡單上傳和下載Object。

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)

# 添加命令列參數 --endpoint，表示其他服務訪問 OSS 時使用的網域名稱，可選項
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)

# 添加命令列參數 --kms_id，表示使用者的 CMK（Customer Master Key）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()

    # 從環境變數中載入憑證資訊（AccessKeyId 和 AccessKeySecret）
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 載入 SDK 的預設配置
    cfg = oss.config.load_default()

    # 設定憑證提供者
    cfg.credentials_provider = credentials_provider

    # 設定儲存空間所在的地區
    cfg.region = args.region

    # 如果使用者提供了自訂的 endpoint，則設定到配置中
    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,  # 從憑證提供者中擷取 AccessKeyId
        secret=credentials_provider._credentials.access_key_secret,  # 從憑證提供者中擷取 AccessKeySecret
        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()

Object Storage Service：用戶端加密（Python SDK V2）