Découvrez comment créer une fonction de rotation pour les secrets génériques avec Function Compute. - Key Management Service

Automatisez la rotation des secrets génériques stockés dans KMS Secrets Manager grâce à Function Compute et CloudFlow. Ce guide couvre deux scénarios de rotation de base de données : les instances ApsaraDB RDS accessibles via l'API RDS et les bases de données MySQL autogérées accessibles via un compte privilégié.

Important

KMS Secrets Manager propose des secrets dynamiques dotés d'une rotation automatique intégrée pour ApsaraDB RDS, RAM et ECS. Si les secrets dynamiques répondent à vos besoins, privilégiez cette solution : elle est moins coûteuse et ne nécessite aucun code personnalisé. Consultez les rubriques Présentation des secrets dynamiques ApsaraDB RDS, Présentation des secrets dynamiques RAM ou Présentation des secrets dynamiques ECS.

Facturation

La rotation des secrets KMS est gratuite. L'utilisation de Function Compute et de CloudFlow est facturée. Pour plus d'informations, consultez les pages Facturation de Function Compute et Facturation de CloudFlow.

Choisir une stratégie de rotation

Les deux scénarios présentés dans ce guide reposent sur la rotation par double compte (utilisateur alterné). Cette stratégie maintient deux comptes de base de données, dont un seul est actif à la fois. Votre application dispose ainsi toujours d'identifiants valides pendant la rotation. Le compte inactif reçoit les nouveaux identifiants, puis devient le compte actif.

Stratégie	Fonctionnement	Disponibilité pendant la rotation	Cas d'utilisation idéal
Compte unique	Met à jour les identifiants d'un seul compte sur place	Brève interruption lors de la mise à jour des identifiants	Configurations simples, utilisateurs ponctuels
Double compte	Alterne entre deux comptes ; l'un est toujours actif	Aucune interruption	Bases de données de production, exigences de haute disponibilité

Le suffixe de rotation _rt identifie le compte alternatif. Par exemple, si le compte actif est appuser, le compte alternatif est appuser_rt. Les noms de compte incluant le suffixe ne doivent pas dépasser 16 caractères.

Fonctionnement

Les deux scénarios suivent le même processus orchestré par CloudFlow. Une planification temporelle de CloudFlow déclenche le flux, qui appelle la même fonction Function Compute plusieurs fois avec différentes valeurs pour le paramètre Step :

Étape	Action de la fonction	Idempotence
`new`	Génère un nouveau mot de passe et l'écrit dans la version en attente du secret (`ACSPending`)	Si `ACSPending` existe déjà pour l'ID de version actuel, ignore la création du secret
`set`	Crée ou met à jour le compte de base de données alternatif avec les identifiants en attente	Sécurisé pour les nouvelles tentatives : vérifie l'état du compte avant d'apporter des modifications
`test`	Vérifie que les identifiants en attente permettent la connexion à la base de données (scénario MySQL uniquement)	Vérification de connectivité en lecture seule
`end`	Fait passer `ACSPending` à `ACSCurrent`, finalisant ainsi la rotation	La mise à jour de l'étape est atomique

Chaque étape étant idempotente, le flux peut être relancé en toute sécurité après des échecs partiels.

Scénario 1 : Rotation d'un secret générique ApsaraDB RDS

Utilisez cette approche lorsque votre base de données fonctionne sur ApsaraDB RDS et que vous souhaitez faire pivoter les identifiants via l'API RDS.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un compte Function Compute actif. Consultez la page Activer Function Compute.
Un compte de base de données ApsaraDB RDS et un secret générique associé dans la console KMS.
Un rôle de service Function Compute disposant des autorisations répertoriées à l'étape 1.

Étape 1 : Configurer les autorisations

Créez un rôle de service pour Function Compute et attachez-y les stratégies suivantes :

AliyunFCDefaultRolePolicy — stratégie système
AliyunSTSAssumeRoleAccess — stratégie système
Une stratégie personnalisée accordant l'accès à KMS et RDS :

{
    "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": "*"
        }
    ]
}

Pour savoir comment attacher ces stratégies, consultez la page Accorder à Function Compute les autorisations d'accès à d'autres services Alibaba Cloud.

Étape 2 : Créer le secret générique

Créez un secret générique dans la console KMS en utilisant le JSON suivant comme valeur de secret. Le nom du secret sera transmis ultérieurement en tant que paramètre de planification à CloudFlow.

