通过Python设置客户端加密实现上传至OSS前已在本地完成数据加密

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 alibabacloud_dkms_transfer.kms_transfer_acs_client import KmsTransferAcsClient
from typing import Optional, Dict
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 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[KmsTransferAcsClient] = 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 = KmsTransferAcsClient(
        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()

客户端加密（Python SDK V2）