データの暗号化と復号化にKMSキーを使用する - Key Management Service

KMSインスタンスSDKクライアントを初期化した後、EncryptおよびDecryptサービスを呼び出してデータを処理できます。このトピックでは、KMSを実装して対称暗号化および復号化を実行するためのコードサンプルについて説明します。

完全なコード例

対称暗号化と復号化のためのKMSの統合は、3つのステップで構成されます。

KMSサービスを呼び出すようにクライアントを初期化します。
クライアントでEncryptサービスを呼び出してデータを暗号化します。
クライアントでDecryptサービスを呼び出して暗号文を復号します。

Python 2バージョンのソースコードGitHubアドレス: aes_encrypt_decrypt_sample.py

Python 3バージョンのソースコードGitHubアドレス: aes_encrypt_decrypt_sample.py

このトピックでは、Python 3バージョンを例として使用します。

暗号化と復号化の完全なコード例

# -*- coding: utf-8 -*-
import os

from openapi.models import Config
from openapi_util.models import RuntimeOptions
from sdk.client import Client
from sdk.models import EncryptRequest, DecryptRequest

config = Config()
# Set the connection protocol to "https". The KMS instance service only allows access through the HTTPS protocol.
config.protocol = "https"
# Client Key.
config.client_key_file = "<CLIENT_KEY_FILE>"
# Client Key decryption password.
config.password = os.getenv('CLIENT_KEY_PASSWORD')
# Set the endpoint to <KMS_INSTANCE_ID>.cryptoservice.kms.aliyuncs.com.
config.endpoint = "<ENDPOINT>"
client = Client(config)


class AESEncryptContext(object):
    """The aes encrypt context may be stored."""

    def __init__(self, key_id, ciphertext_blob, iv, algorithm):
        self.key_id = key_id
        self.ciphertext_blob = ciphertext_blob
        self.iv = iv
        # Use default algorithm value, if the value is not set.
        self.algorithm = algorithm


def encrypt(key_id, plaintext):
    request = EncryptRequest()
    request.plaintext = plaintext
    request.key_id = key_id
    runtime_options = RuntimeOptions()
    # Ignore server certificate
    # runtime_options.ignore_ssl = True
    # verify indicates the path of the instance CA certificate
    runtime_options.verify = "<CA_CERTIFICATE_FILE_PATH>"
    resp = client.encrypt_with_options(request, runtime_options)
    print(resp)
    return AESEncryptContext(resp.key_id, resp.ciphertext_blob, resp.iv, resp.algorithm)


def decrypt(context):
    request = DecryptRequest()
    request.ciphertext_blob = context.ciphertext_blob
    request.key_id = context.key_id
    request.iv = context.iv
    request.algorithm = context.algorithm
    runtime_options = RuntimeOptions()
    # Ignore server certificate
    # runtime_options.ignore_ssl = True
    # verify indicates the path of the instance CA certificate
    runtime_options.verify = "<CA_CERTIFICATE_FILE_PATH>"
    resp = client.decrypt_with_options(request, runtime_options)
    print(resp)


plaintext = "<PLAINTEXT>".encode("utf-8")
key_id = "<KEY_ID>"
context = encrypt(key_id, plaintext)
decrypt(context)

コード例分析

クライアントの初期化

クライアントの初期化の詳細については、「クライアントの初期化」をご参照ください。

# -*- coding: utf-8 -*-
from openapi.models import Config
from sdk.client import Client

config = Config()
# The connection protocol. Set the value to https. The KMS instance service only allows access through the HTTPS protocol.
config.protocol = "https"

# Client Key.
config.client_key_file = "<CLIENT_KEY_FILE>"

# Client Key decryption password.
config.password = os.getenv('CLIENT_KEY_PASSWORD')

# The endpoint of your KMS instance. Set the value in the following format: <ID of your KMS instance>.cryptoservice.kms.aliyuncs.com.
config.endpoint = "<ENDPOINT>"
client = Client(config)

Encryptインターフェイスを呼び出して、対称キーを使用してデータを暗号化します。

def encrypt(key_id, plaintext):
    request = EncryptRequest()
    request.plaintext = plaintext
    request.key_id = key_id
    runtime_options = RuntimeOptions()
    # Ignore server certificate
    # runtime_options.ignore_ssl = True
    # verify indicates the path of the instance CA certificate
    runtime_options.verify = "<CA_CERTIFICATE_FILE_PATH>"
    resp = client.encrypt_with_options(request, runtime_options)
    print(resp)
    return AESEncryptContext(resp.key_id, resp.ciphertext_blob, resp.iv, resp.algorithm)

対称キーを使用して暗号文を復号化するDecryptインターフェイスを呼び出します。

    def decrypt(context):
    request = DecryptRequest()
    request.ciphertext_blob = context.ciphertext_blob
    request.key_id = context.key_id
    request.iv = context.iv
    request.algorithm = context.algorithm
    runtime_options = RuntimeOptions()
    # Ignore server certificate
    # runtime_options.ignore_ssl = True
    # verify indicates the path of the instance CA certificate
    runtime_options.verify = "<CA_CERTIFICATE_FILE_PATH>"
    resp = client.decrypt_with_options(request, runtime_options)
    print(resp)