更新履歴とコメントはAppEngine-Twitter Archive - ゼロと無限の間のログへどうぞ。

サイトに対してパスワードを預けないで済むという画期的なプロトコルOAuth。仕様はややこしいようで簡単、なようで意外に複雑。
日本語情報が少ないのが痛い。

でも1回作ってしまえば使い回しが効く。ということでライブラリ化した。

今のところTwitter APIでの使用しか考えていないが、他のサービスでもある程度使えるんじゃないかなー。

MITライセンスで。

@RT_report

@who_is_near

@who_is_near 六本木

試しにつぶやくには、Twitterにログイン後にここをクリック

@norikae_map

@norikae_map 東京 大阪

試しにつぶやくには、Twitterにログイン後にここをクリック

@norikae_map 木更津駅 羽田空港駅 車

試しにつぶやくには、Twitterにログイン後にここをクリック

@norikae_map 霞が関 霞が関 歩き

試しにつぶやくには、Twitterにログイン後にここをクリック

TwitterにOAuthでログインしてTwitter APIを使うデモ

ソースコード
（※AppEngine BasehandlerとAppEngine Twitterを使っている）

#!/usr/bin/env python
# -*- coding: UTF-8 -*-
'''
Sample for AppEngine-OAuth on Google App Engine
 
See: http://0-oo.net/sbox/python-box/appengine-oauth
See also: http://apiwiki.twitter.com/OAuth-FAQ
'''
 
import logging
import wsgiref.handlers
from appengine_twitter import AppEngineTwitter
from basehandler import BaseHandler, h
from django.utils import simplejson
from google.appengine.ext import db
from google.appengine.ext import webapp
 
 
OAUTH_KEY = 'xxxxx'
OAUTH_SECRET = 'xxxxx'
 
 
class DemoHandler(BaseHandler):
 
  def demo_header(self):
    self.simple_header(u'AppEngine-OAuthのデモ')
 
    self.p(u'<h1>AppEngine-OAuthのデモ</h1>')
 
 
  def demo_footer(self):
    self.p('<br>')
    self.p('<hr>', True)
    self.p(u'このページは')
    self.p('<a href="http://0-oo.net/sbox/python-box/appengine-oauth">')
    self.p(u'AppEngine-OAuth</a> と ')
    self.p('<a href="http://0-oo.net/sbox/python-box/appengine-twitter">')
    self.p(u'AppEngine-Twitter</a> のデモです')
 
    self.simple_footer()
 
 
 
class InitHandler(DemoHandler):
 
  def get(self):
    twitter = AppEngineTwitter()
    twitter.set_oauth(key=OAUTH_KEY, secret=OAUTH_SECRET)
 
    req_info = twitter.prepare_oauth_login()
 
    # request_tokenはcallbackされた後で使うのでDatastoreに保存しておく
    OAuthRequestToken(token=req_info['oauth_token'],
                      secret=req_info['oauth_token_secret']).put()
 
    self.demo_header()
    self.p('<a href="' + h(req_info['url']) + '">')
    self.p(u'TwitterのOAuthログインへ</a>', True)
    self.demo_footer()
 
 
 
class CallbackHandler(DemoHandler):
 
  def get(self):
    twitter = AppEngineTwitter()
    twitter.set_oauth(OAUTH_KEY, OAUTH_SECRET)
 
    # TwitterからHTTP GETでrequest_tokenが渡される
    req_token = self.request.get('oauth_token')
 
    # Datastoreに保存しておいたreqest_token_secretを取り出す
    query = OAuthRequestToken.all()
    query.filter('token = ', req_token)
    req_tokens = query.fetch(1)
 
    # request_tokenとaccess_tokenを交換する
    acc_token = twitter.exchange_oauth_tokens(req_token, req_tokens[0].secret)
 
    # ここまで来ればOAuthを使ってAPIが使える。試しにユーザー名を取得
    twitter.verify()
    name = simplejson.loads(twitter.last_response.content)['screen_name']
 
    self.demo_header()
    self.p(u'こんにちは、' + name + u'さん！', True)
    self.p('<br>')
 
    # 以後は再ログインせずにaccess_tokenを繰り返し使うことができる
    tw2 = AppEngineTwitter()
    tw2.set_oauth(OAUTH_KEY,
                  OAUTH_SECRET,
                  acc_token['oauth_token'],
                  acc_token['oauth_token_secret'])
 
    msg = u'AppEngine-OAuth (on Python and Google App Engine) '
    msg += u'http://0-oo.appspot.com/oauth/ からこんにちは！'
    tw2.update(msg.encode('utf8'))
 
    self.p(u'つぶやいたよ（自分のTLを見てね）', True)
    self.p('<br>')
 
    if tw2.is_following('uresuji_books') == False:
      tw2.follow('uresuji_books')
      self.p(u'@uresuji_books をフォローしたよ', True)
      self.p('<br>')
 
    self.p(u'<a href="./">最初のページへ戻る</a>', True)
    self.demo_footer()
 
 
 
