Gunakan Function Compute untuk merotasi rahasia generik - Key Management Service

Anda dapat menggunakan Function Compute untuk merotasi rahasia generik guna meningkatkan keamanan informasi sensitif. Topik ini menjelaskan cara menggunakan Function Compute untuk merotasi rahasia generik.

Informasi latar belakang

Function Compute adalah layanan komputasi berbasis acara yang sepenuhnya dikelola. Dengan Function Compute, Anda dapat fokus pada penulisan kode tanpa perlu memperoleh atau mengelola sumber daya infrastruktur seperti server. Cukup tulis dan unggah kode Anda, lalu Function Compute akan mengalokasikan sumber daya komputasi, menjalankan tugas secara elastis dan andal, serta menyediakan fitur seperti log kueri, pemantauan kinerja, dan peringatan. Untuk informasi lebih lanjut, lihat Apa itu Function Compute?.

Penagihan

Jika Anda menggunakan Function Compute untuk merotasi rahasia generik, Anda tidak dikenakan biaya untuk rotasi rahasia yang disediakan oleh Key Management Service (KMS). Namun, Anda akan dikenakan biaya untuk Function Compute dan CloudFlow yang diintegrasikan ke dalam Function Compute. Untuk informasi lebih lanjut tentang penagihan, lihat Penagihan Function Compute dan Penagihan CloudFlow.

Penting

Secrets Manager menyediakan rahasia dinamis yang secara otomatis dirotasi. Jika rahasia dinamis dapat memenuhi persyaratan keamanan Anda, kami sarankan Anda menggunakan rahasia dinamis untuk mengurangi biaya. Untuk informasi lebih lanjut, lihat Ikhtisar Rahasia Dinamis ApsaraDB RDS, Ikhtisar, atau Ikhtisar Rahasia Dinamis ECS.

Proses rotasi rahasia

Topik ini menjelaskan cara merotasi rahasia generik dalam skenario berikut:

Rotasi rahasia generik database ApsaraDB RDS dalam mode dua akun menggunakan API ApsaraDB RDS.
Rotasi rahasia generik database MySQL yang dikelola sendiri dalam mode dua akun menggunakan akun istimewa.

Rotasi rahasia generik database ApsaraDB RDS dalam mode dua akun menggunakan API ApsaraDB RDS

Anda dapat menggunakan CloudFlow dan Function Compute untuk merotasi rahasia generik.

1. Persiapkan untuk rotasi rahasia

Aktifkan Function Compute. Untuk informasi lebih lanjut, lihat Aktifkan Function Compute.
Buat akun yang digunakan untuk masuk ke database ApsaraDB RDS Anda. Kemudian, buat rahasia generik untuk akun tersebut di konsol KMS. Jika rahasia generik dirotasi untuk pertama kalinya, akun baru akan dibuat.
Nilai rahasia dalam format JSON. Contoh: Nama rahasia digunakan sebagai parameter penjadwalan yang akan diteruskan ke CloudFlow.
```
{
   "AccountName": "",
 "AccountPassword": ""
}
```
Catatan
AccountName menentukan nama pengguna akun yang digunakan untuk masuk ke database ApsaraDB RDS Anda, dan AccountPassword menentukan kata sandi akun yang digunakan untuk masuk ke database ApsaraDB RDS Anda.

Buat peran layanan normal untuk Function Compute dan berikan izin untuk mengizinkan Function Compute mengakses KMS. Untuk informasi lebih lanjut, lihat Berikan Izin Function Compute untuk Mengakses Layanan Alibaba Cloud Lainnya.

Lampirkan kebijakan AliyunFCDefaultRolePolicy ke peran layanan normal Function Compute.
Lampirkan kebijakan sistem AliyunSTSAssumeRoleAccess ke peran layanan normal.

Lampirkan kebijakan kustom yang mengizinkan akses ke KMS ke peran layanan normal. Skrip berikut menunjukkan isi dari kebijakan:

{
    "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. Buat fungsi rotasi rahasia

Anda harus membuat layanan Function Compute dan fungsi rotasi rahasia di konsol Function Compute. Untuk informasi lebih lanjut, lihat Buat Fungsi dengan Cepat.

Buat layanan Function Compute.
Saat membuat layanan Function Compute, klik Show Advanced Options dan atur Access to VPC menjadi Yes. Kemudian, konfigurasikan VPC, vSwitch, dan Security Group. Pastikan bahwa layanan Function Compute dapat mengakses API ApsaraDB RDS dan KMS menggunakan grup keamanan dan vSwitch tertentu di VPC tertentu.

Buat fungsi.

Dalam contoh ini, Python digunakan. Saat membuat fungsi, pilih Python 3.9 untuk Runtime Environments. Kode berikut memberikan contoh. Anda dapat memodifikasi kode sesuai dengan kebutuhan bisnis Anda.

# -*- 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. Buat alur CloudFlow untuk rotasi rahasia

Buat alur rotasi rahasia. Untuk informasi lebih lanjut, lihat Buat Alur Kerja.

Masuk ke konsol CloudFlow.
Di bilah navigasi atas, pilih wilayah tempat Anda ingin membuat alur.
Penting
Anda harus memilih wilayah tempat fungsi dibuat.
Di halaman Flows, klik Create flow.

Di halaman Create Flow, klik Create Flow with Code, konfigurasikan parameter, lalu klik Next Step.

Anda harus mengubah konten dalam file YAML Definition menjadi kode berikut:

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

Konfigurasikan peran alur.
Klik Create Flow.

Konfigurasikan pengaturan untuk merotasi rahasia generik pada waktu terjadwal. Untuk informasi lebih lanjut, lihat Buat Jadwal Berbasis Waktu.
Jika Anda ingin menggunakan CloudFlow untuk mengonfigurasi rotasi terjadwal rahasia generik, atur Payload menjadi konten berikut:
```
{
  "SecretName": "",
  "RegionId": "",
  "InstanceId":""
}
```
Catatan
SecretName menentukan nama rahasia, RegionId menentukan wilayah, dan InstanceId menentukan ID instance ApsaraDB RDS Anda.

4. (Opsional) Hubungkan aplikasi ke Secrets Manager

Anda dapat menggunakan Secrets Manager JDBC untuk menghubungkan aplikasi ke Secrets Manager. Untuk informasi lebih lanjut, lihat Secrets Manager JDBC.

Rotasi rahasia generik database MySQL yang dikelola sendiri dalam mode dua akun menggunakan akun istimewa