| Index: tools/telemetry/third_party/gsutilz/third_party/oauth2client/tests/test_jwt.py
|
| diff --git a/tools/telemetry/third_party/gsutilz/third_party/oauth2client/tests/test_jwt.py b/tools/telemetry/third_party/gsutilz/third_party/oauth2client/tests/test_jwt.py
|
| deleted file mode 100755
|
| index f58a2947821ce986ed6313ddbbe1151e6d250053..0000000000000000000000000000000000000000
|
| --- a/tools/telemetry/third_party/gsutilz/third_party/oauth2client/tests/test_jwt.py
|
| +++ /dev/null
|
| @@ -1,320 +0,0 @@
|
| -#!/usr/bin/python2.4
|
| -#
|
| -# Copyright 2014 Google Inc. All rights reserved.
|
| -#
|
| -# 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.
|
| -
|
| -
|
| -"""Oauth2client tests
|
| -
|
| -Unit tests for oauth2client.
|
| -"""
|
| -
|
| -__author__ = 'jcgregorio@google.com (Joe Gregorio)'
|
| -
|
| -import os
|
| -import mock
|
| -import sys
|
| -import tempfile
|
| -import time
|
| -import unittest
|
| -
|
| -from .http_mock import HttpMockSequence
|
| -from oauth2client import client
|
| -from oauth2client.client import Credentials
|
| -from oauth2client.client import SignedJwtAssertionCredentials
|
| -from oauth2client.client import VerifyJwtTokenError
|
| -from oauth2client.client import verify_id_token
|
| -from oauth2client.client import HAS_OPENSSL
|
| -from oauth2client.client import HAS_CRYPTO
|
| -from oauth2client import crypt
|
| -from oauth2client.file import Storage
|
| -
|
| -
|
| -def datafile(filename):
|
| - f = open(os.path.join(os.path.dirname(__file__), 'data', filename), 'rb')
|
| - data = f.read()
|
| - f.close()
|
| - return data
|
| -
|
| -
|
| -class CryptTests(unittest.TestCase):
|
| -
|
| - def setUp(self):
|
| - self.format = 'p12'
|
| - self.signer = crypt.OpenSSLSigner
|
| - self.verifier = crypt.OpenSSLVerifier
|
| -
|
| - def test_sign_and_verify(self):
|
| - self._check_sign_and_verify('privatekey.%s' % self.format)
|
| -
|
| - def test_sign_and_verify_from_converted_pkcs12(self):
|
| - """Tests that following instructions to convert from PKCS12 to PEM works."""
|
| - if self.format == 'pem':
|
| - self._check_sign_and_verify('pem_from_pkcs12.pem')
|
| -
|
| - def _check_sign_and_verify(self, private_key_file):
|
| - private_key = datafile(private_key_file)
|
| - public_key = datafile('publickey.pem')
|
| -
|
| - signer = self.signer.from_string(private_key)
|
| - signature = signer.sign('foo')
|
| -
|
| - verifier = self.verifier.from_string(public_key, True)
|
| - self.assertTrue(verifier.verify(b'foo', signature))
|
| -
|
| - self.assertFalse(verifier.verify(b'bar', signature))
|
| - self.assertFalse(verifier.verify(b'foo', 'bad signagure'))
|
| -
|
| - def _check_jwt_failure(self, jwt, expected_error):
|
| - public_key = datafile('publickey.pem')
|
| - certs = {'foo': public_key}
|
| - audience = ('https://www.googleapis.com/auth/id?client_id='
|
| - 'external_public_key@testing.gserviceaccount.com')
|
| - try:
|
| - crypt.verify_signed_jwt_with_certs(jwt, certs, audience)
|
| - self.fail()
|
| - except crypt.AppIdentityError as e:
|
| - self.assertTrue(expected_error in str(e))
|
| -
|
| - def _create_signed_jwt(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| - signer = self.signer.from_string(private_key)
|
| - audience = 'some_audience_address@testing.gserviceaccount.com'
|
| - now = int(time.time())
|
| -
|
| - return crypt.make_signed_jwt(signer, {
|
| - 'aud': audience,
|
| - 'iat': now,
|
| - 'exp': now + 300,
|
| - 'user': 'billy bob',
|
| - 'metadata': {'meta': 'data'},
|
| - })
|
| -
|
| - def test_verify_id_token(self):
|
| - jwt = self._create_signed_jwt()
|
| - public_key = datafile('publickey.pem')
|
| - certs = {'foo': public_key}
|
| - audience = 'some_audience_address@testing.gserviceaccount.com'
|
| - contents = crypt.verify_signed_jwt_with_certs(jwt, certs, audience)
|
| - self.assertEqual('billy bob', contents['user'])
|
| - self.assertEqual('data', contents['metadata']['meta'])
|
| -
|
| - def test_verify_id_token_with_certs_uri(self):
|
| - jwt = self._create_signed_jwt()
|
| -
|
| - http = HttpMockSequence([
|
| - ({'status': '200'}, datafile('certs.json')),
|
| - ])
|
| -
|
| - contents = verify_id_token(
|
| - jwt, 'some_audience_address@testing.gserviceaccount.com', http=http)
|
| - self.assertEqual('billy bob', contents['user'])
|
| - self.assertEqual('data', contents['metadata']['meta'])
|
| -
|
| - def test_verify_id_token_with_certs_uri_fails(self):
|
| - jwt = self._create_signed_jwt()
|
| -
|
| - http = HttpMockSequence([
|
| - ({'status': '404'}, datafile('certs.json')),
|
| - ])
|
| -
|
| - self.assertRaises(VerifyJwtTokenError, verify_id_token, jwt,
|
| - 'some_audience_address@testing.gserviceaccount.com',
|
| - http=http)
|
| -
|
| - def test_verify_id_token_bad_tokens(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| -
|
| - # Wrong number of segments
|
| - self._check_jwt_failure('foo', 'Wrong number of segments')
|
| -
|
| - # Not json
|
| - self._check_jwt_failure('foo.bar.baz', 'Can\'t parse token')
|
| -
|
| - # Bad signature
|
| - jwt = 'foo.%s.baz' % crypt._urlsafe_b64encode('{"a":"b"}')
|
| - self._check_jwt_failure(jwt, 'Invalid token signature')
|
| -
|
| - # No expiration
|
| - signer = self.signer.from_string(private_key)
|
| - audience = ('https:#www.googleapis.com/auth/id?client_id='
|
| - 'external_public_key@testing.gserviceaccount.com')
|
| - jwt = crypt.make_signed_jwt(signer, {
|
| - 'aud': audience,
|
| - 'iat': time.time(),
|
| - })
|
| - self._check_jwt_failure(jwt, 'No exp field in token')
|
| -
|
| - # No issued at
|
| - jwt = crypt.make_signed_jwt(signer, {
|
| - 'aud': 'audience',
|
| - 'exp': time.time() + 400,
|
| - })
|
| - self._check_jwt_failure(jwt, 'No iat field in token')
|
| -
|
| - # Too early
|
| - jwt = crypt.make_signed_jwt(signer, {
|
| - 'aud': 'audience',
|
| - 'iat': time.time() + 301,
|
| - 'exp': time.time() + 400,
|
| - })
|
| - self._check_jwt_failure(jwt, 'Token used too early')
|
| -
|
| - # Too late
|
| - jwt = crypt.make_signed_jwt(signer, {
|
| - 'aud': 'audience',
|
| - 'iat': time.time() - 500,
|
| - 'exp': time.time() - 301,
|
| - })
|
| - self._check_jwt_failure(jwt, 'Token used too late')
|
| -
|
| - # Wrong target
|
| - jwt = crypt.make_signed_jwt(signer, {
|
| - 'aud': 'somebody else',
|
| - 'iat': time.time(),
|
| - 'exp': time.time() + 300,
|
| - })
|
| - self._check_jwt_failure(jwt, 'Wrong recipient')
|
| -
|
| -
|
| -class PEMCryptTestsPyCrypto(CryptTests):
|
| - def setUp(self):
|
| - self.format = 'pem'
|
| - self.signer = crypt.PyCryptoSigner
|
| - self.verifier = crypt.PyCryptoVerifier
|
| -
|
| -
|
| -class PEMCryptTestsOpenSSL(CryptTests):
|
| - def setUp(self):
|
| - self.format = 'pem'
|
| - self.signer = crypt.OpenSSLSigner
|
| - self.verifier = crypt.OpenSSLVerifier
|
| -
|
| -
|
| -class SignedJwtAssertionCredentialsTests(unittest.TestCase):
|
| - def setUp(self):
|
| - self.format = 'p12'
|
| - crypt.Signer = crypt.OpenSSLSigner
|
| -
|
| - def test_credentials_good(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| - credentials = SignedJwtAssertionCredentials(
|
| - 'some_account@example.com',
|
| - private_key,
|
| - scope='read+write',
|
| - sub='joe@example.org')
|
| - http = HttpMockSequence([
|
| - ({'status': '200'}, b'{"access_token":"1/3w","expires_in":3600}'),
|
| - ({'status': '200'}, 'echo_request_headers'),
|
| - ])
|
| - http = credentials.authorize(http)
|
| - resp, content = http.request('http://example.org')
|
| - self.assertEqual(b'Bearer 1/3w', content[b'Authorization'])
|
| -
|
| - def test_credentials_to_from_json(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| - credentials = SignedJwtAssertionCredentials(
|
| - 'some_account@example.com',
|
| - private_key,
|
| - scope='read+write',
|
| - sub='joe@example.org')
|
| - json = credentials.to_json()
|
| - restored = Credentials.new_from_json(json)
|
| - self.assertEqual(credentials.private_key, restored.private_key)
|
| - self.assertEqual(credentials.private_key_password,
|
| - restored.private_key_password)
|
| - self.assertEqual(credentials.kwargs, restored.kwargs)
|
| -
|
| - def _credentials_refresh(self, credentials):
|
| - http = HttpMockSequence([
|
| - ({'status': '200'}, b'{"access_token":"1/3w","expires_in":3600}'),
|
| - ({'status': '401'}, b''),
|
| - ({'status': '200'}, b'{"access_token":"3/3w","expires_in":3600}'),
|
| - ({'status': '200'}, 'echo_request_headers'),
|
| - ])
|
| - http = credentials.authorize(http)
|
| - _, content = http.request('http://example.org')
|
| - return content
|
| -
|
| - def test_credentials_refresh_without_storage(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| - credentials = SignedJwtAssertionCredentials(
|
| - 'some_account@example.com',
|
| - private_key,
|
| - scope='read+write',
|
| - sub='joe@example.org')
|
| -
|
| - content = self._credentials_refresh(credentials)
|
| -
|
| - self.assertEqual(b'Bearer 3/3w', content[b'Authorization'])
|
| -
|
| - def test_credentials_refresh_with_storage(self):
|
| - private_key = datafile('privatekey.%s' % self.format)
|
| - credentials = SignedJwtAssertionCredentials(
|
| - 'some_account@example.com',
|
| - private_key,
|
| - scope='read+write',
|
| - sub='joe@example.org')
|
| -
|
| - (filehandle, filename) = tempfile.mkstemp()
|
| - os.close(filehandle)
|
| - store = Storage(filename)
|
| - store.put(credentials)
|
| - credentials.set_store(store)
|
| -
|
| - content = self._credentials_refresh(credentials)
|
| -
|
| - self.assertEqual(b'Bearer 3/3w', content[b'Authorization'])
|
| - os.unlink(filename)
|
| -
|
| -
|
| -class PEMSignedJwtAssertionCredentialsOpenSSLTests(
|
| - SignedJwtAssertionCredentialsTests):
|
| - def setUp(self):
|
| - self.format = 'pem'
|
| - crypt.Signer = crypt.OpenSSLSigner
|
| -
|
| -
|
| -class PEMSignedJwtAssertionCredentialsPyCryptoTests(
|
| - SignedJwtAssertionCredentialsTests):
|
| - def setUp(self):
|
| - self.format = 'pem'
|
| - crypt.Signer = crypt.PyCryptoSigner
|
| -
|
| -
|
| -class PKCSSignedJwtAssertionCredentialsPyCryptoTests(unittest.TestCase):
|
| -
|
| - def test_for_failure(self):
|
| - crypt.Signer = crypt.PyCryptoSigner
|
| - private_key = datafile('privatekey.p12')
|
| - credentials = SignedJwtAssertionCredentials(
|
| - 'some_account@example.com',
|
| - private_key,
|
| - scope='read+write',
|
| - sub='joe@example.org')
|
| - try:
|
| - credentials._generate_assertion()
|
| - self.fail()
|
| - except NotImplementedError:
|
| - pass
|
| -
|
| -
|
| -class TestHasOpenSSLFlag(unittest.TestCase):
|
| - def test_true(self):
|
| - self.assertEqual(True, HAS_OPENSSL)
|
| - self.assertEqual(True, HAS_CRYPTO)
|
| -
|
| -
|
| -if __name__ == '__main__':
|
| - unittest.main()
|
|
|