"""
dropbox.session.DropboxSession is responsible for holding OAuth authentication info
(app key/secret, request key/secret, access key/secret) as well as configuration information for your app
('app_folder' or 'dropbox' access type, optional locale preference). It knows how to
use all of this information to craft properly constructed requests to Dropbox.
A DropboxSession object must be passed to a dropbox.client.DropboxClient object upon
initialization.
"""
from __future__ import absolute_import
import random
import sys
import time
import urllib
try:
from urlparse import parse_qs
except ImportError:
# fall back for Python 2.5
from cgi import parse_qs
from . import rest
class OAuthToken(object):
__slots__ = ('key', 'secret')
def __init__(self, key, secret):
self.key = key
self.secret = secret
class DropboxSession(object):
API_VERSION = 1
API_HOST = "api.dropbox.com"
WEB_HOST = "www.dropbox.com"
API_CONTENT_HOST = "api-content.dropbox.com"
def __init__(self, consumer_key, consumer_secret, access_type, locale=None, rest_client=rest.RESTClient):
"""Initialize a DropboxSession object.
Your consumer key and secret are available
at https://www.dropbox.com/developers/apps
Args:
- ``access_type``: Either 'dropbox' or 'app_folder'. All path-based operations
will occur relative to either the user's Dropbox root directory
or your application's app folder.
- ``locale``: A locale string ('en', 'pt_PT', etc.) [optional]
The locale setting will be used to translate any user-facing error
messages that the server generates. At this time Dropbox supports
'en', 'es', 'fr', 'de', and 'ja', though we will be supporting more
languages in the future. If you send a language the server doesn't
support, messages will remain in English. Look for these translated
messages in rest.ErrorResponse exceptions as e.user_error_msg.
"""
assert access_type in ['dropbox', 'app_folder'], "expected access_type of 'dropbox' or 'app_folder'"
self.consumer_creds = OAuthToken(consumer_key, consumer_secret)
self.token = None
self.request_token = None
self.root = 'sandbox' if access_type == 'app_folder' else 'dropbox'
self.locale = locale
self.rest_client = rest_client
def is_linked(self):
"""Return whether the DropboxSession has an access token attached."""
return bool(self.token)
def unlink(self):
"""Remove any attached access token from the DropboxSession."""
self.token = None
def set_token(self, access_token, access_token_secret):
"""Attach an access token to the DropboxSession.
Note that the access 'token' is made up of both a token string
and a secret string.
"""
self.token = OAuthToken(access_token, access_token_secret)
def set_request_token(self, request_token, request_token_secret):
"""Attach an request token to the DropboxSession.
Note that the request 'token' is made up of both a token string
and a secret string.
"""
self.request_token = OAuthToken(request_token, request_token_secret)
def build_path(self, target, params=None):
"""Build the path component for an API URL.
This method urlencodes the parameters, adds them
to the end of the target url, and puts a marker for the API
version in front.
Args:
- ``target``: A target url (e.g. '/files') to build upon.
- ``params``: A dictionary of parameters (name to value). [optional]
Returns:
- The path and parameters components of an API URL.
"""
if sys.version_info < (3,) and type(target) == unicode:
target = target.encode("utf8")
target_path = urllib.quote(target)
params = params or {}
params = params.copy()
if self.locale:
params['locale'] = self.locale
if params:
return "/%d%s?%s" % (self.API_VERSION, target_path, urllib.urlencode(params))
else:
return "/%d%s" % (self.API_VERSION, target_path)
def build_url(self, host, target, params=None):
"""Build an API URL.
This method adds scheme and hostname to the path
returned from build_path.
Args:
- ``target``: A target url (e.g. '/files') to build upon.
- ``params``: A dictionary of parameters (name to value). [optional]
Returns:
- The full API URL.
"""
return "https://%s%s" % (host, self.build_path(target, params))
def build_authorize_url(self, request_token, oauth_callback=None):
"""Build a request token authorization URL.
After obtaining a request token, you'll need to send the user to
the URL returned from this function so that they can confirm that
they want to connect their account to your app.
Args:
- ``request_token``: A request token from obtain_request_token.
- ``oauth_callback``: A url to redirect back to with the authorized
request token.
Returns:
- An authorization for the given request token.
"""
params = {'oauth_token': request_token.key,
}
if oauth_callback:
params['oauth_callback'] = oauth_callback
return self.build_url(self.WEB_HOST, '/oauth/authorize', params)
def obtain_request_token(self):
"""Obtain a request token from the Dropbox API.
This is your first step in the OAuth process. You call this to get a
request_token from the Dropbox server that you can then use with
DropboxSession.build_authorize_url() to get the user to authorize it.
After it's authorized you use this token with
DropboxSession.obtain_access_token() to get an access token.
NOTE: You should only need to do this once for each user, and then you
can store the access token for that user for later operations.
Returns:
- An dropbox.session.OAuthToken representing the request token Dropbox assigned
to this app. Also attaches the request token as self.request_token.
"""
self.token = None # clear any token currently on the request
url = self.build_url(self.API_HOST, '/oauth/request_token')
headers, params = self.build_access_headers('POST', url)
response = self.rest_client.POST(url, headers=headers, params=params, raw_response=True)
self.request_token = self._parse_token(response.read())
return self.request_token
def obtain_access_token(self, request_token=None):
"""Obtain an access token for a user.
After you get a request token, and then send the user to the authorize
URL, you can use the authorized request token with this method to get the
access token to use for future operations. The access token is stored on
the session object.
Args:
- ``request_token``: A request token from obtain_request_token. [optional]
The request_token should have been authorized via the
authorization url from build_authorize_url. If you don't pass
a request_token, the fallback is self.request_token, which
will exist if you previously called obtain_request_token on this
DropboxSession instance.
Returns:
- An tuple of (key, secret) representing the access token Dropbox assigned
to this app and user. Also attaches the access token as self.token.
"""
request_token = request_token or self.request_token
assert request_token, "No request_token available on the session. Please pass one."
url = self.build_url(self.API_HOST, '/oauth/access_token')
headers, params = self.build_access_headers('POST', url, request_token=request_token)
response = self.rest_client.POST(url, headers=headers, params=params, raw_response=True)
self.token = self._parse_token(response.read())
return self.token
def build_access_headers(self, method, resource_url, params=None, request_token=None):
"""Build OAuth access headers for a future request.
Args:
- ``method``: The HTTP method being used (e.g. 'GET' or 'POST').
- ``resource_url``: The full url the request will be made to.
- ``params``: A dictionary of parameters to add to what's already on the url.
Typically, this would consist of POST parameters.
Returns:
- A tuple of (header_dict, params) where header_dict is a dictionary
of header names and values appropriate for passing into dropbox.rest.RESTClient
and params is a dictionary like the one that was passed in, but augmented with
oauth-related parameters as appropriate.
"""
if params is None:
params = {}
else:
params = params.copy()
oauth_params = {
'oauth_consumer_key' : self.consumer_creds.key,
'oauth_timestamp' : self._generate_oauth_timestamp(),
'oauth_nonce' : self._generate_oauth_nonce(),
'oauth_version' : self._oauth_version(),
}
token = request_token if request_token is not None else self.token
if token:
oauth_params['oauth_token'] = token.key
self._oauth_sign_request(oauth_params, self.consumer_creds, token)
params.update(oauth_params)
return {}, params
@classmethod
def _oauth_sign_request(cls, params, consumer_pair, token_pair):
params.update({'oauth_signature_method' : 'PLAINTEXT',
'oauth_signature' : ('%s&%s' % (consumer_pair.secret, token_pair.secret)
if token_pair is not None else
'%s&' % (consumer_pair.secret,))})
@classmethod
def _generate_oauth_timestamp(cls):
return int(time.time())
@classmethod
def _generate_oauth_nonce(cls, length=8):
return ''.join([str(random.randint(0, 9)) for i in range(length)])
@classmethod
def _oauth_version(cls):
return '1.0'
@classmethod
def _parse_token(cls, s):
if not s:
raise ValueError("Invalid parameter string.")
params = parse_qs(s, keep_blank_values=False)
if not params:
raise ValueError("Invalid parameter string: %r" % s)
try:
key = params['oauth_token'][0]
except Exception:
raise ValueError("'oauth_token' not found in OAuth request.")
try:
secret = params['oauth_token_secret'][0]
except Exception:
raise ValueError("'oauth_token_secret' not found in "
"OAuth request.")
return OAuthToken(key, secret)