diff --git a/build/lib/oauth2/__init__.py b/build/lib/oauth2/__init__.py new file mode 100644 index 00000000..d8ff20ba --- /dev/null +++ b/build/lib/oauth2/__init__.py @@ -0,0 +1,848 @@ +""" +The MIT License + +Copyright (c) 2007-2010 Leah Culver, Joe Stump, Mark Paschal, Vic Fryzel + +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 base64 +import time +import random +import urllib.parse +import hmac +import binascii +import httplib2 +from hashlib import sha1 + +import oauth2._version + +__version__ = _version.__version__ + +OAUTH_VERSION = '1.0' # Hi Blaine! +HTTP_METHOD = 'GET' +SIGNATURE_METHOD = 'PLAINTEXT' + + +class Error(RuntimeError): + """Generic exception class.""" + + def __init__(self, message='OAuth error occurred.'): + self._message = message + + @property + def message(self): + """A hack to get around the deprecation errors in 2.6.""" + return self._message + + def __str__(self): + return self._message + + +class MissingSignature(Error): + pass + + +def build_authenticate_header(realm=''): + """Optional WWW-Authenticate header (401 error)""" + return {'WWW-Authenticate': 'OAuth realm="%s"' % realm} + + +def build_xoauth_string(url, consumer, token=None): + """Build an XOAUTH string for use in SMTP/IMPA authentication.""" + request = Request.from_consumer_and_token(consumer, token, + "GET", url) + + signing_method = SignatureMethod_HMAC_SHA1() + request.sign_request(signing_method, consumer, token) + + params = [] + for k, v in sorted(request.items()): + if v is not None: + params.append('%s="%s"' % (k, escape(v))) + + return "%s %s %s" % ("GET", url, ','.join(params)) + + +def to_unicode(s): + """ Convert to unicode, raise exception with instructive error + message if s is not unicode, ascii, or utf-8. """ + if not isinstance(s, (bytes, str)): + raise TypeError('You are required to pass either unicode or string here, not: %r (%s)' % (type(s), s)) + if isinstance(s, bytes): + try: + s = s.decode('utf-8') + except UnicodeDecodeError as le: + raise TypeError('You are required to pass either a unicode object or a utf-8 string here. You passed a Python string object which contained non-utf-8: %r. The UnicodeDecodeError that resulted from attempting to interpret it as utf-8 was: %s' % (s, le,)) + return s + +def to_utf8(s): + return to_unicode(s).encode('utf-8') + +def to_unicode_if_string(s): + if isinstance(s, (bytes, str)): + return to_unicode(s) + else: + return s + +def to_utf8_if_string(s): + if isinstance(s, (bytes, str)): + return to_utf8(s) + else: + return s + +def to_unicode_optional_iterator(x): + """ + Raise TypeError if x is a str containing non-utf8 bytes or if x is + an iterable which contains such a str. + """ + if isinstance(x, (bytes, str)): + return to_unicode(x) + + try: + l = list(x) + except TypeError as e: + assert 'is not iterable' in str(e) + return x + else: + return [ to_unicode(e) for e in l ] + +def to_utf8_optional_iterator(x): + """ + Raise TypeError if x is a str or if x is an iterable which + contains a str. + """ + if isinstance(x, (bytes, str)): + return to_utf8(x) + + try: + l = list(x) + except TypeError as e: + assert 'is not iterable' in str(e) + return x + else: + return [ to_utf8_if_string(e) for e in l ] + +def escape(s): + """Escape a URL including any /.""" + return urllib.parse.quote(s, safe='~') + +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)]) + + +def generate_verifier(length=8): + """Generate pseudorandom number.""" + return ''.join([str(random.randint(0, 9)) for i in range(length)]) + + +class Consumer(object): + """A consumer of OAuth-protected services. + + The OAuth consumer is a "third-party" service that wants to access + protected resources from an OAuth service provider on behalf of an end + user. It's kind of the OAuth client. + + Usually a consumer must be registered with the service provider by the + developer of the consumer software. As part of that process, the service + provider gives the consumer a *key* and a *secret* with which the consumer + software can identify itself to the service. The consumer will include its + key in each request to identify itself, but will use its secret only when + signing requests, to prove that the request is from that particular + registered consumer. + + Once registered, the consumer can then use its consumer credentials to ask + the service provider for a request token, kicking off the OAuth + authorization process. + """ + + key = None + secret = None + + def __init__(self, key, secret): + self.key = key + self.secret = secret + + if self.key is None or self.secret is None: + raise ValueError("Key and secret must be set.") + + def __str__(self): + data = {'oauth_consumer_key': self.key, + 'oauth_consumer_secret': self.secret} + + return urllib.parse.urlencode(data) + + +class Token(object): + """An OAuth credential used to request authorization or a protected + resource. + + Tokens in OAuth comprise a *key* and a *secret*. The key is included in + requests to identify the token being used, but the secret is used only in + the signature, to prove that the requester is who the server gave the + token to. + + When first negotiating the authorization, the consumer asks for a *request + token* that the live user authorizes with the service provider. The + consumer then exchanges the request token for an *access token* that can + be used to access protected resources. + """ + + key = None + secret = None + callback = None + callback_confirmed = None + verifier = None + + def __init__(self, key, secret): + self.key = key + self.secret = secret + + if self.key is None or self.secret is None: + raise ValueError("Key and secret must be set.") + + def set_callback(self, callback): + self.callback = callback + self.callback_confirmed = 'true' + + def set_verifier(self, verifier=None): + if verifier is not None: + self.verifier = verifier + else: + self.verifier = generate_verifier() + + def get_callback_url(self): + if self.callback and self.verifier: + # Append the oauth_verifier. + parts = urllib.parse.urlparse(self.callback) + scheme, netloc, path, params, query, fragment = parts[:6] + if query: + query = '%s&oauth_verifier=%s' % (query, self.verifier) + else: + query = 'oauth_verifier=%s' % self.verifier + return urllib.parse.urlunparse((scheme, netloc, path, params, + query, fragment)) + return self.callback + + def to_string(self): + """Returns this token as a plain string, suitable for storage. + + The resulting string includes the token's secret, so you should never + send or store this string where a third party can read it. + """ + + data = { + 'oauth_token': self.key, + 'oauth_token_secret': self.secret, + } + + if self.callback_confirmed is not None: + data['oauth_callback_confirmed'] = self.callback_confirmed + return urllib.parse.urlencode(data) + + @staticmethod + def from_string(s): + """Deserializes a token from a string like one returned by + `to_string()`.""" + + if not len(s): + raise ValueError("Invalid parameter string.") + + params = urllib.parse.parse_qs(s, keep_blank_values=False) + if not len(params): + raise ValueError("Invalid parameter string.") + + 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.") + + token = Token(key, secret) + try: + token.callback_confirmed = params['oauth_callback_confirmed'][0] + except KeyError: + pass # 1.0, no callback confirmed. + return token + + def __str__(self): + return self.to_string() + + +def setter(attr): + name = attr.__name__ + + def getter(self): + try: + return self.__dict__[name] + except KeyError: + raise AttributeError(name) + + def deleter(self): + del self.__dict__[name] + + return property(getter, attr, deleter) + + +class Request(dict): + + """The parameters and information for an HTTP request, suitable for + authorizing with OAuth credentials. + + When a consumer wants to access a service's protected resources, it does + so using a signed HTTP request identifying itself (the consumer) with its + key, and providing an access token authorized by the end user to access + those resources. + + """ + + version = OAUTH_VERSION + + def __init__(self, method=HTTP_METHOD, url=None, parameters=None, + body='', is_form_encoded=False): + if url is not None: + self.url = to_unicode(url) + self.method = method + if parameters is not None: + for k, v in parameters.items(): + k = to_unicode(k) + v = to_unicode_optional_iterator(v) + self[k] = v + self.body = body + self.is_form_encoded = is_form_encoded + + + @setter + def url(self, value): + self.__dict__['url'] = value + if value is not None: + scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(value) + + # Exclude default port numbers. + if scheme == 'http' and netloc[-3:] == ':80': + netloc = netloc[:-3] + elif scheme == 'https' and netloc[-4:] == ':443': + netloc = netloc[:-4] + if scheme not in ('http', 'https'): + raise ValueError("Unsupported URL %s (%s)." % (value, scheme)) + + # Normalized URL excludes params, query, and fragment. + self.normalized_url = urllib.parse.urlunparse((scheme, netloc, path, None, None, None)) + else: + self.normalized_url = None + self.__dict__['url'] = None + + @setter + def method(self, value): + self.__dict__['method'] = value.upper() + + def _get_timestamp_nonce(self): + return self['oauth_timestamp'], self['oauth_nonce'] + + def get_nonoauth_parameters(self): + """Get any non-OAuth parameters.""" + return dict([(k, v) for k, v in self.items() + if not k.startswith('oauth_')]) + + def to_header(self, realm=''): + """Serialize as a header for an HTTPAuth request.""" + oauth_params = ((k, v) for k, v in self.items() + if k.startswith('oauth_')) + stringy_params = ((k, escape(str(v))) for k, v in oauth_params) + header_params = ('%s="%s"' % (k, v) for k, v in stringy_params) + params_header = ', '.join(header_params) + + auth_header = 'OAuth realm="%s"' % realm + if params_header: + auth_header = "%s, %s" % (auth_header, params_header) + + return {'Authorization': auth_header} + + def to_postdata(self): + """Serialize as post data for a POST request.""" + d = {} + for k, v in self.items(): + #d[k.encode('utf-8')] = to_utf8_optional_iterator(v) + d[k] = to_unicode_optional_iterator(v) + + # tell urlencode to deal with sequence values and map them correctly + # to resulting querystring. for example self["k"] = ["v1", "v2"] will + # result in 'k=v1&k=v2' and not k=%5B%27v1%27%2C+%27v2%27%5D + return urllib.parse.urlencode(d, True).replace('+', '%20') + + def to_url(self): + """Serialize as a URL for a GET request.""" + base_url = urllib.parse.urlparse(self.url) + try: + query = base_url.query + except AttributeError: + # must be python <2.5 + query = base_url[4] + query = urllib.parse.parse_qs(query) + for k, v in self.items(): + query.setdefault(k, []).append(v) + + try: + scheme = base_url.scheme + netloc = base_url.netloc + path = base_url.path + params = base_url.params + fragment = base_url.fragment + except AttributeError: + # must be python <2.5 + scheme = base_url[0] + netloc = base_url[1] + path = base_url[2] + params = base_url[3] + fragment = base_url[5] + + url = (scheme, netloc, path, params, + urllib.parse.urlencode(query, True), fragment) + return urllib.parse.urlunparse(url) + + def get_parameter(self, parameter): + ret = self.get(parameter) + if ret is None: + raise Error('Parameter not found: %s' % parameter) + + return ret + + def get_normalized_parameters(self): + """Return a string that contains the parameters that must be signed.""" + items = [] + for key, value in self.items(): + if key == 'oauth_signature': + continue + # 1.0a/9.1.1 states that kvp must be sorted by key, then by value, + # so we unpack sequence values into multiple items for sorting. + if isinstance(value, str): + items.append((to_utf8_if_string(key), to_utf8(value))) + else: + try: + value = list(value) + except TypeError as e: + assert 'is not iterable' in str(e) + items.append((to_utf8_if_string(key), to_utf8_if_string(value))) + else: + items.extend((to_utf8_if_string(key), to_utf8_if_string(item)) for item in value) + + # Include any query string parameters from the provided URL + query = urllib.parse.urlparse(self.url)[4] + + url_items = self._split_url_string(query).items() + url_items = [(to_utf8(k), to_utf8(v)) for k, v in url_items if k != 'oauth_signature' ] + items.extend(url_items) + + items.sort() + encoded_str = urllib.parse.urlencode(items) + # Encode signature parameters per Oauth Core 1.0 protocol + # spec draft 7, section 3.6 + # (http://tools.ietf.org/html/draft-hammer-oauth-07#section-3.6) + # Spaces must be encoded with "%20" instead of "+" + return encoded_str.replace('+', '%20').replace('%7E', '~') + + def sign_request(self, signature_method, consumer, token): + """Set the signature parameter to the result of sign.""" + + if not self.is_form_encoded: + # according to + # http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html + # section 4.1.1 "OAuth Consumers MUST NOT include an + # oauth_body_hash parameter on requests with form-encoded + # request bodies." + self['oauth_body_hash'] = to_unicode(base64.b64encode(sha1(to_utf8(self.body)).digest())) + + if 'oauth_consumer_key' not in self: + self['oauth_consumer_key'] = consumer.key + + if token and 'oauth_token' not in self: + self['oauth_token'] = token.key + + self['oauth_signature_method'] = signature_method.name + self['oauth_signature'] = signature_method.sign(self, consumer, token) + + @classmethod + def make_timestamp(cls): + """Get seconds since epoch (UTC).""" + return str(int(time.time())) + + @classmethod + def make_nonce(cls): + """Generate pseudorandom number.""" + #return str(random.randint(0, 100000000)) + return base64.b64encode(("%0x" % random.getrandbits(256)).encode("utf-8"))[:32] + + @classmethod + def from_request(cls, 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[:6] == 'OAuth ': + auth_header = auth_header[6:] + try: + # Get the parameters from the header. + header_params = cls._split_header(auth_header) + parameters.update(header_params) + except: + raise Error('Unable to parse OAuth parameters from ' + 'Authorization header.') + + # GET or POST query string. + if query_string: + query_params = cls._split_url_string(query_string) + parameters.update(query_params) + + # URL parameters. + param_str = urllib.parse.urlparse(http_url)[4] # query + url_params = cls._split_url_string(param_str) + parameters.update(url_params) + + if parameters: + return cls(http_method, http_url, parameters) + + return None + + @classmethod + def from_consumer_and_token(cls, consumer, token=None, + http_method=HTTP_METHOD, http_url=None, parameters=None, + body='', is_form_encoded=False): + if not parameters: + parameters = {} + + defaults = { + 'oauth_consumer_key': consumer.key, + 'oauth_timestamp': cls.make_timestamp(), + 'oauth_nonce': cls.make_nonce(), + 'oauth_version': cls.version, + } + + defaults.update(parameters) + parameters = defaults + + if token: + parameters['oauth_token'] = token.key + if token.verifier: + parameters['oauth_verifier'] = token.verifier + + return Request(http_method, http_url, parameters, body=body, + is_form_encoded=is_form_encoded) + + @classmethod + def from_token_and_callback(cls, 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 cls(http_method, http_url, parameters) + + @staticmethod + 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.parse.unquote(param_parts[1].strip('\"')) + return params + + @staticmethod + def _split_url_string(param_str): + """Turn URL string into parameters.""" + parameters = urllib.parse.parse_qs(param_str, keep_blank_values=True) + for k, v in parameters.items(): + parameters[k] = urllib.parse.unquote(v[0]) + return parameters + + +class Client(httplib2.Http): + """OAuthClient is a worker to attempt to execute a request.""" + + def __init__(self, consumer, token=None, cache=None, timeout=None, + proxy_info=None): + + if consumer is not None and not isinstance(consumer, Consumer): + raise ValueError("Invalid consumer.") + + if token is not None and not isinstance(token, Token): + raise ValueError("Invalid token.") + + self.consumer = consumer + self.token = token + self.method = SignatureMethod_HMAC_SHA1() + + httplib2.Http.__init__(self, cache=cache, timeout=timeout, proxy_info=proxy_info) + + def set_signature_method(self, method): + if not isinstance(method, SignatureMethod): + raise ValueError("Invalid signature method.") + + self.method = method + + def request(self, uri, method="GET", body='', headers=None, + redirections=httplib2.DEFAULT_MAX_REDIRECTS, connection_type=None): + DEFAULT_POST_CONTENT_TYPE = 'application/x-www-form-urlencoded;charset=UTF-8' + + if not isinstance(headers, dict): + headers = {} + + if method == "POST": + headers['Content-Type'] = headers.get('Content-Type', + DEFAULT_POST_CONTENT_TYPE) + + is_form_encoded = \ + headers.get('Content-Type') == 'application/x-www-form-urlencoded;charset=UTF-8' + + if is_form_encoded and body: + parameters = urllib.parse.parse_qs(body) + else: + parameters = None + + req = Request.from_consumer_and_token(self.consumer, + token=self.token, http_method=method, http_url=uri, + parameters=parameters, body=body, is_form_encoded=is_form_encoded) + + req.sign_request(self.method, self.consumer, self.token) + + schema, rest = urllib.parse.splittype(uri) + if rest.startswith('//'): + hierpart = '//' + else: + hierpart = '' + host, rest = urllib.parse.splithost(rest) + + realm = schema + ':' + hierpart + host + + if is_form_encoded: + body = req.to_postdata() + elif method == "GET": + uri = req.to_url() + else: + headers.update(req.to_header(realm=realm)) + + return httplib2.Http.request(self, uri, method=method, body=body, + headers=headers, redirections=redirections, + connection_type=connection_type) + + +class Server(object): + """A skeletal implementation of a service provider, providing protected + resources to requests from authorized consumers. + + This class implements the logic to check requests for authorization. You + can use it with your web server or web framework to protect certain + resources with OAuth. + """ + + timestamp_threshold = 300 # In seconds, five minutes. + version = OAUTH_VERSION + signature_methods = None + + def __init__(self, signature_methods=None): + self.signature_methods = signature_methods or {} + + def add_signature_method(self, signature_method): + self.signature_methods[signature_method.name] = signature_method + return self.signature_methods + + def verify_request(self, request, consumer, token): + """Verifies an api call and checks all the parameters.""" + + self._check_version(request) + self._check_signature(request, consumer, token) + parameters = request.get_nonoauth_parameters() + return parameters + + def build_authenticate_header(self, realm=''): + """Optional support for the authenticate header.""" + return {'WWW-Authenticate': 'OAuth realm="%s"' % realm} + + def _check_version(self, request): + """Verify the correct version of the request for this server.""" + version = self._get_version(request) + if version and version != self.version: + raise Error('OAuth version %s not supported.' % str(version)) + + def _get_version(self, request): + """Return the version of the request for this server.""" + try: + version = request.get_parameter('oauth_version') + except: + version = OAUTH_VERSION + + return version + + def _get_signature_method(self, request): + """Figure out the signature with some defaults.""" + try: + signature_method = 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 Error('Signature method %s not supported try one of the following: %s' % (signature_method, signature_method_names)) + + return signature_method + + def _get_verifier(self, request): + return request.get_parameter('oauth_verifier') + + def _check_signature(self, request, consumer, token): + timestamp, nonce = request._get_timestamp_nonce() + self._check_timestamp(timestamp) + signature_method = self._get_signature_method(request) + + try: + signature = request.get_parameter('oauth_signature') + except: + raise MissingSignature('Missing oauth_signature.') + + # Validate the signature. + valid = signature_method.check(request, consumer, token, signature) + + if not valid: + key, base = signature_method.signing_base(request, consumer, token) + + raise Error('Invalid signature. Expected signature base ' + 'string: %s' % base) + + 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 Error('Expired timestamp: given %d and now %s has a ' + 'greater difference than threshold %d' % (timestamp, now, + self.timestamp_threshold)) + + +class SignatureMethod(object): + """A way of signing requests. + + The OAuth protocol lets consumers and service providers pick a way to sign + requests. This interface shows the methods expected by the other `oauth` + modules for signing requests. Subclass it and implement its methods to + provide a new way to sign requests. + """ + + def signing_base(self, request, consumer, token): + """Calculates the string that needs to be signed. + + This method returns a 2-tuple containing the starting key for the + signing and the message to be signed. The latter may be used in error + messages to help clients debug their software. + + """ + raise NotImplementedError + + def sign(self, request, consumer, token): + """Returns the signature for the given request, based on the consumer + and token also provided. + + You should use your implementation of `signing_base()` to build the + message to sign. Otherwise it may be less useful for debugging. + + """ + raise NotImplementedError + + def check(self, request, consumer, token, signature): + """Returns whether the given signature is the correct signature for + the given consumer and token signing the given request.""" + built = self.sign(request, consumer, token) + return built == signature + + +class SignatureMethod_HMAC_SHA1(SignatureMethod): + name = 'HMAC-SHA1' + + def signing_base(self, request, consumer, token): + if not hasattr(request, 'normalized_url') or request.normalized_url is None: + raise ValueError("Base URL for request is not set.") + + sig = ( + escape(request.method), + escape(request.normalized_url), + escape(request.get_normalized_parameters()), + ) + + key = '%s&' % escape(consumer.secret) + if token: + key += escape(token.secret) + raw = '&'.join(sig) + return key, raw + + def sign(self, request, consumer, token): + """Builds the base signature string.""" + key, raw = self.signing_base(request, consumer, token) + + hashed = hmac.new(to_utf8(key), to_utf8(raw), sha1) + + # Calculate the digest base 64. + return to_unicode(binascii.b2a_base64(hashed.digest())[:-1]) + + +class SignatureMethod_PLAINTEXT(SignatureMethod): + + name = 'PLAINTEXT' + + def signing_base(self, request, consumer, token): + """Concatenates the consumer key and secret with the token's + secret.""" + sig = '%s&' % escape(consumer.secret) + if token: + sig = sig + escape(token.secret) + return sig, sig + + def sign(self, request, consumer, token): + key, raw = self.signing_base(request, consumer, token) + return raw diff --git a/build/lib/oauth2/_version.py b/build/lib/oauth2/_version.py new file mode 100644 index 00000000..9d779eaa --- /dev/null +++ b/build/lib/oauth2/_version.py @@ -0,0 +1,18 @@ +# This is the version of this source code. + +manual_verstr = "1.5" + + + +auto_build_num = "211" + + + +verstr = manual_verstr + "." + auto_build_num +try: + from pyutil.version_class import Version as pyutil_Version + __version__ = pyutil_Version(verstr) +except (ImportError, ValueError): + # Maybe there is no pyutil installed. + from distutils.version import LooseVersion as distutils_Version + __version__ = distutils_Version(verstr) diff --git a/build/lib/oauth2/clients/__init__.py b/build/lib/oauth2/clients/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/build/lib/oauth2/clients/imap.py b/build/lib/oauth2/clients/imap.py new file mode 100644 index 00000000..68b7cd8c --- /dev/null +++ b/build/lib/oauth2/clients/imap.py @@ -0,0 +1,40 @@ +""" +The MIT License + +Copyright (c) 2007-2010 Leah Culver, Joe Stump, Mark Paschal, Vic Fryzel + +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 oauth2 +import imaplib + + +class IMAP4_SSL(imaplib.IMAP4_SSL): + """IMAP wrapper for imaplib.IMAP4_SSL that implements XOAUTH.""" + + def authenticate(self, url, consumer, token): + if consumer is not None and not isinstance(consumer, oauth2.Consumer): + raise ValueError("Invalid consumer.") + + if token is not None and not isinstance(token, oauth2.Token): + raise ValueError("Invalid token.") + + imaplib.IMAP4_SSL.authenticate(self, 'XOAUTH', + lambda x: oauth2.build_xoauth_string(url, consumer, token)) diff --git a/build/lib/oauth2/clients/smtp.py b/build/lib/oauth2/clients/smtp.py new file mode 100644 index 00000000..3e7bf0b0 --- /dev/null +++ b/build/lib/oauth2/clients/smtp.py @@ -0,0 +1,41 @@ +""" +The MIT License + +Copyright (c) 2007-2010 Leah Culver, Joe Stump, Mark Paschal, Vic Fryzel + +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 oauth2 +import smtplib +import base64 + + +class SMTP(smtplib.SMTP): + """SMTP wrapper for smtplib.SMTP that implements XOAUTH.""" + + def authenticate(self, url, consumer, token): + if consumer is not None and not isinstance(consumer, oauth2.Consumer): + raise ValueError("Invalid consumer.") + + if token is not None and not isinstance(token, oauth2.Token): + raise ValueError("Invalid token.") + + self.docmd('AUTH', 'XOAUTH %s' % \ + base64.b64encode(oauth2.build_xoauth_string(url, consumer, token))) diff --git a/build/lib/tests/__init__.py b/build/lib/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/build/lib/tests/test_oauth.py b/build/lib/tests/test_oauth.py new file mode 100644 index 00000000..c6efeeba --- /dev/null +++ b/build/lib/tests/test_oauth.py @@ -0,0 +1,1301 @@ +# -*- coding: utf-8 -*- + +""" +The MIT License + +Copyright (c) 2009 Vic Fryzel + +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 sys +import os +import unittest +import oauth2 as oauth +import random +import time +import urllib.parse +from types import ListType +import mock +import httplib2 + +sys.path[0:0] = [os.path.join(os.path.dirname(__file__), ".."),] + + +class TestError(unittest.TestCase): + def test_message(self): + try: + raise oauth.Error + except oauth.Error as e: + self.assertEqual(e.message, 'OAuth error occurred.') + + msg = 'OMG THINGS BROKE!!!!' + try: + raise oauth.Error(msg) + except oauth.Error as e: + self.assertEqual(e.message, msg) + + def test_str(self): + try: + raise oauth.Error + except oauth.Error as e: + self.assertEquals(str(e), 'OAuth error occurred.') + +class TestGenerateFunctions(unittest.TestCase): + def test_build_auth_header(self): + header = oauth.build_authenticate_header() + self.assertEqual(header['WWW-Authenticate'], 'OAuth realm=""') + self.assertEqual(len(header), 1) + realm = 'http://example.myrealm.com/' + header = oauth.build_authenticate_header(realm) + self.assertEqual(header['WWW-Authenticate'], 'OAuth realm="%s"' % + realm) + self.assertEqual(len(header), 1) + + def test_build_xoauth_string(self): + consumer = oauth.Consumer('consumer_token', 'consumer_secret') + token = oauth.Token('user_token', 'user_secret') + url = "https://mail.google.com/mail/b/joe@example.com/imap/" + xoauth_string = oauth.build_xoauth_string(url, consumer, token) + + method, oauth_url, oauth_string = xoauth_string.split(' ') + + self.assertEqual("GET", method) + self.assertEqual(url, oauth_url) + + returned = {} + parts = oauth_string.split(',') + for part in parts: + var, val = part.split('=') + returned[var] = val.strip('"') + + self.assertEquals('HMAC-SHA1', returned['oauth_signature_method']) + self.assertEquals('user_token', returned['oauth_token']) + self.assertEquals('consumer_token', returned['oauth_consumer_key']) + self.assertTrue('oauth_signature' in returned, 'oauth_signature') + + def test_escape(self): + string = 'http://whatever.com/~someuser/?test=test&other=other' + self.assert_('~' in oauth.escape(string)) + string = '../../../../../../../etc/passwd' + self.assert_('../' not in oauth.escape(string)) + + def test_gen_nonce(self): + nonce = oauth.generate_nonce() + self.assertEqual(len(nonce), 8) + nonce = oauth.generate_nonce(20) + self.assertEqual(len(nonce), 20) + + def test_gen_verifier(self): + verifier = oauth.generate_verifier() + self.assertEqual(len(verifier), 8) + verifier = oauth.generate_verifier(16) + self.assertEqual(len(verifier), 16) + + def test_gen_timestamp(self): + exp = int(time.time()) + now = oauth.generate_timestamp() + self.assertEqual(exp, now) + +class TestConsumer(unittest.TestCase): + def setUp(self): + self.key = 'my-key' + self.secret = 'my-secret' + self.consumer = oauth.Consumer(key=self.key, secret=self.secret) + + def test_init(self): + self.assertEqual(self.consumer.key, self.key) + self.assertEqual(self.consumer.secret, self.secret) + + def test_basic(self): + self.assertRaises(ValueError, lambda: oauth.Consumer(None, None)) + self.assertRaises(ValueError, lambda: oauth.Consumer('asf', None)) + self.assertRaises(ValueError, lambda: oauth.Consumer(None, 'dasf')) + + def test_str(self): + res = dict(urllib.parse.parse_qsl(str(self.consumer))) + self.assertTrue('oauth_consumer_key' in res) + self.assertTrue('oauth_consumer_secret' in res) + self.assertEquals(res['oauth_consumer_key'], self.consumer.key) + self.assertEquals(res['oauth_consumer_secret'], self.consumer.secret) + +class TestToken(unittest.TestCase): + def setUp(self): + self.key = 'my-key' + self.secret = 'my-secret' + self.token = oauth.Token(self.key, self.secret) + + def test_basic(self): + self.assertRaises(ValueError, lambda: oauth.Token(None, None)) + self.assertRaises(ValueError, lambda: oauth.Token('asf', None)) + self.assertRaises(ValueError, lambda: oauth.Token(None, 'dasf')) + + def test_init(self): + self.assertEqual(self.token.key, self.key) + self.assertEqual(self.token.secret, self.secret) + self.assertEqual(self.token.callback, None) + self.assertEqual(self.token.callback_confirmed, None) + self.assertEqual(self.token.verifier, None) + + def test_set_callback(self): + self.assertEqual(self.token.callback, None) + self.assertEqual(self.token.callback_confirmed, None) + cb = 'http://www.example.com/my-callback' + self.token.set_callback(cb) + self.assertEqual(self.token.callback, cb) + self.assertEqual(self.token.callback_confirmed, 'true') + self.token.set_callback(None) + self.assertEqual(self.token.callback, None) + # TODO: The following test should probably not pass, but it does + # To fix this, check for None and unset 'true' in set_callback + # Additionally, should a confirmation truly be done of the callback? + self.assertEqual(self.token.callback_confirmed, 'true') + + def test_set_verifier(self): + self.assertEqual(self.token.verifier, None) + v = oauth.generate_verifier() + self.token.set_verifier(v) + self.assertEqual(self.token.verifier, v) + self.token.set_verifier() + self.assertNotEqual(self.token.verifier, v) + self.token.set_verifier('') + self.assertEqual(self.token.verifier, '') + + def test_get_callback_url(self): + self.assertEqual(self.token.get_callback_url(), None) + + self.token.set_verifier() + self.assertEqual(self.token.get_callback_url(), None) + + cb = 'http://www.example.com/my-callback?save=1&return=true' + v = oauth.generate_verifier() + self.token.set_callback(cb) + self.token.set_verifier(v) + url = self.token.get_callback_url() + verifier_str = '&oauth_verifier=%s' % v + self.assertEqual(url, '%s%s' % (cb, verifier_str)) + + cb = 'http://www.example.com/my-callback-no-query' + v = oauth.generate_verifier() + self.token.set_callback(cb) + self.token.set_verifier(v) + url = self.token.get_callback_url() + verifier_str = '?oauth_verifier=%s' % v + self.assertEqual(url, '%s%s' % (cb, verifier_str)) + + def test_to_string(self): + string = 'oauth_token_secret=%s&oauth_token=%s' % (self.secret, + self.key) + self.assertEqual(self.token.to_string(), string) + + self.token.set_callback('http://www.example.com/my-callback') + string += '&oauth_callback_confirmed=true' + self.assertEqual(self.token.to_string(), string) + + def _compare_tokens(self, new): + self.assertEqual(self.token.key, new.key) + self.assertEqual(self.token.secret, new.secret) + # TODO: What about copying the callback to the new token? + # self.assertEqual(self.token.callback, new.callback) + self.assertEqual(self.token.callback_confirmed, + new.callback_confirmed) + # TODO: What about copying the verifier to the new token? + # self.assertEqual(self.token.verifier, new.verifier) + + def test_to_string(self): + tok = oauth.Token('tooken', 'seecret') + self.assertEqual(str(tok), 'oauth_token_secret=seecret&oauth_token=tooken') + + def test_from_string(self): + self.assertRaises(ValueError, lambda: oauth.Token.from_string('')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('blahblahblah')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('blah=blah')) + + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token_secret=asfdasf')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token_secret=')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=asfdasf')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=&oauth_token_secret=')) + self.assertRaises(ValueError, lambda: oauth.Token.from_string('oauth_token=tooken%26oauth_token_secret=seecret')) + + string = self.token.to_string() + new = oauth.Token.from_string(string) + self._compare_tokens(new) + + self.token.set_callback('http://www.example.com/my-callback') + string = self.token.to_string() + new = oauth.Token.from_string(string) + self._compare_tokens(new) + +class ReallyEqualMixin: + def failUnlessReallyEqual(self, a, b, msg=None): + self.failUnlessEqual(a, b, msg=msg) + self.failUnlessEqual(type(a), type(b), msg="a :: %r, b :: %r, %r" % (a, b, msg)) + +class TestFuncs(unittest.TestCase): + def test_to_unicode(self): + self.failUnlessRaises(TypeError, oauth.to_unicode, '\xae') + self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, '\xae') + self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, ['\xae']) + + self.failUnlessEqual(oauth.to_unicode(':-)'), ':-)') + self.failUnlessEqual(oauth.to_unicode(b'\u00ae'), b'\u00ae') + self.failUnlessEqual(oauth.to_unicode('\xc2\xae'), b'\u00ae') + self.failUnlessEqual(oauth.to_unicode_optional_iterator([':-)']), [':-)']) + self.failUnlessEqual(oauth.to_unicode_optional_iterator([b'\u00ae']), [b'\u00ae']) + +class TestRequest(unittest.TestCase, ReallyEqualMixin): + def test_setter(self): + url = "http://example.com" + method = "GET" + req = oauth.Request(method) + self.assertTrue(not hasattr(req, 'url') or req.url is None) + self.assertTrue(not hasattr(req, 'normalized_url') or req.normalized_url is None) + + def test_deleter(self): + url = "http://example.com" + method = "GET" + req = oauth.Request(method, url) + + try: + del req.url + url = req.url + self.fail("AttributeError should have been raised on empty url.") + except AttributeError: + pass + except Exception as e: + self.fail(str(e)) + + def test_url(self): + url1 = "http://example.com:80/foo.php" + url2 = "https://example.com:443/foo.php" + exp1 = "http://example.com/foo.php" + exp2 = "https://example.com/foo.php" + method = "GET" + + req = oauth.Request(method, url1) + self.assertEquals(req.normalized_url, exp1) + self.assertEquals(req.url, url1) + + req = oauth.Request(method, url2) + self.assertEquals(req.normalized_url, exp2) + self.assertEquals(req.url, url2) + + def test_bad_url(self): + request = oauth.Request() + try: + request.url = "ftp://example.com" + self.fail("Invalid URL scheme was accepted.") + except ValueError: + pass + + def test_unset_consumer_and_token(self): + consumer = oauth.Consumer('my_consumer_key', 'my_consumer_secret') + token = oauth.Token('my_key', 'my_secret') + request = oauth.Request("GET", "http://example.com/fetch.php") + request.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, + token) + + self.assertEquals(consumer.key, request['oauth_consumer_key']) + self.assertEquals(token.key, request['oauth_token']) + + def test_no_url_set(self): + consumer = oauth.Consumer('my_consumer_key', 'my_consumer_secret') + token = oauth.Token('my_key', 'my_secret') + request = oauth.Request() + + try: + try: + request.sign_request(oauth.SignatureMethod_HMAC_SHA1(), + consumer, token) + except TypeError: + self.fail("Signature method didn't check for a normalized URL.") + except ValueError: + pass + + def test_url_query(self): + url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10" + normalized_url = urlparse.urlunparse(urlparse.urlparse(url)[:3] + (None, None, None)) + method = "GET" + + req = oauth.Request(method, url) + self.assertEquals(req.url, url) + self.assertEquals(req.normalized_url, normalized_url) + + def test_get_parameter(self): + url = "http://example.com" + method = "GET" + params = {'oauth_consumer' : 'asdf'} + req = oauth.Request(method, url, parameters=params) + + self.assertEquals(req.get_parameter('oauth_consumer'), 'asdf') + self.assertRaises(oauth.Error, req.get_parameter, 'blah') + + def test_get_nonoauth_parameters(self): + + oauth_params = { + 'oauth_consumer': 'asdfasdfasdf' + } + + other_params = { + 'foo': 'baz', + 'bar': 'foo', + 'multi': ['FOO','BAR'], + 'uni_utf8': b'\xae', + 'uni_unicode': b'\u00ae', + 'uni_unicode_2': 'åÅøØ', + } + + params = oauth_params + params.update(other_params) + + req = oauth.Request("GET", "http://example.com", params) + self.assertEquals(other_params, req.get_nonoauth_parameters()) + + def test_to_header(self): + realm = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", realm, params) + header, value = list(req.to_header(realm).items())[0] + + parts = value.split('OAuth ') + vars = parts[1].split(', ') + self.assertTrue(len(vars), (len(params) + 1)) + + res = {} + for v in vars: + var, val = v.split('=') + res[var] = urllib.parse.unquote(val.strip('"')) + + self.assertEquals(realm, res['realm']) + del res['realm'] + + self.assertTrue(len(res), len(params)) + + for key, val in res.items(): + self.assertEquals(val, params.get(key)) + + def test_to_postdata_nonascii(self): + realm = "http://sp.example.com/" + + params = { + 'nonasciithing': b'q\xbfu\xe9 ,aasp u?..a.s', + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", realm, params) + + self.failUnlessReallyEqual(req.to_postdata(), 'nonasciithing=q%C2%BFu%C3%A9%20%2Caasp%20u%3F..a.s&oauth_nonce=4572616e48616d6d65724c61686176&oauth_timestamp=137131200&oauth_consumer_key=0685bd9184jfhq22&oauth_signature_method=HMAC-SHA1&oauth_version=1.0&oauth_token=ad180jjd733klru7&oauth_signature=wOJIO9A2W5mFwDgiDvZbTSMK%252FPY%253D') + + def test_to_postdata(self): + realm = "http://sp.example.com/" + + params = { + 'multi': ['FOO','BAR'], + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", realm, params) + + flat = [('multi','FOO'),('multi','BAR')] + del params['multi'] + flat.extend(list(params.items())) + kf = lambda x: x[0] + self.assertEquals(sorted(flat, key=kf), sorted(urllib.parse.parse_qsl(req.to_postdata()), key=kf)) + + def test_to_url(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", url, params) + exp = urlparse.urlparse("%s?%s" % (url, urllib.parse.urlencode(params))) + res = urlparse.urlparse(req.to_url()) + self.assertEquals(exp.scheme, res.scheme) + self.assertEquals(exp.netloc, res.netloc) + self.assertEquals(exp.path, res.path) + + a = urllib.parse.parse_qs(exp.query) + b = urllib.parse.parse_qs(res.query) + self.assertEquals(a, b) + + def test_to_url_with_query(self): + url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", url, params) + # Note: the url above already has query parameters, so append new ones with & + exp = urlparse.urlparse("%s&%s" % (url, urllib.parse.urlencode(params))) + res = urlparse.urlparse(req.to_url()) + self.assertEquals(exp.scheme, res.scheme) + self.assertEquals(exp.netloc, res.netloc) + self.assertEquals(exp.path, res.path) + + a = urllib.parse.parse_qs(exp.query) + b = urllib.parse.parse_qs(res.query) + self.assertTrue('alt' in b) + self.assertTrue('max-contacts' in b) + self.assertEquals(b['alt'], ['json']) + self.assertEquals(b['max-contacts'], ['10']) + self.assertEquals(a, b) + + def test_signature_base_string_nonascii_nonutf8(self): + consumer = oauth.Consumer('consumer_token', 'consumer_secret') + + url = b'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\u2766,+CA' + req = oauth.Request("GET", url) + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') + + url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\xe2\x9d\xa6,+CA' + req = oauth.Request("GET", url) + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') + + url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA' + req = oauth.Request("GET", url) + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') + + url = b'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA' + req = oauth.Request("GET", url) + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') + + def test_signature_base_string_with_query(self): + url = "https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10" + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + req = oauth.Request("GET", url, params) + self.assertEquals(req.normalized_url, 'https://www.google.com/m8/feeds/contacts/default/full/') + self.assertEquals(req.url, 'https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10') + normalized_params = urllib.parse.parse_qsl(req.get_normalized_parameters()) + self.assertTrue(len(normalized_params), len(params) + 2) + normalized_params = dict(normalized_params) + for key, value in params.items(): + if key == 'oauth_signature': + continue + self.assertEquals(value, normalized_params[key]) + self.assertEquals(normalized_params['alt'], 'json') + self.assertEquals(normalized_params['max-contacts'], '10') + + def test_get_normalized_parameters_empty(self): + url = "http://sp.example.com/?empty=" + + req = oauth.Request("GET", url) + + res = req.get_normalized_parameters() + + expected='empty=' + + self.assertEquals(expected, res) + + def test_get_normalized_parameters_duplicate(self): + url = "http://example.com/v2/search/videos?oauth_nonce=79815175&oauth_timestamp=1295397962&oauth_consumer_key=mykey&oauth_signature_method=HMAC-SHA1&q=car&oauth_version=1.0&offset=10&oauth_signature=spWLI%2FGQjid7sQVd5%2FarahRxzJg%3D" + + req = oauth.Request("GET", url) + + res = req.get_normalized_parameters() + + expected='oauth_consumer_key=mykey&oauth_nonce=79815175&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1295397962&oauth_version=1.0&offset=10&q=car' + + self.assertEquals(expected, res) + + def test_get_normalized_parameters_from_url(self): + # example copied from + # https://github.com/ciaranj/node-oauth/blob/master/tests/oauth.js + # which in turns says that it was copied from + # http://oauth.net/core/1.0/#sig_base_example . + url = "http://photos.example.net/photos?file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original" + + req = oauth.Request("GET", url) + + res = req.get_normalized_parameters() + + expected = 'file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original' + + self.assertEquals(expected, res) + + def test_signing_base(self): + # example copied from + # https://github.com/ciaranj/node-oauth/blob/master/tests/oauth.js + # which in turns says that it was copied from + # http://oauth.net/core/1.0/#sig_base_example . + url = "http://photos.example.net/photos?file=vacation.jpg&oauth_consumer_key=dpf43f3p2l4k3l03&oauth_nonce=kllo9940pd9333jh&oauth_signature_method=HMAC-SHA1&oauth_timestamp=1191242096&oauth_token=nnch734d00sl2jdk&oauth_version=1.0&size=original" + + req = oauth.Request("GET", url) + + sm = oauth.SignatureMethod_HMAC_SHA1() + + consumer = oauth.Consumer('dpf43f3p2l4k3l03', 'foo') + key, raw = sm.signing_base(req, consumer, None) + + expected = 'GET&http%3A%2F%2Fphotos.example.net%2Fphotos&file%3Dvacation.jpg%26oauth_consumer_key%3Ddpf43f3p2l4k3l03%26oauth_nonce%3Dkllo9940pd9333jh%26oauth_signature_method%3DHMAC-SHA1%26oauth_timestamp%3D1191242096%26oauth_token%3Dnnch734d00sl2jdk%26oauth_version%3D1.0%26size%3Doriginal' + self.assertEquals(expected, raw) + + def test_get_normalized_parameters(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'multi': ['FOO','BAR', b'\u00ae', '\xc2\xae'], + 'multi_same': ['FOO','FOO'], + 'uni_utf8_bytes': '\xc2\xae', + 'uni_unicode_object': b'\u00ae' + } + + req = oauth.Request("GET", url, params) + + res = req.get_normalized_parameters() + + expected='multi=BAR&multi=FOO&multi=%C2%AE&multi=%C2%AE&multi_same=FOO&multi_same=FOO&oauth_consumer_key=0685bd9184jfhq22&oauth_nonce=4572616e48616d6d65724c61686176&oauth_signature_method=HMAC-SHA1&oauth_timestamp=137131200&oauth_token=ad180jjd733klru7&oauth_version=1.0&uni_unicode_object=%C2%AE&uni_utf8_bytes=%C2%AE' + + self.assertEquals(expected, res) + + def test_get_normalized_parameters_ignores_auth_signature(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_signature': "some-random-signature-%d" % random.randint(1000, 2000), + 'oauth_token': "ad180jjd733klru7", + } + + req = oauth.Request("GET", url, params) + + res = req.get_normalized_parameters() + + self.assertNotEquals(urllib.parse.urlencode(sorted(params.items())), res) + + foo = params.copy() + del foo["oauth_signature"] + self.assertEqual(urllib.parse.urlencode(sorted(foo.items())), res) + + def test_set_signature_method(self): + consumer = oauth.Consumer('key', 'secret') + client = oauth.Client(consumer) + + class Blah: + pass + + try: + client.set_signature_method(Blah()) + self.fail("Client.set_signature_method() accepted invalid method.") + except ValueError: + pass + + m = oauth.SignatureMethod_HMAC_SHA1() + client.set_signature_method(m) + self.assertEquals(m, client.method) + + def test_get_normalized_string_escapes_spaces_properly(self): + url = "http://sp.example.com/" + params = { + "some_random_data": random.randint(100, 1000), + "data": "This data with a random number (%d) has spaces!" % random.randint(1000, 2000), + } + + req = oauth.Request("GET", url, params) + res = req.get_normalized_parameters() + expected = urllib.parse.urlencode(sorted(params.items())).replace('+', '%20') + self.assertEqual(expected, res) + + @mock.patch('oauth2.Request.make_timestamp') + @mock.patch('oauth2.Request.make_nonce') + def test_request_nonutf8_bytes(self, mock_make_nonce, mock_make_timestamp): + mock_make_nonce.return_value = 5 + mock_make_timestamp.return_value = 6 + + tok = oauth.Token(key="tok-test-key", secret="tok-test-secret") + con = oauth.Consumer(key="con-test-key", secret="con-test-secret") + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_token': tok.key, + 'oauth_consumer_key': con.key + } + + # If someone passes a sequence of bytes which is not ascii for + # url, we'll raise an exception as early as possible. + url = "http://sp.example.com/\x92" # It's actually cp1252-encoding... + self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params) + + # And if they pass an unicode, then we'll use it. + url = b'http://sp.example.com/\u2019' + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'cMzvCkhvLL57+sTIxLITTHfkqZk=') + + # And if it is a utf-8-encoded-then-percent-encoded non-ascii + # thing, we'll decode it and use it. + url = "http://sp.example.com/%E2%80%99" + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'yMLKOyNKC/DkyhUOb8DLSvceEWE=') + + # Same thing with the params. + url = "http://sp.example.com/" + + # If someone passes a sequence of bytes which is not ascii in + # params, we'll raise an exception as early as possible. + params['non_oauth_thing'] = '\xae', # It's actually cp1252-encoding... + self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params) + + # And if they pass a unicode, then we'll use it. + params['non_oauth_thing'] = b'\u2019' + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_signature'], '0GU50m0v60CVDB5JnoBXnvvvKx4=') + + # And if it is a utf-8-encoded non-ascii thing, we'll decode + # it and use it. + params['non_oauth_thing'] = '\xc2\xae' + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_signature'], 'pqOCu4qvRTiGiXB8Z61Jsey0pMM=') + + + # Also if there are non-utf8 bytes in the query args. + url = "http://sp.example.com/?q=\x92" # cp1252 + self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params) + + def test_request_hash_of_body(self): + tok = oauth.Token(key="token", secret="tok-test-secret") + con = oauth.Consumer(key="consumer", secret="con-test-secret") + + # Example 1a from Appendix A.1 of + # http://oauth.googlecode.com/svn/spec/ext/body_hash/1.0/oauth-bodyhash.html + # Except that we get a differetn result than they do. + + params = { + 'oauth_version': "1.0", + 'oauth_token': tok.key, + 'oauth_nonce': 10288510250934, + 'oauth_timestamp': 1236874155, + 'oauth_consumer_key': con.key + } + + url = "http://www.example.com/resource" + req = oauth.Request(method="PUT", url=url, parameters=params, body="Hello World!", is_form_encoded=False) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_body_hash'], 'Lve95gjOVATpfV8EL5X4nxwjKHE=') + self.failUnlessReallyEqual(req['oauth_signature'], 't+MX8l/0S8hdbVQL99nD0X1fPnM=') + # oauth-bodyhash.html A.1 has + # '08bUFF%2Fjmp59mWB7cSgCYBUpJ0U%3D', but I don't see how that + # is possible. + + # Example 1b + params = { + 'oauth_version': "1.0", + 'oauth_token': tok.key, + 'oauth_nonce': 10369470270925, + 'oauth_timestamp': 1236874236, + 'oauth_consumer_key': con.key + } + + req = oauth.Request(method="PUT", url=url, parameters=params, body="Hello World!", is_form_encoded=False) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_body_hash'], 'Lve95gjOVATpfV8EL5X4nxwjKHE=') + self.failUnlessReallyEqual(req['oauth_signature'], 'CTFmrqJIGT7NsWJ42OrujahTtTc=') + + # Appendix A.2 + params = { + 'oauth_version': "1.0", + 'oauth_token': tok.key, + 'oauth_nonce': 8628868109991, + 'oauth_timestamp': 1238395022, + 'oauth_consumer_key': con.key + } + + req = oauth.Request(method="GET", url=url, parameters=params, is_form_encoded=False) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) + self.failUnlessReallyEqual(req['oauth_body_hash'], '2jmj7l5rSw0yVb/vlWAYkK/YBwk=') + self.failUnlessReallyEqual(req['oauth_signature'], 'Zhl++aWSP0O3/hYQ0CuBc7jv38I=') + + + def test_sign_request(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200" + } + + tok = oauth.Token(key="tok-test-key", secret="tok-test-secret") + con = oauth.Consumer(key="con-test-key", secret="con-test-secret") + + params['oauth_token'] = tok.key + params['oauth_consumer_key'] = con.key + req = oauth.Request(method="GET", url=url, parameters=params) + + methods = { + 'DX01TdHws7OninCLK9VztNTH1M4=': oauth.SignatureMethod_HMAC_SHA1(), + 'con-test-secret&tok-test-secret': oauth.SignatureMethod_PLAINTEXT() + } + + for exp, method in methods.items(): + req.sign_request(method, con, tok) + self.assertEquals(req['oauth_signature_method'], method.name) + self.assertEquals(req['oauth_signature'], exp) + + # Also if there are non-ascii chars in the URL. + url = "http://sp.example.com/\xe2\x80\x99" # utf-8 bytes + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) + self.assertEquals(req['oauth_signature'], 'loFvp5xC7YbOgd9exIO6TxB7H4s=') + + url = b'http://sp.example.com/\u2019' # Python unicode object + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) + self.assertEquals(req['oauth_signature'], 'loFvp5xC7YbOgd9exIO6TxB7H4s=') + + # Also if there are non-ascii chars in the query args. + url = "http://sp.example.com/?q=\xe2\x80\x99" # utf-8 bytes + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) + self.assertEquals(req['oauth_signature'], 'IBw5mfvoCsDjgpcsVKbyvsDqQaU=') + + url = b'http://sp.example.com/?q=\u2019' # Python unicode object + req = oauth.Request(method="GET", url=url, parameters=params) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) + self.assertEquals(req['oauth_signature'], 'IBw5mfvoCsDjgpcsVKbyvsDqQaU=') + + def test_from_request(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + req = oauth.Request("GET", url, params) + headers = req.to_header() + + # Test from the headers + req = oauth.Request.from_request("GET", url, headers) + self.assertEquals(req.method, "GET") + self.assertEquals(req.url, url) + + self.assertEquals(params, req.copy()) + + # Test with bad OAuth headers + bad_headers = { + 'Authorization' : 'OAuth this is a bad header' + } + + self.assertRaises(oauth.Error, oauth.Request.from_request, "GET", + url, bad_headers) + + # Test getting from query string + qs = urllib.parse.urlencode(params) + req = oauth.Request.from_request("GET", url, query_string=qs) + + exp = urllib.parse.parse_qs(qs, keep_blank_values=False) + for k, v in exp.items(): + exp[k] = urllib.parse.unquote(v[0]) + + self.assertEquals(exp, req.copy()) + + # Test that a boned from_request() call returns None + req = oauth.Request.from_request("GET", url) + self.assertEquals(None, req) + + def test_from_token_and_callback(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': "137131200", + 'oauth_consumer_key': "0685bd9184jfhq22", + 'oauth_signature_method': "HMAC-SHA1", + 'oauth_token': "ad180jjd733klru7", + 'oauth_signature': "wOJIO9A2W5mFwDgiDvZbTSMK%2FPY%3D", + } + + tok = oauth.Token(key="tok-test-key", secret="tok-test-secret") + req = oauth.Request.from_token_and_callback(tok) + self.assertFalse('oauth_callback' in req) + self.assertEquals(req['oauth_token'], tok.key) + + req = oauth.Request.from_token_and_callback(tok, callback=url) + self.assertTrue('oauth_callback' in req) + self.assertEquals(req['oauth_callback'], url) + + def test_from_consumer_and_token(self): + url = "http://sp.example.com/" + + tok = oauth.Token(key="tok-test-key", secret="tok-test-secret") + tok.set_verifier('this_is_a_test_verifier') + con = oauth.Consumer(key="con-test-key", secret="con-test-secret") + req = oauth.Request.from_consumer_and_token(con, token=tok, + http_method="GET", http_url=url) + + self.assertEquals(req['oauth_token'], tok.key) + self.assertEquals(req['oauth_consumer_key'], con.key) + self.assertEquals(tok.verifier, req['oauth_verifier']) + +class SignatureMethod_Bad(oauth.SignatureMethod): + name = "BAD" + + def signing_base(self, request, consumer, token): + return "" + + def sign(self, request, consumer, token): + return "invalid-signature" + + +class TestServer(unittest.TestCase): + def setUp(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': "1.0", + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': int(time.time()), + 'bar': 'blerg', + 'multi': ['FOO','BAR'], + 'foo': 59 + } + + self.consumer = oauth.Consumer(key="consumer-key", + secret="consumer-secret") + self.token = oauth.Token(key="token-key", secret="token-secret") + + params['oauth_token'] = self.token.key + params['oauth_consumer_key'] = self.consumer.key + self.request = oauth.Request(method="GET", url=url, parameters=params) + + signature_method = oauth.SignatureMethod_HMAC_SHA1() + self.request.sign_request(signature_method, self.consumer, self.token) + + def test_init(self): + server = oauth.Server(signature_methods={'HMAC-SHA1' : oauth.SignatureMethod_HMAC_SHA1()}) + self.assertTrue('HMAC-SHA1' in server.signature_methods) + self.assertTrue(isinstance(server.signature_methods['HMAC-SHA1'], + oauth.SignatureMethod_HMAC_SHA1)) + + server = oauth.Server() + self.assertEquals(server.signature_methods, {}) + + def test_add_signature_method(self): + server = oauth.Server() + res = server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + self.assertTrue(len(res) == 1) + self.assertTrue('HMAC-SHA1' in res) + self.assertTrue(isinstance(res['HMAC-SHA1'], + oauth.SignatureMethod_HMAC_SHA1)) + + res = server.add_signature_method(oauth.SignatureMethod_PLAINTEXT()) + self.assertTrue(len(res) == 2) + self.assertTrue('PLAINTEXT' in res) + self.assertTrue(isinstance(res['PLAINTEXT'], + oauth.SignatureMethod_PLAINTEXT)) + + def test_verify_request(self): + server = oauth.Server() + server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + + parameters = server.verify_request(self.request, self.consumer, + self.token) + + self.assertTrue('bar' in parameters) + self.assertTrue('foo' in parameters) + self.assertTrue('multi' in parameters) + self.assertEquals(parameters['bar'], 'blerg') + self.assertEquals(parameters['foo'], 59) + self.assertEquals(parameters['multi'], ['FOO','BAR']) + + def test_build_authenticate_header(self): + server = oauth.Server() + headers = server.build_authenticate_header('example.com') + self.assertTrue('WWW-Authenticate' in headers) + self.assertEquals('OAuth realm="example.com"', + headers['WWW-Authenticate']) + + def test_no_version(self): + url = "http://sp.example.com/" + + params = { + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': int(time.time()), + 'bar': 'blerg', + 'multi': ['FOO','BAR'], + 'foo': 59 + } + + self.consumer = oauth.Consumer(key="consumer-key", + secret="consumer-secret") + self.token = oauth.Token(key="token-key", secret="token-secret") + + params['oauth_token'] = self.token.key + params['oauth_consumer_key'] = self.consumer.key + self.request = oauth.Request(method="GET", url=url, parameters=params) + + signature_method = oauth.SignatureMethod_HMAC_SHA1() + self.request.sign_request(signature_method, self.consumer, self.token) + + server = oauth.Server() + server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + + parameters = server.verify_request(self.request, self.consumer, + self.token) + + def test_invalid_version(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': '222.9922', + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': int(time.time()), + 'bar': 'blerg', + 'multi': ['foo','bar'], + 'foo': 59 + } + + consumer = oauth.Consumer(key="consumer-key", + secret="consumer-secret") + token = oauth.Token(key="token-key", secret="token-secret") + + params['oauth_token'] = token.key + params['oauth_consumer_key'] = consumer.key + request = oauth.Request(method="GET", url=url, parameters=params) + + signature_method = oauth.SignatureMethod_HMAC_SHA1() + request.sign_request(signature_method, consumer, token) + + server = oauth.Server() + server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + + self.assertRaises(oauth.Error, server.verify_request, request, consumer, token) + + def test_invalid_signature_method(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': '1.0', + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': int(time.time()), + 'bar': 'blerg', + 'multi': ['FOO','BAR'], + 'foo': 59 + } + + consumer = oauth.Consumer(key="consumer-key", + secret="consumer-secret") + token = oauth.Token(key="token-key", secret="token-secret") + + params['oauth_token'] = token.key + params['oauth_consumer_key'] = consumer.key + request = oauth.Request(method="GET", url=url, parameters=params) + + signature_method = SignatureMethod_Bad() + request.sign_request(signature_method, consumer, token) + + server = oauth.Server() + server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + + self.assertRaises(oauth.Error, server.verify_request, request, + consumer, token) + + def test_missing_signature(self): + url = "http://sp.example.com/" + + params = { + 'oauth_version': '1.0', + 'oauth_nonce': "4572616e48616d6d65724c61686176", + 'oauth_timestamp': int(time.time()), + 'bar': 'blerg', + 'multi': ['FOO','BAR'], + 'foo': 59 + } + + consumer = oauth.Consumer(key="consumer-key", + secret="consumer-secret") + token = oauth.Token(key="token-key", secret="token-secret") + + params['oauth_token'] = token.key + params['oauth_consumer_key'] = consumer.key + request = oauth.Request(method="GET", url=url, parameters=params) + + signature_method = oauth.SignatureMethod_HMAC_SHA1() + request.sign_request(signature_method, consumer, token) + del request['oauth_signature'] + + server = oauth.Server() + server.add_signature_method(oauth.SignatureMethod_HMAC_SHA1()) + + self.assertRaises(oauth.MissingSignature, server.verify_request, + request, consumer, token) + + +# Request Token: http://oauth-sandbox.sevengoslings.net/request_token +# Auth: http://oauth-sandbox.sevengoslings.net/authorize +# Access Token: http://oauth-sandbox.sevengoslings.net/access_token +# Two-legged: http://oauth-sandbox.sevengoslings.net/two_legged +# Three-legged: http://oauth-sandbox.sevengoslings.net/three_legged +# Key: bd37aed57e15df53 +# Secret: 0e9e6413a9ef49510a4f68ed02cd +class TestClient(unittest.TestCase): +# oauth_uris = { +# 'request_token': '/request_token.php', +# 'access_token': '/access_token.php' +# } + + oauth_uris = { + 'request_token': '/request_token', + 'authorize': '/authorize', + 'access_token': '/access_token', + 'two_legged': '/two_legged', + 'three_legged': '/three_legged' + } + + consumer_key = 'bd37aed57e15df53' + consumer_secret = '0e9e6413a9ef49510a4f68ed02cd' + host = 'http://oauth-sandbox.sevengoslings.net' + + def setUp(self): + self.consumer = oauth.Consumer(key=self.consumer_key, + secret=self.consumer_secret) + + self.body = { + 'foo': 'bar', + 'bar': 'foo', + 'multi': ['FOO','BAR'], + 'blah': 599999 + } + + def _uri(self, type): + uri = self.oauth_uris.get(type) + if uri is None: + raise KeyError("%s is not a valid OAuth URI type." % type) + + return "%s%s" % (self.host, uri) + + def create_simple_multipart_data(self, data): + boundary = '---Boundary-%d' % random.randint(1,1000) + crlf = '\r\n' + items = [] + for key, value in data.items(): + items += [ + '--'+boundary, + 'Content-Disposition: form-data; name="%s"'%str(key), + '', + str(value), + ] + items += ['', '--'+boundary+'--', ''] + content_type = 'multipart/form-data; boundary=%s' % boundary + return content_type, crlf.join(items) + + def test_init(self): + class Blah(): + pass + + try: + client = oauth.Client(Blah()) + self.fail("Client.__init__() accepted invalid Consumer.") + except ValueError: + pass + + consumer = oauth.Consumer('token', 'secret') + try: + client = oauth.Client(consumer, Blah()) + self.fail("Client.__init__() accepted invalid Token.") + except ValueError: + pass + + def test_access_token_get(self): + """Test getting an access token via GET.""" + client = oauth.Client(self.consumer, None) + resp, content = client.request(self._uri('request_token'), "GET") + + self.assertEquals(int(resp['status']), 200) + + def test_access_token_post(self): + """Test getting an access token via POST.""" + client = oauth.Client(self.consumer, None) + resp, content = client.request(self._uri('request_token'), "POST") + + self.assertEquals(int(resp['status']), 200) + + res = dict(urllib.parse.parse_qsl(content)) + self.assertTrue('oauth_token' in res) + self.assertTrue('oauth_token_secret' in res) + + def _two_legged(self, method): + client = oauth.Client(self.consumer, None) + + return client.request(self._uri('two_legged'), method, + body=urllib.parse.urlencode(self.body)) + + def test_two_legged_post(self): + """A test of a two-legged OAuth POST request.""" + resp, content = self._two_legged("POST") + + self.assertEquals(int(resp['status']), 200) + + def test_two_legged_get(self): + """A test of a two-legged OAuth GET request.""" + resp, content = self._two_legged("GET") + self.assertEquals(int(resp['status']), 200) + + @mock.patch('httplib2.Http.request') + def test_multipart_post_does_not_alter_body(self, mockHttpRequest): + random_result = random.randint(1,100) + + data = { + 'rand-%d'%random.randint(1,100):random.randint(1,100), + } + content_type, body = self.create_simple_multipart_data(data) + + client = oauth.Client(self.consumer, None) + uri = self._uri('two_legged') + + def mockrequest(cl, ur, **kw): + self.failUnless(cl is client) + self.failUnless(ur is uri) + self.failUnlessEqual(frozenset(kw.keys()), frozenset(['method', 'body', 'redirections', 'connection_type', 'headers'])) + self.failUnlessEqual(kw['body'], body) + self.failUnlessEqual(kw['connection_type'], None) + self.failUnlessEqual(kw['method'], 'POST') + self.failUnlessEqual(kw['redirections'], httplib2.DEFAULT_MAX_REDIRECTS) + self.failUnless(isinstance(kw['headers'], dict)) + + return random_result + + mockHttpRequest.side_effect = mockrequest + + result = client.request(uri, 'POST', headers={'Content-Type':content_type}, body=body) + self.assertEqual(result, random_result) + + @mock.patch('httplib2.Http.request') + def test_url_with_query_string(self, mockHttpRequest): + uri = 'http://example.com/foo/bar/?show=thundercats&character=snarf' + client = oauth.Client(self.consumer, None) + random_result = random.randint(1,100) + + def mockrequest(cl, ur, **kw): + self.failUnless(cl is client) + self.failUnlessEqual(frozenset(kw.keys()), frozenset(['method', 'body', 'redirections', 'connection_type', 'headers'])) + self.failUnlessEqual(kw['body'], '') + self.failUnlessEqual(kw['connection_type'], None) + self.failUnlessEqual(kw['method'], 'GET') + self.failUnlessEqual(kw['redirections'], httplib2.DEFAULT_MAX_REDIRECTS) + self.failUnless(isinstance(kw['headers'], dict)) + + req = oauth.Request.from_consumer_and_token(self.consumer, None, + http_method='GET', http_url=uri, parameters={}) + req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), self.consumer, None) + expected = urllib.parse.parse_qsl(urlparse.urlparse(req.to_url()).query) + actual = urllib.parse.parse_qsl(urlparse.urlparse(ur).query) + self.failUnlessEqual(len(expected), len(actual)) + actual = dict(actual) + for key, value in expected: + if key not in ('oauth_signature', 'oauth_nonce', 'oauth_timestamp'): + self.failUnlessEqual(actual[key], value) + + return random_result + + mockHttpRequest.side_effect = mockrequest + + client.request(uri, 'GET') + + @mock.patch('httplib2.Http.request') + @mock.patch('oauth2.Request.from_consumer_and_token') + def test_multiple_values_for_a_key(self, mockReqConstructor, mockHttpRequest): + client = oauth.Client(self.consumer, None) + + request = oauth.Request("GET", "http://example.com/fetch.php", parameters={'multi': ['1', '2']}) + mockReqConstructor.return_value = request + + client.request('http://whatever', 'POST', body='multi=1&multi=2') + + self.failUnlessEqual(mockReqConstructor.call_count, 1) + self.failUnlessEqual(mockReqConstructor.call_args[1]['parameters'], {'multi': ['1', '2']}) + + self.failUnless('multi=1' in mockHttpRequest.call_args[1]['body']) + self.failUnless('multi=2' in mockHttpRequest.call_args[1]['body']) + +if __name__ == "__main__": + unittest.main() diff --git a/dist/oauth2-1.5.211-py3.3.egg b/dist/oauth2-1.5.211-py3.3.egg new file mode 100644 index 00000000..5f312500 Binary files /dev/null and b/dist/oauth2-1.5.211-py3.3.egg differ diff --git a/oauth2/__init__.py b/oauth2/__init__.py index 835270e3..d8ff20ba 100644 --- a/oauth2/__init__.py +++ b/oauth2/__init__.py @@ -23,29 +23,15 @@ """ import base64 -import urllib import time import random -import urlparse +import urllib.parse import hmac import binascii import httplib2 +from hashlib import sha1 -try: - from urlparse import parse_qs - parse_qs # placate pyflakes -except ImportError: - # fall back for Python 2.5 - from cgi import parse_qs - -try: - from hashlib import sha1 - sha = sha1 -except ImportError: - # hashlib was added in Python 2.5 - import sha - -import _version +import oauth2._version __version__ = _version.__version__ @@ -87,7 +73,7 @@ def build_xoauth_string(url, consumer, token=None): request.sign_request(signing_method, consumer, token) params = [] - for k, v in sorted(request.iteritems()): + for k, v in sorted(request.items()): if v is not None: params.append('%s="%s"' % (k, escape(v))) @@ -97,12 +83,12 @@ def build_xoauth_string(url, consumer, token=None): def to_unicode(s): """ Convert to unicode, raise exception with instructive error message if s is not unicode, ascii, or utf-8. """ - if not isinstance(s, unicode): - if not isinstance(s, str): - raise TypeError('You are required to pass either unicode or string here, not: %r (%s)' % (type(s), s)) + if not isinstance(s, (bytes, str)): + raise TypeError('You are required to pass either unicode or string here, not: %r (%s)' % (type(s), s)) + if isinstance(s, bytes): try: s = s.decode('utf-8') - except UnicodeDecodeError, le: + except UnicodeDecodeError as le: raise TypeError('You are required to pass either a unicode object or a utf-8 string here. You passed a Python string object which contained non-utf-8: %r. The UnicodeDecodeError that resulted from attempting to interpret it as utf-8 was: %s' % (s, le,)) return s @@ -110,13 +96,13 @@ def to_utf8(s): return to_unicode(s).encode('utf-8') def to_unicode_if_string(s): - if isinstance(s, basestring): + if isinstance(s, (bytes, str)): return to_unicode(s) else: return s def to_utf8_if_string(s): - if isinstance(s, basestring): + if isinstance(s, (bytes, str)): return to_utf8(s) else: return s @@ -126,12 +112,12 @@ def to_unicode_optional_iterator(x): Raise TypeError if x is a str containing non-utf8 bytes or if x is an iterable which contains such a str. """ - if isinstance(x, basestring): + if isinstance(x, (bytes, str)): return to_unicode(x) try: l = list(x) - except TypeError, e: + except TypeError as e: assert 'is not iterable' in str(e) return x else: @@ -142,12 +128,12 @@ def to_utf8_optional_iterator(x): Raise TypeError if x is a str or if x is an iterable which contains a str. """ - if isinstance(x, basestring): + if isinstance(x, (bytes, str)): return to_utf8(x) try: l = list(x) - except TypeError, e: + except TypeError as e: assert 'is not iterable' in str(e) return x else: @@ -155,7 +141,7 @@ def to_utf8_optional_iterator(x): def escape(s): """Escape a URL including any /.""" - return urllib.quote(s.encode('utf-8'), safe='~') + return urllib.parse.quote(s, safe='~') def generate_timestamp(): """Get seconds since epoch (UTC).""" @@ -206,7 +192,7 @@ def __str__(self): data = {'oauth_consumer_key': self.key, 'oauth_consumer_secret': self.secret} - return urllib.urlencode(data) + return urllib.parse.urlencode(data) class Token(object): @@ -250,13 +236,13 @@ def set_verifier(self, verifier=None): def get_callback_url(self): if self.callback and self.verifier: # Append the oauth_verifier. - parts = urlparse.urlparse(self.callback) + parts = urllib.parse.urlparse(self.callback) scheme, netloc, path, params, query, fragment = parts[:6] if query: query = '%s&oauth_verifier=%s' % (query, self.verifier) else: query = 'oauth_verifier=%s' % self.verifier - return urlparse.urlunparse((scheme, netloc, path, params, + return urllib.parse.urlunparse((scheme, netloc, path, params, query, fragment)) return self.callback @@ -274,7 +260,7 @@ def to_string(self): if self.callback_confirmed is not None: data['oauth_callback_confirmed'] = self.callback_confirmed - return urllib.urlencode(data) + return urllib.parse.urlencode(data) @staticmethod def from_string(s): @@ -284,7 +270,7 @@ def from_string(s): if not len(s): raise ValueError("Invalid parameter string.") - params = parse_qs(s, keep_blank_values=False) + params = urllib.parse.parse_qs(s, keep_blank_values=False) if not len(params): raise ValueError("Invalid parameter string.") @@ -345,7 +331,7 @@ def __init__(self, method=HTTP_METHOD, url=None, parameters=None, self.url = to_unicode(url) self.method = method if parameters is not None: - for k, v in parameters.iteritems(): + for k, v in parameters.items(): k = to_unicode(k) v = to_unicode_optional_iterator(v) self[k] = v @@ -357,7 +343,7 @@ def __init__(self, method=HTTP_METHOD, url=None, parameters=None, def url(self, value): self.__dict__['url'] = value if value is not None: - scheme, netloc, path, params, query, fragment = urlparse.urlparse(value) + scheme, netloc, path, params, query, fragment = urllib.parse.urlparse(value) # Exclude default port numbers. if scheme == 'http' and netloc[-3:] == ':80': @@ -368,7 +354,7 @@ def url(self, value): raise ValueError("Unsupported URL %s (%s)." % (value, scheme)) # Normalized URL excludes params, query, and fragment. - self.normalized_url = urlparse.urlunparse((scheme, netloc, path, None, None, None)) + self.normalized_url = urllib.parse.urlunparse((scheme, netloc, path, None, None, None)) else: self.normalized_url = None self.__dict__['url'] = None @@ -382,7 +368,7 @@ def _get_timestamp_nonce(self): def get_nonoauth_parameters(self): """Get any non-OAuth parameters.""" - return dict([(k, v) for k, v in self.iteritems() + return dict([(k, v) for k, v in self.items() if not k.startswith('oauth_')]) def to_header(self, realm=''): @@ -402,23 +388,24 @@ def to_header(self, realm=''): def to_postdata(self): """Serialize as post data for a POST request.""" d = {} - for k, v in self.iteritems(): - d[k.encode('utf-8')] = to_utf8_optional_iterator(v) + for k, v in self.items(): + #d[k.encode('utf-8')] = to_utf8_optional_iterator(v) + d[k] = to_unicode_optional_iterator(v) # tell urlencode to deal with sequence values and map them correctly # to resulting querystring. for example self["k"] = ["v1", "v2"] will # result in 'k=v1&k=v2' and not k=%5B%27v1%27%2C+%27v2%27%5D - return urllib.urlencode(d, True).replace('+', '%20') + return urllib.parse.urlencode(d, True).replace('+', '%20') def to_url(self): """Serialize as a URL for a GET request.""" - base_url = urlparse.urlparse(self.url) + base_url = urllib.parse.urlparse(self.url) try: query = base_url.query except AttributeError: # must be python <2.5 query = base_url[4] - query = parse_qs(query) + query = urllib.parse.parse_qs(query) for k, v in self.items(): query.setdefault(k, []).append(v) @@ -437,8 +424,8 @@ def to_url(self): fragment = base_url[5] url = (scheme, netloc, path, params, - urllib.urlencode(query, True), fragment) - return urlparse.urlunparse(url) + urllib.parse.urlencode(query, True), fragment) + return urllib.parse.urlunparse(url) def get_parameter(self, parameter): ret = self.get(parameter) @@ -450,31 +437,31 @@ def get_parameter(self, parameter): def get_normalized_parameters(self): """Return a string that contains the parameters that must be signed.""" items = [] - for key, value in self.iteritems(): + for key, value in self.items(): if key == 'oauth_signature': continue # 1.0a/9.1.1 states that kvp must be sorted by key, then by value, # so we unpack sequence values into multiple items for sorting. - if isinstance(value, basestring): + if isinstance(value, str): items.append((to_utf8_if_string(key), to_utf8(value))) else: try: value = list(value) - except TypeError, e: + except TypeError as e: assert 'is not iterable' in str(e) items.append((to_utf8_if_string(key), to_utf8_if_string(value))) else: items.extend((to_utf8_if_string(key), to_utf8_if_string(item)) for item in value) # Include any query string parameters from the provided URL - query = urlparse.urlparse(self.url)[4] + query = urllib.parse.urlparse(self.url)[4] url_items = self._split_url_string(query).items() url_items = [(to_utf8(k), to_utf8(v)) for k, v in url_items if k != 'oauth_signature' ] items.extend(url_items) items.sort() - encoded_str = urllib.urlencode(items) + encoded_str = urllib.parse.urlencode(items) # Encode signature parameters per Oauth Core 1.0 protocol # spec draft 7, section 3.6 # (http://tools.ietf.org/html/draft-hammer-oauth-07#section-3.6) @@ -490,7 +477,7 @@ def sign_request(self, signature_method, consumer, token): # section 4.1.1 "OAuth Consumers MUST NOT include an # oauth_body_hash parameter on requests with form-encoded # request bodies." - self['oauth_body_hash'] = base64.b64encode(sha(self.body).digest()) + self['oauth_body_hash'] = to_unicode(base64.b64encode(sha1(to_utf8(self.body)).digest())) if 'oauth_consumer_key' not in self: self['oauth_consumer_key'] = consumer.key @@ -509,7 +496,8 @@ def make_timestamp(cls): @classmethod def make_nonce(cls): """Generate pseudorandom number.""" - return str(random.randint(0, 100000000)) + #return str(random.randint(0, 100000000)) + return base64.b64encode(("%0x" % random.getrandbits(256)).encode("utf-8"))[:32] @classmethod def from_request(cls, http_method, http_url, headers=None, parameters=None, @@ -538,7 +526,7 @@ def from_request(cls, http_method, http_url, headers=None, parameters=None, parameters.update(query_params) # URL parameters. - param_str = urlparse.urlparse(http_url)[4] # query + param_str = urllib.parse.urlparse(http_url)[4] # query url_params = cls._split_url_string(param_str) parameters.update(url_params) @@ -600,15 +588,15 @@ def _split_header(header): # Split key-value. param_parts = param.split('=', 1) # Remove quotes and unescape the value. - params[param_parts[0]] = urllib.unquote(param_parts[1].strip('\"')) + params[param_parts[0]] = urllib.parse.unquote(param_parts[1].strip('\"')) return params @staticmethod def _split_url_string(param_str): """Turn URL string into parameters.""" - parameters = parse_qs(param_str.encode('utf-8'), keep_blank_values=True) - for k, v in parameters.iteritems(): - parameters[k] = urllib.unquote(v[0]) + parameters = urllib.parse.parse_qs(param_str, keep_blank_values=True) + for k, v in parameters.items(): + parameters[k] = urllib.parse.unquote(v[0]) return parameters @@ -638,7 +626,7 @@ def set_signature_method(self, method): def request(self, uri, method="GET", body='', headers=None, redirections=httplib2.DEFAULT_MAX_REDIRECTS, connection_type=None): - DEFAULT_POST_CONTENT_TYPE = 'application/x-www-form-urlencoded' + DEFAULT_POST_CONTENT_TYPE = 'application/x-www-form-urlencoded;charset=UTF-8' if not isinstance(headers, dict): headers = {} @@ -648,10 +636,10 @@ def request(self, uri, method="GET", body='', headers=None, DEFAULT_POST_CONTENT_TYPE) is_form_encoded = \ - headers.get('Content-Type') == 'application/x-www-form-urlencoded' + headers.get('Content-Type') == 'application/x-www-form-urlencoded;charset=UTF-8' if is_form_encoded and body: - parameters = parse_qs(body) + parameters = urllib.parse.parse_qs(body) else: parameters = None @@ -661,12 +649,12 @@ def request(self, uri, method="GET", body='', headers=None, req.sign_request(self.method, self.consumer, self.token) - schema, rest = urllib.splittype(uri) + schema, rest = urllib.parse.splittype(uri) if rest.startswith('//'): hierpart = '//' else: hierpart = '' - host, rest = urllib.splithost(rest) + host, rest = urllib.parse.splithost(rest) realm = schema + ':' + hierpart + host @@ -837,10 +825,10 @@ def sign(self, request, consumer, token): """Builds the base signature string.""" key, raw = self.signing_base(request, consumer, token) - hashed = hmac.new(key, raw, sha) + hashed = hmac.new(to_utf8(key), to_utf8(raw), sha1) # Calculate the digest base 64. - return binascii.b2a_base64(hashed.digest())[:-1] + return to_unicode(binascii.b2a_base64(hashed.digest())[:-1]) class SignatureMethod_PLAINTEXT(SignatureMethod): diff --git a/oauth2/_version.py b/oauth2/_version.py index 9d779eaa..1968f88d 100644 --- a/oauth2/_version.py +++ b/oauth2/_version.py @@ -4,7 +4,7 @@ -auto_build_num = "211" +auto_build_num = "212" diff --git a/setup.py b/setup.py index acc41e17..5e3ecf85 100755 --- a/setup.py +++ b/setup.py @@ -1,40 +1,39 @@ #!/usr/bin/env python -from setuptools import setup, find_packages -import os, re +import os +import re -PKG='oauth2' -VERSIONFILE = os.path.join('oauth2', '_version.py') -verstr = "unknown" -try: - verstrline = open(VERSIONFILE, "rt").read() -except EnvironmentError: - pass # Okay, there is no version file. +from setuptools import find_packages, setup + +PKG = 'oauth2' +VERSIONFILE = os.path.join(os.path.dirname(__file__), 'oauth2', '_version.py') +verstrline = open(VERSIONFILE, "rt").read() +MVSRE = r'''^manual_verstr *= *['"]([^'"]*)['"]''' +mo = re.search(MVSRE, verstrline, re.M) +if mo: + mverstr = mo.group(1) +else: + print("unable to find version in %s" % (VERSIONFILE, )) + raise RuntimeError( + "if %s.py exists, it must be well-formed" % (VERSIONFILE, )) +AVSRE = r'''^auto_build_num *= *['"]([^'"]*)['"]''' +mo = re.search(AVSRE, verstrline, re.M) +if mo: + averstr = mo.group(1) else: - MVSRE = r"^manual_verstr *= *['\"]([^'\"]*)['\"]" - mo = re.search(MVSRE, verstrline, re.M) - if mo: - mverstr = mo.group(1) - else: - print "unable to find version in %s" % (VERSIONFILE,) - raise RuntimeError("if %s.py exists, it must be well-formed" % (VERSIONFILE,)) - AVSRE = r"^auto_build_num *= *['\"]([^'\"]*)['\"]" - mo = re.search(AVSRE, verstrline, re.M) - if mo: - averstr = mo.group(1) - else: - averstr = '' - verstr = '.'.join([mverstr, averstr]) + averstr = '' +verstr = '.'.join([mverstr, averstr]) -setup(name=PKG, - version=verstr, - description="library for OAuth version 1.0", - author="Joe Stump", - author_email="joe@simplegeo.com", - url="http://github.com/simplegeo/python-oauth2", - packages = find_packages(), - install_requires = ['httplib2'], - license = "MIT License", - keywords="oauth", - zip_safe = True, - test_suite="tests", - tests_require=['coverage', 'mock']) +setup( + name=PKG, + version=verstr, + description="library for OAuth version 1.0", + author="Joe Stump", + author_email="joe@simplegeo.com", + url="http://github.com/simplegeo/python-oauth2", + packages=find_packages(), + install_requires=['httplib2'], + license="MIT License", + keywords="oauth", + zip_safe=True, + test_suite="tests", + tests_require=['coverage', 'mock']) diff --git a/tests/test_oauth.py b/tests/test_oauth.py index 099e5794..c6efeeba 100644 --- a/tests/test_oauth.py +++ b/tests/test_oauth.py @@ -1,4 +1,4 @@ -# -*- coding: utf-8 -*- +# -*- coding: utf-8 -*- """ The MIT License @@ -29,19 +29,11 @@ import oauth2 as oauth import random import time -import urllib -import urlparse +import urllib.parse from types import ListType import mock import httplib2 -# Fix for python2.5 compatibility -try: - from urlparse import parse_qs, parse_qsl -except ImportError: - from cgi import parse_qs, parse_qsl - - sys.path[0:0] = [os.path.join(os.path.dirname(__file__), ".."),] @@ -49,19 +41,19 @@ class TestError(unittest.TestCase): def test_message(self): try: raise oauth.Error - except oauth.Error, e: + except oauth.Error as e: self.assertEqual(e.message, 'OAuth error occurred.') msg = 'OMG THINGS BROKE!!!!' try: raise oauth.Error(msg) - except oauth.Error, e: + except oauth.Error as e: self.assertEqual(e.message, msg) def test_str(self): try: raise oauth.Error - except oauth.Error, e: + except oauth.Error as e: self.assertEquals(str(e), 'OAuth error occurred.') class TestGenerateFunctions(unittest.TestCase): @@ -136,7 +128,7 @@ def test_basic(self): self.assertRaises(ValueError, lambda: oauth.Consumer(None, 'dasf')) def test_str(self): - res = dict(parse_qsl(str(self.consumer))) + res = dict(urllib.parse.parse_qsl(str(self.consumer))) self.assertTrue('oauth_consumer_key' in res) self.assertTrue('oauth_consumer_secret' in res) self.assertEquals(res['oauth_consumer_key'], self.consumer.key) @@ -261,11 +253,11 @@ def test_to_unicode(self): self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, '\xae') self.failUnlessRaises(TypeError, oauth.to_unicode_optional_iterator, ['\xae']) - self.failUnlessEqual(oauth.to_unicode(':-)'), u':-)') - self.failUnlessEqual(oauth.to_unicode(u'\u00ae'), u'\u00ae') - self.failUnlessEqual(oauth.to_unicode('\xc2\xae'), u'\u00ae') - self.failUnlessEqual(oauth.to_unicode_optional_iterator([':-)']), [u':-)']) - self.failUnlessEqual(oauth.to_unicode_optional_iterator([u'\u00ae']), [u'\u00ae']) + self.failUnlessEqual(oauth.to_unicode(':-)'), ':-)') + self.failUnlessEqual(oauth.to_unicode(b'\u00ae'), b'\u00ae') + self.failUnlessEqual(oauth.to_unicode('\xc2\xae'), b'\u00ae') + self.failUnlessEqual(oauth.to_unicode_optional_iterator([':-)']), [':-)']) + self.failUnlessEqual(oauth.to_unicode_optional_iterator([b'\u00ae']), [b'\u00ae']) class TestRequest(unittest.TestCase, ReallyEqualMixin): def test_setter(self): @@ -286,7 +278,7 @@ def test_deleter(self): self.fail("AttributeError should have been raised on empty url.") except AttributeError: pass - except Exception, e: + except Exception as e: self.fail(str(e)) def test_url(self): @@ -361,12 +353,12 @@ def test_get_nonoauth_parameters(self): } other_params = { - u'foo': u'baz', - u'bar': u'foo', - u'multi': [u'FOO',u'BAR'], - u'uni_utf8': u'\xae', - u'uni_unicode': u'\u00ae', - u'uni_unicode_2': u'åÅøØ', + 'foo': 'baz', + 'bar': 'foo', + 'multi': ['FOO','BAR'], + 'uni_utf8': b'\xae', + 'uni_unicode': b'\u00ae', + 'uni_unicode_2': 'åÅøØ', } params = oauth_params @@ -389,7 +381,7 @@ def test_to_header(self): } req = oauth.Request("GET", realm, params) - header, value = req.to_header(realm).items()[0] + header, value = list(req.to_header(realm).items())[0] parts = value.split('OAuth ') vars = parts[1].split(', ') @@ -398,7 +390,7 @@ def test_to_header(self): res = {} for v in vars: var, val = v.split('=') - res[var] = urllib.unquote(val.strip('"')) + res[var] = urllib.parse.unquote(val.strip('"')) self.assertEquals(realm, res['realm']) del res['realm'] @@ -412,7 +404,7 @@ def test_to_postdata_nonascii(self): realm = "http://sp.example.com/" params = { - 'nonasciithing': u'q\xbfu\xe9 ,aasp u?..a.s', + 'nonasciithing': b'q\xbfu\xe9 ,aasp u?..a.s', 'oauth_version': "1.0", 'oauth_nonce': "4572616e48616d6d65724c61686176", 'oauth_timestamp': "137131200", @@ -444,9 +436,9 @@ def test_to_postdata(self): flat = [('multi','FOO'),('multi','BAR')] del params['multi'] - flat.extend(params.items()) + flat.extend(list(params.items())) kf = lambda x: x[0] - self.assertEquals(sorted(flat, key=kf), sorted(parse_qsl(req.to_postdata()), key=kf)) + self.assertEquals(sorted(flat, key=kf), sorted(urllib.parse.parse_qsl(req.to_postdata()), key=kf)) def test_to_url(self): url = "http://sp.example.com/" @@ -462,14 +454,14 @@ def test_to_url(self): } req = oauth.Request("GET", url, params) - exp = urlparse.urlparse("%s?%s" % (url, urllib.urlencode(params))) + exp = urlparse.urlparse("%s?%s" % (url, urllib.parse.urlencode(params))) res = urlparse.urlparse(req.to_url()) self.assertEquals(exp.scheme, res.scheme) self.assertEquals(exp.netloc, res.netloc) self.assertEquals(exp.path, res.path) - a = parse_qs(exp.query) - b = parse_qs(res.query) + a = urllib.parse.parse_qs(exp.query) + b = urllib.parse.parse_qs(res.query) self.assertEquals(a, b) def test_to_url_with_query(self): @@ -487,14 +479,14 @@ def test_to_url_with_query(self): req = oauth.Request("GET", url, params) # Note: the url above already has query parameters, so append new ones with & - exp = urlparse.urlparse("%s&%s" % (url, urllib.urlencode(params))) + exp = urlparse.urlparse("%s&%s" % (url, urllib.parse.urlencode(params))) res = urlparse.urlparse(req.to_url()) self.assertEquals(exp.scheme, res.scheme) self.assertEquals(exp.netloc, res.netloc) self.assertEquals(exp.path, res.path) - a = parse_qs(exp.query) - b = parse_qs(res.query) + a = urllib.parse.parse_qs(exp.query) + b = urllib.parse.parse_qs(res.query) self.assertTrue('alt' in b) self.assertTrue('max-contacts' in b) self.assertEquals(b['alt'], ['json']) @@ -504,27 +496,27 @@ def test_to_url_with_query(self): def test_signature_base_string_nonascii_nonutf8(self): consumer = oauth.Consumer('consumer_token', 'consumer_secret') - url = u'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\u2766,+CA' + url = b'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\u2766,+CA' req = oauth.Request("GET", url) - self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json') + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc\xe2\x9d\xa6,+CA' req = oauth.Request("GET", url) - self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json') + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') url = 'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA' req = oauth.Request("GET", url) - self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json') + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') - url = u'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA' + url = b'http://api.simplegeo.com:80/1.0/places/address.json?q=monkeys&category=animal&address=41+Decatur+St,+San+Francisc%E2%9D%A6,+CA' req = oauth.Request("GET", url) - self.failUnlessReallyEqual(req.normalized_url, u'http://api.simplegeo.com/1.0/places/address.json') + self.failUnlessReallyEqual(req.normalized_url, 'http://api.simplegeo.com/1.0/places/address.json') req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), consumer, None) self.failUnlessReallyEqual(req['oauth_signature'], 'WhufgeZKyYpKsI70GZaiDaYwl6g=') @@ -542,10 +534,10 @@ def test_signature_base_string_with_query(self): req = oauth.Request("GET", url, params) self.assertEquals(req.normalized_url, 'https://www.google.com/m8/feeds/contacts/default/full/') self.assertEquals(req.url, 'https://www.google.com/m8/feeds/contacts/default/full/?alt=json&max-contacts=10') - normalized_params = parse_qsl(req.get_normalized_parameters()) + normalized_params = urllib.parse.parse_qsl(req.get_normalized_parameters()) self.assertTrue(len(normalized_params), len(params) + 2) normalized_params = dict(normalized_params) - for key, value in params.iteritems(): + for key, value in params.items(): if key == 'oauth_signature': continue self.assertEquals(value, normalized_params[key]) @@ -616,10 +608,10 @@ def test_get_normalized_parameters(self): 'oauth_consumer_key': "0685bd9184jfhq22", 'oauth_signature_method': "HMAC-SHA1", 'oauth_token': "ad180jjd733klru7", - 'multi': ['FOO','BAR', u'\u00ae', '\xc2\xae'], + 'multi': ['FOO','BAR', b'\u00ae', '\xc2\xae'], 'multi_same': ['FOO','FOO'], 'uni_utf8_bytes': '\xc2\xae', - 'uni_unicode_object': u'\u00ae' + 'uni_unicode_object': b'\u00ae' } req = oauth.Request("GET", url, params) @@ -647,11 +639,11 @@ def test_get_normalized_parameters_ignores_auth_signature(self): res = req.get_normalized_parameters() - self.assertNotEquals(urllib.urlencode(sorted(params.items())), res) + self.assertNotEquals(urllib.parse.urlencode(sorted(params.items())), res) foo = params.copy() del foo["oauth_signature"] - self.assertEqual(urllib.urlencode(sorted(foo.items())), res) + self.assertEqual(urllib.parse.urlencode(sorted(foo.items())), res) def test_set_signature_method(self): consumer = oauth.Consumer('key', 'secret') @@ -679,7 +671,7 @@ def test_get_normalized_string_escapes_spaces_properly(self): req = oauth.Request("GET", url, params) res = req.get_normalized_parameters() - expected = urllib.urlencode(sorted(params.items())).replace('+', '%20') + expected = urllib.parse.urlencode(sorted(params.items())).replace('+', '%20') self.assertEqual(expected, res) @mock.patch('oauth2.Request.make_timestamp') @@ -704,7 +696,7 @@ def test_request_nonutf8_bytes(self, mock_make_nonce, mock_make_timestamp): self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params) # And if they pass an unicode, then we'll use it. - url = u'http://sp.example.com/\u2019' + url = b'http://sp.example.com/\u2019' req = oauth.Request(method="GET", url=url, parameters=params) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) self.failUnlessReallyEqual(req['oauth_signature'], 'cMzvCkhvLL57+sTIxLITTHfkqZk=') @@ -725,7 +717,7 @@ def test_request_nonutf8_bytes(self, mock_make_nonce, mock_make_timestamp): self.assertRaises(TypeError, oauth.Request, method="GET", url=url, parameters=params) # And if they pass a unicode, then we'll use it. - params['non_oauth_thing'] = u'\u2019' + params['non_oauth_thing'] = b'\u2019' req = oauth.Request(method="GET", url=url, parameters=params) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) self.failUnlessReallyEqual(req['oauth_signature'], '0GU50m0v60CVDB5JnoBXnvvvKx4=') @@ -758,7 +750,7 @@ def test_request_hash_of_body(self): 'oauth_consumer_key': con.key } - url = u"http://www.example.com/resource" + url = "http://www.example.com/resource" req = oauth.Request(method="PUT", url=url, parameters=params, body="Hello World!", is_form_encoded=False) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, None) self.failUnlessReallyEqual(req['oauth_body_hash'], 'Lve95gjOVATpfV8EL5X4nxwjKHE=') @@ -828,7 +820,7 @@ def test_sign_request(self): req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) self.assertEquals(req['oauth_signature'], 'loFvp5xC7YbOgd9exIO6TxB7H4s=') - url = u'http://sp.example.com/\u2019' # Python unicode object + url = b'http://sp.example.com/\u2019' # Python unicode object req = oauth.Request(method="GET", url=url, parameters=params) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) self.assertEquals(req['oauth_signature'], 'loFvp5xC7YbOgd9exIO6TxB7H4s=') @@ -839,7 +831,7 @@ def test_sign_request(self): req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) self.assertEquals(req['oauth_signature'], 'IBw5mfvoCsDjgpcsVKbyvsDqQaU=') - url = u'http://sp.example.com/?q=\u2019' # Python unicode object + url = b'http://sp.example.com/?q=\u2019' # Python unicode object req = oauth.Request(method="GET", url=url, parameters=params) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), con, tok) self.assertEquals(req['oauth_signature'], 'IBw5mfvoCsDjgpcsVKbyvsDqQaU=') @@ -876,12 +868,12 @@ def test_from_request(self): url, bad_headers) # Test getting from query string - qs = urllib.urlencode(params) + qs = urllib.parse.urlencode(params) req = oauth.Request.from_request("GET", url, query_string=qs) - exp = parse_qs(qs, keep_blank_values=False) - for k, v in exp.iteritems(): - exp[k] = urllib.unquote(v[0]) + exp = urllib.parse.parse_qs(qs, keep_blank_values=False) + for k, v in exp.items(): + exp[k] = urllib.parse.unquote(v[0]) self.assertEquals(exp, req.copy()) @@ -1165,7 +1157,7 @@ def create_simple_multipart_data(self, data): boundary = '---Boundary-%d' % random.randint(1,1000) crlf = '\r\n' items = [] - for key, value in data.iteritems(): + for key, value in data.items(): items += [ '--'+boundary, 'Content-Disposition: form-data; name="%s"'%str(key), @@ -1207,7 +1199,7 @@ def test_access_token_post(self): self.assertEquals(int(resp['status']), 200) - res = dict(parse_qsl(content)) + res = dict(urllib.parse.parse_qsl(content)) self.assertTrue('oauth_token' in res) self.assertTrue('oauth_token_secret' in res) @@ -1215,7 +1207,7 @@ def _two_legged(self, method): client = oauth.Client(self.consumer, None) return client.request(self._uri('two_legged'), method, - body=urllib.urlencode(self.body)) + body=urllib.parse.urlencode(self.body)) def test_two_legged_post(self): """A test of a two-legged OAuth POST request.""" @@ -1275,8 +1267,8 @@ def mockrequest(cl, ur, **kw): req = oauth.Request.from_consumer_and_token(self.consumer, None, http_method='GET', http_url=uri, parameters={}) req.sign_request(oauth.SignatureMethod_HMAC_SHA1(), self.consumer, None) - expected = parse_qsl(urlparse.urlparse(req.to_url()).query) - actual = parse_qsl(urlparse.urlparse(ur).query) + expected = urllib.parse.parse_qsl(urlparse.urlparse(req.to_url()).query) + actual = urllib.parse.parse_qsl(urlparse.urlparse(ur).query) self.failUnlessEqual(len(expected), len(actual)) actual = dict(actual) for key, value in expected: