如果您需要轮转普通凭据以增强敏感信息的安全性,可以通过函数计算服务来实现。本文介绍如何借助函数计算快速构建普通凭据轮转功能。
背景信息
函数计算是事件驱动的全托管计算服务。使用函数计算,您无需采购和管理服务器等基础设施,只需编写并上传代码。函数计算为您准备好计算资源,帮助您弹性可靠地运行任务,并提供日志查询、性能监控和报警等功能。更多信息,请参见什么是函数计算。
计费说明
使用函数计算轮转普通凭据时,密钥管理服务不会收取凭据轮转费用,但函数计算服务以及函数计算服务中集成的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客户端。