openstack/deb-python-oauth2client
Code Issues Proposed changes

Adding common sign_blob() service account types.

Browse Source 
Also adding service_account_email() property.
This commit is contained in:
Danny Hermes
2016-02-19 14:15:23 -08:00
parent c66e4f201f
commit ce0d71a497
7 changed files with 280 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
oauth2client/client.py
View File

@@ -1617,6 +1617,18 @@ class AssertionCredentials(GoogleCredentials):
"""
self._do_revoke(http_request, self.access_token)
def sign_blob(self, blob):
"""Cryptographically sign a blob (of bytes).
Args:
blob: bytes, Message to be signed.
Returns:
tuple, A pair of the private key ID used to sign the blob and
the signed contents.
"""
raise NotImplementedError('This method is abstract.')
def _RequireCryptoOrDie():
"""Ensure we have a crypto library, or throw CryptoUnavailableError.

29
oauth2client/contrib/appengine.py
View File

@@ -166,6 +166,7 @@ class AppAssertionCredentials(AssertionCredentials):
self.scope = util.scopes_to_string(scope)
self._kwargs = kwargs
self.service_account_id = kwargs.get('service_account_id', None)
self._service_account_email = None
# Assertion type is no longer used, but still in the
# parent class signature.
@@ -210,6 +211,34 @@ class AppAssertionCredentials(AssertionCredentials):
def create_scoped(self, scopes):
return AppAssertionCredentials(scopes, **self._kwargs)
def sign_blob(self, blob):
"""Cryptographically sign a blob (of bytes).
Implements abstract method
:meth:`oauth2client.client.AssertionCredentials.sign_blob`.
Args:
blob: bytes, Message to be signed.
Returns:
tuple, A pair of the private key ID used to sign the blob and
the signed contents.
"""
return app_identity.sign_blob(blob)
@property
def service_account_email(self):
"""Get the email for the current service account.
Returns:
string, The email associated with the Google App Engine
service account.
"""
if self._service_account_email is None:
self._service_account_email = (
app_identity.get_service_account_name())
return self._service_account_email
class FlowProperty(db.Property):
"""App Engine datastore Property for Flow.

73
oauth2client/contrib/gce.py
View File

@@ -21,6 +21,7 @@ import json
import logging
import warnings
import httplib2
from six.moves import http_client
from six.moves import urllib
@@ -35,8 +36,10 @@ __author__ = 'jcgregorio@google.com (Joe Gregorio)'
logger = logging.getLogger(__name__)
# URI Template for the endpoint that returns access_tokens.
META = ('http://metadata.google.internal/computeMetadata/v1/instance/'
'service-accounts/default/token')
_METADATA_ROOT = ('http://metadata.google.internal/computeMetadata/v1/'
'instance/service-accounts/default/')
META = _METADATA_ROOT + 'token'
_DEFAULT_EMAIL_METADATA = _METADATA_ROOT + 'email'
_SCOPES_WARNING = """\
You have requested explicit scopes to be used with a GCE service account.
Using this argument will have no effect on the actual scopes for tokens
@@ -45,6 +48,30 @@ can't be overridden in the request.
"""
def _get_service_account_email(http_request=None):
"""Get the GCE service account email from the current environment.
Args:
http_request: callable, (Optional) a callable that matches the method
signature of httplib2.Http.request, used to make
the request to the metadata service.
Returns:
tuple, A pair where the first entry is an optional response (from a
failed request) and the second is service account email found (as
a string).
"""
if http_request is None:
http_request = httplib2.Http().request
response, content = http_request(
_DEFAULT_EMAIL_METADATA, headers={'Metadata-Flavor': 'Google'})
if response.status == http_client.OK:
content = _from_bytes(content)
return None, content
else:
return response, content
class AppAssertionCredentials(AssertionCredentials):
"""Credentials object for Compute Engine Assertion Grants
@@ -78,6 +105,7 @@ class AppAssertionCredentials(AssertionCredentials):
# Assertion type is no longer used, but still in the
# parent class signature.
super(AppAssertionCredentials, self).__init__(None)
self._service_account_email = None
@classmethod
def from_json(cls, json_data):
@@ -123,3 +151,44 @@ class AppAssertionCredentials(AssertionCredentials):
def create_scoped(self, scopes):
return AppAssertionCredentials(scopes, **self.kwargs)
def sign_blob(self, blob):
"""Cryptographically sign a blob (of bytes).
This method is provided to support a common interface, but
the actual key used for a Google Compute Engine service account
is not available, so it can't be used to sign content.
Args:
blob: bytes, Message to be signed.
Raises:
NotImplementedError, always.
"""
raise NotImplementedError(
'Compute Engine service accounts cannot sign blobs')
@property
def service_account_email(self):
"""Get the email for the current service account.
Uses the Google Compute Engine metadata service to retrieve the email
of the default service account.
Returns:
string, The email associated with the Google Compute Engine
service account.
Raises:
AttributeError, if the email can not be retrieved from the Google
Compute Engine metadata service.
"""
if self._service_account_email is None:
failure, email = _get_service_account_email()
if failure is None:
self._service_account_email = email
else:
raise AttributeError('Failed to retrieve the email from the '
'Google Compute Engine metadata service',
failure, email)
return self._service_account_email