#Model(s)
 
class OAuthRequestToken(db.Model):
  token = db.StringProperty()
  secret = db.StringProperty()
 
 
 
logging.getLogger()
routing = [('/oauth/', InitHandler),
           ('/oauth/callback', CallbackHandler)]
application = webapp.WSGIApplication(routing, debug=False)
wsgiref.handlers.CGIHandler().run(application)

#!/usr/bin/env python
# -*- coding: utf-8 -*-
'''
AppEngine-OAuth
 
OAuth utility for applications on Google App Engine
 
See: http://0-oo.net/sbox/python-box/appengine-oauth
License: http://0-oo.net/pryn/MIT_license.txt (The MIT license)
'''
 
__author__ = 'dgbadmin@gmail.com'
__version__ = '0.1.1'
 
 
import hmac
import urllib
from google.appengine.api import urlfetch
from hashlib import sha1
from random import getrandbits
from time import time
 
 
class AppEngineOAuth(object):
 
  def __init__(self, key, secret, acs_token='', acs_token_secret=''):
    self._key = key
    self._secret = secret
    self._token = acs_token
    self._token_secret = acs_token_secret
 
    # Be understandable which type token is (request or access)
    if acs_token == '':
      self._token_type = None
    else:
      self._token_type = 'access'
 
 
  def prepare_login(self, req_token_url):
    '''
    Return request_token, request_token_secret and params of authorize url.
    '''
    # Get request token
    params = self.get_oauth_params(req_token_url, {})
    res = urlfetch.fetch(url=req_token_url + '?' + urllib.urlencode(params),
                         method='GET')
    self.last_response = res
    if res.status_code != 200:
      raise Exception('OAuth Request Token Error: ' + res.content)
    # Response content is request_token
    dic = self._qs2dict(res.content)
    self._token = dic['oauth_token']
    self._token_secret = dic['oauth_token_secret']
    self._token_type = 'request'
 
    # Get params with signature
    sig_params = {'oauth_signature': params['oauth_signature']}
    dic['params'] = urllib.urlencode(self.get_oauth_params(req_token_url,
                                                           sig_params))
 
    return dic
 
 
  def exchange_tokens(self, acs_token_url, req_token, req_token_secret):
    self._token = req_token
    self._token_secret = req_token_secret
    self._token_type = 'request'
 
    params = urllib.urlencode(self.get_oauth_params(acs_token_url, {}))
    res = urlfetch.fetch(url=acs_token_url, payload=params, method='POST')
    self.last_response = res
    if res.status_code != 200:
      raise Exception('OAuth Access Token Error: ' + res.content)
    # Response content is access_token
    dic = self._qs2dict(res.content)
    self._token = dic['oauth_token']
    self._token_secret = dic['oauth_token_secret']
    self._token_type = 'access'
 
    return dic
 
 
  def get_oauth_params(self, url, params, method='GET'):
    oauth_params = {'oauth_consumer_key': self._key,
                    'oauth_signature_method': 'HMAC-SHA1',
                    'oauth_timestamp': int(time()),
                    'oauth_nonce': getrandbits(64),
                    'oauth_version': '1.0'}
    if self._token_type != None:
      oauth_params['oauth_token'] = self._token
 
    # Add other params
    params.update(oauth_params)
 
    # Sort and concat
    s = ''
    for k in sorted(params):
      s += self._quote(k) + '=' + self._quote(params[k]) + '&'
    msg = method + '&' + self._quote(url) + '&' + self._quote(s[:-1])
 
    # Maybe token_secret is empty
    key = self._secret + '&' + self._token_secret
 
    digest = hmac.new(key, msg, sha1).digest()
    params['oauth_signature'] = digest.encode('base64')[:-1]
 
    return params
 
 
  def _quote(self, s):
    return urllib.quote(str(s), '')
 
 
  def _qs2dict(self, s):
    dic = {}
    for param in s.split('&'):
      (key, value) = param.split('=')
      dic[key] = value
    return dic