OSS客户端加密是在数据上传至OSS之前,由用户在本地对数据进行加密处理,确保只有密钥持有者才能解密数据,增强数据在传输和存储过程中的安全性。
注意事项
本文示例代码以华东1(杭州)的地域ID
cn-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()