Index: tools/telemetry/third_party/gsutil/third_party/oauth2client/tests/test_jwt.py |
diff --git a/tools/telemetry/third_party/gsutil/third_party/oauth2client/tests/test_jwt.py b/tools/telemetry/third_party/gsutil/third_party/oauth2client/tests/test_jwt.py |
deleted file mode 100644 |
index f58a2947821ce986ed6313ddbbe1151e6d250053..0000000000000000000000000000000000000000 |
--- a/tools/telemetry/third_party/gsutil/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() |