452      小米  MIUI米柚

通過客戶端加密保護數據__數據安全_最佳實踐_對象存儲 OSS-阿裏雲

客戶端加密是指用戶數據在發送給遠端服務器之前就完成加密，而加密所用的密鑰的明文隻保留在本地，從而可以保證用戶數據安全，即使數據泄漏別人也無法解密得到原始數據。

本文介紹如何基於OSS的現有Python SDK版本，通過客戶端加密來保護數據。

原理介紹

1. 用戶本地維護一對RSA密鑰(rsa_private_key和rsa_public_key)。
2. 每次上傳Object時，隨機生成一個AES256類型的對稱密鑰data_key，然後用data_key加密原始content得到encrypt_content。
3. 用rsa_public_key加密data_key，得到encrypt_data_key，作為用戶的自定義meta放入請求頭部，和encrypt_content一起發送到OSS。
4. Get Object時，首先得到encrypt_content以及用戶自定義meta中的encrypt_data_key。
5. 用戶使用rsa_private_key解密encrypt_data_key得到data_key，然後用data_key解密encrypt_content得到原始content。

注意：本文用戶的密鑰為非對稱的RSA密鑰, 加密object content時用的AES256-CTR算法, 詳情可參考PyCrypto Document。本文旨在介紹如何通過object的自定義meta來實現客戶端加密，至於加密密匙類型及加密算法，用戶可以根據自己的需要進行選擇。

架構圖

準備工作

1. Python SDK的安裝和使用，參考 Python SDK 快速安裝。

2. 安裝PyCrypto庫。

   pip install pycrypto

完整 Python 示例代碼

# -*- coding: utf-8 -*-
import os
import shutil
import base64
import random
import oss2
from Crypto.Cipher import PKCS1_OAEP
from Crypto.PublicKey import RSA
from Crypto.Cipher import AES
from Crypto import Random
from Crypto.Util import Counter
# aes 256, key always is 32 bytes
_AES_256_KEY_SIZE = 32
_AES_CTR_COUNTER_BITS_LEN = 8 * 16
class AESCipher:
    def __init__(self, key=None, start=None):
        self.key = key
        self.start = start
        if not self.key:
            self.key = Random.new().read(_AES_256_KEY_SIZE)
        if not self.start:
            self.start = random.randint(1, 10)
        ctr = Counter.new(_AES_CTR_COUNTER_BITS_LEN, initial_value=self.start)
        self.cipher = AES.new(self.key, AES.MODE_CTR, counter=ctr)
    def encrypt(self, raw):
        return self.cipher.encrypt(raw)
    def decrypt(self, enc):
        return self.cipher.decrypt(enc)
# 首先初始化AccessKeyId、AccessKeySecret、Endpoint等信息。
# 通過環境變量獲取，或者把諸如“<您的AccessKeyId>”替換成真實的AccessKeyId等。
#
# 以杭州區域為例，Endpoint可以是：
#   https://oss-cn-hangzhou.aliyuncs.com
#   https://oss-cn-hangzhou.aliyuncs.com
# 分別以HTTP、HTTPS協議訪問。
access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '<您的AccessKeyId>')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '<您的AccessKeySecret>')
bucket_name = os.getenv('OSS_TEST_BUCKET', '<您的Bucket>')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '<您的訪問域名>')
# 確認上麵的參數都填寫正確了
for param in (access_key_id, access_key_secret, bucket_name, endpoint):
    assert '<' not in param, '請設置參數：' + param
##### 0 prepare ########
# 0.1 生成rsa key文件並保存到disk 
rsa_private_key_obj = RSA.generate(2048)
rsa_public_key_obj = rsa_private_key_obj.publickey()
encrypt_obj = PKCS1_OAEP.new(rsa_public_key_obj)
decrypt_obj = PKCS1_OAEP.new(rsa_private_key_obj)
# save to local disk 
file_out = open("private_key.pem", "w")
file_out.write(rsa_private_key_obj.exportKey())
file_out.close()
file_out = open("public_key.pem", "w")
file_out.write(rsa_public_key_obj.exportKey())
file_out.close()
# 0.2 創建Bucket對象，所有Object相關的接口都可以通過Bucket對象來進行
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)
obj_name = 'test-sig-1'
content = "test content"
#### 1 Put Object  ####
# 1.1 生成加密這個object所用的一次性的對稱密鑰 encrypt_cipher, 其中的key 和 start為隨機生成的value
encrypt_cipher = AESCipher()
# 1.2 將輔助解密的信息用公鑰加密後存到object的自定義meta中. 後續當我們get object時，就可以根據自定義meta，用私鑰解密得到原始content
headers = {}
headers['x-oss-meta-x-oss-key'] = base64.b64encode(encrypt_obj.encrypt(encrypt_cipher.key))
headers['x-oss-meta-x-oss-start'] = base64.b64encode(encrypt_obj.encrypt(str(encrypt_cipher.start)))
# 1.3. 用 encrypt_cipher 對原始content加密得到encrypt_content
encryt_content = encrypt_cipher.encrypt(content)
# 1.4 上傳object
result = bucket.put_object(obj_name, encryt_content, headers)
if result.status / 100 != 2:
    exit(1)
#### 2 Get Object ####
# 2.1 下載得到加密後的object
result = bucket.get_object(obj_name)
if result.status / 100 != 2:
    exit(1)
resp = result.resp
download_encrypt_content = resp.read()
# 2.2 從自定義meta中解析出之前加密這個object所用的key 和 start 
download_encrypt_key = base64.b64decode(resp.headers.get('x-oss-meta-x-oss-key', ''))
key = decrypt_obj.decrypt(download_encrypt_key)
download_encrypt_start = base64.b64decode(resp.headers.get('x-oss-meta-x-oss-start', ''))
start = int(decrypt_obj.decrypt(download_encrypt_start))
# 2.3 生成解密用的cipher, 並解密得到原始content
decrypt_cipher = AESCipher(key, start)
download_content = decrypt_cipher.decrypt(download_encrypt_content)
if download_content != content:
    print "Error!"
else:
    print "Decrypt ok. Content is: %s" % download_content