サードパーティDBのシークレット情報をAWS SecretsManagerで管理する

February 5, 2024

#AWS

#CDK

#Python

#SecretsManager

AWS Secrets Managerとは

AWS Secrets Managerは、AWSのマネージドサービスで、機密情報やAPIキーなどの秘密情報を安全に管理します。シークレット情報の安全な保管と自動ローテーション、AWSリソースとのシームレスな統合、アクセス管理と監査、そしてAPIやSDKを提供しています。これにより、セキュリティを強化し、アプリケーションの機密情報を効果的に管理できます。

今回すること

サードパーティのDBであるTiDBをAWS Lambdaで使用するため、TiDBのシークレット情報を管理し、実際にLambdaでどう動かすかの例を紹介したいと思います。

イメージ図1

使用するサービス・ツール

AWS CDK
AWS Secrets Manager
AWS Lambda

記事の構成

CDK実行環境にシークレット情報の環境変数を登録
CloudFormation(CDK)を実行してAWSリソースを作成

AWS Sercrets Manager
AWS Lambda
LambdaでSecrets Managerを操作するためのIAMロール

Lambda実行時、SDKでSecretsManagerからシークレット情報を取得、TiDBへアクセスする

1.CDK実行環境にシークレット情報の環境変数を登録

以下のようなRDBMSのシークレット情報を環境変数に登録します

RDBMS_HOST_NAME
RDBMS_USER_NAME
RDBMS_PASSWARD
RDBMS_REGION
RDBMS_PORT
RDBMS_DATABASE_NAME

2.CloudFormation(CDK)を実行してAWSリソースを作成

CDKでSecretsManagerリソースを定義

シークレットの識別子となる値としてsecret_nameと登録したいシークレット情報をここで定義します

import os

from aws_cdk import SecretValue
from aws_cdk import aws_secretsmanager as secretsmanager
from constructs import Construct

rdbms_host_name = os.getenv("RDBMS_HOST_NAME", "rdbms_host_name")
rdbms_user_name = os.getenv("RDBMS_USER_NAME", "rdbms_user_name")
rdbms_passward = os.getenv("RDBMS_PASSWARD", "rdbms_passward")
rdbms_region = os.getenv("RDBMS_REGION", "rdbms_region")
rdbms_port = os.getenv("RDBMS_PORT", "rdbms_port")
rdbms_database_name = os.getenv("RDBMS_DATABASE_NAME", "rdbms_database_name")


class RdbmsSecretsmanagerResource(Construct):
    def __init__(self, scope: Construct, id: str, secret_name):
        super().__init__(scope, id)

        self.secret: secretsmanager.Secret = secretsmanager.Secret(
            self,
            "RdbmsTokenSecret",
            secret_object_value={
                "RDBMS_HOST_NAME": SecretValue.unsafe_plain_text(rdbms_host_name),
                "RDBMS_USER_NAME": SecretValue.unsafe_plain_text(rdbms_user_name),
                "RDBMS_PASSWARD": SecretValue.unsafe_plain_text(rdbms_passward),
                "RDBMS_REGION": SecretValue.unsafe_plain_text(rdbms_region),
                "RDBMS_PORT": SecretValue.unsafe_plain_text(rdbms_port),
                "RDBMS_DATABASE_NAME": SecretValue.unsafe_plain_text(
                    rdbms_database_name
                ),
            },
            secret_name=secret_name,
        )

CDKでLambdaを定義(SecretsManagerを連携)

SecretsManagerのsecret_full_arnをLambdaの環境変数へ登録

注）secret.secret_nameだと余計なsuffixがついてしまい、後続のSecretIdに使えないようでしたのでsecret_full_arnを使用しました。secretから取らずに直接上記で設定したsecret_nameを使用しても動きますが、secretから取る方がコードがスッキリするのでsecret_full_arnを使用しています。

lambda_function = lambda_.Function(
    self,
    "TestFunction",
    environment={
      "RDBMS_SECRET_ARN": secret.secret_full_arn
    },
    ...
)

SecretsManagerを操作する権限をLambdaのロールへ付与

grant_readメソッドを使うと読み込み権限を簡単に付与できる

secret.grant_read(lambda_function)

3.Lambda実行時、SDKでSecretsManagerからシークレット情報を取得、TiDBへアクセスする

ここのSecretIdではsecret_arnかsecret_nameを使用できます(suffix付きのsecret_nameは使用できませんでした)

secretsmanager_module.py

class SecretsmanagerModule:
    def __init__(self, secretsmanager_client, secret_id) -> None:
        self.secretsmanager_client = secretsmanager_client
        self.secret_id = secret_id

    def get_secret_value(self):
        response = self.secretsmanager_client.get_secret_value(SecretId=self.secret_id)
        return response["SecretString"]

取得したSecretStringはjson文字列になっているのでjson.loadsでパースする必要があります。

今回はsqlalchemyを使用してDBへアクセスします

rdbms_module.py

import json
import os

import boto3
from secretsmanager_module import SecretsmanagerModule
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker


RDBMS_DATABASE_NAME = os.environ.get("RDBMS_DATABASE_NAME")

secretsmanager_client = boto3.client(
    "secretsmanager", region_name=os.environ["AWS_REGION"]
)

secret = json.loads(
    SecretsmanagerModule(
        secretsmanager_client=secretsmanager_client,
        secret_id=os.environ["RDBMS_SECRET_ARN"],
    ).get_secret_value()
)


class RdbmsModule:
    def __init__(self):
        engine = create_engine(
            f"mysql+mysqlconnector://{secret['RDBMS_USER_NAME']}:"
            + f"{secret['RDBMS_PASSWARD']}@{secret['RDBMS_HOST_NAME']}:"
            + f"{secret['RDBMS_PORT']}/{RDBMS_DATABASE_NAME}"
        )
        self.Session = sessionmaker(bind=engine)

    def create_session(self):
        return self.Session()

上記で生成したsessionでusersテーブルからuserリストを取得します

from rdbms.rdbms_module import RdbmsModule

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    age = Column(Integer)

class GetUserProcessor:
    def __init__(self) -> None:
        self.rdbms_module = RdbmsModule()

    def get_user_list(self, parameter):
        session = self.rdbms_module.create_session()
        try:
            users = session.query(User).all()
            return users
        except Exception as e:
            session.rollback()
            raise e
        finally:
            session.close()

まとめ

今回はCDKでSecretsManagerにシークレットを登録する方法、Lambdaで登録したシークレットを取得しDBへアクセスした例を紹介しました

SecretsManagerではパスワードの自動ローテーションの設定なども出来るので今後も上手く付き合っていきたいサービスの一つですね

← Back to home