毎日Learning

学んだことを共有します

Google App Engineの認証にTwitterのOAuthを使ってみた

Python AppEngine Tiwtter

Google App Engineの認証は、Googleアカウントを使うのが最も簡単だが、今回、Twitterと連携するWebアプリケーションを作ったので、認証をTwitterで済ましてしまおうと思いやってみた。

使用したライブラリは、tweepy joshthecoder/tweepy · GitHub と、simple_cookie.py

参考にさせていただいたサイトは、 tweepyでtwitterの3-legged OAuth認証を試してみた(GoogleAppEngine) | taichino.com 。

他にもいくつか見たが、ここが一番参考になった。

pythonで認証といえば、デコレータ。

という、Djangoを使ってて感動した部分を踏襲させていただき、webappのgetやpostメソッドにdecorateすれば認証した結果を、requestに追加するようにした。

先に使い方は、以下のような感じのコードでできる。

from google.appengine.ext import webapp

from common.twitter import oauth_twitter

from utils import render_to_response

class HomePage(webapp.RequestHandler):
    @oauth_twitter(redirect=True)
    def get(self):
        screen_name = self.request.screen_name

        # TwitterAPI- ユーザー情報を取得
        user_info =self.request.twitter["api"].get_user(screen_name=screen_name)

        self.response.out.write(render_to_response("home.html", {"user_info":user_info}))

self.requestに、twitterという辞書を作り、辞書に、authとapiを入れている。

screen_nameは、認証後、画面表示や、DBからの情報取得に必ず使うので、認証後に取得してrequestにセットするようにした。

oauth_twitterに、redirect=Trueを渡すことで、認証されていない状態でこのURLが呼びだされた場合、Twitterの認証画面にリダイレクトされる。

Twitterの認証画面でログインし、自分のページにリダイレクトされるのだが、そのリダイレクト先を、oauth_twitterの引数に、callback_urlで指定することもできる。

callback_urlはデフォルトで、「http://<ドメイン>/verity/」にしている。

このリダイレクト先の処理は、以下のように書く。

from google.appengine.ext import webapp

from common.twitter import verity_twitter, logout_twitter

class VerityAction(webapp.RequestHandler):
    @verity_twitter
    def get(self):
        return self.redirect("/home/")

class LogoutAction(webapp.RequestHandler):
    @logout_twitter
    def get(self):
        return self.redirect("/")

まとめてログアウトも書いた。

「http://<ドメイン>/verity/」に、Twitterで認証した後、リダイレクトしてもらい、その処理についてもデコレータで処理するようにした。

verity_twitterでは、Twitterからリダイレクトされた際にGETパラメータで渡される、oauth_tokenとoauth_verifierを取得し、request_tokenからaccess_tokenを取得して、cookieからsid取ってきて、memcacheに保存…って感じの処理をする。

で、その後、/home/にリダイレクトして、認証済みの状態で表示されるってわけ。

でまあついでに書いたログアウトも、logout_twitterデコレータが、cookieやらmemcacheから認証に必要な情報を消してくれるってわけ。

こんな感じの使い方をする。

ちなみに、Ajaxで使用する場合、oauth_twitterデコレータにredirect=Falseを渡せば良い。

こんな感じで。

class UserTimelineAction(webapp.RequestHandler):
    @oauth_twitter(redirect=False)
    def get(self, since_id=None):

        # 認証中
        if self.request.is_auth:

            d = {"since_id": since_id, "count": 10} if since_id else {"count": 10}
            t_l = self.request.twitter["api"].user_timeline(screen_name=self.request.screen_name, **d)
            
            self.response.headers["Content-Type"] = "text/javascript; charset=UTF-8"
            return self.response.out.write(simplejson.dumps([{"id": t.id, "text": t.text, "created_at": t.created_at} for t in t_l]))

        # 認証不可
        else:
            return self.error(401)

redirect=Falseを渡した場合は、self.request.is_authにTrueかFalseをセットするようにしている。

認証不可のエラーコードを返すもよし。認証不可を伝えるjsonで返すもよし。といった感じだ。

使い方は以上。以下がそのデコレータ。

# -*- coding: utf-8 -*-

import uuid

from google.appengine.api import memcache
from google.appengine.api import urlfetch
from google.appengine.ext import db

import tweepy
from lib.simple_cookie import Cookies

#API登録時に表示されるConsumer keyとConsumer secretを指定してください
CONSUMER_KEY = "XXXXXXXXXXXXXXXXXXXXXX" #Consumer key
CONSUMER_SECRET = "XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX" #Consumer secret

#Cookieの有効期間（秒）
COOKIE_EXPIRE_TIME = 30*60

class RequestToken(db.Model):
    token_key    = db.StringProperty(required=True)
    token_secret = db.StringProperty(required=True)

def oauth_twitter(callback_url="/verity/", redirect=True):
    def _deco(_func):
        def _f(_self, *args, **kwargs):
            cookie = Cookies(_self, max_age=COOKIE_EXPIRE_TIME)
            if not cookie.has_key("sid"):
                cookie["sid"] = str(uuid.uuid4())
            access_token = memcache.get(cookie["sid"])
            if access_token:
                auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
                auth.set_access_token(access_token.key, access_token.secret)
                api = tweepy.API(auth_handler=auth)
                cookie["sid"] = cookie["sid"]
                memcache.set(cookie["sid"], memcache.get(cookie["sid"]), COOKIE_EXPIRE_TIME)
                memcache.set("screen_name_" + cookie["sid"], memcache.get("screen_name_" + cookie["sid"]), COOKIE_EXPIRE_TIME)
                _self.request.twitter = {
                    "auth": auth,
                    "api": api,
                }
                _self.request.cookie = cookie
                _self.request.screen_name = memcache.get("screen_name_" + cookie["sid"])
                _self.request.is_auth = True
                _func(_self, *args, **kwargs)
            else:
                if redirect:
                    auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET, "%s%s" % (_self.request.host_url, callback_url))
                    auth_url = auth.get_authorization_url()
                    request_token = RequestToken(token_key=auth.request_token.key, token_secret=auth.request_token.secret)
                    request_token.put()
                    _self.redirect(auth_url)
                else:
                    _self.request.is_auth = False
                    _func(_self, *args, **kwargs)
        return _f
    return _deco

def verity_twitter(func):
    def _deco(_self):
        request_token_key = _self.request.get("oauth_token")
        request_verifier  = _self.request.get('oauth_verifier')
        auth = tweepy.OAuthHandler(CONSUMER_KEY, CONSUMER_SECRET)
        request_token = RequestToken.gql("WHERE token_key=:1", request_token_key).get()
        auth.set_request_token(request_token.token_key, request_token.token_secret)
        access_token = auth.get_access_token(request_verifier)
        cookie = Cookies(_self)
        memcache.set(cookie["sid"], access_token, COOKIE_EXPIRE_TIME)
        memcache.set("screen_name_" + cookie["sid"], auth.get_username(), COOKIE_EXPIRE_TIME)
        func(_self)
    return _deco

def logout_twitter(func):
    def _deco(_self):
        cookie = Cookies(_self)
        if cookie.has_key('sid'):
            memcache.delete(cookie['sid'])
            memcache.delete("screen_name_" + cookie['sid'])
            del cookie['sid']
        func(_self)
    return _deco

以上。