今までの内容は以下の通り

Twitter APIを使って自動でTweetする（１）
Request Tokenの取得
Twitter APIを使って自動でTweetする（２）

AWSのAPI GatewayとLambdaへのデプロイ
Twitter APIを使って自動でTweetする（３）

Vueを使ったTwitterの認証・認可の取得
Twitter APIを使って自動でTweetする（４）

AWS S3のホスティングでコンテンツの公開

今回は、Twitterの認証で取得したTokenを使ってTwitterのAPIでAccess Tokenに交換する。このAccess Tokenを取得さえすれば、それを使って自由にTweetできるはずだ。

##Access Tokenの取得APIの概要

Resource URL
https://api.twitter.com/oauth/access_token
Method
Post
Authorization Header Parameters
以下のパラメータをAuthorizationヘッダーにセットする
Authorization: OAuth oauth_consumer_key="xxxxxxx", oauth_nonce="...
- oauth_token（必須）
  前回取得したRequest Token
- oauth_consumer_key（必須）
  Twitter APIのAPI Key
- oauth_signature_method（必須）
  このリクエストの署名のアルゴリズム（HMAC-SHA1など）
- oauth_signature（必須）
  署名
- oauth_timestamp（必須）
  リクエスト発行時のUNIXTIME
- oauth_nonce（必須）
  リクエスト毎に一意なランダムな値
- oauth_version（オプション）
  1.0
Request Body
なし

##前回までのコードの整理（Lambda Layersの利用）

Access Token取得のAPIは、上記の通りsignatureなどRequest Token取得APIと共通部分が多い。そこで、複数のAWS Lambda関数間でコードを共有できるLayerを使ってみる。まず、Layer用のコードを以下のように作成した。

utilities.py

import json
import time
import base64
import os
import uuid
import hashlib
import hmac
import urllib.parse
import requests

def create_timestamp():
    return str(int(time.time()))

def create_nonce():
    return str(uuid.uuid4())

def create_base_string(params: dict):
    string = ''
    for key, val in sorted(params.items()):
        string += key + '=' + val + '&'

    return string.rstrip('&')

def convert_dict(dict_: dict):
    string = ''
    for key, val in dict_.items():
        string += key + '="' + val + '", '

    return string.rstrip(', ')

def create_signature(base_string: str, method: str, uri:str, api_secret:str, token_secret:str):
    base_string = method + '&' + urllib.parse.quote(uri, safe='') + '&' + urllib.parse.quote(base_string, safe='')
    key = api_secret + '&' + token_secret
    hashed_string = hmac.new(key.encode(), base_string.encode(), hashlib.sha1)
    return urllib.parse.quote(base64.standard_b64encode(hashed_string.digest()), safe='')

この、SignatureやNonce・Timestampの作成をするUtilityモジュールをLambdaで利用できるように、Layerを作成する。また、Layerを作成をするにあたり、requestsもLayerに入れてそれぞれのLambda関数では含める必要がないようにする。
Layer用のデプロイパッケージは以下のように作成する。pythonサブディレクトリ配下に配置する必要がある点に注意する必要がある。

mkdir layer/python
cp utilities.py layer/python/.
pip install requests -t layer/python/.
cd layer
zip -r layer.zip *

作成したパッケージからLayerを作成する。まず、Lambdaのコンソールにアクセスし、「Layers」、「レイヤーの作成」をクリックする。

「レイヤー名」を入力し、先ほど作成した「layer.zip」をアップロードする。「互換性のあるランタイム」で「Python 3.7」を選択し、「作成」をクリックする。

これでLayerの作成は完了。次にLayerを使う側の設定を行う。
Layerを使うLambda関数の画面で、「Layers」を選択し、「レイヤーの追加」をクリックする。

先ほど作成したレイヤーを選択し、「追加」をクリックする。

これで、先ほど作成したutilitiesモジュールやrequestsパッケージを利用することができる。

##Lambda関数用コードの作成

まず、すでに作成済みのget_request_token.pyのコードを以下のように修正し、Layerで提供するutilitiesモジュールとrequestsパッケージを利用するようにする。

get_request_token.py

import os
import urllib.parse
import requests
import utilities

REQUEST_URL = 'https://api.twitter.com/oauth/request_token'
REQUEST_METHOD = 'POST'
API_KEY = os.environ.get('API_KEY')
API_SECRET = os.environ.get('API_SECRET')
TOKEN_SECRET = ''
CALLBACK_URL = os.environ.get('CALLBACK_URL')
SIGNATURE_METHOD = 'HMAC-SHA1'
TOKEN_TABLE = 'token_table'
    
