目次
AWS Secrets Managerとは
AWS Secrets Managerは、AWSのマネージドサービスで、機密情報やAPIキーなどの秘密情報を安全に管理します。シークレット情報の安全な保管と自動ローテーション、AWSリソースとのシームレスな統合、アクセス管理と監査、そしてAPIやSDKを提供しています。これにより、セキュリティを強化し、アプリケーションの機密情報を効果的に管理できます。
今回すること
サードパーティのDBであるTiDBをAWS Lambdaで使用するため、TiDBのシークレット情報を管理し、実際にLambdaでどう動かすかの例を紹介したいと思います。
記事の構成
1. CDK実行環境にシークレット情報の環境変数を登録
2. CloudFormation(CDK)を実行してAWSリソースを作成
3. Lambda実行時、SDKでSecretsManagerからシークレット情報を取得、TiDBへアクセスする
1. CDK実行環境にシークレット情報の環境変数を登録
以下のようなRDBMSのシークレット情報を環境変数に登録します
RDBMS_HOST_NAME
RDBMS_USER_NAME
RDBMS_PASSWARD
RDBMS_REGION
RDBMS_PORT
RDBMS_DATABASE_NAME2.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は使用できませんでした)
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へアクセスします
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ではパスワードの自動ローテーションの設定なども出来るので今後も上手く付き合っていきたいサービスの一つですね