{
    "AccountName": "<rds-account-name>",
    "AccountPassword": "<rds-account-password>"
}

Champ	Description
`AccountName`	Nom d'utilisateur du compte de base de données RDS
`AccountPassword`	Mot de passe du compte de base de données RDS

Étape 3 : Créer la fonction de rotation

Dans la console Function Compute, créez un service. Cliquez sur Show Advanced Options, définissez Access to VPC sur Yes, puis configurez les paramètres VPC, vSwitch et Security Group. Le groupe de sécurité doit autoriser la fonction à atteindre à la fois l'API RDS et le point de terminaison VPC KMS (kms-vpc.<region-id>.aliyuncs.com).
Créez une fonction avec l'Runtime Environments défini sur Python 3.9. Utilisez le code suivant :

# -*- 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)

Étape 4 : Créer le flux de rotation CloudFlow

Connectez-vous à la console CloudFlow et sélectionnez la même région que vos fonctions Function Compute.
Sur la page Flows, cliquez sur Create flow.
Sur la page Create Flow, cliquez sur Create Flow with Code, définissez le YAML de la section Definition comme suit, puis cliquez sur Next Step :

version: v1
type: flow
steps:
  - type: task
    name: RotateSecretNew
    resourceArn: <function-arn>
    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: <function-arn>
    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: <function-arn>
    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

Remplacez <function-arn> par le nom de ressource Alibaba Cloud (ARN) de la fonction que vous avez créée.

Configurez un rôle de flux, puis cliquez sur Create Flow.
Pour planifier la rotation automatique, créez une planification basée sur le temps. Consultez la page Créer une planification basée sur le temps. Définissez le champ Payload comme suit :

{
    "SecretName": "<secret-name>",
    "RegionId": "<region-id>",
    "InstanceId": "<rds-instance-id>"
}

Champ	Description
`SecretName`	Nom du secret générique dans KMS
`RegionId`	Région où le secret et la fonction sont déployés
`InstanceId`	ID de l'instance ApsaraDB RDS

Étape 5 : Vérifier la rotation

Après avoir configuré le flux, déclenchez manuellement une exécution pour confirmer que tout fonctionne correctement :

Dans la console CloudFlow, ouvrez votre flux et cliquez sur Start Execution en utilisant le même payload JSON qu'à l'étape 4.
Attendez la fin de l'exécution. Les trois étapes (RotateSecretNew, RotateSecretSet, RotateSecretEnd) doivent afficher l'état Succeeded.
Dans la console KMS, ouvrez le secret et vérifiez son historique de versions. L'étiquette ACSCurrent doit désormais pointer vers une nouvelle version.

Étape 6 : (Facultatif) Connecter les applications à Secrets Manager

Utilisez Secrets Manager JDBC pour récupérer les identifiants ayant fait l'objet d'une rotation dans votre application sans modifier le code. Consultez la page Secrets Manager JDBC.

Scénario 2 : Rotation d'un secret générique MySQL autogéré

Utilisez cette approche lorsque votre base de données est une instance MySQL autogérée. Ce scénario utilise un compte privilégié pour gérer la rotation des identifiants et ajoute une étape test pour vérifier la connectivité avant de finaliser la rotation.

Prérequis

Avant de commencer, assurez-vous de disposer des éléments suivants :

Un compte Function Compute actif. Consultez la page Activer Function Compute.
Un compte MySQL privilégié et un secret générique associé dans la console KMS.
Un second secret générique pour le compte régulier à faire pivoter.
Un rôle de service Function Compute disposant des autorisations répertoriées à l'étape 1.

Étape 1 : Configurer les autorisations

Créez un rôle de service pour Function Compute et attachez-y les stratégies suivantes :

AliyunFCDefaultRolePolicy — stratégie système
AliyunSTSAssumeRoleAccess — stratégie système
Une stratégie personnalisée accordant l'accès à KMS :

{
    "Version": "1",
    "Statement": [
        {
            "Effect": "Allow",
            "Action": [
                "kms:GetSecretValue",
                "kms:GetRandomPassword",
                "kms:Decrypt",
                "kms:GenerateDataKey",
                "kms:PutSecretValue",
                "kms:UpdateSecretVersionStage"
            ],
            "Resource": "*"
        }
    ]
}

Étape 2 : Créer les secrets génériques

Créez deux secrets génériques dans la console KMS.

Secret du compte privilégié — utilisé pour créer et gérer les comptes réguliers :

