尝试使用Tornado Web进行Dropbox OAuth认证
这是我的混合文件：
import base64
import binascii
import hashlib
import hmac
import logging
import time
import urllib
import urlparse
import uuid
import tornado.web
import tornado.auth
from tornado import httpclient
from tornado import escape
from tornado.httputil import url_concat
from tornado.util import bytes_type, b
class DropboxMixin(tornado.auth.OAuthMixin):
""" Dropbox OAuth authentication.
"""
_OAUTH_REQUEST_TOKEN_URL = "https://api.dropbox.com/1/oauth/request_token"
_OAUTH_ACCESS_TOKEN_URL = "https://api.dropbox.com/1/oauth/access_token"
_OAUTH_AUTHORIZE_URL = "https://www.dropbox.com/1/oauth/authorize"
def authorize_redirect(self, callback_uri=None, extra_params=None,
http_client=None):
"""Redirects the user to obtain OAuth authorization for this service.
Twitter and FriendFeed both require that you register a Callback
URL with your application. You should call this method to log the
user in, and then call get_authenticated_user() in the handler
you registered as your Callback URL to complete the authorization
process.
This method sets a cookie called _oauth_request_token which is
subsequently used (and cleared) in get_authenticated_user for
security purposes.
"""
if callback_uri and getattr(self, "_OAUTH_NO_CALLBACKS", False):
raise Exception("This service does not support oauth_callback")
if http_client is None:
http_client = httpclient.AsyncHTTPClient()
http_client.fetch(
self._oauth_request_token_url(),
self.async_callback(
self._on_request_token, self._OAUTH_AUTHORIZE_URL,
callback_uri))
def get_authenticated_user(self, callback, http_client=None):
"""Gets the OAuth authorized user and access token on callback.
This method should be called from the handler for your registered
OAuth Callback URL to complete the registration process. We call
callback with the authenticated user, which in addition to standard
attributes like 'name' includes the 'access_key' attribute, which
contains the OAuth access you can use to make authorized requests
to this service on behalf of the user.
"""
request_key = escape.utf8(self.get_argument("oauth_token"))
oauth_verifier = self.get_argument("oauth_verifier", None)
request_cookie = self.get_cookie("_oauth_request_token")
if not request_cookie:
logging.warning("Missing OAuth request token cookie")
callback(None)
return
self.clear_cookie("_oauth_request_token")
cookie_key, cookie_secret = [base64.b64decode(escape.utf8(i)) for i in request_cookie.split("|")]
if cookie_key != request_key:
logging.info((cookie_key, request_key, request_cookie))
logging.warning("Request token does not match cookie")
callback(None)
return
token = dict(key=cookie_key, secret=cookie_secret)
if oauth_verifier:
token["verifier"] = oauth_verifier
if http_client is None:
http_client = httpclient.AsyncHTTPClient()
http_client.fetch(self._oauth_access_token_url(token),
self.async_callback(self._on_access_token, callback))
def dropbox_request(self, path, callback, access_token=None,
post_args=None, **args):
# Add the OAuth resource request signature if we have credentials
url = "http://api.dropbox.com/1" + path
如果有人关心的话，网址应该是 https://api.dropbox.com/1" + 路径
if access_token:
all_args = {}
all_args.update(args)
all_args.update(post_args or {})
method = "POST" if post_args is not None else "GET"
oauth = self._oauth_request_parameters(
url, access_token, all_args, method=method)
args.update(oauth)
if args: url += "?" + urllib.urlencode(args)
callback = self.async_callback(self._on_dropbox_request, callback)
http = httpclient.AsyncHTTPClient()
if post_args is not None:
http.fetch(url, method="POST", body=urllib.urlencode(post_args),
callback=callback)
else:
http.fetch(url, callback=callback)
def _oauth_request_token_url(self, callback_uri= None, extra_params=None):
consumer_token = self._oauth_consumer_token()
url = self._OAUTH_REQUEST_TOKEN_URL
args = dict(
oauth_consumer_key=consumer_token["key"],
oauth_signature_method="HMAC-SHA1",
oauth_timestamp=str(int(time.time())),
oauth_nonce=binascii.b2a_hex(uuid.uuid4().bytes),
oauth_version=getattr(self, "_OAUTH_VERSION", "1.0"),
)
signature = _oauth_signature(consumer_token, "GET", url, args)
args["oauth_signature"] = signature
return url + "?" + urllib.urlencode(args)
def _oauth_access_token_url(self, request_token):
consumer_token = self._oauth_consumer_token()
url = self._OAUTH_ACCESS_TOKEN_URL
args = dict(
oauth_consumer_key=consumer_token["key"],
oauth_token=request_token["key"],
oauth_signature_method="HMAC-SHA1",
oauth_timestamp=str(int(time.time())),
oauth_nonce=binascii.b2a_hex(uuid.uuid4().bytes),
oauth_version=getattr(self, "_OAUTH_VERSION", "1.0"),
)
if "verifier" in request_token:
args["oauth_verifier"]=request_token["verifier"]
signature = _oauth_signature(consumer_token, "GET", url, args,
request_token)
args["oauth_signature"] = signature
return url + "?" + urllib.urlencode(args)
def _on_dropbox_request(self, callback, response):
if response.error:
print("Error response %s fetching %s", response.error,
response.request.url)
callback(None)
return
callback(escape.json_decode(response.body))
def _oauth_consumer_token(self):
self.require_setting("dropbox_consumer_key", "Dropbox OAuth")
self.require_setting("dropbox_consumer_secret", "Dropbox OAuth")
return dict(
key=self.settings["dropbox_consumer_key"],
secret=self.settings["dropbox_consumer_secret"])
def _oauth_get_user(self, access_token, callback):
callback = self.async_callback(self._parse_user_response, callback)
self.dropbox_request(
"/account/info",
access_token=access_token,
callback=callback)
def _oauth_request_parameters(self, url, access_token, parameters={},
method="GET"):
"""Returns the OAuth parameters as a dict for the given request.
parameters should include all POST arguments and query string arguments
that will be sent with the request.
"""
consumer_token = self._oauth_consumer_token()
base_args = dict(
oauth_consumer_key=consumer_token["key"],
oauth_token=access_token["key"],
oauth_signature_method="HMAC-SHA1",
oauth_timestamp=str(int(time.time())),
oauth_nonce=binascii.b2a_hex(uuid.uuid4().bytes),
oauth_version=getattr(self, "_OAUTH_VERSION", "1.0"),
)
args = {}
args.update(base_args)
args.update(parameters)
signature = _oauth_signature(consumer_token, method, url, args,
access_token)
base_args["oauth_signature"] = signature
return base_args
def _parse_user_response(self, callback, user):
if user:
user["username"] = user["display_name"]
callback(user)
def _oauth_signature(consumer_token, method, url, parameters={}, token=None):
"""Calculates the HMAC-SHA1 OAuth signature for the given request.
See http://oauth.net/core/1.0/#signing_process
"""
parts = urlparse.urlparse(url)
scheme, netloc, path = parts[:3]
normalized_url = scheme.lower() + "://" + netloc.lower() + path
base_elems = []
base_elems.append(method.upper())
base_elems.append(normalized_url)
base_elems.append("&".join("%s=%s" % (k, _oauth_escape(str(v)))
for k, v in sorted(parameters.items())))
base_string = "&".join(_oauth_escape(e) for e in base_elems)
key_elems = [escape.utf8(consumer_token["secret"])]
key_elems.append(escape.utf8(token["secret"] if token else ""))
key = b("&").join(key_elems)
hash = hmac.new(key, escape.utf8(base_string), hashlib.sha1)
return binascii.b2a_base64(hash.digest())[:-1]
def _oauth_escape(val):
if isinstance(val, unicode):
val = val.encode("utf-8")
return urllib.quote(val, safe="~")
def _oauth_parse_response(body):
p = escape.parse_qs(body, keep_blank_values=False)
token = dict(key=p[b("oauth_token")][0], secret=p[b("oauth_token_secret")][0])
# Add the extra parameters the Provider included to the token
special = (b("oauth_token"), b("oauth_token_secret"))
token.update((k, p[k][0]) for k in p if k not in special)
return token
我的视图
class DropboxIndex(BaseHandler, DropboxMixin):
@tornado.web.asynchronous
def get(self):
if self.get_argument("oauth_token", None):
self.get_authenticated_user(self.async_callback(self._on_dbox_auth))
return
self.authorize_redirect()
def _on_dbox_auth(self, token):
from pprint import pprint
pprint(token)
self.redirect("/app/dropbox")
我的网址模式
patterns = [
(r"/", Index),
(r"/help/?", Help),
# User authentication
(r"/user/login/?", Login),
(r"/user/logout/?", LogoutHandler),
(r"/user/edit/?", IdentityIndex),
(r"/user/register/?", Register),
(r"/user/twitter/auth/?", TwitterLogin),
#(r"/user/google/auth/?", GoogleLogin),
(r"/user/facebook/auth/?", FacebookLogin),
(r"/app/dropbox/?", DropboxIndex),
(r"/app/dropbox/account/?", DropboxAccount),
(r"/social/?", SocialIndex),
(r"/api/tweets/?", apiSocialTweets),
(r"/media/(.*)", tornado.web.StaticFileHandler, {"path" : static_serve}),
]
一切都正常，除了无法返回到我的回调网址，并带回oauth_token。我在查询中看到了oauth_token，并且可以用它来授权我的应用程序与Dropbox连接……但就是无法把oauth_token拿回来并使用。
1 个回答
0
好吧，这段代码挺复杂的，所以如果我的回答没什么帮助，请见谅 -
Urllib 和 Urllib2 这两个库在处理异步请求时基本上是不能兼容的。如果你想知道原因和解决办法，可以看看这个问题。
我建议你在代码中使用 httplib， 如果这样有帮助或者你想了解更多，欢迎留言。
谢谢，
Sushant