Function Computeを使用したジェネリックシークレットのローテーション - Key Management Service

Function Computeを使用して汎用シークレットをローテーションし、機密情報のセキュリティを向上させることができます。このトピックでは、Function Computeを使用してジェネリックシークレットをローテーションする方法について説明します。

背景情報

Function Computeは、完全マネージド型のイベント駆動型コンピューティングサービスです。 Function Computeを使用すると、サーバーなどのインフラストラクチャリソースを取得または管理する必要なく、コーディングに集中できます。この方法では、コードを記述してアップロードするだけです。 Function Computeは、コンピューティングリソースを割り当て、柔軟で信頼性の高い方法でタスクを実行し、ログクエリ、パフォーマンスモニタリング、アラートなどの機能を提供します。詳細については、「Function Computeとは」をご参照ください。

課金

Function Computeを使用して一般的なシークレットをローテーションする場合、Key Management Service (KMS) によって提供されるシークレットローテーションに対しては課金されません。 Function Computeと、Function Computeに統合されたサーバーレスワークフローに対して課金されます。課金の詳細については、「Function Computeの課金」および「Serverless Workflowの課金」をご参照ください。

重要

Secrets Managerは、自動的にローテーションされる動的シークレットを提供します。動的シークレットがセキュリティ要件を満たすことができる場合は、動的シークレットを使用してコストを削減することを推奨します。詳細については、「ダイナミックApsaraDB RDSシークレットの概要」、「概要」、または「ダイナミックECSシークレットの概要」をご参照ください。

秘密の回転プロセス

このトピックでは、次のシナリオでジェネリックシークレットをローテーションする方法について説明します。

ApsaraDB RDS APIを使用して、デュアルアカウントモードでApsaraDB RDSデータベースのジェネリックシークレットを回転させる
特権アカウントを使用して、デュアルアカウントモードで自己管理型MySQLデータベースの汎用シークレットを回転させる

ApsaraDB RDS APIを使用して、デュアルアカウントモードでApsaraDB RDSデータベースのジェネリックシークレットを回転させる

Serverless WorkflowとFunction Computeを使用して、ジェネリックシークレットをローテーションできます。

1. 秘密のローテーションの準備

Function Computeを有効にします。詳細については、「Function Computeの有効化」をご参照ください。
ApsaraDB RDSデータベースへのログインに使用するアカウントを作成します。次に、KMSコンソールでアカウントのジェネリックシークレットを作成します。ジェネリックシークレットが初めてローテーションされた場合、新しいアカウントが生成されます。
シークレット値はJSON形式です。例: シークレット名は、Serverless Workflowに渡されるスケジューリングパラメーターとして使用されます。
```
{
   "AccountName": "",
 "AccountPassword": ""
}
```
説明
AccountNameは、ApsaraDB RDSデータベースへのログインに使用されるアカウントのユーザー名を指定し、AccountPasswordは、ApsaraDB RDSデータベースへのログインに使用されるアカウントのパスワードを指定します。
Function Computeの通常のサービスロールを作成し、Function ComputeがKMSにアクセスできるようにする権限を付与します。詳細については、「他のAlibaba Cloudサービスへのアクセス権限付与」をご参照ください。
- AliyunFCDefaultRolePolicyポリシーをFunction Computeの通常のサービスロールにアタッチします。
- AliyunSTSAssumeRoleAccessシステムポリシーを通常のサービスロールにアタッチします。
- KMSへのアクセスを許可するカスタムポリシーを通常のサービスロールにアタッチします。次のスクリプトは、ポリシーの内容を示しています。
```
{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kms:GetSecretValue",
                "kms:GetRandomPassword",
                "kms:PutSecretValue",
                "kms:UpdateSecretVersionStage"
            ],
            "Resource": "*"
        },
        {
            "Effect": "Allow",
            "Action": [
                "rds:GrantAccountPrivilege",
                "rds:DescribeAccounts",
                "rds:ResetAccountPassword",
                "rds:CreateAccount"
            ],
            "Resource": "*"
        }
    ]
}
```

2. 秘密回転関数の作成

Function Computeコンソールで、Function Computeサービスと秘密のローテーション関数を作成する必要があります。詳細については、「関数の迅速な作成」をご参照ください。

Function Computeサービスを作成します。
Function Computeサービスを作成するときは、[詳細オプションの表示] をクリックし、[VPCへのアクセス] を [はい] に設定する必要があります。次に、VPC、vSwitch、セキュリティグループを設定します。 Function Computeサービスが、指定されたセキュリティグループと指定されたVPCのvSwitchを使用してApsaraDB RDS APIとKMSにアクセスできることを確認します。

関数を作成します。

この例では、Pythonが使用されます。関数を作成するときは、ランタイム環境で [Python 3.9] を選択します。次のコードは例を提供します。ビジネス要件に基づいてコードを変更できます。

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

from aliyunsdkrds.request.v20140815.CreateAccountRequest import CreateAccountRequest
from aliyunsdkrds.request.v20140815.DescribeAccountsRequest import DescribeAccountsRequest
from aliyunsdkrds.request.v20140815.GrantAccountPrivilegeRequest import GrantAccountPrivilegeRequest
from aliyunsdkrds.request.v20140815.ResetAccountPasswordRequest import ResetAccountPasswordRequest

from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkcore.auth.credentials import StsTokenCredential
from aliyunsdkcore.client import AcsClient
from aliyunsdkkms.request.v20160120.GetRandomPasswordRequest import GetRandomPasswordRequest
from aliyunsdkkms.request.v20160120.GetSecretValueRequest import GetSecretValueRequest
from aliyunsdkkms.request.v20160120.PutSecretValueRequest import PutSecretValueRequest
from aliyunsdkkms.request.v20160120.UpdateSecretVersionStageRequest import UpdateSecretVersionStageRequest

logger = logging.getLogger()
logger.setLevel(logging.INFO)


def handler(event, context):
    evt = json.loads(event)
    secret_name = evt['SecretName']
    region_id = evt['RegionId']
    step = evt['Step']
    instance_id = evt['InstanceId']
    version_id = evt.get('VersionId')
    if not version_id:
        version_id = context.requestId
    credentials = StsTokenCredential(context.credentials.accessKeyId, context.credentials.accessKeySecret,
                                     context.credentials.securityToken)
    client = AcsClient(region_id=region_id, credential=credentials)

    endpoint = "kms-vpc." + region_id + ".aliyuncs.com"
    client.add_endpoint(region_id, 'kms', endpoint)
    resp = get_secret_value(client, secret_name)
    if "Generic" != resp['SecretType']:
        logger.error("Secret %s is not enabled for rotation" % secret_name)
        raise ValueError("Secret %s is not enabled for rotation" % secret_name)

    if step == "new":
        new_phase(client, secret_name, version_id)

    elif step == "set":
        set_phase(client, instance_id, secret_name, version_id)

    elif step == "end":
        end_phase(client, secret_name, version_id)

    else:
        logger.error("handler: Invalid step parameter %s for secret %s" % (step, secret_name))
        raise ValueError("Invalid step parameter %s for secret %s" % (step, secret_name))
    return {"VersionId": version_id}


def new_phase(client, secret_name, version_id):
    current_dict = get_secret_dict(client, secret_name, "ACSCurrent")
    try:
        get_secret_dict(client, secret_name, "ACSPending", version_id)
        logger.info("new: Successfully retrieved secret for %s." % secret_name)
    except ServerException as e:
        if e.error_code != 'Forbidden.ResourceNotFound':
            raise ValueError("Can to find secret %s " % (secret_name))
        current_dict['AccountName'] = get_alt_account_name(current_dict['AccountName'])

        exclude_characters = os.environ[
            'EXCLUDE_CHARACTERS'] if 'EXCLUDE_CHARACTERS' in os.environ else "\\\"\',./:;<>?[]{|}~`"
        passwd = get_random_password(client, exclude_characters)
        current_dict['AccountPassword'] = passwd['RandomPassword']
        put_secret_value(client, secret_name, version_id, json.dumps(current_dict),
                         json.dumps(['ACSPending']))
        logger.info(
            "new: Successfully put secret for secret_name %s and version %s." % (secret_name, version_id))