{
    "Endpoint": "<mysql-host>",
    "AccountName": "<privileged-account-name>",
    "AccountPassword": "<privileged-account-password>",
    "SSL": false
}

Champ	Description	Par défaut
`Endpoint`	Nom de domaine ou adresse IP de la base de données MySQL	—
`AccountName`	Nom d'utilisateur du compte privilégié	—
`AccountPassword`	Mot de passe du compte privilégié	—
`SSL`	Indique s'il faut utiliser un certificat SSL pour la connexion	`false`

Secret du compte régulier — le secret qui fera l'objet d'une rotation. Son nom est transmis en tant que paramètre de planification à CloudFlow :

{
    "Endpoint": "<mysql-host>",
    "AccountName": "<account-name>",
    "AccountPassword": "<account-password>",
    "MasterSecret": "<privileged-secret-name>",
    "SSL": false
}

Champ	Description	Par défaut
`Endpoint`	Nom de domaine ou adresse IP de la base de données MySQL	—
`AccountName`	Nom d'utilisateur du compte à faire pivoter	—
`AccountPassword`	Mot de passe du compte à faire pivoter	—
`MasterSecret`	Nom du secret du compte privilégié dans KMS	—
`SSL`	Indique s'il faut utiliser un certificat SSL pour la connexion	`false`

Lorsque SSL est défini sur true , la fonction attend le certificat SSL à l'emplacement /opt/python/certs/cert.pem . Ajoutez le certificat à une couche personnalisée (voir Étape 3).

Étape 3 : Créer la fonction de rotation

Créez un service Function Compute avec l'accès VPC activé (même configuration que le Scénario 1). Assurez-vous que la fonction peut atteindre l'instance MySQL et le point de terminaison VPC KMS.
Créez une fonction avec l'Runtime Environments défini sur Python 3.9. Utilisez le code suivant :

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

try:
    import pymysql
except:
    os.system('pip install pymysql -t ./')
    import pymysql
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
from aliyunsdkrds.request.v20140815.DescribeDBInstancesRequest import DescribeDBInstancesRequest

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']
    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, secret_name, version_id)

    elif step == "test":
        test_phase(client, 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
        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, secret_name, version_id):
    current_dict = get_secret_dict(client, secret_name, "ACSCurrent")
    pending_dict = get_secret_dict(client, secret_name, "ACSPending", version_id)

    conn = get_connection(pending_dict)
    if conn:
        conn.close()
        logger.info(
            "set: ACSPending secret is already set as password in MySQL DB for secret secret_name %s." % secret_name)
        return

    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']))

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

    conn = get_connection(current_dict)
    if not conn:
        logger.error("set: Unable to access the given database using current credentials for secret %s" % secret_name)
        raise ValueError("Unable to access the given database using current credentials for secret %s" % secret_name)
    conn.close()

    master_secret = current_dict['MasterSecret']
    master_dict = get_secret_dict(client, master_secret, "ACSCurrent")
    if current_dict['Endpoint'] != master_dict['Endpoint'] and not is_rds_replica_database(current_dict, master_dict):
        logger.error("set: Current database Endpoint %s is not the same Endpoint as/rds replica of master %s" % (
            current_dict['Endpoint'], master_dict['Endpoint']))
        raise ValueError("Current database Endpoint %s is not the same Endpoint as/rds replica of master %s" % (
            current_dict['Endpoint'], master_dict['Endpoint']))

    conn = get_connection(master_dict)
    if not conn:
        logger.error(
            "set: Unable to access the given database using credentials in master secret secret %s" % master_secret)
        raise ValueError("Unable to access the given database using credentials in master secret secret %s" % master_secret)

    try:
        with conn.cursor() as cur:
            cur.execute("SELECT User FROM mysql.user WHERE User = %s", pending_dict['AccountName'])
            if cur.rowcount == 0:
                cur.execute("CREATE USER %s IDENTIFIED BY %s",
                            (pending_dict['AccountName'], pending_dict['AccountPassword']))

            cur.execute("SHOW GRANTS FOR %s", current_dict['AccountName'])
            for row in cur.fetchall():
                if 'XA_RECOVER_ADMIN' in row[0]:
                    continue
                grant = row[0].split(' TO ')
                new_grant_escaped = grant[0].replace('%', '%%')  # % is a special character in Python format strings.
                cur.execute(new_grant_escaped + " TO %s ", (pending_dict['AccountName'],))
            cur.execute("SELECT VERSION()")
            ver = cur.fetchone()[0]

            escaped_encryption_statement = get_escaped_encryption_statement(ver)
            cur.execute("SELECT ssl_type, ssl_cipher, x509_issuer, x509_subject FROM mysql.user WHERE User = %s",
                        current_dict['AccountName'])
            tls_options = cur.fetchone()
            ssl_type = tls_options[0]
            if not ssl_type:
                cur.execute(escaped_encryption_statement + " NONE", pending_dict['AccountName'])
            elif ssl_type == "ANY":
                cur.execute(escaped_encryption_statement + " SSL", pending_dict['AccountName'])
            elif ssl_type == "X509":
                cur.execute(escaped_encryption_statement + " X509", pending_dict['AccountName'])
            else:
                cur.execute(escaped_encryption_statement + " CIPHER %s AND ISSUER %s AND SUBJECT %s",
                            (pending_dict['AccountName'], tls_options[1], tls_options[2], tls_options[3]))

            password_option = get_password_option(ver)
            cur.execute("SET PASSWORD FOR %s = " + password_option,
                        (pending_dict['AccountName'], pending_dict['AccountPassword']))
            conn.commit()
            logger.info("set: Successfully changed password for %s in MySQL DB for secret secret_name %s." % (
                pending_dict['AccountName'], secret_name))
    finally:
        conn.close()

