mirror of
https://github.com/jrconlin/oauthsimple.git
synced 2026-07-28 04:06:51 +00:00
Restructured to be more python friendly
This commit is contained in:
jr conlin
@@ -1,232 +0,0 @@
|
||||
#!/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());
|
||||
@@ -1,8 +0,0 @@
|
||||
#! /usr/bin/env python
|
||||
class OAuthSimpleException(Exception):
|
||||
def __init__(self,value):
|
||||
self.msg = value
|
||||
|
||||
def __str__(self):
|
||||
return repr(self.msg)
|
||||
|
||||
@@ -1,63 +0,0 @@
|
||||
#!/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()
|
||||
|
||||
Reference in New Issue
Block a user