From 02c375640cd50c3945bc5f4fe0654dff3fe100b6 Mon Sep 17 00:00:00 2001 From: jrconlin Date: Mon, 28 Nov 2011 15:04:43 -0800 Subject: [PATCH] Cleaned up python --- python/OAuthSimple/OAuthSimple.py | 474 +++++++++++---------- python/OAuthSimple/OAuthSimpleException.py | 17 +- python/OAuthSimple/tests.py | 137 +++--- 3 files changed, 325 insertions(+), 303 deletions(-) diff --git a/python/OAuthSimple/OAuthSimple.py b/python/OAuthSimple/OAuthSimple.py index ed9332a..4d78046 100644 --- a/python/OAuthSimple/OAuthSimple.py +++ b/python/OAuthSimple/OAuthSimple.py @@ -1,232 +1,242 @@ -#!/usr/bin/env python -tt - -import base64; -import hmac; -import hashlib; -import random; -import re; -import time; -import urlparse; -import urllib2; -import pdb - -class OAuthSimpleException(Exception): - def __init__(self,value): - self.msg = value - - def __str__(self): - return repr(self.msg) - -class OAuthSimple: - from OAuthSimple import OAuthSimpleException - _secrets={} - _parameters={} - _default_signature_method="HMAC-SHA1" - _action="GET" - _nonce_characters='0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz' - sbs="" - _path="" - - def __init__(self, apiKey = None, sharedSecret = None): - if apiKey: - self._secrets['oauth_consumer_key']=apiKey - if sharedSecret: - self._secrets['shared_secret']=sharedSecret - return ; - - def reset (self): - self._parameters={} - self.path={} - self.sbs={} - return self - - def setParameters(self,parameters={}): - if parameters: - if type(parameters)==type(""): - parameters = self._parseParameterString(parameters) - if not self._parameters: - self._parameters = parameters - elif parameters: - self._parameters = self._arrayMerge(self._parameters,parameters) - if not self._parameters.get('oauth_nonce'): - self._getNonce() - if not self._parameters.get('oauth_timestamp'): - self._getTimeStamp(); - if not self._parameters.get('oauth_consumer_key'): - self._getApiKey(); - if not self._parameters.get('oauth_token'): - self._getAccessToken(); - if not self._parameters.get('oauth_signature_method'): - self.setSignatureMethod() - if not self._parameters.get('oauth_version'): - self._parameters['oauth_version']='1.0'; - - return self; - - def setPath (self,path): - if not path: - raise OAuthSimpleException('No path specified'); - self._path = path; - return self; - - def setAction(self, action): - action = action.upper(); - if re.match('[^A-Z]',action): - raise OAuthSimpleException('Invalid action specified for OAuthSimple.setAction'); - self._action = action; - return self; - - def signatures(self,signatures): - if signatures and (type(signatures) != type({})): - raise OAuthSimpleException('Must pass Dict to OAuthSimple.signatures') - - if signatures: - self._secrets = self._arrayMerge(signatures,self._secrets) - - # swap keys - swap = (('api_key','oauth_consumer_key'), - ('consumer_key','oauth_consumer_key'), - ('access_token','oauth_token'), - ('access_token_secret','oauth_secret')) - for swapable in swap: - if (self._secrets.get(swapable[0])): - self._secrets[swapable[1]]=self._secrets[swapable[0]] - del self._secrets[swapable[0]] - if not self._secrets.get('oauth_consumer_key'): - raise OAuthSimpleException('Missing required oauth_consumer_key') - if not self._secrets.get('shared_secret'): - raise OAuthSimpleException('Missing required shared_secret') - if self._secrets.get('oauth_token') and not self._secrets.get('oauth_secret'): - raise OAuthSimpleException('Missing oauth_secret for supplied oauth_token') - - return self; - - def setSignatureMethod(self,method=None): - if not method: - method=self._default_signature_method - method = method.upper() - if method not in ('PLAINTEXT','HMAC-SHA1'): - raise OAuthSimpleException('Invalid Signature method specified ') - self._parameters['oauth_signature_method']=method; - return self; - - def sign(self,args={}): - if args.get('action'): - self.setAction(args['action']); - if args.get('path'): - self.setPath(args['path']); - if args.get('method'): - self.setSignatureMethod(args['method']); - if args.get('signatures'): - self.signatures(args.get('signatures')); - self.setParameters(args.get('parameters')); - normParamString = self._normalizeParameters(); - self._parameters['oauth_signature'] = self._generateSignature(normParamString) - return {'parameters':self._parameters, - 'signature':self._oauthEscape(self._parameters['oauth_signature']), - 'signed_url':self._path+'?'+self._normalizeParameters(), - 'header':self.getHeaderString(), - 'sbs':self.sbs - }; - - def getHeaderString(self,args={}): - result = []; - - if not self._parameters.get('oauth_signature'): - self.sign(args); - - for pName in self._parameters: - if not pName.startswith('oauth_'): - continue; - pValue = self._parameters.get(pName); - if type(pValue) == type([]): - for val in pValue: - result.append(pName + '="' + self._oauthEscape(val) + '"') - else: - result.append(pName + '="' + self._oauthEscape(str(pValue)) + '"'); - return 'OAuth '+', '.join(result); - - def _arrayMerge(self,target,source): - for skey in source.keys(): - target[skey]=source.get(skey); - return target; - - def _parseParameterString(self,paramString): - return urlparse.parse_qs(paramString,True); - - def _oauthEscape (self,string): - if not string: - return ''; - - string = urllib2.quote(string); - return string.replace('/','%2F').replace('+','%20').replace('!','%21').replace('*','%2A').replace('\\','%27').replace('(','%28').replace(')','%29') - - def _getApiKey(self): - if not self._parameters.has_key('oauth_consumer_key'): - if not self._secrets.get('oauth_consumer_key'): - raise OAuthSimpleException('No oauth_consumer_key set') - self._parameters['oauth_consumer_key']=self._secrets.get('oauth_consumer_key') - return self._parameters.get('oauth_consumer_key'); - - def _getAccessToken(self): - if not self._secrets.has_key('oauth_secret'): - return '' - if not self._parameters.has_key('oauth_access_token'): - if not self._secrets.get('oauth_access_token'): - raise OAuthSimpleException('No oauth_access_token set') - self._parameters['oauth_access_token'] = self._secrets.get('oauth_access_token') - return self._parameters.get('oauth_access_token') - - def _getNonce(self,length=5): - result = []; - cLength = len(self._nonce_characters); - - for i in range(0,length): - rnd = random.randint(0,cLength-1) - result.append(self._nonce_characters[rnd]) - self._parameters['oauth_nonce']=''.join(result); - return self._parameters['oauth_nonce'] - - def _getTimeStamp(self): - """ return the top UTC time integer as a string """ - self._parameters['oauth_timestamp']=int(time.time()); - return self._parameters['oauth_timestamp'] - - def _normalizeParameters(self): - elements=[] - if not self._parameters: - return ''; - pKeys = self._parameters.keys(); - pKeys.sort(); - for paramName in pKeys: - if paramName.find('_secret') > 0: - continue; - paramValue = self._parameters.get(paramName) - if type(paramValue) == type([]): - paramValue.sort(); - for value in paramValue: - elements.append(self._oauthEscape(paramName)+'='+self._oauthEscape(value)) - next - else: - elements.append(self._oauthEscape(paramName)+'='+self._oauthEscape(str(paramValue))) - return '&'.join(elements) - - def _generateSignature(self,normParamString): - secretKey=''; - - if self._secrets.get('shared_secret'): - secretKey = self._oauthEscape(self._secrets['shared_secret']); - - secretKey += '&'; - - if self._secrets.get('oauth_secret'): - secretKey += self._oauthEscape(self._secrets['oauth_secret']); - - if (self._parameters['oauth_signature_method'] == 'PLAINTEXT'): - return secretKey; - elif (self._parameters['oauth_signature_method'] == 'HMAC-SHA1'): - self.sbs = '&'.join([self._oauthEscape(self._action), - self._oauthEscape(self._path), - self._oauthEscape(normParamString)] - ) - return base64.b64encode(hmac.new(secretKey,self.sbs,hashlib.sha1).digest()); +#!/usr/bin/env python -tt + +import base64 +import hashlib +import hmac +import random +import re +import time +import urllib2 +import urlparse +import string + +import OAuthSimpleException + + +class OAuthSimple: + _secrets = {} + _parameters = {} + _default_signature_method = "HMAC-SHA1" + _action = "GET" + _nonce_characters = string.digits + string.letters + sbs = "" + _path = "" + + def __init__(self, apiKey=None, sharedSecret=None): + if apiKey: + self._secrets['oauth_consumer_key'] = apiKey + if sharedSecret: + self._secrets['shared_secret'] = sharedSecret + return + + def reset(self): + self._parameters = {} + self.path = {} + self.sbs = {} + return self + + def setParameters(self, parameters={}): + if parameters: + if type(parameters) == type(""): + parameters = self._parseParameterString(parameters) + if not self._parameters: + self._parameters = parameters + elif parameters: + self._parameters = self._arrayMerge(self._parameters, + parameters) + if not self._parameters.get('oauth_nonce'): + self._getNonce() + if not self._parameters.get('oauth_timestamp'): + self._getTimeStamp() + if not self._parameters.get('oauth_consumer_key'): + self._getApiKey() + if not self._parameters.get('oauth_token'): + self._getAccessToken() + if not self._parameters.get('oauth_signature_method'): + self.setSignatureMethod() + if not self._parameters.get('oauth_version'): + self._parameters['oauth_version'] = '1.0' + return self + + def setPath(self, path): + if not path: + raise OAuthSimpleException('No path specified') + self._path = path + return self + + def setAction(self, action): + action = action.upper() + if re.match('[^A-Z]', action): + raise OAuthSimpleException( + 'Invalid action specified for OAuthSimple.setAction') + self._action = action + return self + + def signatures(self, signatures): + if signatures and (type(signatures) != type({})): + raise OAuthSimpleException( + 'Must pass Dict to OAuthSimple.signatures') + + if signatures: + self._secrets = self._arrayMerge(signatures, self._secrets) + + # swap keys + swap = (('api_key', 'oauth_consumer_key'), + ('consumer_key', 'oauth_consumer_key'), + ('access_token', 'oauth_token'), + ('access_token_secret', 'oauth_secret')) + for swapable in swap: + if (self._secrets.get(swapable[0])): + self._secrets[swapable[1]] = self._secrets[swapable[0]] + del self._secrets[swapable[0]] + if not self._secrets.get('oauth_consumer_key'): + raise OAuthSimpleException('Missing required oauth_consumer_key') + if not self._secrets.get('shared_secret'): + raise OAuthSimpleException('Missing required shared_secret') + if (self._secrets.get('oauth_token') and + not self._secrets.get('oauth_secret')): + raise OAuthSimpleException( + 'Missing oauth_secret for supplied oauth_token') + return self + + def setSignatureMethod(self, method=None): + if not method: + method = self._default_signature_method + method = method.upper() + if method not in ('PLAINTEXT', 'HMAC-SHA1'): + raise OAuthSimpleException('Invalid Signature method specified ') + self._parameters['oauth_signature_method'] = method + return self + + def sign(self, args={}): + if args.get('action'): + self.setAction(args['action']) + if args.get('path'): + self.setPath(args['path']) + if args.get('method'): + self.setSignatureMethod(args['method']) + if args.get('signatures'): + self.signatures(args.get('signatures')) + self.setParameters(args.get('parameters')) + normParamString = self._normalizeParameters() + self._parameters['oauth_signature'] = \ + self._generateSignature(normParamString) + return {'parameters': self._parameters, + 'signature': self._oauthEscape( + self._parameters['oauth_signature']), + 'signed_url': '%s?%s' % (self._path, + self._normalizeParameters()), + 'header': self.getHeaderString(), + 'sbs': self.sbs} + + def getHeaderString(self, args={}): + result = [] + + if not self._parameters.get('oauth_signature'): + self.sign(args) + + for pName in self._parameters: + if not pName.startswith('oauth_'): + continue + pValue = self._parameters.get(pName) + if type(pValue) == type([]): + for val in pValue: + result.append('%s="%s"' % (pName, + self._oauthEscape(val))) + else: + result.append('%s="%s"' % (pName, + self._oauthEscape(str(pValue)))) + return 'OAuth %s' % (', '.join(result)) + + def _arrayMerge(self, target, source): + for skey in source.keys(): + target[skey] = source.get(skey) + return target + + def _parseParameterString(self, paramString): + return urlparse.parse_qs(paramString, True) + + def _oauthEscape(self, string): + if not string: + return '' + + string = urllib2.quote(string) + return string.replace('/', '%2F').replace('+', '%20')\ + .replace('!', '%21').replace('*', '%2A')\ + .replace('\\', '%27').replace('(', '%28').\ + replace(')', '%29') + + def _getApiKey(self): + if 'oauth_consumer_key' not in self._parameters: + if not self._secrets.get('oauth_consumer_key'): + raise OAuthSimpleException('No oauth_consumer_key set') + self._parameters['oauth_consumer_key'] = \ + self._secrets.get('oauth_consumer_key') + return self._parameters.get('oauth_consumer_key') + + def _getAccessToken(self): + if 'oauth_secret' not in self._secrets: + return '' + if 'oauth_access_token' not in self._parameters: + if not self._secrets.get('oauth_access_token'): + raise OAuthSimpleException('No oauth_access_token set') + self._parameters['oauth_access_token'] = \ + self._secrets.get('oauth_access_token') + return self._parameters.get('oauth_access_token') + + def _getNonce(self, length=5): + result = [] + cLength = len(self._nonce_characters) + + for i in range(0, length): + rnd = random.randint(0, cLength - 1) + result.append(self._nonce_characters[rnd]) + self._parameters['oauth_nonce'] = ''.join(result) + return self._parameters['oauth_nonce'] + + def _getTimeStamp(self): + """ return the top UTC time integer as a string """ + self._parameters['oauth_timestamp'] = int(time.time()) + return self._parameters['oauth_timestamp'] + + def _normalizeParameters(self): + elements = [] + if not self._parameters: + return '' + pKeys = self._parameters.keys() + pKeys.sort() + for paramName in pKeys: + if paramName.find('_secret') > 0: + continue + paramValue = self._parameters.get(paramName) + if type(paramValue) == type([]): + paramValue.sort() + for value in paramValue: + elements.append('%s=%s' % (self._oauthEscape(paramName), + self._oauthEscape(value))) + next + else: + elements.append('%s=%s' % (self._oauthEscape(paramName), + self._oauthEscape(str(paramValue)))) + return '&'.join(elements) + + def _generateSignature(self, normParamString): + secretKey = '' + + if self._secrets.get('shared_secret'): + secretKey = self._oauthEscape(self._secrets['shared_secret']) + + secretKey += '&' + + if self._secrets.get('oauth_secret'): + secretKey += self._oauthEscape(self._secrets['oauth_secret']) + + if (self._parameters['oauth_signature_method'] == 'PLAINTEXT'): + return secretKey + elif (self._parameters['oauth_signature_method'] == 'HMAC-SHA1'): + self.sbs = '&'.join([self._oauthEscape(self._action), + self._oauthEscape(self._path), + self._oauthEscape(normParamString)]) + return base64.b64encode(hmac.new(secretKey, + self.sbs, + hashlib.sha1).digest()) diff --git a/python/OAuthSimple/OAuthSimpleException.py b/python/OAuthSimple/OAuthSimpleException.py index 27aa8fd..d8689bd 100644 --- a/python/OAuthSimple/OAuthSimpleException.py +++ b/python/OAuthSimple/OAuthSimpleException.py @@ -1,8 +1,9 @@ -#! /usr/bin/env python -class OAuthSimpleException(Exception): - def __init__(self,value): - self.msg = value - - def __str__(self): - return repr(self.msg) - +#! /usr/bin/env python + + +class OAuthSimpleException(Exception): + def __init__(self, value): + self.msg = value + + def __str__(self): + return repr(self.msg) diff --git a/python/OAuthSimple/tests.py b/python/OAuthSimple/tests.py index 2e33225..959e2c5 100644 --- a/python/OAuthSimple/tests.py +++ b/python/OAuthSimple/tests.py @@ -1,63 +1,74 @@ -#!/usr/bin/env python -tt -import sys; -import unittest; -import urllib2; -import json; - -from OAuthSimple import OAuthSimple - -class TestOAuth(unittest.TestCase): - - signatures = {'consumer_key':'v9s778n692e9qvd83wfj9t8c','shared_secret':'54XqbMADta'}; - test_url = 'http://api.netflix.com/catalog/titles'; - parameters = 'term=mac%20and+me&expand=formats,synopsis&max_results=1&v=2.0&output=json'; - - def setUp(self): - self.o1 = OAuthSimple() - - def testUtils(self): - self.assertEqual(self.o1._oauthEscape('a b+c!d*\\e(f)g+h'),'a%20b%2Bc%21d%2A%5Ce%28f%29g%2Bh'); - self.assertEqual(self.o1._arrayMerge({'a':0,'b':2,'c':3},{'a':1,'d':4}),{'a':1,'b':2,'c':3,'d':4}); - self.o1.signatures({'api_key':'123','shared_secret':'456'}) - self.o1.setParameters({'a':1,'b':2}); - self.assertEqual(self.o1._secrets['oauth_consumer_key'],'123'); - self.assertEqual(self.o1._secrets['shared_secret'],'456') - self.assertEqual(len(self.o1._getNonce(10)),10); - self.assertNotEqual(self.o1._getNonce(),self.o1._getNonce()) - self.o1.reset(); - self.assertTrue(len(self.o1._parameters)==0) - - def testSimple(self): - self.o1.reset(); - signed = self.o1.sign({'signatures':self.signatures,'parameters':self.parameters,'path':self.test_url}); - # import pdb; pdb.set_trace(); - self.assertTrue(len(signed.get('signature')) > 0) - self.assertTrue(len(signed.get('signed_url'))>0) - # send the link. - try: - rsp=urllib2.urlopen(signed.get('signed_url')) - except urllib2.HTTPError as ex: - if (ex.headers.get('x-mashery-error-code') == 'ERR_401_TIMESTAMP_IS_INVALID'): - print "Your clock is off. Check and retry." - raise - print ex - raise - content = json.loads(rsp.read()) - self.assertTrue(content.get('catalog')[0].get('id') == u'http://api.netflix.com/catalog/titles/movies/60035973') - - def testComplex(self): - self.o1.reset(); - self.o1.signatures(self.signatures); - self.o1.setPath(u'http://api.netflix.com/catalog/titles/movies/60035973'); - self.o1.setParameters({'expand':'all', - 'v':'2.0', - 'output':'json'}); - signed = self.o1.sign(); - content = json.loads(urllib2.urlopen(signed.get('signed_url')).read()) - # This is probably testing Netflix's API more than mine. - self.assertEqual(content['catalog_title']['directors'][0]['name'],u'Stewart Raffill') - - -if (__name__ == '__main__'): - unittest.main() - +#!/usr/bin/env python -tt +import unittest; +import urllib2; +import json; + +from OAuthSimple import OAuthSimple + +class TestOAuth(unittest.TestCase): + + signatures = {'consumer_key': 'v9s778n692e9qvd83wfj9t8c', + 'shared_secret':'54XqbMADta'}; + test_url = 'http://api.netflix.com/catalog/titles'; + parameters = \ + 'term=mac%20and+me&expand=formats,' +\ + 'synopsis&max_results=1&v=2.0&output=json'; + + def setUp(self): + self.o1 = OAuthSimple() + + def testUtils(self): + self.assertEqual(self.o1._oauthEscape('a b+c!d*\\e(f)g+h'), + 'a%20b%2Bc%21d%2A%5Ce%28f%29g%2Bh'); + self.assertEqual(self.o1._arrayMerge({'a': 0, 'b': 2, 'c': 3}, + {'a': 1, 'd': 4}), + {'a': 1, 'b': 2, 'c': 3, 'd': 4}); + self.o1.signatures({'api_key': '123', 'shared_secret': '456'}) + self.o1.setParameters({'a': 1, 'b': 2}); + self.assertEqual(self.o1._secrets['oauth_consumer_key'], '123'); + self.assertEqual(self.o1._secrets['shared_secret'], '456') + self.assertEqual(len(self.o1._getNonce(10)), 10); + self.assertNotEqual(self.o1._getNonce(),self.o1._getNonce()) + self.o1.reset(); + self.assertTrue(len(self.o1._parameters) == 0) + + def testSimple(self): + self.o1.reset(); + signed = self.o1.sign({'signatures': self.signatures, + 'parameters': self.parameters, + 'path': self.test_url}); + # import pdb; pdb.set_trace(); + self.assertTrue(len(signed.get('signature')) > 0) + self.assertTrue(len(signed.get('signed_url')) > 0) + # send the link. + try: + rsp=urllib2.urlopen(signed.get('signed_url')) + except urllib2.HTTPError as ex: + if (ex.headers.get('x-mashery-error-code') == + 'ERR_401_TIMESTAMP_IS_INVALID'): + print "Your clock is off. Check and retry." + raise + print ex + raise + content = json.loads(rsp.read()) + self.assertTrue(content.get('catalog')[0].get('id') == + u'http://api.netflix.com/catalog/titles/movies/60035973') + + def testComplex(self): + self.o1.reset(); + self.o1.signatures(self.signatures); + self.o1.setPath( + 'http://api.netflix.com/catalog/titles/movies/60035973'); + self.o1.setParameters({'expand': 'all', + 'v': '2.0', + 'output': 'json'}); + signed = self.o1.sign(); + content = json.loads(urllib2.urlopen(signed.get('signed_url')).read()) + # This is probably testing Netflix's API more than mine. + self.assertEqual(content['catalog_title']['directors'][0]['name'], + u'Stewart Raffill') + + +if (__name__ == '__main__'): + unittest.main() +