Use Function Compute to rotate generic secrets - Key Management Service

You can use Function Compute to rotate generic secrets to improve the security of sensitive information. This topic describes how to use Function Compute to rotate generic secrets.

Background information

Function Compute is a fully managed, event-driven computing service. Function Compute allows you to focus on coding without the need to obtain or manage infrastructure resources such as servers. This way, you need to only write and upload your code. Function Compute allocates computing resources, runs tasks in an elastic and reliable manner, and provides features such as log query, performance monitoring, and alerting. For more information, see What is Function Compute?.

Billing

If you use Function Compute to rotate generic secrets, you are not charged for secret rotation that is provided by Key Management Service (KMS). You are charged for Function Compute and Serverless Workflow that is integrated into Function Compute. For more information about billing, see Billing of Function Compute and Billing of Serverless Workflow.

Important

Secrets Manager provides dynamic secrets that are automatically rotated. If dynamic secrets can meet your requirements for security, we recommend that you use dynamic secrets to reduce costs. For more information, see Overview of Dynamic ApsaraDB RDS secrets,Overview of Dynamic RAM secrets, or Overview of Dynamic ECS secrets.

Secret rotation process

This topic describes how to rotate a generic secret in the following scenarios:

Rotate a generic secret of an ApsaraDB RDS database in dual-account mode by using ApsaraDB RDS API
Rotate generic secrets of a self-managed MySQL database in dual-account mode by using a privileged account

Rotate a generic secret of an ApsaraDB RDS database in dual-account mode by using ApsaraDB RDS API

You can use Serverless Workflow and Function Compute to rotate a generic secret.

1. Prepare for secret rotation

Activate Function Compute. For more information, see Activate Function Compute.
Create an account that is used to log on to your ApsaraDB RDS database. Then, create a generic secret for the account in the KMS console. If the generic secret is rotated for the first time, a new account is generated.
The secret value is in the JSON format. Example: The secret name is used as the scheduling parameter that will be passed to Serverless Workflow.
```
{
   "AccountName": "",
 "AccountPassword": ""
}
```
Note
AccountName specifies the username of the account that is used to log on to your ApsaraDB RDS database, and AccountPassword specifies the password of the account that is used to log on to your ApsaraDB RDS database.

Create a normal service role for Function Compute and grant permissions to allow Function Compute to access KMS. For more information, see Grant Function Compute permissions to access other Alibaba Cloud services.

Attach the AliyunFCDefaultRolePolicy policy to the normal service role of Function Compute.
Attach the AliyunSTSAssumeRoleAccess system policy to the normal service role.

Attach a custom policy that allows access to KMS to the normal service role. The following script shows the content of the policy:

{
    "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. Create secret rotation functions

You must create a Function Compute service and secret rotation functions in the Function Compute console. For more information, see Quickly create a function.

Create a Function Compute service.
When you create a Function Compute service, you must click Show Advanced Options and set Access to VPC to Yes. Then, configure VPC, vSwitch, and Security Group. Make sure that the Function Compute service can access ApsaraDB RDS API and KMS by using the specified security group and vSwitch in the specified VPC.

Create functions.

In this example, Python is used. When you create the functions, select Python 3.9 for Runtime Environments. The following code provides an example. You can modify the code based on your business requirements.

# -*- 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. Create a Serverless Workflow flow for secret rotation

Create a secret rotation flow. For more information, see Create a flow.

Log on to the Serverless Workflow console.
In the top navigation bar, select the region where you want to create a flow.
Important
You must select the region where the functions are created.
On the Flows page, click Create flow.

On the Create Flow page, click Create Flow with Code, configure the parameters, and then click Next Step.

You must change the content in the YAML file of Definition to the following code:

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

Configure a flow role.
Click Create Flow.

Configure settings to rotate the generic secret at a scheduled time. For more information, see Create a time-based schedule.
If you want to use Serverless Workflow to configure scheduled rotation of the generic secret, set Payload to the following content:
```
{
  "SecretName": "",
  "RegionId": "",
  "InstanceId":""
}
```
Note
SecretName specifies the secret name, RegionId specifies the region, and InstanceId specifies the ID of your ApsaraDB RDS instance.

4. (Optional) Connect applications to Secrets Manager

You can use Secrets Manager JDBC to connect applications to Secrets Manager. For more information, see Secrets Manager JDBC.

Rotate generic secrets of a self-managed MySQL database in dual-account mode by using a privileged account

You can use Serverless Workflow and Function Compute to rotate generic secrets.