class GetRequestToken:
    
    def __init__(self):
        pass
        
    def _create_authorization_params(self):
        authorization_params = {}
        authorization_params['oauth_timestamp'] = utilities.create_timestamp()
        authorization_params['oauth_consumer_key'] = urllib.parse.quote(API_KEY, safe='')
        authorization_params['oauth_nonce'] = utilities.create_nonce()
        authorization_params['oauth_callback'] = urllib.parse.quote(CALLBACK_URL, safe='')
        authorization_params['oauth_signature_method'] = SIGNATURE_METHOD
        authorization_params['oauth_version'] = '1.0'
        authorization_params['oauth_signature'] = utilities.create_signature(utilities.create_base_string(authorization_params), REQUEST_METHOD, REQUEST_URL, API_SECRET, TOKEN_SECRET)
    
        return authorization_params
    
    def _create_authorization_header(self):
        authorization_header_val = 'OAuth '
        params = self._create_authorization_params()
        authorization_header_val += utilities.convert_dict(params)
        
        return {'Authorization': authorization_header_val}
    
    def _execute(self):
        headers = self._create_authorization_header()
    
        return requests.post(REQUEST_URL, headers=headers)

def lambda_handler(event, context):
    
    get_request_token = GetRequestToken()
    
    response_data = get_request_token._execute()
    token = response_data.text.split('&')[0]

    return {
        'statusCode': 200,
        'headers': {'Access-Control-Allow-Origin': '*'},
        'body': token
    }

次に、Access Token取得のためのget_access_token関数のコードを作成する。

get_access_token.py

import os
import urllib.parse
import requests
import utilities

REQUEST_URL = 'https://api.twitter.com/oauth/access_token'
REQUEST_METHOD = 'POST'
API_KEY = os.environ.get('API_KEY')
API_SECRET = os.environ.get('API_SECRET')
TOKEN_SECRET = ''
SIGNATURE_METHOD = 'HMAC-SHA1'

class GetAccessToken:
    
    def __init__(self, event, context):
        self.oauth_token = event.get('queryStringParameters').get('oauth_token')
        self.oauth_verifier = event.get('queryStringParameters').get('oauth_verifier')
    
    def _create_authorization_params(self):
        authorization_params = {}
        authorization_params['oauth_timestamp'] = utilities.create_timestamp()
        authorization_params['oauth_consumer_key'] = urllib.parse.quote(API_KEY, safe='')
        authorization_params['oauth_nonce'] = utilities.create_nonce()
        authorization_params['oauth_signature_method'] = SIGNATURE_METHOD
        authorization_params['oauth_version'] = '1.0'
        authorization_params['oauth_token'] = self.oauth_token
        authorization_params['oauth_verifier'] = self.oauth_verifier
        authorization_params['oauth_signature'] = utilities.create_signature(utilities.create_base_string(authorization_params), REQUEST_METHOD, REQUEST_URL, API_SECRET, TOKEN_SECRET)
    
        return authorization_params
    
    def _create_authorization_header(self):
        authorization_header_val = 'OAuth '
        params = self._create_authorization_params()
        authorization_header_val += utilities.convert_dict(params)
        
        return {'Authorization': authorization_header_val}
        
    
    def _execute(self):
        headers = self._create_authorization_header()
    
        return requests.post(REQUEST_URL, headers=headers)

def lambda_handler(event, context):
    
    get_access_token = GetAccessToken(event, context)
    response_data = get_access_token._execute()

    return {
        'statusCode': 200,
        'headers': {'Access-Control-Allow-Origin': '*'},
        'body': response_data.text
    }

これらのコードのLambdaへのデプロイ方法は第二回を参照してほしい。（Lambdaの環境変数の設定やLayerの追加も忘れないこと）
次に、今回新たに作成したget_access_tokenのLambda関数を利用するためにAPI Gatewayを設定する必要がある。こちらの詳細も第二回を参照してほしいが、リソースパスとして「oauth_access_token」、メソッドを「GET」にする点に注意。

最後に、今回作成したAPI Gatewayのリソース「oauth_access_token」のURIをTwitter APIのコールバックURLとget_request_tokenのlambda関数の環境変数に設定することを忘れないこと。
これで、前回までで作成したアプリで「Twitterでログイン」からログインするとAccess Tokenが取得でき、画面に表示されるはずだ。

続編の今回得られたAccess Tokenを利用してTweetする記事を公開しました。

前回までの投稿は以下
Twitter APIを使って自動でTweetする(1)
Twitter APIを使って自動でTweetする(2)
Twitter APIを使って自動でTweetする(3)
Twitter APIを使って自動でTweetする(4)