| #!/usr/bin/env python | |
| # | |
| # Copyright (C) 2007, 2008 Google Inc. | |
| # | |
| # Licensed under the Apache License, Version 2.0 (the "License"); | |
| # you may not use this file except in compliance with the License. | |
| # You may obtain a copy of the License at | |
| # | |
| # http://www.apache.org/licenses/LICENSE-2.0 | |
| # | |
| # Unless required by applicable law or agreed to in writing, software | |
| # distributed under the License is distributed on an "AS IS" BASIS, | |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | |
| # See the License for the specific language governing permissions and | |
| # limitations under the License. | |
| __author__ = 'api.jscudder (Jeff Scudder)' | |
| import re | |
| import unittest | |
| import urllib | |
| import gdata.auth | |
| CONSUMER_KEY = 'www.yourwebapp.com' | |
| CONSUMER_SECRET = 'qB1P2kCFDpRjF+/Iww4' | |
| RSA_KEY = """-----BEGIN RSA PRIVATE KEY----- | |
| MIICXAIBAAKBgQDVbOaFW+KXecfFJn1PIzYHnNXFxhaQ36QM0K5uSb0Y8NeQUlD2 | |
| 6t8aKgnm6mcb4vaopHjjdIGWgAzM5Dt0oPIiDXo+jSQbvCIXRduuAt+0cFGb2d+L | |
| hALk4AwB8IVIkDJWwgo5Z2OLsP2r/wQlUYKm/tnvQaevK24jNYMLWVJl2QIDAQAB | |
| AoGAU93ERBlUVEPFjaJPUX67p4gotNvfWDSZiXOjZ7FQPnG9s3e1WyH2Y5irZXMs | |
| 61dnp+NhobfRiGtvHEB/YJgyLRk/CJDnMKslo95e7o65IE9VkcyY6Yvt7YTslsRX | |
| Eu7T0xLEA7ON46ypCwNLeWxpJ9SWisEKu2yZJnWauCXEsgUCQQD7b2ZuhGx3msoP | |
| YEnwvucp0UxneCvb68otfERZ1J6NfNP47QJw6OwD3r1sWCJ27QZmpvtQH1f8sCk9 | |
| t22anGG7AkEA2UzXdtQ8H1uLAN/XXX2qoLuvJK5jRswHS4GeOg4pnnDSiHg3Vbva | |
| AxmMIL93ufvIy/xdoENwDPfcI4CbYlrDewJAGWy7W+OSIEoLsqBW+bwkHetnIXNa | |
| ZAOkzxKoyrigS8hamupEe+xhqUaFuwXyfjobkpfCA+kXeZrKoM4CjEbR7wJAHMbf | |
| Vd4/ZAu0edYq6DenLAgO5rWtcge9A5PTx25utovMZcQ917273mM4unGAwoGEkvcF | |
| 0x57LUx5u73hVgIdFwJBAKWGuHRwGPgTWYvhpHM0qveH+8KdU9BUt/kV4ONxIVDB | |
| ftetEmJirqOGLECbImoLcUwQrgfMW4ZCxOioJMz/gY0= | |
| -----END RSA PRIVATE KEY----- | |
| """ | |
| class AuthModuleUtilitiesTest(unittest.TestCase): | |
| def testGenerateClientLoginRequestBody(self): | |
| body = gdata.auth.GenerateClientLoginRequestBody('jo@gmail.com', | |
| 'password', 'test service', 'gdata.auth test') | |
| expected_parameters = {'Email':r'jo%40gmail.com', 'Passwd':'password', | |
| 'service':'test+service', 'source':'gdata.auth+test', | |
| 'accountType':'HOSTED_OR_GOOGLE'} | |
| self.__matchBody(body, expected_parameters) | |
| body = gdata.auth.GenerateClientLoginRequestBody('jo@gmail.com', | |
| 'password', 'test service', 'gdata.auth test', account_type='A TEST', | |
| captcha_token='12345', captcha_response='test') | |
| expected_parameters['accountType'] = 'A+TEST' | |
| expected_parameters['logintoken'] = '12345' | |
| expected_parameters['logincaptcha'] = 'test' | |
| self.__matchBody(body, expected_parameters) | |
| def __matchBody(self, body, expected_name_value_pairs): | |
| parameters = body.split('&') | |
| for param in parameters: | |
| (name, value) = param.split('=') | |
| self.assert_(expected_name_value_pairs[name] == value) | |
| def testGenerateClientLoginAuthToken(self): | |
| http_body = ('SID=DQAAAGgA7Zg8CTN\r\n' | |
| 'LSID=DQAAAGsAlk8BBbG\r\n' | |
| 'Auth=DQAAAGgAdk3fA5N') | |
| self.assert_(gdata.auth.GenerateClientLoginAuthToken(http_body) == | |
| 'GoogleLogin auth=DQAAAGgAdk3fA5N') | |
| class GenerateClientLoginRequestBodyTest(unittest.TestCase): | |
| def testPostBodyShouldMatchShortExample(self): | |
| auth_body = gdata.auth.GenerateClientLoginRequestBody('johndoe@gmail.com', | |
| 'north23AZ', 'cl', 'Gulp-CalGulp-1.05') | |
| self.assert_(-1 < auth_body.find('Email=johndoe%40gmail.com')) | |
| self.assert_(-1 < auth_body.find('Passwd=north23AZ')) | |
| self.assert_(-1 < auth_body.find('service=cl')) | |
| self.assert_(-1 < auth_body.find('source=Gulp-CalGulp-1.05')) | |
| def testPostBodyShouldMatchLongExample(self): | |
| auth_body = gdata.auth.GenerateClientLoginRequestBody('johndoe@gmail.com', | |
| 'north23AZ', 'cl', 'Gulp-CalGulp-1.05', | |
| captcha_token='DQAAAGgA...dkI1', captcha_response='brinmar') | |
| self.assert_(-1 < auth_body.find('logintoken=DQAAAGgA...dkI1')) | |
| self.assert_(-1 < auth_body.find('logincaptcha=brinmar')) | |
| def testEquivalenceWithOldLogic(self): | |
| email = 'jo@gmail.com' | |
| password = 'password' | |
| account_type = 'HOSTED' | |
| service = 'test' | |
| source = 'auth test' | |
| old_request_body = urllib.urlencode({'Email': email, | |
| 'Passwd': password, | |
| 'accountType': account_type, | |
| 'service': service, | |
| 'source': source}) | |
| new_request_body = gdata.auth.GenerateClientLoginRequestBody(email, | |
| password, service, source, account_type=account_type) | |
| for parameter in old_request_body.split('&'): | |
| self.assert_(-1 < new_request_body.find(parameter)) | |
| class GenerateAuthSubUrlTest(unittest.TestCase): | |
| def testDefaultParameters(self): | |
| url = gdata.auth.GenerateAuthSubUrl('http://example.com/xyz?x=5', | |
| 'http://www.google.com/test/feeds') | |
| self.assert_(-1 < url.find( | |
| r'scope=http%3A%2F%2Fwww.google.com%2Ftest%2Ffeeds')) | |
| self.assert_(-1 < url.find( | |
| r'next=http%3A%2F%2Fexample.com%2Fxyz%3Fx%3D5')) | |
| self.assert_(-1 < url.find('secure=0')) | |
| self.assert_(-1 < url.find('session=1')) | |
| def testAllParameters(self): | |
| url = gdata.auth.GenerateAuthSubUrl('http://example.com/xyz?x=5', | |
| 'http://www.google.com/test/feeds', secure=True, session=False, | |
| request_url='https://example.com/auth') | |
| self.assert_(-1 < url.find( | |
| r'scope=http%3A%2F%2Fwww.google.com%2Ftest%2Ffeeds')) | |
| self.assert_(-1 < url.find( | |
| r'next=http%3A%2F%2Fexample.com%2Fxyz%3Fx%3D5')) | |
| self.assert_(-1 < url.find('secure=1')) | |
| self.assert_(-1 < url.find('session=0')) | |
| self.assert_(url.startswith('https://example.com/auth')) | |
| class GenerateOAuthRequestTokenUrlTest(unittest.TestCase): | |
| def testDefaultParameters(self): | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.RSA_SHA1, CONSUMER_KEY, | |
| rsa_key=RSA_KEY) | |
| scopes = [ | |
| 'http://abcd.example.com/feeds', | |
| 'http://www.example.com/abcd/feeds' | |
| ] | |
| url = gdata.auth.GenerateOAuthRequestTokenUrl( | |
| oauth_input_params, scopes=scopes) | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.google.com', url.host) | |
| self.assertEquals('/accounts/OAuthGetRequestToken', url.path) | |
| self.assertEquals('1.0', url.params['oauth_version']) | |
| self.assertEquals('RSA-SHA1', url.params['oauth_signature_method']) | |
| self.assert_(url.params['oauth_nonce']) | |
| self.assert_(url.params['oauth_timestamp']) | |
| actual_scopes = url.params['scope'].split(' ') | |
| self.assertEquals(2, len(actual_scopes)) | |
| for scope in actual_scopes: | |
| self.assert_(scope in scopes) | |
| self.assertEquals(CONSUMER_KEY, url.params['oauth_consumer_key']) | |
| self.assert_(url.params['oauth_signature']) | |
| def testAllParameters(self): | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.HMAC_SHA1, CONSUMER_KEY, | |
| consumer_secret=CONSUMER_SECRET) | |
| scopes = ['http://abcd.example.com/feeds'] | |
| url = gdata.auth.GenerateOAuthRequestTokenUrl( | |
| oauth_input_params, scopes=scopes, | |
| request_token_url='https://www.example.com/accounts/OAuthRequestToken', | |
| extra_parameters={'oauth_version': '2.0', 'my_param': 'my_value'}) | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.example.com', url.host) | |
| self.assertEquals('/accounts/OAuthRequestToken', url.path) | |
| self.assertEquals('2.0', url.params['oauth_version']) | |
| self.assertEquals('HMAC-SHA1', url.params['oauth_signature_method']) | |
| self.assert_(url.params['oauth_nonce']) | |
| self.assert_(url.params['oauth_timestamp']) | |
| actual_scopes = url.params['scope'].split(' ') | |
| self.assertEquals(1, len(actual_scopes)) | |
| for scope in actual_scopes: | |
| self.assert_(scope in scopes) | |
| self.assertEquals(CONSUMER_KEY, url.params['oauth_consumer_key']) | |
| self.assert_(url.params['oauth_signature']) | |
| self.assertEquals('my_value', url.params['my_param']) | |
| class GenerateOAuthAuthorizationUrlTest(unittest.TestCase): | |
| def testDefaultParameters(self): | |
| token_key = 'ABCDDSFFDSG' | |
| token_secret = 'SDFDSGSDADADSAF' | |
| request_token = gdata.auth.OAuthToken(key=token_key, secret=token_secret) | |
| url = gdata.auth.GenerateOAuthAuthorizationUrl(request_token) | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.google.com', url.host) | |
| self.assertEquals('/accounts/OAuthAuthorizeToken', url.path) | |
| self.assertEquals(token_key, url.params['oauth_token']) | |
| def testAllParameters(self): | |
| token_key = 'ABCDDSFFDSG' | |
| token_secret = 'SDFDSGSDADADSAF' | |
| scopes = [ | |
| 'http://abcd.example.com/feeds', | |
| 'http://www.example.com/abcd/feeds' | |
| ] | |
| request_token = gdata.auth.OAuthToken(key=token_key, secret=token_secret, | |
| scopes=scopes) | |
| url = gdata.auth.GenerateOAuthAuthorizationUrl( | |
| request_token, | |
| authorization_url='https://www.example.com/accounts/OAuthAuthToken', | |
| callback_url='http://www.yourwebapp.com/print', | |
| extra_params={'permission': '1'}, | |
| include_scopes_in_callback=True, scopes_param_prefix='token_scope') | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.example.com', url.host) | |
| self.assertEquals('/accounts/OAuthAuthToken', url.path) | |
| self.assertEquals(token_key, url.params['oauth_token']) | |
| expected_callback_url = ('http://www.yourwebapp.com/print?' | |
| 'token_scope=http%3A%2F%2Fabcd.example.com%2Ffeeds' | |
| '+http%3A%2F%2Fwww.example.com%2Fabcd%2Ffeeds') | |
| self.assertEquals(expected_callback_url, url.params['oauth_callback']) | |
| class GenerateOAuthAccessTokenUrlTest(unittest.TestCase): | |
| def testDefaultParameters(self): | |
| token_key = 'ABCDDSFFDSG' | |
| token_secret = 'SDFDSGSDADADSAF' | |
| authorized_request_token = gdata.auth.OAuthToken(key=token_key, | |
| secret=token_secret) | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.HMAC_SHA1, CONSUMER_KEY, | |
| consumer_secret=CONSUMER_SECRET) | |
| url = gdata.auth.GenerateOAuthAccessTokenUrl(authorized_request_token, | |
| oauth_input_params) | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.google.com', url.host) | |
| self.assertEquals('/accounts/OAuthGetAccessToken', url.path) | |
| self.assertEquals(token_key, url.params['oauth_token']) | |
| self.assertEquals('1.0', url.params['oauth_version']) | |
| self.assertEquals('HMAC-SHA1', url.params['oauth_signature_method']) | |
| self.assert_(url.params['oauth_nonce']) | |
| self.assert_(url.params['oauth_timestamp']) | |
| self.assertEquals(CONSUMER_KEY, url.params['oauth_consumer_key']) | |
| self.assert_(url.params['oauth_signature']) | |
| def testAllParameters(self): | |
| token_key = 'ABCDDSFFDSG' | |
| authorized_request_token = gdata.auth.OAuthToken(key=token_key) | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.RSA_SHA1, CONSUMER_KEY, | |
| rsa_key=RSA_KEY) | |
| url = gdata.auth.GenerateOAuthAccessTokenUrl( | |
| authorized_request_token, oauth_input_params, | |
| access_token_url='https://www.example.com/accounts/OAuthGetAccessToken', | |
| oauth_version= '2.0') | |
| self.assertEquals('https', url.protocol) | |
| self.assertEquals('www.example.com', url.host) | |
| self.assertEquals('/accounts/OAuthGetAccessToken', url.path) | |
| self.assertEquals(token_key, url.params['oauth_token']) | |
| self.assertEquals('2.0', url.params['oauth_version']) | |
| self.assertEquals('RSA-SHA1', url.params['oauth_signature_method']) | |
| self.assert_(url.params['oauth_nonce']) | |
| self.assert_(url.params['oauth_timestamp']) | |
| self.assertEquals(CONSUMER_KEY, url.params['oauth_consumer_key']) | |
| self.assert_(url.params['oauth_signature']) | |
| class ExtractAuthSubTokensTest(unittest.TestCase): | |
| def testGetTokenFromUrl(self): | |
| url = 'http://www.yourwebapp.com/showcalendar.html?token=CKF50YzIH' | |
| self.assert_(gdata.auth.AuthSubTokenFromUrl(url) == | |
| 'AuthSub token=CKF50YzIH') | |
| self.assert_(gdata.auth.TokenFromUrl(url) == 'CKF50YzIH') | |
| url = 'http://www.yourwebapp.com/showcalendar.html?token==tokenCKF50YzIH=' | |
| self.assert_(gdata.auth.AuthSubTokenFromUrl(url) == | |
| 'AuthSub token==tokenCKF50YzIH=') | |
| self.assert_(gdata.auth.TokenFromUrl(url) == '=tokenCKF50YzIH=') | |
| def testGetTokenFromHttpResponse(self): | |
| response_body = ('Token=DQAA...7DCTN\r\n' | |
| 'Expiration=20061004T123456Z') | |
| self.assert_(gdata.auth.AuthSubTokenFromHttpBody(response_body) == | |
| 'AuthSub token=DQAA...7DCTN') | |
| class CreateAuthSubTokenFlowTest(unittest.TestCase): | |
| def testGenerateRequest(self): | |
| request_url = gdata.auth.generate_auth_sub_url(next='http://example.com', | |
| scopes=['http://www.blogger.com/feeds/', | |
| 'http://www.google.com/base/feeds/']) | |
| self.assertEquals(request_url.protocol, 'https') | |
| self.assertEquals(request_url.host, 'www.google.com') | |
| self.assertEquals(request_url.params['scope'], | |
| 'http://www.blogger.com/feeds/ http://www.google.com/base/feeds/') | |
| self.assertEquals(request_url.params['hd'], 'default') | |
| self.assert_(request_url.params['next'].find('auth_sub_scopes') > -1) | |
| self.assert_(request_url.params['next'].startswith('http://example.com')) | |
| # Use a more complicated 'next' URL. | |
| request_url = gdata.auth.generate_auth_sub_url( | |
| next='http://example.com/?token_scope=http://www.blogger.com/feeds/', | |
| scopes=['http://www.blogger.com/feeds/', | |
| 'http://www.google.com/base/feeds/']) | |
| self.assert_(request_url.params['next'].find('auth_sub_scopes') > -1) | |
| self.assert_(request_url.params['next'].find('token_scope') > -1) | |
| self.assert_(request_url.params['next'].startswith('http://example.com/')) | |
| def testParseNextUrl(self): | |
| url = ('http://example.com/?auth_sub_scopes=http%3A%2F%2Fwww.blogger.com' | |
| '%2Ffeeds%2F+http%3A%2F%2Fwww.google.com%2Fbase%2Ffeeds%2F&' | |
| 'token=my_nifty_token') | |
| token = gdata.auth.extract_auth_sub_token_from_url(url) | |
| self.assertEquals(token.get_token_string(), 'my_nifty_token') | |
| self.assert_(isinstance(token, gdata.auth.AuthSubToken)) | |
| self.assert_(token.valid_for_scope('http://www.blogger.com/feeds/')) | |
| self.assert_(token.valid_for_scope('http://www.google.com/base/feeds/')) | |
| self.assert_( | |
| not token.valid_for_scope('http://www.google.com/calendar/feeds/')) | |
| # Parse a more complicated response. | |
| url = ('http://example.com/?auth_sub_scopes=http%3A%2F%2Fwww.blogger.com' | |
| '%2Ffeeds%2F+http%3A%2F%2Fwww.google.com%2Fbase%2Ffeeds%2F&' | |
| 'token_scope=http%3A%2F%2Fwww.blogger.com%2Ffeeds%2F&' | |
| 'token=second_token') | |
| token = gdata.auth.extract_auth_sub_token_from_url(url) | |
| self.assertEquals(token.get_token_string(), 'second_token') | |
| self.assert_(isinstance(token, gdata.auth.AuthSubToken)) | |
| self.assert_(token.valid_for_scope('http://www.blogger.com/feeds/')) | |
| self.assert_(token.valid_for_scope('http://www.google.com/base/feeds/')) | |
| self.assert_( | |
| not token.valid_for_scope('http://www.google.com/calendar/feeds/')) | |
| def testParseNextWithNoToken(self): | |
| token = gdata.auth.extract_auth_sub_token_from_url('http://example.com/') | |
| self.assert_(token is None) | |
| token = gdata.auth.extract_auth_sub_token_from_url( | |
| 'http://example.com/?no_token=foo&other=1') | |
| self.assert_(token is None) | |
| class ExtractClientLoginTokenTest(unittest.TestCase): | |
| def testExtractFromBodyWithScopes(self): | |
| http_body_string = ('SID=DQAAAGgA7Zg8CTN\r\n' | |
| 'LSID=DQAAAGsAlk8BBbG\r\n' | |
| 'Auth=DQAAAGgAdk3fA5N') | |
| token = gdata.auth.extract_client_login_token(http_body_string, | |
| ['http://docs.google.com/feeds/']) | |
| self.assertEquals(token.get_token_string(), 'DQAAAGgAdk3fA5N') | |
| self.assert_(isinstance(token, gdata.auth.ClientLoginToken)) | |
| self.assert_(token.valid_for_scope('http://docs.google.com/feeds/')) | |
| self.assert_(not token.valid_for_scope('http://www.blogger.com/feeds')) | |
| class ExtractOAuthTokensTest(unittest.TestCase): | |
| def testOAuthTokenFromUrl(self): | |
| scope_1 = 'http://docs.google.com/feeds/' | |
| scope_2 = 'http://www.blogger.com/feeds/' | |
| # Case 1: token and scopes both are present. | |
| url = ('http://dummy.com/?oauth_token_scope=http%3A%2F%2Fwww.blogger.com' | |
| '%2Ffeeds%2F+http%3A%2F%2Fdocs.google.com%2Ffeeds%2F&' | |
| 'oauth_token=CMns6t7MCxDz__8B') | |
| token = gdata.auth.OAuthTokenFromUrl(url) | |
| self.assertEquals('CMns6t7MCxDz__8B', token.key) | |
| self.assertEquals(2, len(token.scopes)) | |
| self.assert_(scope_1 in token.scopes) | |
| self.assert_(scope_2 in token.scopes) | |
| # Case 2: token and scopes both are present but scope_param_prefix | |
| # passed does not match the one present in the URL. | |
| url = ('http://dummy.com/?oauth_token_scope=http%3A%2F%2Fwww.blogger.com' | |
| '%2Ffeeds%2F+http%3A%2F%2Fdocs.google.com%2Ffeeds%2F&' | |
| 'oauth_token=CMns6t7MCxDz__8B') | |
| token = gdata.auth.OAuthTokenFromUrl(url, | |
| scopes_param_prefix='token_scope') | |
| self.assertEquals('CMns6t7MCxDz__8B', token.key) | |
| self.assert_(not token.scopes) | |
| # Case 3: None present. | |
| url = ('http://dummy.com/?no_oauth_token_scope=http%3A%2F%2Fwww.blogger.com' | |
| '%2Ffeeds%2F+http%3A%2F%2Fdocs.google.com%2Ffeeds%2F&' | |
| 'no_oauth_token=CMns6t7MCxDz__8B') | |
| token = gdata.auth.OAuthTokenFromUrl(url) | |
| self.assert_(token is None) | |
| def testOAuthTokenFromHttpBody(self): | |
| token_key = 'ABCD' | |
| token_secret = 'XYZ' | |
| # Case 1: token key and secret both present single time. | |
| http_body = 'oauth_token=%s&oauth_token_secret=%s' % (token_key, | |
| token_secret) | |
| token = gdata.auth.OAuthTokenFromHttpBody(http_body) | |
| self.assertEquals(token_key, token.key) | |
| self.assertEquals(token_secret, token.secret) | |
| class OAuthInputParametersTest(unittest.TestCase): | |
| def setUp(self): | |
| self.oauth_input_parameters_hmac = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.HMAC_SHA1, CONSUMER_KEY, | |
| consumer_secret=CONSUMER_SECRET) | |
| self.oauth_input_parameters_rsa = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.RSA_SHA1, CONSUMER_KEY, | |
| rsa_key=RSA_KEY) | |
| def testGetSignatureMethod(self): | |
| self.assertEquals( | |
| 'HMAC-SHA1', | |
| self.oauth_input_parameters_hmac.GetSignatureMethod().get_name()) | |
| rsa_signature_method = self.oauth_input_parameters_rsa.GetSignatureMethod() | |
| self.assertEquals('RSA-SHA1', rsa_signature_method.get_name()) | |
| self.assertEquals(RSA_KEY, rsa_signature_method._fetch_private_cert(None)) | |
| def testGetConsumer(self): | |
| self.assertEquals(CONSUMER_KEY, | |
| self.oauth_input_parameters_hmac.GetConsumer().key) | |
| self.assertEquals(CONSUMER_KEY, | |
| self.oauth_input_parameters_rsa.GetConsumer().key) | |
| self.assertEquals(CONSUMER_SECRET, | |
| self.oauth_input_parameters_hmac.GetConsumer().secret) | |
| self.assert_(self.oauth_input_parameters_rsa.GetConsumer().secret is None) | |
| class TokenClassesTest(unittest.TestCase): | |
| def testClientLoginToAndFromString(self): | |
| token = gdata.auth.ClientLoginToken() | |
| token.set_token_string('foo') | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| self.assertEquals(token.auth_header, '%s%s' % ( | |
| gdata.auth.PROGRAMMATIC_AUTH_LABEL, 'foo')) | |
| token.set_token_string(token.get_token_string()) | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| def testAuthSubToAndFromString(self): | |
| token = gdata.auth.AuthSubToken() | |
| token.set_token_string('foo') | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| self.assertEquals(token.auth_header, '%s%s' % ( | |
| gdata.auth.AUTHSUB_AUTH_LABEL, 'foo')) | |
| token.set_token_string(token.get_token_string()) | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| def testSecureAuthSubToAndFromString(self): | |
| # Case 1: no token. | |
| token = gdata.auth.SecureAuthSubToken(RSA_KEY) | |
| token.set_token_string('foo') | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| token.set_token_string(token.get_token_string()) | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| self.assertEquals(str(token), 'foo') | |
| # Case 2: token is a string | |
| token = gdata.auth.SecureAuthSubToken(RSA_KEY, token_string='foo') | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| token.set_token_string(token.get_token_string()) | |
| self.assertEquals(token.get_token_string(), 'foo') | |
| self.assertEquals(str(token), 'foo') | |
| def testOAuthToAndFromString(self): | |
| token_key = 'ABCD' | |
| token_secret = 'XYZ' | |
| # Case 1: token key and secret both present single time. | |
| token_string = 'oauth_token=%s&oauth_token_secret=%s' % (token_key, | |
| token_secret) | |
| token = gdata.auth.OAuthToken() | |
| token.set_token_string(token_string) | |
| self.assert_(-1 < token.get_token_string().find(token_string.split('&')[0])) | |
| self.assert_(-1 < token.get_token_string().find(token_string.split('&')[1])) | |
| self.assertEquals(token_key, token.key) | |
| self.assertEquals(token_secret, token.secret) | |
| # Case 2: token key and secret both present multiple times with unwanted | |
| # parameters. | |
| token_string = ('oauth_token=%s&oauth_token_secret=%s&' | |
| 'oauth_token=%s&ExtraParams=GarbageString' % (token_key, | |
| token_secret, | |
| 'LMNO')) | |
| token = gdata.auth.OAuthToken() | |
| token.set_token_string(token_string) | |
| self.assert_(-1 < token.get_token_string().find(token_string.split('&')[0])) | |
| self.assert_(-1 < token.get_token_string().find(token_string.split('&')[1])) | |
| self.assertEquals(token_key, token.key) | |
| self.assertEquals(token_secret, token.secret) | |
| # Case 3: Only token key present. | |
| token_string = 'oauth_token=%s' % (token_key,) | |
| token = gdata.auth.OAuthToken() | |
| token.set_token_string(token_string) | |
| self.assertEquals(token_string, token.get_token_string()) | |
| self.assertEquals(token_key, token.key) | |
| self.assert_(not token.secret) | |
| # Case 4: Only token key present. | |
| token_string = 'oauth_token_secret=%s' % (token_secret,) | |
| token = gdata.auth.OAuthToken() | |
| token.set_token_string(token_string) | |
| self.assertEquals(token_string, token.get_token_string()) | |
| self.assertEquals(token_secret, token.secret) | |
| self.assert_(not token.key) | |
| # Case 5: None present. | |
| token_string = '' | |
| token = gdata.auth.OAuthToken() | |
| token.set_token_string(token_string) | |
| self.assert_(token.get_token_string() is None) | |
| self.assert_(not token.key) | |
| self.assert_(not token.secret) | |
| def testSecureAuthSubGetAuthHeader(self): | |
| # Case 1: Presence of OAuth token (in case of 3-legged OAuth) | |
| url = 'http://dummy.com/?q=notebook&s=true' | |
| token = gdata.auth.SecureAuthSubToken(RSA_KEY, token_string='foo') | |
| auth_header = token.GetAuthHeader('GET', url) | |
| self.assert_('Authorization' in auth_header) | |
| header_value = auth_header['Authorization'] | |
| self.assert_(header_value.startswith(r'AuthSub token="foo"')) | |
| self.assert_(-1 < header_value.find(r'sigalg="rsa-sha1"')) | |
| self.assert_(-1 < header_value.find(r'data="')) | |
| self.assert_(-1 < header_value.find(r'sig="')) | |
| m = re.search(r'data="(.*?)"', header_value) | |
| self.assert_(m is not None) | |
| data = m.group(1) | |
| self.assert_(data.startswith('GET')) | |
| self.assert_(-1 < data.find(url)) | |
| def testOAuthGetAuthHeader(self): | |
| # Case 1: Presence of OAuth token (in case of 3-legged OAuth) | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.RSA_SHA1, CONSUMER_KEY, | |
| rsa_key=RSA_KEY) | |
| token = gdata.auth.OAuthToken(key='ABCDDSFFDSG', | |
| oauth_input_params=oauth_input_params) | |
| auth_header = token.GetAuthHeader('GET', | |
| 'http://dummy.com/?q=notebook&s=true', | |
| realm='http://dummy.com') | |
| self.assert_('Authorization' in auth_header) | |
| header_value = auth_header['Authorization'] | |
| self.assert_(-1 < header_value.find(r'OAuth realm="http://dummy.com"')) | |
| self.assert_(-1 < header_value.find(r'oauth_version="1.0"')) | |
| self.assert_(-1 < header_value.find(r'oauth_token="ABCDDSFFDSG"')) | |
| self.assert_(-1 < header_value.find(r'oauth_nonce="')) | |
| self.assert_(-1 < header_value.find(r'oauth_timestamp="')) | |
| self.assert_(-1 < header_value.find(r'oauth_signature="')) | |
| self.assert_(-1 < header_value.find( | |
| r'oauth_consumer_key="%s"' % CONSUMER_KEY)) | |
| self.assert_(-1 < header_value.find(r'oauth_signature_method="RSA-SHA1"')) | |
| # Case 2: Absence of OAuth token (in case of 2-legged OAuth) | |
| oauth_input_params = gdata.auth.OAuthInputParams( | |
| gdata.auth.OAuthSignatureMethod.HMAC_SHA1, CONSUMER_KEY, | |
| consumer_secret=CONSUMER_SECRET) | |
| token = gdata.auth.OAuthToken(oauth_input_params=oauth_input_params) | |
| auth_header = token.GetAuthHeader( | |
| 'GET', 'http://dummy.com/?xoauth_requestor_id=user@gmail.com&q=book') | |
| self.assert_('Authorization' in auth_header) | |
| header_value = auth_header['Authorization'] | |
| self.assert_(-1 < header_value.find(r'OAuth realm=""')) | |
| self.assert_(-1 < header_value.find(r'oauth_version="1.0"')) | |
| self.assertEquals(-1, header_value.find(r'oauth_token=')) | |
| self.assert_(-1 < header_value.find(r'oauth_nonce="')) | |
| self.assert_(-1 < header_value.find(r'oauth_timestamp="')) | |
| self.assert_(-1 < header_value.find(r'oauth_signature="')) | |
| self.assert_(-1 < header_value.find( | |
| r'oauth_consumer_key="%s"' % CONSUMER_KEY)) | |
| self.assert_(-1 < header_value.find(r'oauth_signature_method="HMAC-SHA1"')) | |
| if __name__ == '__main__': | |
| unittest.main() |