def test_phase(client, secret_name, version_id):
    conn = get_connection(get_secret_dict(client, secret_name, "ACSPending", version_id))
    if conn:
        try:
            with conn.cursor() as cur:
                cur.execute("SELECT NOW()")
                conn.commit()
        finally:
            conn.close()

        logger.info("test: Successfully accessed into MySQL DB with ACSPending secret in %s." % secret_name)
        return
    else:
        logger.error(
            "test: Unable to access the given database with pending secret of secret secret_name %s" % secret_name)
        raise ValueError("Unable to access the given database with pending secret of secret secret_name %s" % secret_name)

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 update ACSCurrent stage to version %s for secret %s." % (version_id, secret_name))

def get_connection(secret_dict):
    port = int(secret_dict['Port']) if 'Port' in secret_dict else 3306
    dbname = secret_dict['DBName'] if 'DBName' in secret_dict else None

    use_ssl, fall_back = get_ssl_config(secret_dict)

    conn = connect_and_authenticate(secret_dict, port, dbname, use_ssl)
    if conn or not fall_back:
        return conn
    else:
        return connect_and_authenticate(secret_dict, port, dbname, False)

def get_ssl_config(secret_dict):
    if 'SSL' not in secret_dict:
        return True, True

    if isinstance(secret_dict['SSL'], bool):
        return secret_dict['SSL'], False

    if isinstance(secret_dict['SSL'], str):
        ssl = secret_dict['SSL'].lower()
        if ssl == "true":
            return True, False
        elif ssl == "false":
            return False, False
        else:
            return True, True

    return True, True

def connect_and_authenticate(secret_dict, port, dbname, use_ssl):
    ssl = {'ca': '/opt/python/certs/cert.pem'} if use_ssl else None

    try:
        conn = pymysql.connect(host=secret_dict['Endpoint'], user=secret_dict['AccountName'],
                               password=secret_dict['AccountPassword'],
                               port=port, database=dbname, connect_timeout=5, ssl=ssl)
        logger.info("Successfully established %s connection as user '%s' with Endpoint: '%s'" % (
            "SSL/TLS" if use_ssl else "non SSL/TLS", secret_dict['AccountName'], secret_dict['Endpoint']))
        return conn
    except pymysql.OperationalError as e:
        if 'certificate verify failed: IP address mismatch' in e.args[1]:
            logger.error(
                "Hostname verification failed when estlablishing SSL/TLS Handshake with Endpoint: %s" % secret_dict[
                    'Endpoint'])
        return None

