SecretManagerからSecretを取得して外部APIを実行させるLambda関数サンプル

概要

・OAuth2.0のクライアント認証で使うトークンを取得する。
・SecretManagerにClientSecretとClientIdを保持している
・confidentialを作成してAPI Gateway経由で外部APIを実行し、取得したトークンをAPI Gatewayのステージ変数へ格納する。

import json
import ast
import boto3
import urllib.request
import urllib.parse
import base64
import os
import logging
from botocore.exceptions import ClientError

logging.basicConfig(
    format='%(asctime)-15s %(levelname)s %(message)s',
    level=logging.INFO
)

# AWSクライアント
apigw_client = boto3.client("apigateway")

# リージョン取得
region_name = os.environ['AWS_DEFAULT_REGION']

# API GatewayのIDを環境変数から取得
apigw_id = os.environ['APIGW_ID']

# 格納先のAPI Gatewayステージ変数キー
apigw_stagevalue_key = os.environ['APIGW_VALUE_KEY']

# 外部APIエンドポイント
target_url = os.environ['TARGET_API_URL']

# 格納先変数名
value_name = os.environ['VALUE_NAME']

# Secrets Managerで管理する値の名前
secret_name = "target-clientsecret"
target_client_id = "target-clientid"

# 外部API実行処理
def get_target_token():
    session = boto3.session.Session()
    client = session.client(
        service_name='secretsmanager',
        region_name=region_name
    )
    # Secrets Manager からSecretを取得
    try:
        get_secret_client_secrets_response = client.get_secret_value(
            SecretId=secret_name
        )
        get_secret_secrets_id_response = client.get_secret_value(
            SecretId=target_client_id
        )
        secret_client_secrets_data = get_secret_client_secrets_response['SecretString']
        secret_secrets_id_data = get_secret_secrets_id_response['SecretString']
        secretClientSecrets = ast.literal_eval(secret_client_secrets_data)
        secretClientID = ast.literal_eval(secret_secrets_id_data)
        clientSecret = secretClientSecrets['targetClientSecret'] #シークレット
        clientId = secretClientID['targetClientId'] #クライアントID
    except ClientError as e:
        logging.error(e)
        return "error"
        
    # トークン取得リクエスト（対象の外部APIによって編集する）
    credentials = base64.b64encode((clientId + ":" + clientSecret).encode()).decode()
    headers = {
        'Accept': 'application/json',
        'Authorization': 'Basic ' + credentials,
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    data = {'grant_type':'client_credentials'}
    req = urllib.request.Request(target_url, headers=headers, data=urllib.parse.urlencode(data).encode('utf-8'), method='POST')
    
    # リクエスト実行
    try:
        with urllib.request.urlopen(req) as response:
            body = response.read()
            value = json.loads(body.decode('utf-8'))["access_token"]
            return value
    except urllib.error.HTTPError as err:
        logging.error(err)
        return "error"

def lambda_handler(event, context):
    # 外部APIを実行してtokenを取得
    token = get_target_token()
  
    # 設定ファイル書き換える
    if len(token) < 500:
        response = apigw_client.update_stage(
            restApiId=apigw_id,
            stageName=apigw_stagevalue_key,
            patchOperations=[
                {
                    'op': 'replace',
                    'path': '/variables/' + value_name,
                    'value': token
                }
            ]
        )
    else:
        # ステージ変数が500文字制限があるので、500文字で分割
        split_token_arr = [token[x:x+500] for x in range(0, len(token), 500)]
        patch_operations = [
            {
                'op': 'replace',
                'path': '/variables/' + value_name + "0",
                'value': split_token_arr[0]
            }
        ]
        if len(split_token_arr) > 1:
            patch_operations.append(
                {
                    'op': 'replace',
                    'path': '/variables/' + value_name + "1",
                    'value': split_token_arr[1]
                }
            )
        #  ステージ変数に値を格納
        response = apigw_client.update_stage(
            restApiId=apigw_id,
            stageName=apigw_stagevalue_key,
            patchOperations=patch_operations
        )
    # 終了
    return {
        'statusCode': 200,
        'body': token
    }