今回もAWSでの疎通確認についてです。タイトルにあるようにEC2またはECSコンテナからRDS Proxy経由でAurora PostgreSQLに接続できるかの疎通確認をpythonで行った備忘録になります。
今回は前に説明した以下の記事のElastiCache Redisの疎通確認以上に前提が複雑なので、同じコードを実行する場合はしっかり確認した上で行ってください。
AWSリソースの作り方などは説明しませんが、より詳細な情報については参考資料に挙げているページ様を参照してください。
では、始めます。
０：前提
今回の前提は以下のようになります。前提条件が多いですが、頑張ってAWSリソース構築を行ってください
- DB接続情報(dbname, port, username, password)がAWS Secretsに登録済み
- 疎通元のEC2またはECSコンテナがAWS Secretsにアクセス可能
- RDS ProxyのTLS接続が有効
- RDS ProxyはIAM認証
- Aurora PostgreSQLは通常のユーザ名/パスワード認証
- その他AWSリソースは全て正常に作成されている
１：実行環境構築
前の記事と同じくEC2ならAnsible、ECSならDockerfileによるイメージ作成で、まずはpythonとpipの環境を構築します。
そしてpipを使って以下の2つをインストールします。
pip install psycopg2-binary pip install boto3
次にTLS接続を行うために証明書を用意します。
適切な証明書がない場合は以下のAmazonのページにある「Root CAs」の表の一番上の行の「Self-Signed Certificate」の「PEM」をクリックしてダウンロードしたものを使用してください。
用意した証明書をEC2またはECSコンテナにアップロードします。
これで実行環境の構築は終了です。
２：疎通確認コード
実行環境ができたら以下のファイルをコピペし、証明書ファイルパス、RDS Proxyエンドポイント、シークレット名、リージョン名をそれぞれ適切なものに書き換えて保存します。
・connect_db_via_proxy.py
# -*- coding: utf-8 -*- import psycopg2 import boto3 import json def main(): # 以下の4つの変数を適切なものに書き換える sslcert_path = "./xxxxxx.pem" # SSL証明書ファイルパス db_endpoint = "xxxxxxxxxxxxxxx.rds.amazonaws.com" # RDS Proxyエンドポイント secret_name = "db-secrets" # DB情報のシークレット名 region_name = "ap-northeast-1" # リージョン名 # Secrets Manager から認証情報を取得 db_credentials = get_secret(secret_name, region_name) # データベース接続情報 db_port = db_credentials["port"] # ポート番号 db_name = db_credentials["dbname"] # データベース名 username = db_credentials["username"] # ユーザー名 # 取得したDB情報を表示 print(db_name) print(username) print(db_port) print("Get DB Secrets.") # IAM認証用にトークンを作成 token = get_iam_token(db_endpoint, db_port, username, region_name) print(token) print("Create token.") # データベースに接続 connection = psycopg2.connect( sslmode="require", # RDS ProxyのTLS接続を有効にする sslrootcert=sslcert_path, # 証明書設定 connect_timeout=10, host=db_endpoint, port=db_port, dbname=db_name, user=username, password=token ) print("Successfully connected to the database!") # カーソルを作成してクエリを実行(ここではDB名を取得) cursor = connection.cursor() cursor.execute("SELECT current_database();") result = cursor.fetchone() print(f"Current database: {result[0]}") # 接続を閉じる cursor.close() connection.close() # Secrets Manager から認証情報を取得する関数 def get_secret(secret_name, region_name): # セッションとクライアント作成 session = boto3.session.Session() client = session.client( service_name='secretsmanager', region_name=region_name ) try: # シークレット取得 response = client.get_secret_value(SecretId=secret_name) # Secrets の値を JSON 形式でデコード secret = json.loads(response['SecretString']) return secret except Exception as e: print(f"Failed to retrieve secret: {e}") raise # IAMトークンを生成する関数 def get_iam_token(proxy_endpoint, port, username, region_name): rds_client = boto3.client("rds", region_name=region_name) token = rds_client.generate_db_auth_token( DBHostname=proxy_endpoint, Port=port, DBUsername=username, Region=region_name ) return token if __name__ == '__main__': main()
上記のコードを実行すると色々とprintで表示されますが、最終的に以下の表示が出ていればRDS Proxy経由でのDBへの疎通ができています。
Successfully connected to the database! Current database: [データベース名]
以上がRDS Proxy経由でAurora PostgreSQLへの疎通確認をpythonで行う方法になります。
正直まだAWS自体もちゃんと理解はできていないですが、このやり方で疎通確認はできたので困っている人の参考になれば幸いです。
・参考資料