def get_secret_dict(client, secret_name, stage, version_id=None):
    required_fields = ['Endpoint', '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 rotate user, account_name length with _rotation appended would exceed 16 characters")
        return new_account_name

def get_password_option(version):
    if version.startswith("8"):
        return "%s"
    else:
        return "PASSWORD(%s)"

def get_escaped_encryption_statement(version):
    if version.startswith("5.6"):
        return "GRANT USAGE ON *.* TO %s@'%%' REQUIRE"
    else:
        return "ALTER USER %s@'%%' REQUIRE"

def is_rds_replica_database(client, replica_dict, master_dict):
    replica_instance_id = replica_dict['Endpoint'].split(".")[0].replace('io', '')
    master_instance_id = master_dict['Endpoint'].split(".")[0].replace('io', '')
    try:
        describe_response = describe_db_instances(client, replica_instance_id)
    except Exception as err:
        logger.warning("Encountered error while verifying rds replica status: %s" % err)
        return False
    items = describe_response['Items']
    instances = items.get("DBInstance")
    if not instances:
        logger.info("Cannot verify replica status - no RDS instance found with identifier: %s" % replica_instance_id)
        return False

    current_instance = instances[0]
    return master_instance_id == current_instance.get('DBInstanceId')

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 describe_db_instances(client, db_instance_id):
    request = DescribeDBInstancesRequest()
    request.set_accept_format('json')
    request.set_DBInstanceId(db_instance_id)
    response = client.do_action_with_exception(request)
    return json.loads(response)

Créez une couche personnalisée pour empaqueter la dépendance PyMySQL (et éventuellement le certificat SSL) :
1. Exécutez les commandes suivantes pour construire le package de la couche :
```
mkdir my-secret-rotate
cd my-secret-rotate
pip install --target ./python pymysql
# If SSL is required, create the certs directory and add your certificate:
# mkdir python/certs && cp cert.pem python/certs/
zip -r my-secret-rotate.zip python
```
2. Dans la console Function Compute, créez une couche personnalisée en utilisant le package ZIP. Consultez la page Créer une couche personnalisée.
Attachez la couche personnalisée à votre fonction. Sur la page Services, cliquez sur Functions pour votre service, localisez la fonction, cliquez sur Configure, puis ajoutez la couche dans la section Layers. Consultez la page Gérer les couches.

Étape 4 : Créer le flux de rotation CloudFlow

Connectez-vous à la console CloudFlow et sélectionnez la même région que vos fonctions.
Sur la page Flows, cliquez sur Create flow.
Sur la page Create Flow, cliquez sur Create Flow with Code, définissez le YAML de la section Definition comme suit, puis cliquez sur Next Step :

version: v1
type: flow
steps:
  - type: task
    name: RotateSecretNew
    resourceArn: <function-arn>
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: Step
        source: new
  - type: task
    name: RotateSecretSet
    resourceArn: <function-arn>
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: Step
        source: set
      - target: VersionId
        source: $local.VersionId
  - type: task
    name: RotateSecretTest
    resourceArn: <function-arn>
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: Step
        source: test
      - target: VersionId
        source: $local.VersionId
  - type: task
    name: RotateSecretEnd
    resourceArn: <function-arn>
    inputMappings:
      - target: SecretName
        source: $input.payload.SecretName
      - target: RegionId
        source: $input.payload.RegionId
      - target: Step
        source: end
      - target: VersionId
        source: $local.VersionId

Remplacez <function-arn> par l'ARN de la fonction que vous avez créée.

Configurez un rôle de flux, puis cliquez sur Create Flow.
Pour planifier la rotation automatique, créez une planification basée sur le temps et définissez le champ Payload comme suit :

{
    "SecretName": "<secret-name>",
    "RegionId": "<region-id>"
}

Champ	Description
`SecretName`	Nom du secret générique du compte régulier dans KMS
`RegionId`	Région où le secret et la fonction sont déployés

Étape 5 : Vérifier la rotation

Déclenchez manuellement une exécution pour confirmer la configuration :

Dans la console CloudFlow, ouvrez votre flux et cliquez sur Start Execution en utilisant le payload JSON de l'étape 4.
Attendez la fin de l'exécution. Les quatre étapes (RotateSecretNew, RotateSecretSet, RotateSecretTest, RotateSecretEnd) doivent afficher l'état Succeeded.
Dans la console KMS, ouvrez le secret et vérifiez son historique de versions. L'étiquette ACSCurrent doit pointer vers une nouvelle version.

Étape 6 : (Facultatif) Connecter les applications à Secrets Manager

Utilisez Secrets Manager JDBC pour récupérer automatiquement les identifiants ayant fait l'objet d'une rotation. Consultez la page Secrets Manager JDBC.

Étapes suivantes

Secrets Manager JDBC — intégrez les identifiants rotatifs aux applications basées sur JDBC
Présentation des secrets dynamiques ApsaraDB RDS — envisagez les secrets dynamiques si votre base de données est hébergée sur ApsaraDB RDS
Qu'est-ce que Function Compute ? — découvrez davantage le service de calcul serverless utilisé dans ce guide