17
oauth2client/service_account.py
View File

@@ -320,10 +320,27 @@ class ServiceAccountCredentials(AssertionCredentials):
key_id=self._private_key_id)
def sign_blob(self, blob):
"""Cryptographically sign a blob (of bytes).
Implements abstract method
:meth:`oauth2client.client.AssertionCredentials.sign_blob`.
Args:
blob: bytes, Message to be signed.
Returns:
tuple, A pair of the private key ID used to sign the blob and
the signed contents.
"""
return self._private_key_id, self._signer.sign(blob)
@property
def service_account_email(self):
"""Get the email for the current service account.
Returns:
string, The email associated with the service account.
"""
return self._service_account_email
@property

60
tests/contrib/test_appengine.py
View File

@@ -116,14 +116,29 @@ class TestAppAssertionCredentials(unittest.TestCase):
class AppIdentityStubImpl(apiproxy_stub.APIProxyStub):
def __init__(self):
def __init__(self, key_name=None, sig_bytes=None,
svc_acct=None):
super(TestAppAssertionCredentials.AppIdentityStubImpl,
self).__init__('app_identity_service')
self._key_name = key_name
self._sig_bytes = sig_bytes
self._sign_calls = []
self._svc_acct = svc_acct
self._get_acct_name_calls = 0
def _Dynamic_GetAccessToken(self, request, response):
response.set_access_token('a_token_123')
response.set_expiration_time(time.time() + 1800)
def _Dynamic_SignForApp(self, request, response):
response.set_key_name(self._key_name)
response.set_signature_bytes(self._sig_bytes)
self._sign_calls.append(request.bytes_to_sign())
def _Dynamic_GetServiceAccountName(self, request, response):
response.set_service_account_name(self._svc_acct)
self._get_acct_name_calls += 1
class ErroringAppIdentityStubImpl(apiproxy_stub.APIProxyStub):
def __init__(self):
@@ -210,6 +225,49 @@ class TestAppAssertionCredentials(unittest.TestCase):
self.assertTrue(isinstance(new_credentials, AppAssertionCredentials))
self.assertEqual('dummy_scope', new_credentials.scope)
def test_sign_blob(self):
key_name = b'1234567890'
sig_bytes = b'himom'
app_identity_stub = self.AppIdentityStubImpl(
key_name=key_name, sig_bytes=sig_bytes)
apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
apiproxy_stub_map.apiproxy.RegisterStub('app_identity_service',
app_identity_stub)
credentials = AppAssertionCredentials([])
to_sign = b'blob'
self.assertEqual(app_identity_stub._sign_calls, [])
result = credentials.sign_blob(to_sign)
self.assertEqual(result, (key_name, sig_bytes))
self.assertEqual(app_identity_stub._sign_calls, [to_sign])
def test_service_account_email(self):
acct_name = 'new-value@appspot.gserviceaccount.com'
app_identity_stub = self.AppIdentityStubImpl(svc_acct=acct_name)
apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
apiproxy_stub_map.apiproxy.RegisterStub('app_identity_service',
app_identity_stub)
credentials = AppAssertionCredentials([])
self.assertIsNone(credentials._service_account_email)
self.assertEqual(app_identity_stub._get_acct_name_calls, 0)
self.assertEqual(credentials.service_account_email, acct_name)
self.assertIsNotNone(credentials._service_account_email)
self.assertEqual(app_identity_stub._get_acct_name_calls, 1)
def test_service_account_email_already_set(self):
acct_name = 'existing@appspot.gserviceaccount.com'
credentials = AppAssertionCredentials([])
credentials._service_account_email = acct_name
app_identity_stub = self.AppIdentityStubImpl(svc_acct=acct_name)
apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
apiproxy_stub_map.apiproxy.RegisterStub('app_identity_service',
app_identity_stub)
self.assertEqual(app_identity_stub._get_acct_name_calls, 0)
self.assertEqual(credentials.service_account_email, acct_name)
self.assertEqual(app_identity_stub._get_acct_name_calls, 0)
def test_get_access_token(self):
app_identity_stub = self.AppIdentityStubImpl()
apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()

90
tests/contrib/test_gce.py
View File

