1. Lambda 와 Python을 이용한 Twitter 3 legged oauth

트위터에서 어플리케이션을 사용할 때 다음과 같은 인증창을 보신 적이 있으실 겁니다.

어플리케이션에 계정을 인증하며 이 어플리케이션이 무슨 권한을 가지는지를 설명하는 창이죠.

이 것을 요청하고 여기서 연계앱인증을 누르면 무슨 일이 일어나는지 또 그 것들을 어떻게 관리하는지에 대해 설명하려고 합니다.

1. 무엇이 필요한가?

API KEY and SECRET

먼저 어플리케이션과 어플리케이션의 API KEY 와 SECRET이 필요합니다.

T witter Developer Dashboard 에서 원하는 어플리케이션의 API와 SECRET을 기록해 둡니다. 또는 어플리케이션을 만듭니다.

이 API KEY는 트위터 API와 통신을 하며 어플리케이션을 식별하기 위해 사용되어집니다.

AWS Account

AWS의 Lambda와 API Gateway 그리고 S3서비스를 이용합니다.
일반적인 Free tier의 범위 내에서 작동하기에 추가적으로 사용하고 계신 서비스가 없으면 기본적으로 비용은 발생하지 않습니다.
다만 프로그램의 오작동, 설정미스, 그 외 서비스의 사용등으로 인한 비용발생에는 책임을 지지 않습니다.

약 한달 간 가동하며 lambda, API Gateway, S3에서의 비용은 발생하지 않고있습니다.

Lambda에 Layer 추가하기

Lambda에서 Pythond을 사용할 때는 표준Library이거나 AWS제공 Library가 아니면 import를 할 수 없습니다.
기동할 때 마다 라이브러리를 설치하여 작동시키는 방법이 존재하지만 해당 방식은 상당히 비효율적이기 때문에 Layer라는 기능을 통해 미리 Lambda에 Library를 Install 해둡니다.

oauth2모듈과 requests 모듈을 사용합니다.

Lambda 관련 간략한 설명을 포함한 제 이전 Lambda+S3를 활용한 Twitter 움짤 봇 만들기 포스트

또는 이 한국어 블로그 를 참고해주세요
글이랑 자료 친절하고 예쁘시네요 부럽습니다 ㅜㅜ

2. 3-legged-oauth란 무엇이며 왜 필요한가

from Twitter Developer

API Key를 가지고 기본적인 API에 접근은 가능하지만 유저와 관련된 API는 사용할 수 없습니다.

예를들면 공개된 유저의 검색, 공개된 글의 검색과 같은 작업은 API Key를 가지고 실행이 가능하지만
유저의 이름 변경, 유저로 트윗 작성 등등은 API Key만가지고는 진행할 수 없으며 해당 유저가 이 어플리케이션이 해당 행동을 하는 것을 허가했다는 인증키가 필요합니다.

이 인증키 또는 Access Token이라 불리는 것을 얻기 위해 3-legged-oauth가 필요합니다.

다음과 같은 과정을 거칩니다.

API KEY와 SECRET으로 oauth_token과 oauth_secret을 발행하여 해당 oauth_token을 사용하여 유저에게 인증 창을 띄움
API URL ) https://api.twitter.com/oauth/request_token
유저를 redirect 시킬 URL ) “https://api.twitter.com/oauth/authorize?oauth_token={oauth_token}”
유저가 승인하면 callback url로 GET을 사용해 redirect 되어짐 이때 url에는 oath_token과 oauth_verifier가 포함
oauth_verifier와 oauth_token을 oauth_secret을 사용하여 인증이 타당한지 검증을 POST로 access token발행 API에 요청함
API URL ) https://api.twitter.com/oauth/access_token
받은 ACCESS_TOKEN과 ACCESS_TOKEN_SECRET를 저장

3. 실제구현

프론트페이지는 최대한 띄우기 귀찮은 사람들을 위한 방식입니다.
Lambda statemachine이라는 기능 좀 써볼 껄 하는 후회도 있네요

3.1 유저에게 인증 창 띄우기

lambda를 활용한 구현을 중심을 설명하자면

우선 총 2개의 lambda가 필요합니다. 유저에게 oauth token을 부여한 인증페이지를 제공하는 lambda와 callback을 통해 불려서 실질적인 accesstoken을 취득하는 lambda

3.1 에서는 유저에게 oauth token이 포함한 인증페이지를 제공하는 부분입니다.

실제 코드를 보며 설명하겠습니다

