PythonでAWS Secrets ManagerからAPIキーを取得するときのちょっとしたポイント
データアナリティクス事業本部＠札幌の佐藤です。
個人の趣味としてAWS LambdaからTwitterへ自動投稿の練習をしています。 自動投稿するにはTwitterのAPIキーが必要なのですが、さすがに遊びとはいえソースコード上にAPIキーがハードコーディングされている状況はよろしくないため、練習のためAWS Secrets ManagerでTwitterのAPIキーを管理することにしました。
AWS Secrets ManagerからAPIキーを取得するところで少し考えたところがあるため、些末な事項ではありますが記載したいと思います。
AWS Secrets ManagerへのAPIキー登録
事前にTwitterDeveloperでAPIキーを取得しておいてください。
AWS Secrets Managerを新規作成します。
今回は事前に取得したAPIキーを登録するので、「その他のシークレット」を選択します。
シークレットの名前は今後使用するので任意の名前を設定してください。
自動ローテーションは無効にしておきます。
レビュー画面下部にサンプルコードがあるので、基本的にはこのコードを使えばAWS Secrets Managerからキーを取得するので楽ですね。
今回はPythonで実装しているためPythonのサンプルコードを利用したいと思います。
# Use this code snippet in your app.
# If you need more information about configurations or implementing the sample code, visit the AWS docs:
# https://aws.amazon.com/developers/getting-started/python/
import boto3
import base64
from botocore.exceptions import ClientError
def get_secret():
secret_name = "AIKATSU_TWEET"
region_name = "ap-northeast-1"
# Create a Secrets Manager client
session = boto3.session.Session()
client = session.client(
service_name='secretsmanager',
region_name=region_name
)
# In this sample we only handle the specific exceptions for the 'GetSecretValue' API.
# See https://docs.aws.amazon.com/secretsmanager/latest/apireference/API_GetSecretValue.html
# We rethrow the exception by default.
try:
get_secret_value_response = client.get_secret_value(
SecretId=secret_name
)
except ClientError as e:
if e.response['Error']['Code'] == 'DecryptionFailureException':
# Secrets Manager can't decrypt the protected secret text using the provided KMS key.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InternalServiceErrorException':
# An error occurred on the server side.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidParameterException':
# You provided an invalid value for a parameter.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'InvalidRequestException':
# You provided a parameter value that is not valid for the current state of the resource.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
elif e.response['Error']['Code'] == 'ResourceNotFoundException':
# We can't find the resource that you asked for.
# Deal with the exception here, and/or rethrow at your discretion.
raise e
else:
# Decrypts secret using the associated KMS CMK.
# Depending on whether the secret is a string or binary, one of these fields will be populated.
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
decoded_binary_secret = base64.b64decode(get_secret_value_response['SecretBinary'])
# Your code goes here.
変数secretの中身
変数 get_secret_value_response の中身って何だろうということで中身を見てみると、ARNから先ほど登録したAPIキーの情報までを持っている辞書型の変数であることが分かります。
{ 'ARN': 'arn:aws:secretsmanager:ap-northeast-1:XXXXXXX:secret:AIKATSU_TWEET-XXXXXX',
'Name': 'AIKATSU_TWEET',
'VersionId': '0c90934e-6221-49eb-XXXX-5bc4768316e7',
'SecretString': '{ "API_KEY":"XXXXXXXXXXXX",
"API_SECRET_KEY":"XXXXXXXXXXXX",
"ACCESS_TOKEN":"XXXXXXXXXXXX",
"ACCESS_TOKEN_SECRET":"XXXXXXXXXXXX"
}',
'VersionStages': ['AWSCURRENT'],
'CreatedDate': datetime.datetime(2020, 6, 22, 14, 31, 8, 276000, tzinfo=tzlocal()),
'ResponseMetadata': { 'RequestId': '632e0f98-6462-40d3-XXXX-3ab20c407ef2',
'HTTPStatusCode': 200,
'HTTPHeaders': { 'date': 'Mon, 06 Jul 2020 06:36:30 GMT',
'content-type': 'application/x-amz-json-1.1',
'content-length': '505',
'connection': 'keep-alive',
'x-amzn-requestid': '632e0f98-6462-40d3-XXXX-3ab20c407ef2'
},
'RetryAttempts': 0
}
}
後続の secret = get_secret_value_response['SecretString'] でAPIキーの情報を取得していますので、変数 secret には SecretString のValueがセットされていますね。
'{ "API_KEY":"XXXXXXXXXXXX",
"API_SECRET_KEY":"XXXXXXXXXXXX",
"ACCESS_TOKEN":"XXXXXXXXXXXX",
"ACCESS_TOKEN_SECRET":"XXXXXXXXXXXX"
}',
ここまで情報がそろっているので、後は変数 secret から各APIキーの情報を取得するだけです。
ただ、変数 get_secret_value_response の中身を見てもわかるように、変数 secret は辞書型ではなく文字列型です。
そのため、変数 secret から単純に要素を取得することはできません。
私は、パッと見で辞書型と勘違いしいたため、Keyを指定しても取得できなくて悩んでしまいました。
文字列型を辞書型にする
格納値的にもシンプルな辞書の構文であることが分かっているため、文字列型をやりくりしてするよりも、素直に辞書型にするほうが楽だと思います。
そのため astモジュール の literal_eval() を使用することで辞書型となります。
式ノードまたは Python のリテラルまたはコンテナのディスプレイ表現を表す文字列を安全に評価します。与えられる文字列またはノードは次のリテラルのみからなるものに限られます: 文字列、バイト列、数、タプル、リスト、辞書、集合、ブール値、 None 。
ast --- 抽象構文木
以下の38行目～46行目の内容となります。
import ast
import base64
import boto3
import json
from botocore.exceptions import ClientError
from requests_oauthlib import OAuth1Session
# AWS Secrets ManagerからAPIキーを取得
def get_apikey():
session = boto3.session.Session()
client = session.client(
service_name = 'secretsmanager',
region_name = 'ap-northeast-1'
)
try:
get_secret_value_response = client.get_secret_value(SecretId='AIKATSU_TWEET')
except ClientError as e:
raise e
else:
if 'SecretString' in get_secret_value_response:
secret = get_secret_value_response['SecretString']
else:
secret = base64.b64decode(get_secret_value_response['SecretBinary'])
return secret
# パラメータストアの取得
def get_ssm_parameter_store(param, WithDecryption=True):
ssm_client = boto3.client('ssm')
res = ssm_client.get_parameters(Names=[param], WithDecryption=WithDecryption)
return res['Parameters'][0]['Value']
def lambda_handler(event, context):
# 文字列型を辞書型にする
secret = ast.literal_eval(get_apikey())
# Twitterの接続
twitter = OAuth1Session(
secret['API_KEY'],
secret['API_SECRET_KEY'],
secret['ACCESS_TOKEN'],
secret['ACCESS_TOKEN_SECRET']
)
url_update = 'https://api.twitter.com/1.1/statuses/update.json'
url_media = 'https://upload.twitter.com/1.1/media/upload.json'
AWS_S3_BUCKET_NAME = get_ssm_parameter_store('AIKATSU_TWEET_MORP_OUTPUT_BUCKET')
file = 'aikatsu_tweet_wordcloud.jpg'
s3 = boto3.resource('s3')
bucket = s3.Bucket(AWS_S3_BUCKET_NAME)
file_path = '/tmp/' + file
bucket.download_file(file, file_path)
param = { "media" : open(file_path, 'rb').read() }
res = twitter.post(url_media, files = param)
if res.status_code == 200 :
requestJson = json.loads(res.text)
mediaId = requestJson['media_id']
params = { "status" : '昨日はアイカツ！について{}人がツイートしていました。こんなワードがホットだったようです。 #aikatsu'.format(tweet_user_num), "media_ids" : (mediaId) }
res = twitter.post(url_update, params = params)
if res.status_code == 200 :
pass
else:
print('error:' + str(res.status_code))
return True
最後に
実際に使用してみて、AWS Secrets Managerは低コストで簡単にAPIキーを管理できるようにでき、実装もサンプルコードを利用することで簡単に実装できるなという印象でした。
