OLD | NEW |
| (Empty) |
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 . import GOOGLE_REVOKE_URI | |
29 from . import GOOGLE_TOKEN_URI | |
30 from . import util | |
31 from client import AssertionCredentials | |
32 from third_party import six | |
33 | |
34 | |
35 class _ServiceAccountCredentials(AssertionCredentials): | |
36 """Class representing a service account (signed JWT) credential.""" | |
37 | |
38 MAX_TOKEN_LIFETIME_SECS = 3600 # 1 hour in seconds | |
39 | |
40 def __init__(self, service_account_id, service_account_email, private_key_id, | |
41 private_key_pkcs8_text, scopes, user_agent=None, | |
42 token_uri=GOOGLE_TOKEN_URI, revoke_uri=GOOGLE_REVOKE_URI, | |
43 **kwargs): | |
44 | |
45 super(_ServiceAccountCredentials, self).__init__( | |
46 None, user_agent=user_agent, token_uri=token_uri, revoke_uri=revoke_uri) | |
47 | |
48 self._service_account_id = service_account_id | |
49 self._service_account_email = service_account_email | |
50 self._private_key_id = private_key_id | |
51 self._private_key = _get_private_key(private_key_pkcs8_text) | |
52 self._private_key_pkcs8_text = private_key_pkcs8_text | |
53 self._scopes = util.scopes_to_string(scopes) | |
54 self._user_agent = user_agent | |
55 self._token_uri = token_uri | |
56 self._revoke_uri = revoke_uri | |
57 self._kwargs = kwargs | |
58 | |
59 def _generate_assertion(self): | |
60 """Generate the assertion that will be used in the request.""" | |
61 | |
62 header = { | |
63 'alg': 'RS256', | |
64 'typ': 'JWT', | |
65 'kid': self._private_key_id | |
66 } | |
67 | |
68 now = int(time.time()) | |
69 payload = { | |
70 'aud': self._token_uri, | |
71 'scope': self._scopes, | |
72 'iat': now, | |
73 'exp': now + _ServiceAccountCredentials.MAX_TOKEN_LIFETIME_SECS, | |
74 'iss': self._service_account_email | |
75 } | |
76 payload.update(self._kwargs) | |
77 | |
78 assertion_input = (_urlsafe_b64encode(header) + b'.' + | |
79 _urlsafe_b64encode(payload)) | |
80 | |
81 # Sign the assertion. | |
82 rsa_bytes = rsa.pkcs1.sign(assertion_input, self._private_key, 'SHA-256') | |
83 signature = base64.urlsafe_b64encode(rsa_bytes).rstrip(b'=') | |
84 | |
85 return assertion_input + b'.' + signature | |
86 | |
87 def sign_blob(self, blob): | |
88 # Ensure that it is bytes | |
89 try: | |
90 blob = blob.encode('utf-8') | |
91 except AttributeError: | |
92 pass | |
93 return (self._private_key_id, | |
94 rsa.pkcs1.sign(blob, self._private_key, 'SHA-256')) | |
95 | |
96 @property | |
97 def service_account_email(self): | |
98 return self._service_account_email | |
99 | |
100 @property | |
101 def serialization_data(self): | |
102 return { | |
103 'type': 'service_account', | |
104 'client_id': self._service_account_id, | |
105 'client_email': self._service_account_email, | |
106 'private_key_id': self._private_key_id, | |
107 'private_key': self._private_key_pkcs8_text | |
108 } | |
109 | |
110 def create_scoped_required(self): | |
111 return not self._scopes | |
112 | |
113 def create_scoped(self, scopes): | |
114 return _ServiceAccountCredentials(self._service_account_id, | |
115 self._service_account_email, | |
116 self._private_key_id, | |
117 self._private_key_pkcs8_text, | |
118 scopes, | |
119 user_agent=self._user_agent, | |
120 token_uri=self._token_uri, | |
121 revoke_uri=self._revoke_uri, | |
122 **self._kwargs) | |
123 | |
124 | |
125 def _urlsafe_b64encode(data): | |
126 return base64.urlsafe_b64encode( | |
127 json.dumps(data, separators=(',', ':')).encode('UTF-8')).rstrip(b'=') | |
128 | |
129 | |
130 def _get_private_key(private_key_pkcs8_text): | |
131 """Get an RSA private key object from a pkcs8 representation.""" | |
132 | |
133 if not isinstance(private_key_pkcs8_text, six.binary_type): | |
134 private_key_pkcs8_text = private_key_pkcs8_text.encode('ascii') | |
135 der = rsa.pem.load_pem(private_key_pkcs8_text, 'PRIVATE KEY') | |
136 asn1_private_key, _ = decoder.decode(der, asn1Spec=PrivateKeyInfo()) | |
137 return rsa.PrivateKey.load_pkcs1( | |
138 asn1_private_key.getComponentByName('privateKey').asOctets(), | |
139 format='DER') | |
OLD | NEW |