@@ -17,14 +17,17 @@
import json
from six.moves import http_client
from six.moves import urllib
import unittest
import unittest2
import mock
import httplib2
from oauth2client._helpers import _to_bytes
from oauth2client.client import AccessTokenRefreshError
from oauth2client.client import Credentials
from oauth2client.client import save_to_well_known_file
from oauth2client.contrib.gce import _DEFAULT_EMAIL_METADATA
from oauth2client.contrib.gce import _get_service_account_email
from oauth2client.contrib.gce import _SCOPES_WARNING
from oauth2client.contrib.gce import AppAssertionCredentials
@@ -32,7 +35,7 @@ from oauth2client.contrib.gce import AppAssertionCredentials
__author__ = 'jcgregorio@google.com (Joe Gregorio)'
class AppAssertionCredentialsTests(unittest.TestCase):
class AppAssertionCredentialsTests(unittest2.TestCase):
def test_constructor(self):
credentials = AppAssertionCredentials(foo='bar')
@@ -150,6 +153,49 @@ class AppAssertionCredentialsTests(unittest.TestCase):
self.assertEqual('dummy_scope', new_credentials.scope)
warn_mock.assert_called_once_with(_SCOPES_WARNING)
def test_sign_blob_not_implemented(self):
credentials = AppAssertionCredentials([])
with self.assertRaises(NotImplementedError):
credentials.sign_blob(b'blob')
@mock.patch('oauth2client.contrib.gce._get_service_account_email',
return_value=(None, 'retrieved@email.com'))
def test_service_account_email(self, get_email):
credentials = AppAssertionCredentials([])
self.assertIsNone(credentials._service_account_email)
self.assertEqual(credentials.service_account_email,
get_email.return_value[1])
self.assertIsNotNone(credentials._service_account_email)
get_email.assert_called_once_with()
@mock.patch('oauth2client.contrib.gce._get_service_account_email')
def test_service_account_email_already_set(self, get_email):
credentials = AppAssertionCredentials([])
acct_name = 'existing@email.com'
credentials._service_account_email = acct_name
self.assertEqual(credentials.service_account_email, acct_name)
get_email.assert_not_called()
@mock.patch('oauth2client.contrib.gce._get_service_account_email')
def test_service_account_email_failure(self, get_email):
# Set-up the mock.
bad_response = httplib2.Response({'status': http_client.NOT_FOUND})
content = b'bad-bytes-nothing-here'
get_email.return_value = (bad_response, content)
# Test the failure.
credentials = AppAssertionCredentials([])
self.assertIsNone(credentials._service_account_email)
with self.assertRaises(AttributeError) as exc_manager:
getattr(credentials, 'service_account_email')
error_msg = ('Failed to retrieve the email from the '
'Google Compute Engine metadata service')
self.assertEqual(
exc_manager.exception.args,
(error_msg, bad_response, content))
self.assertIsNone(credentials._service_account_email)
get_email.assert_called_once_with()
def test_get_access_token(self):
http = mock.MagicMock()
http.request = mock.MagicMock(
@@ -178,5 +224,43 @@ class AppAssertionCredentialsTests(unittest.TestCase):
os.path.isdir = ORIGINAL_ISDIR
class Test__get_service_account_email(unittest2.TestCase):
def test_success(self):
http_request = mock.MagicMock()
acct_name = b'1234567890@developer.gserviceaccount.com'
http_request.return_value = (
httplib2.Response({'status': http_client.OK}), acct_name)
result = _get_service_account_email(http_request)
self.assertEqual(result, (None, acct_name.decode('utf-8')))
http_request.assert_called_once_with(
_DEFAULT_EMAIL_METADATA,
headers={'Metadata-Flavor': 'Google'})
@mock.patch.object(httplib2.Http, 'request')
def test_success_default_http(self, http_request):
# Don't make _from_bytes() work too hard.
acct_name = u'1234567890@developer.gserviceaccount.com'
http_request.return_value = (
httplib2.Response({'status': http_client.OK}), acct_name)
result = _get_service_account_email()
self.assertEqual(result, (None, acct_name))
http_request.assert_called_once_with(
_DEFAULT_EMAIL_METADATA,
headers={'Metadata-Flavor': 'Google'})
def test_failure(self):
http_request = mock.MagicMock()
response = httplib2.Response({'status': http_client.NOT_FOUND})
content = b'Not found'
http_request.return_value = (response, content)
result = _get_service_account_email(http_request)
self.assertEqual(result, (response, content))
http_request.assert_called_once_with(
_DEFAULT_EMAIL_METADATA,
headers={'Metadata-Flavor': 'Google'})
if __name__ == '__main__': # pragma: NO COVER
unittest.main()
unittest2.main()

5
tests/test_client.py
View File

@@ -1075,6 +1075,11 @@ class TestAssertionCredentials(unittest2.TestCase):
self, '400', revoke_raise=True,
valid_bool_value=False, token_attr='access_token')
def test_sign_blob_abstract(self):
credentials = AssertionCredentials(None)
with self.assertRaises(NotImplementedError):
credentials.sign_blob(b'blob')
class UpdateQueryParamsTest(unittest2.TestCase):
def test_update_query_params_no_params(self):