-"""\r
-The MIT License\r
-\r
-Copyright (c) 2007 Leah Culver\r
-\r
-Permission is hereby granted, free of charge, to any person obtaining a copy\r
-of this software and associated documentation files (the "Software"), to deal\r
-in the Software without restriction, including without limitation the rights\r
-to use, copy, modify, merge, publish, distribute, sublicense, and/or sell\r
-copies of the Software, and to permit persons to whom the Software is\r
-furnished to do so, subject to the following conditions:\r
-\r
-The above copyright notice and this permission notice shall be included in\r
-all copies or substantial portions of the Software.\r
-\r
-THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR\r
-IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,\r
-FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE\r
-AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER\r
-LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,\r
-OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN\r
-THE SOFTWARE.\r
-"""\r
-\r
-import cgi\r
-import urllib\r
-import time\r
-import random\r
-import urlparse\r
-import hmac\r
-import binascii\r
-\r
-\r
-VERSION = '1.0' # Hi Blaine!\r
-HTTP_METHOD = 'GET'\r
-SIGNATURE_METHOD = 'PLAINTEXT'\r
-\r
-\r
-class OAuthError(RuntimeError):\r
- """Generic exception class."""\r
- def __init__(self, message='OAuth error occured.'):\r
- self.message = message\r
-\r
-def build_authenticate_header(realm=''):\r
- """Optional WWW-Authenticate header (401 error)"""\r
- return {'WWW-Authenticate': 'OAuth realm="%s"' % realm}\r
-\r
-def escape(s):\r
- """Escape a URL including any /."""\r
- return urllib.quote(s, safe='~')\r
-\r
-def _utf8_str(s):\r
- """Convert unicode to utf-8."""\r
- if isinstance(s, unicode):\r
- return s.encode("utf-8")\r
- else:\r
- return str(s)\r
-\r
-def generate_timestamp():\r
- """Get seconds since epoch (UTC)."""\r
- return int(time.time())\r
-\r
-def generate_nonce(length=8):\r
- """Generate pseudorandom number."""\r
- return ''.join([str(random.randint(0, 9)) for i in range(length)])\r
-\r
-\r
-class OAuthConsumer(object):\r
- """Consumer of OAuth authentication.\r
-\r
- OAuthConsumer is a data type that represents the identity of the Consumer\r
- via its shared secret with the Service Provider.\r
-\r
- """\r
- key = None\r
- secret = None\r
-\r
- def __init__(self, key, secret):\r
- self.key = key\r
- self.secret = secret\r
-\r
-\r
-class OAuthToken(object):\r
- """OAuthToken is a data type that represents an End User via either an access\r
- or request token.\r
-\r
- key -- the token\r
- secret -- the token secret\r
-\r
- """\r
- key = None\r
- secret = None\r
-\r
- def __init__(self, key, secret):\r
- self.key = key\r
- self.secret = secret\r
-\r
- def to_string(self):\r
- return urllib.urlencode({'oauth_token': self.key,\r
- 'oauth_token_secret': self.secret})\r
-\r
- def from_string(s):\r
- """ Returns a token from something like:\r
- oauth_token_secret=xxx&oauth_token=xxx\r
- """\r
- params = cgi.parse_qs(s, keep_blank_values=False)\r
- key = params['oauth_token'][0]\r
- secret = params['oauth_token_secret'][0]\r
- return OAuthToken(key, secret)\r
- from_string = staticmethod(from_string)\r
-\r
- def __str__(self):\r
- return self.to_string()\r
-\r
-\r
-class OAuthRequest(object):\r
- """OAuthRequest represents the request and can be serialized.\r
-\r
- OAuth parameters:\r
- - oauth_consumer_key\r
- - oauth_token\r
- - oauth_signature_method\r
- - oauth_signature\r
- - oauth_timestamp\r
- - oauth_nonce\r
- - oauth_version\r
- ... any additional parameters, as defined by the Service Provider.\r
- """\r
- parameters = None # OAuth parameters.\r
- http_method = HTTP_METHOD\r
- http_url = None\r
- version = VERSION\r
-\r
- def __init__(self, http_method=HTTP_METHOD, http_url=None, parameters=None):\r
- self.http_method = http_method\r
- self.http_url = http_url\r
- self.parameters = parameters or {}\r
-\r
- def set_parameter(self, parameter, value):\r
- self.parameters[parameter] = value\r
-\r
- def get_parameter(self, parameter):\r
- try:\r
- return self.parameters[parameter]\r
- except:\r
- raise OAuthError('Parameter not found: %s' % parameter)\r
-\r
- def _get_timestamp_nonce(self):\r
- return self.get_parameter('oauth_timestamp'), self.get_parameter(\r
- 'oauth_nonce')\r
-\r
- def get_nonoauth_parameters(self):\r
- """Get any non-OAuth parameters."""\r
- parameters = {}\r
- for k, v in self.parameters.iteritems():\r
- # Ignore oauth parameters.\r
- if k.find('oauth_') < 0:\r
- parameters[k] = v\r
- return parameters\r
-\r
- def to_header(self, realm=''):\r
- """Serialize as a header for an HTTPAuth request."""\r
- auth_header = 'OAuth realm="%s"' % realm\r
- # Add the oauth parameters.\r
- if self.parameters:\r
- for k, v in self.parameters.iteritems():\r
- if k[:6] == 'oauth_':\r
- auth_header += ', %s="%s"' % (k, escape(str(v)))\r
- return {'Authorization': auth_header}\r
-\r
- def to_postdata(self):\r
- """Serialize as post data for a POST request."""\r
- return '&'.join(['%s=%s' % (escape(str(k)), escape(str(v))) \\r
- for k, v in self.parameters.iteritems()])\r
-\r
- def to_url(self):\r
- """Serialize as a URL for a GET request."""\r
- return '%s?%s' % (self.get_normalized_http_url(), self.to_postdata())\r
-\r
- def get_normalized_parameters(self):\r
- """Return a string that contains the parameters that must be signed."""\r
- params = self.parameters\r
- try:\r
- # Exclude the signature if it exists.\r
- del params['oauth_signature']\r
- except:\r
- pass\r
- # Escape key values before sorting.\r
- key_values = [(escape(_utf8_str(k)), escape(_utf8_str(v))) \\r
- for k,v in params.items()]\r
- # Sort lexicographically, first after key, then after value.\r
- key_values.sort()\r
- # Combine key value pairs into a string.\r
- return '&'.join(['%s=%s' % (k, v) for k, v in key_values])\r
-\r
- def get_normalized_http_method(self):\r
- """Uppercases the http method."""\r
- return self.http_method.upper()\r
-\r
- def get_normalized_http_url(self):\r
- """Parses the URL and rebuilds it to be scheme://host/path."""\r
- parts = urlparse.urlparse(self.http_url)\r
- scheme, netloc, path = parts[:3]\r
- # Exclude default port numbers.\r
- if scheme == 'http' and netloc[-3:] == ':80':\r
- netloc = netloc[:-3]\r
- elif scheme == 'https' and netloc[-4:] == ':443':\r
- netloc = netloc[:-4]\r
- return '%s://%s%s' % (scheme, netloc, path)\r
-\r
- def sign_request(self, signature_method, consumer, token):\r
- """Set the signature parameter to the result of build_signature."""\r
- # Set the signature method.\r
- self.set_parameter('oauth_signature_method',\r
- signature_method.get_name())\r
- # Set the signature.\r
- self.set_parameter('oauth_signature',\r
- self.build_signature(signature_method, consumer, token))\r
-\r
- def build_signature(self, signature_method, consumer, token):\r
- """Calls the build signature method within the signature method."""\r
- return signature_method.build_signature(self, consumer, token)\r
-\r
- def from_request(http_method, http_url, headers=None, parameters=None,\r
- query_string=None):\r
- """Combines multiple parameter sources."""\r
- if parameters is None:\r
- parameters = {}\r
-\r
- # Headers\r
- if headers and 'Authorization' in headers:\r
- auth_header = headers['Authorization']\r
- # Check that the authorization header is OAuth.\r
- if auth_header.index('OAuth') > -1:\r
- auth_header = auth_header.lstrip('OAuth ')\r
- try:\r
- # Get the parameters from the header.\r
- header_params = OAuthRequest._split_header(auth_header)\r
- parameters.update(header_params)\r
- except:\r
- raise OAuthError('Unable to parse OAuth parameters from '\r
- 'Authorization header.')\r
-\r
- # GET or POST query string.\r
- if query_string:\r
- query_params = OAuthRequest._split_url_string(query_string)\r
- parameters.update(query_params)\r
-\r
- # URL parameters.\r
- param_str = urlparse.urlparse(http_url)[4] # query\r
- url_params = OAuthRequest._split_url_string(param_str)\r
- parameters.update(url_params)\r
-\r
- if parameters:\r
- return OAuthRequest(http_method, http_url, parameters)\r
-\r
- return None\r
- from_request = staticmethod(from_request)\r
-\r
- def from_consumer_and_token(oauth_consumer, token=None,\r
- http_method=HTTP_METHOD, http_url=None, parameters=None):\r
- if not parameters:\r
- parameters = {}\r
-\r
- defaults = {\r
- 'oauth_consumer_key': oauth_consumer.key,\r
- 'oauth_timestamp': generate_timestamp(),\r
- 'oauth_nonce': generate_nonce(),\r
- 'oauth_version': OAuthRequest.version,\r
- }\r
-\r
- defaults.update(parameters)\r
- parameters = defaults\r
-\r
- if token:\r
- parameters['oauth_token'] = token.key\r
-\r
- return OAuthRequest(http_method, http_url, parameters)\r
- from_consumer_and_token = staticmethod(from_consumer_and_token)\r
-\r
- def from_token_and_callback(token, callback=None, http_method=HTTP_METHOD,\r
- http_url=None, parameters=None):\r
- if not parameters:\r
- parameters = {}\r
-\r
- parameters['oauth_token'] = token.key\r
-\r
- if callback:\r
- parameters['oauth_callback'] = callback\r
-\r
- return OAuthRequest(http_method, http_url, parameters)\r
- from_token_and_callback = staticmethod(from_token_and_callback)\r
-\r
- def _split_header(header):\r
- """Turn Authorization: header into parameters."""\r
- params = {}\r
- parts = header.split(',')\r
- for param in parts:\r
- # Ignore realm parameter.\r
- if param.find('realm') > -1:\r
- continue\r
- # Remove whitespace.\r
- param = param.strip()\r
- # Split key-value.\r
- param_parts = param.split('=', 1)\r
- # Remove quotes and unescape the value.\r
- params[param_parts[0]] = urllib.unquote(param_parts[1].strip('\"'))\r
- return params\r
- _split_header = staticmethod(_split_header)\r
-\r
- def _split_url_string(param_str):\r
- """Turn URL string into parameters."""\r
- parameters = cgi.parse_qs(param_str, keep_blank_values=False)\r
- for k, v in parameters.iteritems():\r
- parameters[k] = urllib.unquote(v[0])\r
- return parameters\r
- _split_url_string = staticmethod(_split_url_string)\r
-\r
-class OAuthServer(object):\r
- """A worker to check the validity of a request against a data store."""\r
- timestamp_threshold = 300 # In seconds, five minutes.\r
- version = VERSION\r
- signature_methods = None\r
- data_store = None\r
-\r
- def __init__(self, data_store=None, signature_methods=None):\r
- self.data_store = data_store\r
- self.signature_methods = signature_methods or {}\r
-\r
- def set_data_store(self, data_store):\r
- self.data_store = data_store\r
-\r
- def get_data_store(self):\r
- return self.data_store\r
-\r
- def add_signature_method(self, signature_method):\r
- self.signature_methods[signature_method.get_name()] = signature_method\r
- return self.signature_methods\r
-\r
- def fetch_request_token(self, oauth_request):\r
- """Processes a request_token request and returns the\r
- request token on success.\r
- """\r
- try:\r
- # Get the request token for authorization.\r
- token = self._get_token(oauth_request, 'request')\r
- except OAuthError:\r
- # No token required for the initial token request.\r
- version = self._get_version(oauth_request)\r
- consumer = self._get_consumer(oauth_request)\r
- self._check_signature(oauth_request, consumer, None)\r
- # Fetch a new token.\r
- token = self.data_store.fetch_request_token(consumer)\r
- return token\r
-\r
- def fetch_access_token(self, oauth_request):\r
- """Processes an access_token request and returns the\r
- access token on success.\r
- """\r
- version = self._get_version(oauth_request)\r
- consumer = self._get_consumer(oauth_request)\r
- # Get the request token.\r
- token = self._get_token(oauth_request, 'request')\r
- self._check_signature(oauth_request, consumer, token)\r
- new_token = self.data_store.fetch_access_token(consumer, token)\r
- return new_token\r
-\r
- def verify_request(self, oauth_request):\r
- """Verifies an api call and checks all the parameters."""\r
- # -> consumer and token\r
- version = self._get_version(oauth_request)\r
- consumer = self._get_consumer(oauth_request)\r
- # Get the access token.\r
- token = self._get_token(oauth_request, 'access')\r
- self._check_signature(oauth_request, consumer, token)\r
- parameters = oauth_request.get_nonoauth_parameters()\r
- return consumer, token, parameters\r
-\r
- def authorize_token(self, token, user):\r
- """Authorize a request token."""\r
- return self.data_store.authorize_request_token(token, user)\r
-\r
- def get_callback(self, oauth_request):\r
- """Get the callback URL."""\r
- return oauth_request.get_parameter('oauth_callback')\r
-\r
- def build_authenticate_header(self, realm=''):\r
- """Optional support for the authenticate header."""\r
- return {'WWW-Authenticate': 'OAuth realm="%s"' % realm}\r
-\r
- def _get_version(self, oauth_request):\r
- """Verify the correct version request for this server."""\r
- try:\r
- version = oauth_request.get_parameter('oauth_version')\r
- except:\r
- version = VERSION\r
- if version and version != self.version:\r
- raise OAuthError('OAuth version %s not supported.' % str(version))\r
- return version\r
-\r
- def _get_signature_method(self, oauth_request):\r
- """Figure out the signature with some defaults."""\r
- try:\r
- signature_method = oauth_request.get_parameter(\r
- 'oauth_signature_method')\r
- except:\r
- signature_method = SIGNATURE_METHOD\r
- try:\r
- # Get the signature method object.\r
- signature_method = self.signature_methods[signature_method]\r
- except:\r
- signature_method_names = ', '.join(self.signature_methods.keys())\r
- raise OAuthError('Signature method %s not supported try one of the '\r
- 'following: %s' % (signature_method, signature_method_names))\r
-\r
- return signature_method\r
-\r
- def _get_consumer(self, oauth_request):\r
- consumer_key = oauth_request.get_parameter('oauth_consumer_key')\r
- consumer = self.data_store.lookup_consumer(consumer_key)\r
- if not consumer:\r
- raise OAuthError('Invalid consumer.')\r
- return consumer\r
-\r
- def _get_token(self, oauth_request, token_type='access'):\r
- """Try to find the token for the provided request token key."""\r
- token_field = oauth_request.get_parameter('oauth_token')\r
- token = self.data_store.lookup_token(token_type, token_field)\r
- if not token:\r
- raise OAuthError('Invalid %s token: %s' % (token_type, token_field))\r
- return token\r
-\r
- def _check_signature(self, oauth_request, consumer, token):\r
- timestamp, nonce = oauth_request._get_timestamp_nonce()\r
- self._check_timestamp(timestamp)\r
- self._check_nonce(consumer, token, nonce)\r
- signature_method = self._get_signature_method(oauth_request)\r
- try:\r
- signature = oauth_request.get_parameter('oauth_signature')\r
- except:\r
- raise OAuthError('Missing signature.')\r
- # Validate the signature.\r
- valid_sig = signature_method.check_signature(oauth_request, consumer,\r
- token, signature)\r
- if not valid_sig:\r
- key, base = signature_method.build_signature_base_string(\r
- oauth_request, consumer, token)\r
- raise OAuthError('Invalid signature. Expected signature base '\r
- 'string: %s' % base)\r
- built = signature_method.build_signature(oauth_request, consumer, token)\r
-\r
- def _check_timestamp(self, timestamp):\r
- """Verify that timestamp is recentish."""\r
- timestamp = int(timestamp)\r
- now = int(time.time())\r
- lapsed = now - timestamp\r
- if lapsed > self.timestamp_threshold:\r
- raise OAuthError('Expired timestamp: given %d and now %s has a '\r
- 'greater difference than threshold %d' %\r
- (timestamp, now, self.timestamp_threshold))\r
-\r
- def _check_nonce(self, consumer, token, nonce):\r
- """Verify that the nonce is uniqueish."""\r
- nonce = self.data_store.lookup_nonce(consumer, token, nonce)\r
- if nonce:\r
- raise OAuthError('Nonce already used: %s' % str(nonce))\r
-\r
-\r
-class OAuthClient(object):\r
- """OAuthClient is a worker to attempt to execute a request."""\r
- consumer = None\r
- token = None\r
-\r
- def __init__(self, oauth_consumer, oauth_token):\r
- self.consumer = oauth_consumer\r
- self.token = oauth_token\r
-\r
- def get_consumer(self):\r
- return self.consumer\r
-\r
- def get_token(self):\r
- return self.token\r
-\r
- def fetch_request_token(self, oauth_request):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def fetch_access_token(self, oauth_request):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def access_resource(self, oauth_request):\r
- """-> Some protected resource."""\r
- raise NotImplementedError\r
-\r
-\r
-class OAuthDataStore(object):\r
- """A database abstraction used to lookup consumers and tokens."""\r
-\r
- def lookup_consumer(self, key):\r
- """-> OAuthConsumer."""\r
- raise NotImplementedError\r
-\r
- def lookup_token(self, oauth_consumer, token_type, token_token):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def lookup_nonce(self, oauth_consumer, oauth_token, nonce):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def fetch_request_token(self, oauth_consumer):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def fetch_access_token(self, oauth_consumer, oauth_token):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
- def authorize_request_token(self, oauth_token, user):\r
- """-> OAuthToken."""\r
- raise NotImplementedError\r
-\r
-\r
-class OAuthSignatureMethod(object):\r
- """A strategy class that implements a signature method."""\r
- def get_name(self):\r
- """-> str."""\r
- raise NotImplementedError\r
-\r
- def build_signature_base_string(self, oauth_request, oauth_consumer, oauth_token):\r
- """-> str key, str raw."""\r
- raise NotImplementedError\r
-\r
- def build_signature(self, oauth_request, oauth_consumer, oauth_token):\r
- """-> str."""\r
- raise NotImplementedError\r
-\r
- def check_signature(self, oauth_request, consumer, token, signature):\r
- built = self.build_signature(oauth_request, consumer, token)\r
- return built == signature\r
-\r
-\r
-class OAuthSignatureMethod_HMAC_SHA1(OAuthSignatureMethod):\r
-\r
- def get_name(self):\r
- return 'HMAC-SHA1'\r
-\r
- def build_signature_base_string(self, oauth_request, consumer, token):\r
- sig = (\r
- escape(oauth_request.get_normalized_http_method()),\r
- escape(oauth_request.get_normalized_http_url()),\r
- escape(oauth_request.get_normalized_parameters()),\r
- )\r
-\r
- key = '%s&' % escape(consumer.secret)\r
- if token:\r
- key += escape(token.secret)\r
- raw = '&'.join(sig)\r
- return key, raw\r
-\r
- def build_signature(self, oauth_request, consumer, token):\r
- """Builds the base signature string."""\r
- key, raw = self.build_signature_base_string(oauth_request, consumer,\r
- token)\r
-\r
- # HMAC object.\r
- try:\r
- import hashlib # 2.5\r
- hashed = hmac.new(key, raw, hashlib.sha1)\r
- except:\r
- import sha # Deprecated\r
- hashed = hmac.new(key, raw, sha)\r
-\r
- # Calculate the digest base 64.\r
- return binascii.b2a_base64(hashed.digest())[:-1]\r
-\r
-\r
-class OAuthSignatureMethod_PLAINTEXT(OAuthSignatureMethod):\r
-\r
- def get_name(self):\r
- return 'PLAINTEXT'\r
-\r
- def build_signature_base_string(self, oauth_request, consumer, token):\r
- """Concatenates the consumer key and secret."""\r
- sig = '%s&' % escape(consumer.secret)\r
- if token:\r
- sig = sig + escape(token.secret)\r
- return sig, sig\r
-\r
- def build_signature(self, oauth_request, consumer, token):\r
- key, raw = self.build_signature_base_string(oauth_request, consumer,\r
- token)\r
+"""
+The MIT License
+
+Copyright (c) 2007 Leah Culver
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in
+all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
+THE SOFTWARE.
+"""
+
+import cgi
+import urllib
+import time
+import random
+import urlparse
+import hmac
+import binascii
+
+
+VERSION = '1.0' # Hi Blaine!
+HTTP_METHOD = 'GET'
+SIGNATURE_METHOD = 'PLAINTEXT'
+
+
+class OAuthError(RuntimeError):
+ """Generic exception class."""
+ def __init__(self, message='OAuth error occured.'):
+ self.message = message
+
+def build_authenticate_header(realm=''):
+ """Optional WWW-Authenticate header (401 error)"""
+ return {'WWW-Authenticate': 'OAuth realm="%s"' % realm}
+
+def escape(s):
+ """Escape a URL including any /."""
+ return urllib.quote(s, safe='~')
+
+def _utf8_str(s):
+ """Convert unicode to utf-8."""
+ if isinstance(s, unicode):
+ return s.encode("utf-8")
+ else:
+ return str(s)
+
+def generate_timestamp():
+ """Get seconds since epoch (UTC)."""
+ return int(time.time())
+
+def generate_nonce(length=8):
+ """Generate pseudorandom number."""
+ return ''.join([str(random.randint(0, 9)) for i in range(length)])
+
+
+class OAuthConsumer(object):
+ """Consumer of OAuth authentication.
+
+ OAuthConsumer is a data type that represents the identity of the Consumer
+ via its shared secret with the Service Provider.
+
+ """
+ key = None
+ secret = None
+
+ def __init__(self, key, secret):
+ self.key = key
+ self.secret = secret
+
+
+class OAuthToken(object):
+ """OAuthToken is a data type that represents an End User via either an access
+ or request token.
+
+ key -- the token
+ secret -- the token secret
+
+ """
+ key = None
+ secret = None
+
+ def __init__(self, key, secret):
+ self.key = key
+ self.secret = secret
+
+ def to_string(self):
+ return urllib.urlencode({'oauth_token': self.key,
+ 'oauth_token_secret': self.secret})
+
+ def from_string(s):
+ """ Returns a token from something like:
+ oauth_token_secret=xxx&oauth_token=xxx
+ """
+ params = cgi.parse_qs(s, keep_blank_values=False)
+ key = params['oauth_token'][0]
+ secret = params['oauth_token_secret'][0]
+ return OAuthToken(key, secret)
+ from_string = staticmethod(from_string)
+
+ def __str__(self):
+ return self.to_string()
+
+
+class OAuthRequest(object):
+ """OAuthRequest represents the request and can be serialized.
+
+ OAuth parameters:
+ - oauth_consumer_key
+ - oauth_token
+ - oauth_signature_method
+ - oauth_signature
+ - oauth_timestamp
+ - oauth_nonce
+ - oauth_version
+ ... any additional parameters, as defined by the Service Provider.
+ """
+ parameters = None # OAuth parameters.
+ http_method = HTTP_METHOD
+ http_url = None
+ version = VERSION
+
+ def __init__(self, http_method=HTTP_METHOD, http_url=None, parameters=None):
+ self.http_method = http_method
+ self.http_url = http_url
+ self.parameters = parameters or {}
+
+ def set_parameter(self, parameter, value):
+ self.parameters[parameter] = value
+
+ def get_parameter(self, parameter):
+ try:
+ return self.parameters[parameter]
+ except:
+ raise OAuthError('Parameter not found: %s' % parameter)
+
+ def _get_timestamp_nonce(self):
+ return self.get_parameter('oauth_timestamp'), self.get_parameter(
+ 'oauth_nonce')
+
+ def get_nonoauth_parameters(self):
+ """Get any non-OAuth parameters."""
+ parameters = {}
+ for k, v in self.parameters.iteritems():
+ # Ignore oauth parameters.
+ if k.find('oauth_') < 0:
+ parameters[k] = v
+ return parameters
+
+ def to_header(self, realm=''):
+ """Serialize as a header for an HTTPAuth request."""
+ auth_header = 'OAuth realm="%s"' % realm
+ # Add the oauth parameters.
+ if self.parameters:
+ for k, v in self.parameters.iteritems():
+ if k[:6] == 'oauth_':
+ auth_header += ', %s="%s"' % (k, escape(str(v)))
+ return {'Authorization': auth_header}
+
+ def to_postdata(self):
+ """Serialize as post data for a POST request."""
+ return '&'.join(['%s=%s' % (escape(str(k)), escape(str(v))) \
+ for k, v in self.parameters.iteritems()])
+
+ def to_url(self):
+ """Serialize as a URL for a GET request."""
+ return '%s?%s' % (self.get_normalized_http_url(), self.to_postdata())
+
+ def get_normalized_parameters(self):
+ """Return a string that contains the parameters that must be signed."""
+ params = self.parameters
+ try:
+ # Exclude the signature if it exists.
+ del params['oauth_signature']
+ except:
+ pass
+ # Escape key values before sorting.
+ key_values = [(escape(_utf8_str(k)), escape(_utf8_str(v))) \
+ for k,v in params.items()]
+ # Sort lexicographically, first after key, then after value.
+ key_values.sort()
+ # Combine key value pairs into a string.
+ return '&'.join(['%s=%s' % (k, v) for k, v in key_values])
+
+ def get_normalized_http_method(self):
+ """Uppercases the http method."""
+ return self.http_method.upper()
+
+ def get_normalized_http_url(self):
+ """Parses the URL and rebuilds it to be scheme://host/path."""
+ parts = urlparse.urlparse(self.http_url)
+ scheme, netloc, path = parts[:3]
+ # Exclude default port numbers.
+ if scheme == 'http' and netloc[-3:] == ':80':
+ netloc = netloc[:-3]
+ elif scheme == 'https' and netloc[-4:] == ':443':
+ netloc = netloc[:-4]
+ return '%s://%s%s' % (scheme, netloc, path)
+
+ def sign_request(self, signature_method, consumer, token):
+ """Set the signature parameter to the result of build_signature."""
+ # Set the signature method.
+ self.set_parameter('oauth_signature_method',
+ signature_method.get_name())
+ # Set the signature.
+ self.set_parameter('oauth_signature',
+ self.build_signature(signature_method, consumer, token))
+
+ def build_signature(self, signature_method, consumer, token):
+ """Calls the build signature method within the signature method."""
+ return signature_method.build_signature(self, consumer, token)
+
+ def from_request(http_method, http_url, headers=None, parameters=None,
+ query_string=None):
+ """Combines multiple parameter sources."""
+ if parameters is None:
+ parameters = {}
+
+ # Headers
+ if headers and 'Authorization' in headers:
+ auth_header = headers['Authorization']
+ # Check that the authorization header is OAuth.
+ if auth_header.index('OAuth') > -1:
+ auth_header = auth_header.lstrip('OAuth ')
+ try:
+ # Get the parameters from the header.
+ header_params = OAuthRequest._split_header(auth_header)
+ parameters.update(header_params)
+ except:
+ raise OAuthError('Unable to parse OAuth parameters from '
+ 'Authorization header.')
+
+ # GET or POST query string.
+ if query_string:
+ query_params = OAuthRequest._split_url_string(query_string)
+ parameters.update(query_params)
+
+ # URL parameters.
+ param_str = urlparse.urlparse(http_url)[4] # query
+ url_params = OAuthRequest._split_url_string(param_str)
+ parameters.update(url_params)
+
+ if parameters:
+ return OAuthRequest(http_method, http_url, parameters)
+
+ return None
+ from_request = staticmethod(from_request)
+
+ def from_consumer_and_token(oauth_consumer, token=None,
+ http_method=HTTP_METHOD, http_url=None, parameters=None):
+ if not parameters:
+ parameters = {}
+
+ defaults = {
+ 'oauth_consumer_key': oauth_consumer.key,
+ 'oauth_timestamp': generate_timestamp(),
+ 'oauth_nonce': generate_nonce(),
+ 'oauth_version': OAuthRequest.version,
+ }
+
+ defaults.update(parameters)
+ parameters = defaults
+
+ if token:
+ parameters['oauth_token'] = token.key
+
+ return OAuthRequest(http_method, http_url, parameters)
+ from_consumer_and_token = staticmethod(from_consumer_and_token)
+
+ def from_token_and_callback(token, callback=None, http_method=HTTP_METHOD,
+ http_url=None, parameters=None):
+ if not parameters:
+ parameters = {}
+
+ parameters['oauth_token'] = token.key
+
+ if callback:
+ parameters['oauth_callback'] = callback
+
+ return OAuthRequest(http_method, http_url, parameters)
+ from_token_and_callback = staticmethod(from_token_and_callback)
+
+ def _split_header(header):
+ """Turn Authorization: header into parameters."""
+ params = {}
+ parts = header.split(',')
+ for param in parts:
+ # Ignore realm parameter.
+ if param.find('realm') > -1:
+ continue
+ # Remove whitespace.
+ param = param.strip()
+ # Split key-value.
+ param_parts = param.split('=', 1)
+ # Remove quotes and unescape the value.
+ params[param_parts[0]] = urllib.unquote(param_parts[1].strip('\"'))
+ return params
+ _split_header = staticmethod(_split_header)
+
+ def _split_url_string(param_str):
+ """Turn URL string into parameters."""
+ parameters = cgi.parse_qs(param_str, keep_blank_values=False)
+ for k, v in parameters.iteritems():
+ parameters[k] = urllib.unquote(v[0])
+ return parameters
+ _split_url_string = staticmethod(_split_url_string)
+
+class OAuthServer(object):
+ """A worker to check the validity of a request against a data store."""
+ timestamp_threshold = 300 # In seconds, five minutes.
+ version = VERSION
+ signature_methods = None
+ data_store = None
+
+ def __init__(self, data_store=None, signature_methods=None):
+ self.data_store = data_store
+ self.signature_methods = signature_methods or {}
+
+ def set_data_store(self, data_store):
+ self.data_store = data_store
+
+ def get_data_store(self):
+ return self.data_store
+
+ def add_signature_method(self, signature_method):
+ self.signature_methods[signature_method.get_name()] = signature_method
+ return self.signature_methods
+
+ def fetch_request_token(self, oauth_request):
+ """Processes a request_token request and returns the
+ request token on success.
+ """
+ try:
+ # Get the request token for authorization.
+ token = self._get_token(oauth_request, 'request')
+ except OAuthError:
+ # No token required for the initial token request.
+ version = self._get_version(oauth_request)
+ consumer = self._get_consumer(oauth_request)
+ self._check_signature(oauth_request, consumer, None)
+ # Fetch a new token.
+ token = self.data_store.fetch_request_token(consumer)
+ return token
+
+ def fetch_access_token(self, oauth_request):
+ """Processes an access_token request and returns the
+ access token on success.
+ """
+ version = self._get_version(oauth_request)
+ consumer = self._get_consumer(oauth_request)
+ # Get the request token.
+ token = self._get_token(oauth_request, 'request')
+ self._check_signature(oauth_request, consumer, token)
+ new_token = self.data_store.fetch_access_token(consumer, token)
+ return new_token
+
+ def verify_request(self, oauth_request):
+ """Verifies an api call and checks all the parameters."""
+ # -> consumer and token
+ version = self._get_version(oauth_request)
+ consumer = self._get_consumer(oauth_request)
+ # Get the access token.
+ token = self._get_token(oauth_request, 'access')
+ self._check_signature(oauth_request, consumer, token)
+ parameters = oauth_request.get_nonoauth_parameters()
+ return consumer, token, parameters
+
+ def authorize_token(self, token, user):
+ """Authorize a request token."""
+ return self.data_store.authorize_request_token(token, user)
+
+ def get_callback(self, oauth_request):
+ """Get the callback URL."""
+ return oauth_request.get_parameter('oauth_callback')
+
+ def build_authenticate_header(self, realm=''):
+ """Optional support for the authenticate header."""
+ return {'WWW-Authenticate': 'OAuth realm="%s"' % realm}
+
+ def _get_version(self, oauth_request):
+ """Verify the correct version request for this server."""
+ try:
+ version = oauth_request.get_parameter('oauth_version')
+ except:
+ version = VERSION
+ if version and version != self.version:
+ raise OAuthError('OAuth version %s not supported.' % str(version))
+ return version
+
+ def _get_signature_method(self, oauth_request):
+ """Figure out the signature with some defaults."""
+ try:
+ signature_method = oauth_request.get_parameter(
+ 'oauth_signature_method')
+ except:
+ signature_method = SIGNATURE_METHOD
+ try:
+ # Get the signature method object.
+ signature_method = self.signature_methods[signature_method]
+ except:
+ signature_method_names = ', '.join(self.signature_methods.keys())
+ raise OAuthError('Signature method %s not supported try one of the '
+ 'following: %s' % (signature_method, signature_method_names))
+
+ return signature_method
+
+ def _get_consumer(self, oauth_request):
+ consumer_key = oauth_request.get_parameter('oauth_consumer_key')
+ consumer = self.data_store.lookup_consumer(consumer_key)
+ if not consumer:
+ raise OAuthError('Invalid consumer.')
+ return consumer
+
+ def _get_token(self, oauth_request, token_type='access'):
+ """Try to find the token for the provided request token key."""
+ token_field = oauth_request.get_parameter('oauth_token')
+ token = self.data_store.lookup_token(token_type, token_field)
+ if not token:
+ raise OAuthError('Invalid %s token: %s' % (token_type, token_field))
+ return token
+
+ def _check_signature(self, oauth_request, consumer, token):
+ timestamp, nonce = oauth_request._get_timestamp_nonce()
+ self._check_timestamp(timestamp)
+ self._check_nonce(consumer, token, nonce)
+ signature_method = self._get_signature_method(oauth_request)
+ try:
+ signature = oauth_request.get_parameter('oauth_signature')
+ except:
+ raise OAuthError('Missing signature.')
+ # Validate the signature.
+ valid_sig = signature_method.check_signature(oauth_request, consumer,
+ token, signature)
+ if not valid_sig:
+ key, base = signature_method.build_signature_base_string(
+ oauth_request, consumer, token)
+ raise OAuthError('Invalid signature. Expected signature base '
+ 'string: %s' % base)
+ built = signature_method.build_signature(oauth_request, consumer, token)
+
+ def _check_timestamp(self, timestamp):
+ """Verify that timestamp is recentish."""
+ timestamp = int(timestamp)
+ now = int(time.time())
+ lapsed = now - timestamp
+ if lapsed > self.timestamp_threshold:
+ raise OAuthError('Expired timestamp: given %d and now %s has a '
+ 'greater difference than threshold %d' %
+ (timestamp, now, self.timestamp_threshold))
+
+ def _check_nonce(self, consumer, token, nonce):
+ """Verify that the nonce is uniqueish."""
+ nonce = self.data_store.lookup_nonce(consumer, token, nonce)
+ if nonce:
+ raise OAuthError('Nonce already used: %s' % str(nonce))
+
+
+class OAuthClient(object):
+ """OAuthClient is a worker to attempt to execute a request."""
+ consumer = None
+ token = None
+
+ def __init__(self, oauth_consumer, oauth_token):
+ self.consumer = oauth_consumer
+ self.token = oauth_token
+
+ def get_consumer(self):
+ return self.consumer
+
+ def get_token(self):
+ return self.token
+
+ def fetch_request_token(self, oauth_request):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def fetch_access_token(self, oauth_request):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def access_resource(self, oauth_request):
+ """-> Some protected resource."""
+ raise NotImplementedError
+
+
+class OAuthDataStore(object):
+ """A database abstraction used to lookup consumers and tokens."""
+
+ def lookup_consumer(self, key):
+ """-> OAuthConsumer."""
+ raise NotImplementedError
+
+ def lookup_token(self, oauth_consumer, token_type, token_token):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def lookup_nonce(self, oauth_consumer, oauth_token, nonce):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def fetch_request_token(self, oauth_consumer):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def fetch_access_token(self, oauth_consumer, oauth_token):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+ def authorize_request_token(self, oauth_token, user):
+ """-> OAuthToken."""
+ raise NotImplementedError
+
+
+class OAuthSignatureMethod(object):
+ """A strategy class that implements a signature method."""
+ def get_name(self):
+ """-> str."""
+ raise NotImplementedError
+
+ def build_signature_base_string(self, oauth_request, oauth_consumer, oauth_token):
+ """-> str key, str raw."""
+ raise NotImplementedError
+
+ def build_signature(self, oauth_request, oauth_consumer, oauth_token):
+ """-> str."""
+ raise NotImplementedError
+
+ def check_signature(self, oauth_request, consumer, token, signature):
+ built = self.build_signature(oauth_request, consumer, token)
+ return built == signature
+
+
+class OAuthSignatureMethod_HMAC_SHA1(OAuthSignatureMethod):
+
+ def get_name(self):
+ return 'HMAC-SHA1'
+
+ def build_signature_base_string(self, oauth_request, consumer, token):
+ sig = (
+ escape(oauth_request.get_normalized_http_method()),
+ escape(oauth_request.get_normalized_http_url()),
+ escape(oauth_request.get_normalized_parameters()),
+ )
+
+ key = '%s&' % escape(consumer.secret)
+ if token:
+ key += escape(token.secret)
+ raw = '&'.join(sig)
+ return key, raw
+
+ def build_signature(self, oauth_request, consumer, token):
+ """Builds the base signature string."""
+ key, raw = self.build_signature_base_string(oauth_request, consumer,
+ token)
+
+ # HMAC object.
+ try:
+ import hashlib # 2.5
+ hashed = hmac.new(key, raw, hashlib.sha1)
+ except:
+ import sha # Deprecated
+ hashed = hmac.new(key, raw, sha)
+
+ # Calculate the digest base 64.
+ return binascii.b2a_base64(hashed.digest())[:-1]
+
+
+class OAuthSignatureMethod_PLAINTEXT(OAuthSignatureMethod):
+
+ def get_name(self):
+ return 'PLAINTEXT'
+
+ def build_signature_base_string(self, oauth_request, consumer, token):
+ """Concatenates the consumer key and secret."""
+ sig = '%s&' % escape(consumer.secret)
+ if token:
+ sig = sig + escape(token.secret)
+ return sig, sig
+
+ def build_signature(self, oauth_request, consumer, token):
+ key, raw = self.build_signature_base_string(oauth_request, consumer,
+ token)
return key
\ No newline at end of file
projects / osqa.git / blobdiff