코드는 twitterdev에서 제공하는 코드를 lambda에 맞게 변형하여 사용하였습니다.

https://github.com/twitterdev/twauth-web/blob/master/twauth-web.py

import json
import os
import oauth2 as oauth
import urllib.request
import urllib.parse
import urllib.error
import boto3

json으로 데이터통신을 하기 때문에 json
환경변수를 위한 os
oauth 관리의 편리성을 위한 oauth2 라이브러리
url 관련처리를 위한 urllib
aws 종합 라이브러리인 boto3

s3 = boto3.client('s3')
request_token_url = 'https://api.twitter.com/oauth/request_token'
access_token_url = 'https://api.twitter.com/oauth/access_token'
authorize_url = 'https://api.twitter.com/oauth/authorize'

먼저 고정으로 필요한 변수들을 정의합니다. aws의 저장소 서비스인 s3를 접근하기 위한 s3 client 와 트위터의 API 주소들을 적어둡니다

    parm=event['queryStringParameters']
    print(parm)
    auth_info={}
    auth_info["email"]=parm["email"]
    auth_info["pin"]=parm["pin"]

lambda_handler의 event에는 API GATEWAY 또는 Function URL로 labmda를 호출 했을 때 그 http 리퀘스트에 대한 정보가 담겨있습니다.
개인적인 용도로 email 과 pin을 입력받고 인증을 진행했기 때문에 request 정보에서 해당 부분을 가져오는 코드입니다.
참고로 API의 rate 제한을 걸 수 있는 점에서 API GATEWAY 사용을 추천하긴 합니다.

s3_res = boto3.resource('s3')
try:
    content_object = s3_res.Object('tweet-spotify-temp-oauth-token', 'temp_oauth_token.json')
    file_content = content_object.get()['Body'].read().decode('utf-8')
    oauth_store = json.loads(file_content)
except:
    oauth_store = {}

try:
    content_object = s3_res.Object('tweet-spotify-temp-oauth-token', 'token_auth.json')
    file_content = content_object.get()['Body'].read().decode('utf-8')
    token_auth = json.loads(file_content)
except:
    token_auth = {}

s3에서 temp_oauth_token 과 token_auth 파일을 불러옵니다.
lambda는 기본적으로 state를 가지지 않기 때문에 한번 불리고 나면 정보는 사라집니다. 그 정보를 보관하기 위해 s3와 연계할 필요가 있습니다.
temp_oauth_token은 oauth_token과 oauth_token_secret을 dictionary로 저장하며 token_auth는 이메일과 핀 정보를 저장합니다 token_auth 이름을 왜이렇게 지었지?
oauth_token은 한번 인증이 끝나면 사용되어지지 않고 token_auth의 정보도 인증이 완료하면 데이터베이스로 이동하기 때문에 s3에서 주기적으로 삭제하고 있습니다.

# Generate the OAuth request tokens, then display them
app_callback_url = "콜백URL. 다음 access token을 얻고 저장하기 위한 lambda의 API Gateway 또는 functionURL을 입력"
consumer = oauth.Consumer(
    os.getenv("API_KEY"), os.getenv("API_SECRET"))
client = oauth.Client(consumer)
resp, content = client.request(request_token_url, "POST", body=urllib.parse.urlencode({
                                "oauth_callback": app_callback_url}))

if resp['status'] != '200':
    error_message = 'Invalid response, status {status}, {message}'.format(
        status=resp['status'], message=content.decode('utf-8'))

request_token = dict(urllib.parse.parse_qsl(content))
oauth_token = request_token[b'oauth_token'].decode('utf-8')
oauth_token_secret = request_token[b'oauth_token_secret'].decode('utf-8')

oauth_store[oauth_token] = oauth_token_secret
token_auth[oauth_token] = auth_info

app_callback_url은 인증이 성공할 경우 해당 url로 인증관련 정보들을 들고 redirect 됩니다. 저희는 이 callback url을 또다른 lambda function에 연결하여 그 정보들을 받고 저장해야 합니다. 3.2에서 설명할 lambda함수의 function url 또는 API Gateway 주소를 입력합니다.
이후는 어플리케이션의 API KEY를 이용하여 oauth token을 발급받습니다.
이후 그 oauth_token과 oauth_token_secret쌍을 oauth_store에 token_auth에는 핀과 이메일을 추가합니다.

s3.put_object(
 Body=json.dumps(oauth_store),
 Bucket='tweet-spotify-temp-oauth-token',
 Key='temp_oauth_token.json'
)