def set_phase(client, instance_id, secret_name, version_id):
    current_dict = get_secret_dict(client, secret_name, "ACSCurrent")
    pending_dict = get_secret_dict(client, secret_name, "ACSPending", version_id)
    pending_resp = describe_accounts(client, instance_id, pending_dict["AccountName"])
    pending_accounts = pending_resp["Accounts"]["DBInstanceAccount"]

    if get_alt_account_name(current_dict['AccountName']) != pending_dict['AccountName']:
        logger.error("set: Attempting to modify user %s other than current user or rotation %s" % (
            pending_dict['AccountName'], current_dict['AccountName']))
        raise ValueError("Attempting to modify user %s other than current user or rotation %s" % (
            pending_dict['AccountName'], current_dict['AccountName']))

    current_resp = describe_accounts(client, instance_id, current_dict["AccountName"])
    current_accounts = current_resp["Accounts"]["DBInstanceAccount"]
    if len(current_accounts) == 0:
        logger.error("set: Unable to log into database using current credentials for secret %s" % secret_name)
        raise ValueError("Unable to log into database using current credentials for secret %s" % secret_name)
    if len(pending_accounts) == 0:
        create_rds_account(client, instance_id, pending_dict["AccountName"],
                           pending_dict["AccountPassword"], current_accounts[0]["AccountType"])
        pending_accounts = describe_accounts(client, instance_id, pending_dict["AccountName"])["Accounts"][
            "DBInstanceAccount"]
    else:
        # reset password
        reset_account_password(client, instance_id, pending_dict["AccountName"], pending_dict["AccountPassword"])

    current_privileges = current_accounts[0]["DatabasePrivileges"]["DatabasePrivilege"]
    pending_privileges = pending_accounts[0]["DatabasePrivileges"]["DatabasePrivilege"]
    if len(current_privileges) > 0:
        for current_privilege in current_privileges:
            is_contains = False
            for pending_privilege in pending_privileges:
                if current_privilege["DBName"] == pending_privilege["DBName"] and current_privilege[
                    "AccountPrivilege"] == pending_privilege["AccountPrivilege"]:
                    is_contains = True
                    continue
            if not is_contains:
                grant_account_privilege(client, instance_id, pending_dict["AccountName"], current_privilege["DBName"],
                                        current_privilege["AccountPrivilege"])


def end_phase(client, secret_name, version_id):
    update_secret_version_stage(client, secret_name, 'ACSCurrent', move_to_version=version_id)
    update_secret_version_stage(client, secret_name, 'ACSPending', remove_from_version=version_id)
    logger.info(
        "end: Successfully set ACSCurrent stage to version %s for secret %s." % (version_id, secret_name))


def get_secret_dict(client, secret_name, stage, version_id=None):
    required_fields = ['AccountName', 'AccountPassword']
    if version_id:
        secret = get_secret_value(client, secret_name, version_id, stage)
    else:
        secret = get_secret_value(client, secret_name, stage=stage)
    plaintext = secret['SecretData']
    secret_dict = json.loads(plaintext)
    for field in required_fields:
        if field not in secret_dict:
            raise KeyError("%s key is missing from secret JSON" % field)
    return secret_dict


def get_alt_account_name(current_account_name):
    rotation_suffix = "_rt"
    if current_account_name.endswith(rotation_suffix):
        return current_account_name[:(len(rotation_suffix) * -1)]
    else:
        new_account_name = current_account_name + rotation_suffix
        if len(new_account_name) > 16:
            raise ValueError(
                "Unable to rotation user, account_name length with _rotation appended would exceed 16 characters")
        return new_account_name


