[Google App Engine][twitter][oauth]ログインを自作する

twitterのUsing OAuth 1.0aを読んでいて、認証を行なうには、oauthを使う必要があるとのこと。

oauthを使うにあたって、すでに、サンプルがころがっているのだが、理由はわからないが、すぐに受け入れることはできなかった。

Djangoや、JavaScriptのjQueryは受け入れられたのに。

きっと、3rdパーティーの不確定要素が多いライブラリからだと思うのだが。
(この判断も一方的なのだが。)

前置きが長くなってしまったが、というわけで、Google App EngineのPythonを使って自作をすることに。

#!/usr/bin/env python # # Copyright 2007 Google Inc. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # import datetime,time,random,urllib,hmac,hashlib,re,os from google.appengine.ext import webapp from google.appengine.ext.webapp import util from google.appengine.api import urlfetch #django from google.appengine.dist import use_library use_library('django', '1.0') from django.utils import simplejson from google.appengine.ext.webapp import template # Request Token URL reqt_url = 'https://api.twitter.com/oauth/request_token' # Authorize URL auth_url = 'https://api.twitter.com/oauth/authorize' # Access Token URL acct_url = 'https://api.twitter.com/oauth/access_token' class oauth(): def __init__(self,secret= "",method="POST"): self.consumer_secret = '●●●●●●' self.token_secret = secret self.method = method self.param = { 'oauth_consumer_key':'●●●●●●', 'oauth_nonce':self.getNonce(), 'oauth_signature_method':'HMAC-SHA1', 'oauth_version':'1.0', 'oauth_timestamp':self.getTimeStamp() } def setParam(self,param): for i in param: self.param[i] = param[i] self.param['oauth_signature'] = self.makeSignature() def callRequest(self,requrl): param = urllib.urlencode(self.param) if self.method == 'POST': result = urlfetch.fetch( url=requrl, payload=param, method=urlfetch.POST, headers={'Content-Type': 'application/x-www-form-urlencoded','User-Agent':'***'} ) else: param = re.sub("&", ",", param) result = urlfetch.fetch( url=requrl, method=urlfetch.GET, headers={'Authorization': 'OAuth realm="",'+param,'User-Agent':'***'} ) if result.status_code == 200: return {'result':True,'content':result.content} else: return {'result':False,'content':result.content} #return {'result':False,'content':str(result.status_code)} def getNonce(self): str = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz123456789[]{}!$%&'()-^\:;*+><" strlist = list(str) string = "" for i in range(20): x = random.randint(0,len(strlist)-1) string += strlist[x] return string def getTimeStamp(self): d = datetime.datetime.today() d = time.mktime(d.timetuple()) d = str(int(d)) return d def makeSignature(self): pks = sorted(self.param.keys()) param = '' for i in pks: param = param + '&' + i + '=' + urllib.quote_plus(self.param[i]) else: param = param[1:] param = urllib.quote_plus(self.method) + '&' + urllib.quote_plus(reqt_url) + '&' + urllib.quote_plus(param) h = hmac.new("%s&%s" % (urllib.quote(self.consumer_secret), urllib.quote(self.token_secret)), param, hashlib.sha1) sig = h.digest().encode("base64").strip() return sig class LoginHandler(webapp.RequestHandler): def get(self): oauth_token = self.request.get('oauth_token') oauth_verifier = self.request.get('oauth_verifier') if oauth_token == '' and oauth_verifier == '': ioauth = oauth() param = {'oauth_callback':'この処理が走るURL'} ioauth.setParam(param) ret = ioauth.callRequest(reqt_url) if ret['result']: content = ret['content'] content = content.split('&') for i in content: temp = i.split('=') ioauth.param[temp[0]] = temp[1] self.response.headers.add_header('Set-Cookie', 'ts='+ioauth.param['oauth_token_secret']) self.redirect(auth_url+"?oauth_token="+ioauth.param['oauth_token'], permanent=True) else: ioauth = oauth(secret=self.request.cookies.get('ts')) param = {'oauth_token':oauth_token,'oauth_verifier':oauth_verifier} ioauth.setParam(param) ret = ioauth.callRequest(acct_url) if ret['result']: content = ret['content'] content = content.split('&') for i in content: temp = i.split('=') ioauth.param[temp[0]] = temp[1] #認証成功!! path = os.path.join(os.path.dirname(__file__), 'redirect.html') template_values = { 'oauth_token':ioauth.param['oauth_token'], 'oauth_token_secret':ioauth.param['oauth_token_secret'], 'user_id':ioauth.param['user_id'], 'screen_name':ioauth.param['screen_name'] } self.response.out.write(template.render(path,template_values)) def main(): application = webapp.WSGIApplication( [('/twitter/login', LoginHandler)], debug=True ) util.run_wsgi_app(application) if __name__ == '__main__': main()

http://hoge/twitter/loginにアクセスすると、LoginHandlerクラスが反応します。

最初、直アクセスした場合は、当然、

oauth_token
と、
oauth_verifier
がないので、最初のif文が実行されます。

その中で、oauthクラスのインスタンス変数を宣言するのですが、コンストラクタの時点で、作成できる変数群をすべて設定しておきます。

setParamクラスでsignatureを作成し、httpのPOSTで通信を行ないます。

oauth_token、oauth_token_secret返ってくるので、oauth_token_secretをクッキーに保存しoauth_tokenをquery stringに設定して、リダイレクトをします。

リダイレクトの結果、oauth_token、oauth_verifierがコールバック先のquery stringに付与されて呼び出されるので、それを元に、再度、signatureを作成、access_tokenをゲットします。

今回、上記の処理を書くにあたって、Twitter API を OAuth で認証するスクリプトを 0 から書いてみたを参考にしました。

ただ参照先サイトは、コールバックの設定がなかったので、そこだけ、ちょっとプログラムの変更をする必要がありましたが。

こうしてまとめておけば、後でいつでも、振り返るので、便利ですね。