如果您需要輪轉普通憑據以增強敏感信息的安全性，可以通過函數計算服務來實現。本文介紹如何借助函數計算快速構建普通憑據輪轉功能。
背景信息
函數計算是事件驅動的全托管計算服務。使用函數計算，您無需采購和管理服務器等基礎設施，只需編寫并上傳代碼。函數計算為您準備好計算資源，幫助您彈性可靠地運行任務，并提供日志查詢、性能監控和報警等功能。更多信息，請參見什么是函數計算。
計費說明
使用函數計算輪轉普通憑據時，密鑰管理服務不會收取憑據輪轉費用，但函數計算服務以及函數計算服務中集成的Serverless工作流會產生費用。關于費用詳情，請參見函數計算計費說明和Serverless計費說明。
輪轉操作流程
下文以兩個場景為例，介紹如何輪轉普通憑據。
使用阿里云RDS OpenAPI輪轉RDS雙賬號賬密憑據
使用高權限賬號輪轉用戶自建MySQL數據庫雙賬號賬密憑據
使用阿里云RDS OpenAPI輪轉RDS雙賬號賬密憑據
通過函數計算集成Serverless工作流實現普通憑據輪轉功能。
1.輪轉憑據準備工作
開通函數計算服務。具體操作，請參見開通函數計算服務。
創建一個阿里云RDS數據庫賬號及密碼（輪轉程序首次輪轉會自動創建另外一個賬號及密碼），且在密鑰管理服務控制臺創建該普通憑據。具體操作，請參見管理憑據。
憑據值為JSON格式，憑據內容如下。該憑據名稱后續會作為Serverless定時調度參數傳入。
{ "AccountName": "", "AccountPassword": "" }說明其中AccountName表示數據庫賬號，AccountPassword表示數據庫密碼。
創建函數計算普通服務角色，并授權訪問專屬KMS。具體操作，請參見授予函數計算訪問其他云服務的權限。
通過精確授權，添加函數計算服務角色授權策略AliyunFCDefaultRolePolicy。
添加系統策略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.創建輪轉憑據函數
您需要在函數計算控制臺創建服務以及輪轉憑據函數。具體操作，請參見快速創建函數。
創建函數計算的服務。
您在創建服務時需要單擊顯示高級選項，設置允許訪問 VPC為是，并選擇專有網絡、交換機以及安全組。確保您的函數計算服務在指定的安全組下通過指定VPC中的交換機訪問RDS OpenAPI和密鑰管理服務。
創建函數。
本文以Python為例，您在創建函數時運行環境選擇Python3.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工作流
創建憑據輪轉流程。具體操作，請參見創建流程。
在頂部菜單欄，選擇地域。
重要選擇的地域需和創建函數的地域相同。
在流程頁面，單擊創建流程。
在創建流程頁面，單擊使用代碼段創作，填寫相關信息，單擊下一步。
您需要將流程定義的YAML內容修改為以下內容。
version: v1 type: flow steps: - type: task name: RotateSecretNew resourceArn: 已創建的函數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: 已創建的函數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: 已創建的函數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配置流程角色。
在創建流程頁面，單擊創建流程。
定時輪轉憑據。具體操作，請參見創建定時調度。
通過Serverless工作流實現普通憑據定時輪轉，觸發消息需要配置為如下內容。
{ "SecretName": "", "RegionId": "", "InstanceId":"" }說明其中SecretName表示憑據名稱，RegionId表示地域，InstanceId表示RDS實例ID。
（可選）4.應用接入
您可以使用憑據管家JDBC客戶端完成應用接入。具體操作，請參見憑據管家JDBC客戶端。
使用高權限賬號輪轉用戶自建MySQL數據庫雙賬號賬密憑據
通過函數計算集成Serverless工作流實現普通憑據輪轉功能。
1.輪轉憑據準備工作
開通函數計算服務。具體操作，請參見開通函數計算服務。
創建一個高級別權限的MySQL數據庫賬號及密碼（用于創建和修改數據庫賬號密碼），并在密鑰管理服務控制臺創建該普通憑據。具體操作，請參見管理憑據。
憑據值為JSON格式，憑據內容如下。
{ "Endpoint": "", "AccountName": "", "AccountPassword": "", "SSL":false }說明其中Endpoint表示數據庫連接域名或地址，AccountName表示數據庫賬號，AccountPassword表示數據庫密碼，SSL表示是否使用證書（取值為true或false，默認為false。證書默認存放路徑為/opt/python/certs/cert.pem）。
創建一個MySQL數據庫賬號及密碼（輪轉程序首次輪轉會自動創建另外一個賬號及密碼），并在密鑰管理服務控制臺創建該普通憑據。具體操作，請參見管理憑據。
憑據值為JSON格式，憑據內容如下。該憑據名稱會作為Serverless定時調度參數傳入。
{ "Endpoint": "", "AccountName": "", "AccountPassword": "", "MasterSecret":"", "SSL":false }說明其中Endpoint表示數據庫連接域名或地址，AccountName表示數據庫賬號，AccountPassword表示數據庫密碼，MasterSecret表示高級別權限的MySQL賬號憑據，SSL表示是否使用證書（取值為true或false，默認為false。證書默認存放路徑為/opt/python/certs/cert.pem）。
創建普通服務角色，并授權函數計算訪問專屬KMS。具體操作，請參見授予函數計算訪問其他云服務的權限。
通過精確授權，添加函數計算服務角色授權策略AliyunFCDefaultRolePolicy。
添加系統策略AliyunSTSAssumeRoleAccess。
添加訪問KMS的權限策略（該策略為自定義權限策略），配置如下。
{ "Version": "1", "Statement": [ { "Effect": "Allow", "Action": [ "kms:GetSecretValue", "kms:GetRandomPassword", "kms:Decrypt", "kms:GenerateDataKey", "kms:PutSecretValue", "kms:UpdateSecretVersionStage" ], "Resource": "*" } ] }
2.創建輪轉憑據函數
創建函數計算的服務。具體操作，請參見快速創建函數。
您在創建服務時需要單擊顯示高級選項，設置允許訪問 VPC為是，并選擇專有網絡、交換機以及安全組。確保您的函數計算服務在指定的安全組下通過指定VPC中的交換機訪問RDS OpenAPI和密鑰管理服務。
創建函數。具體操作，請參見快速創建函數。
本文以Python為例，您在創建函數時運行環境選擇Python3.9，示例代碼如下。（您可以根據實際應用情況修改）
# -*- 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 cha30racter 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)創建自定義層。具體操作，請參見創建自定義層。
通過自定義層可避免頻繁安裝依賴。本文以Python為例，代碼示例中需要添加pymysql依賴，如果MySQL數據庫使用SSL證書，也支持在層中添加SSL證書。
構建層的ZIP包。
執行
mkdir my-secret-rotate命令創建工作目錄。執行
cd my-secret-rotate命令進入已創建的工作目錄。執行
pip install --target ./python pymysql命令安裝依賴庫到my-secret-rotate/python。如果MySQL數據庫使用SSL證書，您需要在python目錄下添加certs目錄，并將證書文件cert.pem放入certs目錄下。
在my-secret-rotate目錄，執行
zip -r my-secret-rotate.zip python命令打包依賴。
通過函數計算控制臺創建自定義層。
在函數中配置自定義層。具體操作，請參見在函數中配置自定義層。
在服務與函數頁簽單擊目標服務操作列的函數管理。
在函數管理頁簽，單擊目標函數操作列的配置，在層區域為函數添加自定義層。
3.創建輪轉憑據Serverless工作流
創建憑據輪轉流程。具體操作，請參見創建流程。
在頂部菜單欄，選擇地域。
重要選擇的地域需和創建函數的地域相同。
在流程頁面，單擊創建流程。
在創建流程頁面，單擊使用代碼段創作，填寫相關信息，單擊下一步。
您需要將流程定義的YAML內容修改為以下內容。
version: v1 type: flow steps: - type: task name: RotateSecretNew resourceArn: 已創建的函數arn inputMappings: - target: SecretName source: $input.payload.SecretName - target: RegionId source: $input.payload.RegionId - target: Step source: new - type: task name: RotateSecretSet resourceArn: 已創建的函數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: 已創建的函數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: 已創建的函數arn inputMappings: - target: SecretName source: $input.payload.SecretName - target: RegionId source: $input.payload.RegionId - target: Step source: end - target: VersionId source: $local.VersionId配置流程角色。
在創建流程頁面，單擊創建流程。
定時輪轉憑據。具體操作，請參見創建定時調度。
通過Serverless工作流實現普通憑據定時輪轉，觸發消息需要配置為如下內容。
{ "SecretName": "", "RegionId": "" }說明其中SecretName表示憑據名稱，RegionId表示地域。
（可選）4.應用接入
您可以使用憑據管家JDBC客戶端完成應用接入。具體操作，請參見憑據JDBC客戶端。
文檔內容是否對您有幫助？