s3.put_object(
 Body=json.dumps(token_auth),
 Bucket='tweet-spotify-temp-oauth-token',
 Key='token_auth.json'
)

이후 해당 dictionary를 json으로 다시 s3에 돌려둡니다.
여기서 어차피 temporary 파일이라면서 왜 굳이 다시 불러와서 append하고 돌려두는 작업을 하는가 라고 생각하실 수 있습니다.
그냥 바로바로 최신 정보만 넣어두면 문제는 2명이 거의 동시에 인증을 시도할 때 입니다.
한명이 인증이 끝나지 않았는데 다른 사람이 인증을 시작하면 앞 사람의 인증정보가 사라져서 인증이 불가하게 됩니다. 그래서 일정 기간동안 계속 가지고 있게 관리합니다.

body = f"https://api.twitter.com/oauth/authorize?oauth_token={oauth_token}"
response = {}
response["statusCode"]=302
response["headers"]={'Location': body}
data = {}
response["body"]=json.dumps(data)

print(oauth_store)
return response

다음은 이 oauth_token을 유저에게 보내서 redirect시킵니다.
302로 사이트가 이동됨으로 하고 그 주소를 oauth_token을 포함한 트위터 인증 URL로 지정하면 이 작업이 끝나면 유저는 해당 URL로 연결되게 됩니다.

여기서 유저는 다음과 같은 화면을 보게됩니다.

3.1 사용자를 인증페이지로 redirect 시키는 코드 전문

import json
import os
import oauth2 as oauth
import urllib.request
import urllib.parse
import urllib.error
import boto3

s3 = boto3.client('s3')
request_token_url = 'https://api.twitter.com/oauth/request_token'
access_token_url = 'https://api.twitter.com/oauth/access_token'
authorize_url = 'https://api.twitter.com/oauth/authorize'
def lambda_handler(event, context):
    parm=event['queryStringParameters']
    print(parm)
    auth_info={}
    auth_info["email"]=parm["email"]
    auth_info["pin"]=parm["pin"]

    s3_res = boto3.resource('s3')
    try:
        content_object = s3_res.Object('tweet-spotify-temp-oauth-token', 'temp_oauth_token.json')
        file_content = content_object.get()['Body'].read().decode('utf-8')
        oauth_store = json.loads(file_content)
    except:
        oauth_store = {}

    try:
        content_object = s3_res.Object('tweet-spotify-temp-oauth-token', 'token_auth.json')
        file_content = content_object.get()['Body'].read().decode('utf-8')
        token_auth = json.loads(file_content)
    except:
        token_auth = {}

    app_callback_url = "콜백URL. 다음 access token을 얻고 저장하기 위한 lambda의 API Gateway 또는 functionURL을 입력"

    # Generate the OAuth request tokens, then display them

    consumer = oauth.Consumer(
        os.getenv("API_KEY"), os.getenv("API_SECRET"))
    client = oauth.Client(consumer)
    resp, content = client.request(request_token_url, "POST", body=urllib.parse.urlencode({
                                    "oauth_callback": app_callback_url}))

    if resp['status'] != '200':
        error_message = 'Invalid response, status {status}, {message}'.format(
            status=resp['status'], message=content.decode('utf-8'))

    request_token = dict(urllib.parse.parse_qsl(content))
    oauth_token = request_token[b'oauth_token'].decode('utf-8')
    oauth_token_secret = request_token[b'oauth_token_secret'].decode('utf-8')

    oauth_store[oauth_token] = oauth_token_secret
    token_auth[oauth_token] = auth_info

    s3.put_object(
     Body=json.dumps(oauth_store),
     Bucket='tweet-spotify-temp-oauth-token',
     Key='temp_oauth_token.json'
    )

    s3.put_object(
     Body=json.dumps(token_auth),
     Bucket='tweet-spotify-temp-oauth-token',
     Key='token_auth.json'
    )

    body = f"https://api.twitter.com/oauth/authorize?oauth_token={oauth_token}"
    response = {}
    response["statusCode"]=302
    response["headers"]={'Location': body}
    data = {}
    response["body"]=json.dumps(data)

    print(oauth_store)
    return response

3.2 트위터 인증완료 정보 저장, spotify 인증 개시

유저가 인증을 완료하면 oauth_token_verifier 를 돌려줍니다. 이 verifier와 가지고있는 secret을 이용하여 정말 우리가 보내준 token에대한 인증완료인지 검사하고 access _token을 요청하여 저장후 spotify인증을 개시하는 코드입니다.

