Source Code for Module oauth2client.service_account

  1  # Copyright 2014 Google Inc. All rights reserved. 
  2  # 
  3  # Licensed under the Apache License, Version 2.0 (the "License"); 
  4  # you may not use this file except in compliance with the License. 
  5  # You may obtain a copy of the License at 
  6  # 
  7  #      http://www.apache.org/licenses/LICENSE-2.0 
  8  # 
  9  # Unless required by applicable law or agreed to in writing, software 
 10  # distributed under the License is distributed on an "AS IS" BASIS, 
 11  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 
 12  # See the License for the specific language governing permissions and 
 13  # limitations under the License. 
 14   
 15  """A service account credentials class. 
 16   
 17  This credentials class is implemented on top of rsa library. 
 18  """ 
 19   
 20  import base64 
 21  import json 
 22  import time 
 23   
 24  from pyasn1.codec.ber import decoder 
 25  from pyasn1_modules.rfc5208 import PrivateKeyInfo 
 26  import rsa 
 27   
 28  from oauth2client import GOOGLE_REVOKE_URI 
 29  from oauth2client import GOOGLE_TOKEN_URI 
 30  from oauth2client import util 
 31  from oauth2client.client import AssertionCredentials 
 32   
 33   
 34 -class _ServiceAccountCredentials(AssertionCredentials): 
 35    """Class representing a service account (signed JWT) credential.""" 
 36   
 37    MAX_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds 
 38   
 39 -  def __init__(self, service_account_id, service_account_email, private_key_id, 
 40                 private_key_pkcs8_text, scopes, user_agent=None, 
 41                 token_uri=GOOGLE_TOKEN_URI, revoke_uri=GOOGLE_REVOKE_URI, 
 42                 **kwargs): 
 43   
 44      super(_ServiceAccountCredentials, self).__init__( 
 45          None, user_agent=user_agent, token_uri=token_uri, revoke_uri=revoke_uri) 
 46   
 47      self._service_account_id = service_account_id 
 48      self._service_account_email = service_account_email 
 49      self._private_key_id = private_key_id 
 50      self._private_key = _get_private_key(private_key_pkcs8_text) 
 51      self._private_key_pkcs8_text = private_key_pkcs8_text 
 52      self._scopes = util.scopes_to_string(scopes) 
 53      self._user_agent = user_agent 
 54      self._token_uri = token_uri 
 55      self._revoke_uri = revoke_uri 
 56      self._kwargs = kwargs 
 57   
 58 -  def _generate_assertion(self): 
 59      """Generate the assertion that will be used in the request.""" 
 60   
 61      header = { 
 62          'alg': 'RS256', 
 63          'typ': 'JWT', 
 64          'kid': self._private_key_id 
 65      } 
 66   
 67      now = long(time.time()) 
 68      payload = { 
 69          'aud': self._token_uri, 
 70          'scope': self._scopes, 
 71          'iat': now, 
 72          'exp': now + _ServiceAccountCredentials.MAX_TOKEN_LIFETIME_SECS, 
 73          'iss': self._service_account_email 
 74      } 
 75      payload.update(self._kwargs) 
 76   
 77      assertion_input = '%s.%s' % ( 
 78          _urlsafe_b64encode(header), 
 79          _urlsafe_b64encode(payload)) 
 80   
 81      # Sign the assertion. 
 82      signature = base64.urlsafe_b64encode(rsa.pkcs1.sign( 
 83          assertion_input, self._private_key, 'SHA-256')).rstrip('=') 
 84   
 85      return '%s.%s' % (assertion_input, signature) 
 86   
 87 -  def sign_blob(self, blob): 
 88      return (self._private_key_id, 
 89              rsa.pkcs1.sign(blob, self._private_key, 'SHA-256')) 
 90   
 91    @property 
 92 -  def service_account_email(self): 
 93      return self._service_account_email 
 94   
 95    @property 
 96 -  def serialization_data(self): 
 97      return { 
 98          'type': 'service_account', 
 99          'client_id': self._service_account_id, 
100          'client_email': self._service_account_email, 
101          'private_key_id': self._private_key_id, 
102          'private_key': self._private_key_pkcs8_text 
103      } 
104   
105 -  def create_scoped_required(self): 
106      return not self._scopes 
107   
108 -  def create_scoped(self, scopes): 
109      return _ServiceAccountCredentials(self._service_account_id, 
110                                        self._service_account_email, 
111                                        self._private_key_id, 
112                                        self._private_key_pkcs8_text, 
113                                        scopes, 
114                                        user_agent=self._user_agent, 
115                                        token_uri=self._token_uri, 
116                                        revoke_uri=self._revoke_uri, 
117                                        **self._kwargs) 
118   
119   
120 -def _urlsafe_b64encode(data): 
121    return base64.urlsafe_b64encode( 
122        json.dumps(data, separators=(',', ':')).encode('UTF-8')).rstrip('=') 
123   
124   
125 -def _get_private_key(private_key_pkcs8_text): 
126    """Get an RSA private key object from a pkcs8 representation.""" 
127   
128    der = rsa.pem.load_pem(private_key_pkcs8_text, 'PRIVATE KEY') 
129    asn1_private_key, _ = decoder.decode(der, asn1Spec=PrivateKeyInfo()) 
130    return rsa.PrivateKey.load_pkcs1( 
131        asn1_private_key.getComponentByName('privateKey').asOctets(), 
132        format='DER') 
133