OAuthが必要になったので勉強してます。
とりあえず、Google先生に聞いて概要は理解したけど
ソースを読まないことにはなんともならん、と。
で、
OAuth Examplesを読んでみた。
Pythonなので、
AppEngine-OAuth-Library by Mike Knapp (more)ってところです。
OAuthの流れ。
自サイトとTwitter連携の話。
Consumerは自サイト
Service ProviderはTwitter
Userはあなた
twt-tnkというサイトを作りました。
これが自サイトとします。
0.ConsumerはService ProviderからあらかじめOAuth利用許可を得る
このステップは，具体的にはService ProviderにConsumer登録を行い，Consumer KeyとConsumer Secretという値を取得することで行います。
これは、事前の登録で、Twitterにそのページがあります。
開発者
連携アプリの追加と設定はこちらから変更できます。
ってところです。
1.UserがConsumerに，Service Providerから認可が必要な情報へのアクセス権を取得するように指示する。
自サイトにある「Twitterにログイン」的なリンクを用意し、それをUserが押す行為です。
twt-tnk画面上部に「Twitterアカウントでログインする 」ってのがあり、
「/login」というURLになってます。
2.ConsumerはバックグラウンドでService Providerにアクセスし，未認可のRequest Tokenを取得する
自サイトがバックグラウンドでTwitterにアクセスし、未認可のRequest Tokenを取得します。
- def get(self, mode=""):
-
-
- (client, cookie) = CreateClientAndCookie(self)
-
-
- if mode == "login":
-
- return self.redirect(client.get_authorization_url())
-
- def CreateClientAndCookie(self):
-
-
- callback_url = "%s/verify" % self.request.host_url
-
-
- client = oauth.TwitterClient(CONSUMER_KEY, CONSUMER_SECRET,
- callback_url)
-
- cookie = Cookies(self, max_age=COOKIE_EXPIRE_TIME)
-
- return client, cookie
def get(self, mode=""):
#TwitterClient、Cookiesクラスの作成
(client, cookie) = CreateClientAndCookie(self)
#Twitterの認証画面表示
if mode == "login":
#Twitterの認証画面へリダイレクトする
return self.redirect(client.get_authorization_url())
def CreateClientAndCookie(self):
#Twitter認証画面からのコールバック用URLを設定
callback_url = "%s/verify" % self.request.host_url
#TwitterClientクラスの作成
client = oauth.TwitterClient(CONSUMER_KEY, CONSUMER_SECRET,
callback_url)
#Cookiesクラスの作成
cookie = Cookies(self, max_age=COOKIE_EXPIRE_TIME)
return client, cookie
3.ConsumerはUserをService Providerにリダイレクトさせる。この際Consumerは未認可のRequest TokenをURL Parameterに付加する
自サイトはUserをTwitterにリダイレクトさせます。Request TokenをGETパラメータに付与して。
上のソースを見ると、loginときたら、すぐにself.redirect(client.get_authorization_url())としていますが、
実は、client.get_authorization_url() の中でちゃんとTwitterにアクセスしてRequest Tokenを取得しています。
自サイトの「Twitterアカウントでログインする」を押すと、ちゃんと
http://twitter.com/oauth/authorize?oauth_token=
にアクセスしているのがブラウザ上で分かります。
- def get_authorization_url(self):
-
-
- token = self._get_auth_token()
- return "http://twitter.com/oauth/authorize?oauth_token=%s" % token
-
- def _get_auth_token(self):
-
-
-
-
-
-
-
- response = self.make_request(self.request_url)
- result = self._extract_credentials(response)
-
- auth_token = result["token"]
- auth_secret = result["secret"]
-
-
- auth = AuthToken(service=self.service_name,
- token=auth_token,
- secret=auth_secret)
- auth.put()
-
-
- memcache.set(self._get_memcache_auth_key(auth_token), auth_secret,
- time=20*60)
-
- return auth_token
def get_authorization_url(self):
"""Get Authorization URL."""
token = self._get_auth_token()
return "http://twitter.com/oauth/authorize?oauth_token=%s" % token
def _get_auth_token(self):
"""Get Authorization Token.
Actually gets the authorization token and secret from the service. The
token and secret are stored in our database, and the auth token is
returned.
"""
response = self.make_request(self.request_url)
result = self._extract_credentials(response)
auth_token = result["token"]
auth_secret = result["secret"]
# Save the auth token and secret in our database.
auth = AuthToken(service=self.service_name,
token=auth_token,
secret=auth_secret)
auth.put()
# Add the secret to memcache as well.
memcache.set(self._get_memcache_auth_key(auth_token), auth_secret,
time=20*60)
return auth_token
ちなみに、TwitterClientはこんなかんじ。
- class TwitterClient(OAuthClient):
-
-
-
-
-
-
- def __init__(self, consumer_key, consumer_secret, callback_url):
-
-
- OAuthClient.__init__(self,
- "twitter",
- consumer_key,
- consumer_secret,
- "http://twitter.com/oauth/request_token",
- "http://twitter.com/oauth/access_token",
- callback_url)
class TwitterClient(OAuthClient):
"""Twitter Client.
A client for talking to the Twitter API using OAuth as the
authentication model.
"""
def __init__(self, consumer_key, consumer_secret, callback_url):
"""Constructor."""
OAuthClient.__init__(self,
"twitter",
consumer_key,
consumer_secret,
"http://twitter.com/oauth/request_token",
"http://twitter.com/oauth/access_token",
callback_url)
ちなみにちなみに、バックグラウンドでリクエスト送ってる箇所は↓のかんじ
- def make_async_request(self, url, token="", secret="", additional_params=None,
- protected=False, method=urlfetch.GET):
-
-
-
-
-
-
-
-
- payload = self.prepare_request(url, token, secret, additional_params,
- method)
- if method == urlfetch.GET:
- url = "%s?%s" % (url, payload)
- payload = None
- headers = {"Authorization": "OAuth"} if protected else {}
- rpc = urlfetch.create_rpc(deadline=10.0)
- urlfetch.make_fetch_call(rpc, url, method=method, headers=headers, payload=payload)
- return rpc
-
- def make_request(self, url, token="", secret="", additional_params=None,
- protected=False, method=urlfetch.GET):
- return self.make_async_request(url, token, secret, additional_params, protected, method).get_result()
-
def make_async_request(self, url, token="", secret="", additional_params=None,
protected=False, method=urlfetch.GET):
"""Make Request.
Make an authenticated request to any OAuth protected resource.
If protected is equal to True, the Authorization: OAuth header will be set.
A urlfetch response object is returned.
"""
payload = self.prepare_request(url, token, secret, additional_params,
method)
if method == urlfetch.GET:
url = "%s?%s" % (url, payload)
payload = None
headers = {"Authorization": "OAuth"} if protected else {}
rpc = urlfetch.create_rpc(deadline=10.0)
urlfetch.make_fetch_call(rpc, url, method=method, headers=headers, payload=payload)
return rpc
def make_request(self, url, token="", secret="", additional_params=None,
protected=False, method=urlfetch.GET):
return self.make_async_request(url, token, secret, additional_params, protected, method).get_result()
4.UserはService Provider上でConsumerへのアクセス権委譲を許可する。この際Service Providerは未認可のRequest Tokenを認可済とする
Userは、Twitterのページで、「許可する」といったボタンを押下し、アクセス権委譲を許可します。
すると、TwitterはRequest Tokenを認可済としてくれます。
5.Service ProviderはUserをConsumerにリダイレクトさせる。この際Service Providerは認可済のRequest TokenをURLに含める
TwitterはUserを自サイトにリダイレクトしてくれます。
このとき、GETパラメータに認可済みのRequest Tokenが付与されてます。
6.ConsumerはバックグラウンドでService Providerと通信を行い，認可済のRequest Tokenを実際のアクセス権を示すAccess Tokenと交換する
自サイトはバックグラウンドでTwitterと通信を行い、認可済のRequest TokenをAccess Tokenと交換してもらいます。
リダイレクトされたら、自サイトは以下の処理をします。
-
-
-
- def VerityAuth(self, client, cookie):
- auth_token = self.request.get("oauth_token")
- auth_verifier = self.request.get("oauth_verifier")
- user_info = client.get_user_info(auth_token, auth_verifier=auth_verifier)
-
-
- cookie["user_token"] = user_info["token"]
- cookie["user_secret"] = user_info["secret"]
- cookie["screen_name"] = user_info["username"]
#-----------------------------------------------------------
#OAuthの認証後に、アクセストークンをCookieに保存
#-----------------------------------------------------------
def VerityAuth(self, client, cookie):
auth_token = self.request.get("oauth_token")
auth_verifier = self.request.get("oauth_verifier")
user_info = client.get_user_info(auth_token, auth_verifier=auth_verifier)
#アクセストークンをCookieへ保存
cookie["user_token"] = user_info["token"]
cookie["user_secret"] = user_info["secret"]
cookie["screen_name"] = user_info["username"]
client.get_user_info(auth_token, auth_verifier=auth_verifier)で、認可済のRequest TokenをAccess Tokenと交換してもらっています。
- def get_user_info(self, auth_token, auth_verifier=""):
-
-
-
-
-
-
- auth_token = urlunquote(auth_token)
- auth_verifier = urlunquote(auth_verifier)
-
- auth_secret = memcache.get(self._get_memcache_auth_key(auth_token))
-
- if not auth_secret:
- result = AuthToken.gql(
-
-
-
-
-
- , self.service_name, auth_token).get()
-
- if not result:
- logging.error("The auth token %s was not found in our db" % auth_token)
- raise Exception, "Could not find Auth Token in database"
- else:
- auth_secret = result.secret
-
- response = self.make_request(self.access_url,
- token=auth_token,
- secret=auth_secret,
- additional_params={"oauth_verifier":
- auth_verifier})
-
-
- result = self._extract_credentials(response)
-
-
- user_info = self._lookup_user_info(result["token"], result["secret"])
- user_info.update(result)
-
- return user_info
def get_user_info(self, auth_token, auth_verifier=""):
"""Get User Info.
Exchanges the auth token for an access token and returns a dictionary
of information about the authenticated user.
"""
auth_token = urlunquote(auth_token)
auth_verifier = urlunquote(auth_verifier)
auth_secret = memcache.get(self._get_memcache_auth_key(auth_token))
if not auth_secret:
result = AuthToken.gql("""
WHERE
service = :1 AND
token = :2
LIMIT
1
""", self.service_name, auth_token).get()
if not result:
logging.error("The auth token %s was not found in our db" % auth_token)
raise Exception, "Could not find Auth Token in database"
else:
auth_secret = result.secret
response = self.make_request(self.access_url,
token=auth_token,
secret=auth_secret,
additional_params={"oauth_verifier":
auth_verifier})
# Extract the access token/secret from the response.
result = self._extract_credentials(response)
# Try to collect some information about this user from the service.
user_info = self._lookup_user_info(result["token"], result["secret"])
user_info.update(result)
return user_info
7.Consumerは6)で得られたTokenを利用して，特定の情報にアクセスする
その他のメソッドなど。
- def _get_memcache_auth_key(self, auth_token):
-
- return "oauth_%s_%s" % (self.service_name, auth_token)
-
- def _extract_credentials(self, result):
-
-
-
-
-
-
- token = None
- secret = None
- parsed_results = parse_qs(result.content)
-
- if "oauth_token" in parsed_results:
- token = parsed_results["oauth_token"][0]
-
- if "oauth_token_secret" in parsed_results:
- secret = parsed_results["oauth_token_secret"][0]
-
- if not (token and secret) or result.status_code != 200:
- logging.error("Could not extract token/secret: %s" % result.content)
- raise OAuthException("Problem talking to the service")
-
- return {
- "service": self.service_name,
- "token": token,
- "secret": secret
- }
def _get_memcache_auth_key(self, auth_token):
return "oauth_%s_%s" % (self.service_name, auth_token)
def _extract_credentials(self, result):
"""Extract Credentials.
Returns an dictionary containing the token and secret (if present).
Throws an Exception otherwise.
"""
token = None
secret = None
parsed_results = parse_qs(result.content)
if "oauth_token" in parsed_results:
token = parsed_results["oauth_token"][0]
if "oauth_token_secret" in parsed_results:
secret = parsed_results["oauth_token_secret"][0]
if not (token and secret) or result.status_code != 200:
logging.error("Could not extract token/secret: %s" % result.content)
raise OAuthException("Problem talking to the service")
return {
"service": self.service_name,
"token": token,
"secret": secret
}
Twitterからもらえるuser_infoは↓のかんじ
- def _lookup_user_info(self, access_token, access_secret):
-
-
-
-
-
- response = self.make_request(
- "http://twitter.com/account/verify_credentials.json",
- token=access_token, secret=access_secret, protected=True)
-
- data = json.loads(response.content)
-
- user_info = self._get_default_user_info()
- user_info["id"] = data["id"]
- user_info["username"] = data["screen_name"]
- user_info["name"] = data["name"]
- user_info["picture"] = data["profile_image_url"]
-
- return user_info
def _lookup_user_info(self, access_token, access_secret):
"""Lookup User Info.
Lookup the user on Twitter.
"""
response = self.make_request(
"http://twitter.com/account/verify_credentials.json",
token=access_token, secret=access_secret, protected=True)
data = json.loads(response.content)
user_info = self._get_default_user_info()
user_info["id"] = data["id"]
user_info["username"] = data["screen_name"]
user_info["name"] = data["name"]
user_info["picture"] = data["profile_image_url"]
return user_info