간단하게 요약하자면
s3에서 앞에서 저장한 정보들을 가져온다 -> oauth_verifier를 사용하여 타당한 token인지 검사한다 ( oauth 라이브러리가 잘 해준다 )
-> 해당 정보로 트위터에 accesstoken을 요청한다 -> 돌아온 access token과 앞서 저장한 이메일과 핀번호를 dynamodb에 저장한다

import os
import oauth2 as oauth
import urllib.request
import urllib.parse
import urllib.error
import boto3
import json

s3 = boto3.resource('s3')
content_object = s3.Object('tweet-spotify-temp-oauth-token', 'temp_oauth_token.json')
access_token_url = 'https://api.twitter.com/oauth/access_token'
dynamoDB = boto3.resource('dynamodb')
table= dynamoDB.Table('tweet_spotify')

def lambda_handler(event, context):
    if table.item_count>25:
        return {
            'statusCode': 400,
            'body': "죄송합니다. 사용 가능 유저 수가 상한입니다. Spotify에 API확장을 요청하는동안 기다려주세요(최대6주소요)"
        }

    s3 = boto3.resource('s3')
    content_object = s3.Object('tweet-spotify-temp-oauth-token', 'temp_oauth_token.json')
    file_content = content_object.get()['Body'].read().decode('utf-8')
    oauth_store = json.loads(file_content)

    content_object = s3.Object('tweet-spotify-temp-oauth-token', 'token_auth.json')
    file_content = content_object.get()['Body'].read().decode('utf-8')
    token_auth = json.loads(file_content)

    oauth_token=""
    oauth_verifier=""
    oauth_denied=""
    parm=event['queryStringParameters']

    try:
        oauth_token = parm['oauth_token']
        oauth_verifier = parm['oauth_verifier']
        oauth_denied = parm['denied']
    except:
        print("pass")

    # if the OAuth request was denied, delete our local token
    # and show an error message
    if oauth_denied:
        print("ERR0")
        end_with_error()
        exit(-1)

    if not oauth_token or not oauth_verifier:
        print("ERR1")
        end_with_error()
        exit(-1)

    # unless oauth_token is still stored locally, return error
    if oauth_token not in oauth_store:
        print("ERR2")
        end_with_error()
        exit(-1)

    oauth_token_secret = oauth_store[oauth_token]

    # if we got this far, we have both callback params and we have
    # found this token locally

    consumer = oauth.Consumer(
        os.getenv("API_KEY"), os.getenv("API_SECRET"))
    token = oauth.Token(oauth_token, oauth_token_secret)
    token.set_verifier(oauth_verifier)
    client = oauth.Client(consumer, token)

    resp, content = client.request(access_token_url, "POST")
    auth_info = dict(urllib.parse.parse_qsl(content))
    access_token = auth_info[b'oauth_token'].decode('utf-8')
    access_token_secret = auth_info[b'oauth_token_secret'].decode('utf-8')

    tweet_user_id = auth_info[b'user_id'].decode('utf-8')
    table.put_item(
        Item = {
            'tweet_user_id': tweet_user_id,
            'tweet_id' : auth_info[b'screen_name'].decode('utf-8'),
            'tweet_access_token': access_token,
            'tweet_access_token_secret': access_token_secret,
            'spotify_email': token_auth[oauth_token]["email"],
            'pin' : token_auth[oauth_token]["pin"]
        }
    )

    body = f"스포티파이 인증을 시작하기위한 lambda의 API GATEWAY 또는 functionURL?tweet_user_id={tweet_user_id}"
    response = {}
    response["statusCode"]=302
    response["headers"]={'Location': body}
    data = {}
    response["body"]=json.dumps(data)
    return response

def end_with_error():
    print("ERR")
    return {
        'statusCode': 400,
        'body': "Failed"
    }

dynamodb에 저장되거나 얻은 access token을 활용하여 twitter api를 사용하여 해당 사용자에게 적절한 처리를 하시면 됩니다.

다음은 스포티파이 API 인증과정으로 찾아오겠습니다

언제오지

1. 무엇이 필요한가?#

API KEY and SECRET#

AWS Account#

Lambda에 Layer 추가하기#

2. 3-legged-oauth란 무엇이며 왜 필요한가#

3. 실제구현#

3.1 유저에게 인증 창 띄우기#

3.1 사용자를 인증페이지로 redirect 시키는 코드 전문#

3.2 트위터 인증완료 정보 저장, spotify 인증 개시#