def get_secret_value(client, secret_name, version_id=None, stage=None):
    request = GetSecretValueRequest()
    request.set_accept_format('json')
    request.set_SecretName(secret_name)
    if version_id:
        request.set_VersionId(version_id)
    if stage:
        request.set_VersionStage(stage)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def put_secret_value(client, secret_name, version_id, secret_data, version_stages=None):
    request = PutSecretValueRequest()
    request.set_accept_format('json')
    request.set_SecretName(secret_name)
    request.set_VersionId(version_id)
    if version_stages:
        request.set_VersionStages(version_stages)
    request.set_SecretData(secret_data)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def get_random_password(client, exclude_characters=None):
    request = GetRandomPasswordRequest()
    request.set_accept_format('json')
    if exclude_characters:
        request.set_ExcludeCharacters(exclude_characters)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def update_secret_version_stage(client, secret_name, version_stage, remove_from_version=None, move_to_version=None):
    request = UpdateSecretVersionStageRequest()
    request.set_accept_format('json')
    request.set_VersionStage(version_stage)
    request.set_SecretName(secret_name)
    if remove_from_version:
        request.set_RemoveFromVersion(remove_from_version)
    if move_to_version:
        request.set_MoveToVersion(move_to_version)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def create_rds_account(client, db_instance_id, account_name, account_password, account_type):
    request = CreateAccountRequest()
    request.set_accept_format('json')
    request.set_DBInstanceId(db_instance_id)
    request.set_AccountName(account_name)
    request.set_AccountPassword(account_password)
    request.set_AccountType(account_type)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def grant_account_privilege(client, db_instance_id, account_name, db_name, account_privilege):
    request = GrantAccountPrivilegeRequest()
    request.set_accept_format('json')
    request.set_DBInstanceId(db_instance_id)
    request.set_AccountName(account_name)
    request.set_DBName(db_name)
    request.set_AccountPrivilege(account_privilege)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def describe_accounts(client, db_instance_id, account_name):
    request = DescribeAccountsRequest()
    request.set_accept_format('json')
    request.set_DBInstanceId(db_instance_id)
    request.set_AccountName(account_name)
    response = client.do_action_with_exception(request)
    return json.loads(response)


def reset_account_password(client, db_instance_id, account_name, account_password):
    request = ResetAccountPasswordRequest()
    request.set_accept_format('json')
    request.set_DBInstanceId(db_instance_id)
    request.set_AccountName(account_name)
    request.set_AccountPassword(account_password)
    response = client.do_action_with_exception(request)
    return json.loads(response)

3. シークレットローテーション用のサーバーレスワークフローフローの作成

秘密のローテーションフローを作成します。詳細については、「ワークフローの作成」をご参照ください。

Serverless Workflowコンソールにログインします。
上部のナビゲーションバーで、フローを作成するリージョンを選択します。
重要
関数が作成されるリージョンを選択する必要があります。
[フロー] ページで、[フローの作成] をクリックします。

[フローの作成] ページで、[コードによるフローの作成] をクリックし、パラメーターを設定し、[次のステップ] をクリックします。

DefinitionのYAMLファイルの内容を次のコードに変更する必要があります。

version: v1
type: flow
steps:
  - type: task
    name: RotateSecretNew
    resourceArn: the Alibaba Cloud Resource Name (ARN) of the function that you created
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: InstanceId
        source: $input.payload.InstanceId
      - target: Step
        source: new
  - type: task
    name: RotateSecretSet
    resourceArn: the ARN of the function that you created
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: InstanceId
        source: $input.payload.InstanceId
      - target: Step
        source: set
      - target: VersionId
        source: $local.VersionId
  - type: task
    name: RotateSecretEnd
    resourceArn: the ARN of the function that you created
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: InstanceId
        source: $input.payload.InstanceId
      - target: Step
        source: end
      - target: VersionId
        source: $local.VersionId

フローロールを設定します。
[フローの作成] をクリックします。

スケジュールされた時間にジェネリックシークレットをローテーションするように設定します。詳細については、「時間ベースのスケジュールの作成」をご参照ください。
Serverless Workflowを使用してジェネリックシークレットのスケジュールされたローテーションを設定する場合は、[ペイロード] を次の内容に設定します。
```
{
  "SecretName": "",
  "RegionId": "",
  "InstanceId":""
}
```
説明
SecretNameはシークレット名、RegionIdはリージョン、InstanceIdはApsaraDB RDSインスタンスのIDを指定します。

4. (オプション) アプリケーションをSecrets Managerに接続する

Secrets Manager JDBCを使用して、アプリケーションをSecrets Managerに接続できます。詳細については、「Secrets Manager JDBC」をご参照ください。

特権アカウントを使用して、デュアルアカウントモードで自己管理型MySQLデータベースの汎用シークレットを回転させる