Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.8/site-packages/google/auth/credentials.py: 60%

104 statements  

« prev     ^ index     » next       coverage.py v7.3.1, created at 2023-09-25 06:37 +0000

1# Copyright 2016 Google LLC 

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 

16"""Interfaces for credentials.""" 

17 

18import abc 

19import os 

20 

21from google.auth import _helpers, environment_vars 

22from google.auth import exceptions 

23from google.auth import metrics 

24 

25 

26class Credentials(metaclass=abc.ABCMeta): 

27 """Base class for all credentials. 

28 

29 All credentials have a :attr:`token` that is used for authentication and 

30 may also optionally set an :attr:`expiry` to indicate when the token will 

31 no longer be valid. 

32 

33 Most credentials will be :attr:`invalid` until :meth:`refresh` is called. 

34 Credentials can do this automatically before the first HTTP request in 

35 :meth:`before_request`. 

36 

37 Although the token and expiration will change as the credentials are 

38 :meth:`refreshed <refresh>` and used, credentials should be considered 

39 immutable. Various credentials will accept configuration such as private 

40 keys, scopes, and other options. These options are not changeable after 

41 construction. Some classes will provide mechanisms to copy the credentials 

42 with modifications such as :meth:`ScopedCredentials.with_scopes`. 

43 """ 

44 

45 def __init__(self): 

46 self.token = None 

47 """str: The bearer token that can be used in HTTP headers to make 

48 authenticated requests.""" 

49 self.expiry = None 

50 """Optional[datetime]: When the token expires and is no longer valid. 

51 If this is None, the token is assumed to never expire.""" 

52 self._quota_project_id = None 

53 """Optional[str]: Project to use for quota and billing purposes.""" 

54 self._trust_boundary = None 

55 """Optional[str]: Encoded string representation of credentials trust 

56 boundary.""" 

57 self._universe_domain = "googleapis.com" 

58 """Optional[str]: The universe domain value, default is googleapis.com 

59 """ 

60 

61 @property 

62 def expired(self): 

63 """Checks if the credentials are expired. 

64 

65 Note that credentials can be invalid but not expired because 

66 Credentials with :attr:`expiry` set to None is considered to never 

67 expire. 

68 """ 

69 if not self.expiry: 

70 return False 

71 

72 # Remove some threshold from expiry to err on the side of reporting 

73 # expiration early so that we avoid the 401-refresh-retry loop. 

74 skewed_expiry = self.expiry - _helpers.REFRESH_THRESHOLD 

75 return _helpers.utcnow() >= skewed_expiry 

76 

77 @property 

78 def valid(self): 

79 """Checks the validity of the credentials. 

80 

81 This is True if the credentials have a :attr:`token` and the token 

82 is not :attr:`expired`. 

83 """ 

84 return self.token is not None and not self.expired 

85 

86 @property 

87 def quota_project_id(self): 

88 """Project to use for quota and billing purposes.""" 

89 return self._quota_project_id 

90 

91 @property 

92 def universe_domain(self): 

93 """The universe domain value.""" 

94 return self._universe_domain 

95 

96 @abc.abstractmethod 

97 def refresh(self, request): 

98 """Refreshes the access token. 

99 

100 Args: 

101 request (google.auth.transport.Request): The object used to make 

102 HTTP requests. 

103 

104 Raises: 

105 google.auth.exceptions.RefreshError: If the credentials could 

106 not be refreshed. 

107 """ 

108 # pylint: disable=missing-raises-doc 

109 # (pylint doesn't recognize that this is abstract) 

110 raise NotImplementedError("Refresh must be implemented") 

111 

112 def _metric_header_for_usage(self): 

113 """The x-goog-api-client header for token usage metric. 

114 

115 This header will be added to the API service requests in before_request 

116 method. For example, "cred-type/sa-jwt" means service account self 

117 signed jwt access token is used in the API service request 

118 authorization header. Children credentials classes need to override 

119 this method to provide the header value, if the token usage metric is 

120 needed. 

121 

122 Returns: 

123 str: The x-goog-api-client header value. 

124 """ 

125 return None 

126 

127 def apply(self, headers, token=None): 

128 """Apply the token to the authentication header. 

129 

130 Args: 

131 headers (Mapping): The HTTP request headers. 

132 token (Optional[str]): If specified, overrides the current access 

133 token. 

134 """ 

135 headers["authorization"] = "Bearer {}".format( 

136 _helpers.from_bytes(token or self.token) 

137 ) 

138 if self._trust_boundary is not None: 

139 headers["x-identity-trust-boundary"] = self._trust_boundary 

140 if self.quota_project_id: 

141 headers["x-goog-user-project"] = self.quota_project_id 

142 

143 def before_request(self, request, method, url, headers): 

144 """Performs credential-specific before request logic. 

145 

146 Refreshes the credentials if necessary, then calls :meth:`apply` to 

147 apply the token to the authentication header. 

148 

149 Args: 

150 request (google.auth.transport.Request): The object used to make 

151 HTTP requests. 

152 method (str): The request's HTTP method or the RPC method being 

153 invoked. 

154 url (str): The request's URI or the RPC service's URI. 

155 headers (Mapping): The request's headers. 

156 """ 

157 # pylint: disable=unused-argument 

158 # (Subclasses may use these arguments to ascertain information about 

159 # the http request.) 

160 if not self.valid: 

161 self.refresh(request) 

162 metrics.add_metric_header(headers, self._metric_header_for_usage()) 

163 self.apply(headers) 

164 

165 

166class CredentialsWithQuotaProject(Credentials): 

167 """Abstract base for credentials supporting ``with_quota_project`` factory""" 

168 

169 def with_quota_project(self, quota_project_id): 

170 """Returns a copy of these credentials with a modified quota project. 

171 

172 Args: 

173 quota_project_id (str): The project to use for quota and 

174 billing purposes 

175 

176 Returns: 

177 google.oauth2.credentials.Credentials: A new credentials instance. 

178 """ 

179 raise NotImplementedError("This credential does not support quota project.") 

180 

181 def with_quota_project_from_environment(self): 

182 quota_from_env = os.environ.get(environment_vars.GOOGLE_CLOUD_QUOTA_PROJECT) 

183 if quota_from_env: 

184 return self.with_quota_project(quota_from_env) 

185 return self 

186 

187 

188class CredentialsWithTokenUri(Credentials): 

189 """Abstract base for credentials supporting ``with_token_uri`` factory""" 

190 

191 def with_token_uri(self, token_uri): 

192 """Returns a copy of these credentials with a modified token uri. 

193 

194 Args: 

195 token_uri (str): The uri to use for fetching/exchanging tokens 

196 

197 Returns: 

198 google.oauth2.credentials.Credentials: A new credentials instance. 

199 """ 

200 raise NotImplementedError("This credential does not use token uri.") 

201 

202 

203class AnonymousCredentials(Credentials): 

204 """Credentials that do not provide any authentication information. 

205 

206 These are useful in the case of services that support anonymous access or 

207 local service emulators that do not use credentials. 

208 """ 

209 

210 @property 

211 def expired(self): 

212 """Returns `False`, anonymous credentials never expire.""" 

213 return False 

214 

215 @property 

216 def valid(self): 

217 """Returns `True`, anonymous credentials are always valid.""" 

218 return True 

219 

220 def refresh(self, request): 

221 """Raises :class:``InvalidOperation``, anonymous credentials cannot be 

222 refreshed.""" 

223 raise exceptions.InvalidOperation("Anonymous credentials cannot be refreshed.") 

224 

225 def apply(self, headers, token=None): 

226 """Anonymous credentials do nothing to the request. 

227 

228 The optional ``token`` argument is not supported. 

229 

230 Raises: 

231 google.auth.exceptions.InvalidValue: If a token was specified. 

232 """ 

233 if token is not None: 

234 raise exceptions.InvalidValue("Anonymous credentials don't support tokens.") 

235 

236 def before_request(self, request, method, url, headers): 

237 """Anonymous credentials do nothing to the request.""" 

238 

239 

240class ReadOnlyScoped(metaclass=abc.ABCMeta): 

241 """Interface for credentials whose scopes can be queried. 

242 

243 OAuth 2.0-based credentials allow limiting access using scopes as described 

244 in `RFC6749 Section 3.3`_. 

245 If a credential class implements this interface then the credentials either 

246 use scopes in their implementation. 

247 

248 Some credentials require scopes in order to obtain a token. You can check 

249 if scoping is necessary with :attr:`requires_scopes`:: 

250 

251 if credentials.requires_scopes: 

252 # Scoping is required. 

253 credentials = credentials.with_scopes(scopes=['one', 'two']) 

254 

255 Credentials that require scopes must either be constructed with scopes:: 

256 

257 credentials = SomeScopedCredentials(scopes=['one', 'two']) 

258 

259 Or must copy an existing instance using :meth:`with_scopes`:: 

260 

261 scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) 

262 

263 Some credentials have scopes but do not allow or require scopes to be set, 

264 these credentials can be used as-is. 

265 

266 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 

267 """ 

268 

269 def __init__(self): 

270 super(ReadOnlyScoped, self).__init__() 

271 self._scopes = None 

272 self._default_scopes = None 

273 

274 @property 

275 def scopes(self): 

276 """Sequence[str]: the credentials' current set of scopes.""" 

277 return self._scopes 

278 

279 @property 

280 def default_scopes(self): 

281 """Sequence[str]: the credentials' current set of default scopes.""" 

282 return self._default_scopes 

283 

284 @abc.abstractproperty 

285 def requires_scopes(self): 

286 """True if these credentials require scopes to obtain an access token. 

287 """ 

288 return False 

289 

290 def has_scopes(self, scopes): 

291 """Checks if the credentials have the given scopes. 

292 

293 .. warning: This method is not guaranteed to be accurate if the 

294 credentials are :attr:`~Credentials.invalid`. 

295 

296 Args: 

297 scopes (Sequence[str]): The list of scopes to check. 

298 

299 Returns: 

300 bool: True if the credentials have the given scopes. 

301 """ 

302 credential_scopes = ( 

303 self._scopes if self._scopes is not None else self._default_scopes 

304 ) 

305 return set(scopes).issubset(set(credential_scopes or [])) 

306 

307 

308class Scoped(ReadOnlyScoped): 

309 """Interface for credentials whose scopes can be replaced while copying. 

310 

311 OAuth 2.0-based credentials allow limiting access using scopes as described 

312 in `RFC6749 Section 3.3`_. 

313 If a credential class implements this interface then the credentials either 

314 use scopes in their implementation. 

315 

316 Some credentials require scopes in order to obtain a token. You can check 

317 if scoping is necessary with :attr:`requires_scopes`:: 

318 

319 if credentials.requires_scopes: 

320 # Scoping is required. 

321 credentials = credentials.create_scoped(['one', 'two']) 

322 

323 Credentials that require scopes must either be constructed with scopes:: 

324 

325 credentials = SomeScopedCredentials(scopes=['one', 'two']) 

326 

327 Or must copy an existing instance using :meth:`with_scopes`:: 

328 

329 scoped_credentials = credentials.with_scopes(scopes=['one', 'two']) 

330 

331 Some credentials have scopes but do not allow or require scopes to be set, 

332 these credentials can be used as-is. 

333 

334 .. _RFC6749 Section 3.3: https://tools.ietf.org/html/rfc6749#section-3.3 

335 """ 

336 

337 @abc.abstractmethod 

338 def with_scopes(self, scopes, default_scopes=None): 

339 """Create a copy of these credentials with the specified scopes. 

340 

341 Args: 

342 scopes (Sequence[str]): The list of scopes to attach to the 

343 current credentials. 

344 

345 Raises: 

346 NotImplementedError: If the credentials' scopes can not be changed. 

347 This can be avoided by checking :attr:`requires_scopes` before 

348 calling this method. 

349 """ 

350 raise NotImplementedError("This class does not require scoping.") 

351 

352 

353def with_scopes_if_required(credentials, scopes, default_scopes=None): 

354 """Creates a copy of the credentials with scopes if scoping is required. 

355 

356 This helper function is useful when you do not know (or care to know) the 

357 specific type of credentials you are using (such as when you use 

358 :func:`google.auth.default`). This function will call 

359 :meth:`Scoped.with_scopes` if the credentials are scoped credentials and if 

360 the credentials require scoping. Otherwise, it will return the credentials 

361 as-is. 

362 

363 Args: 

364 credentials (google.auth.credentials.Credentials): The credentials to 

365 scope if necessary. 

366 scopes (Sequence[str]): The list of scopes to use. 

367 default_scopes (Sequence[str]): Default scopes passed by a 

368 Google client library. Use 'scopes' for user-defined scopes. 

369 

370 Returns: 

371 google.auth.credentials.Credentials: Either a new set of scoped 

372 credentials, or the passed in credentials instance if no scoping 

373 was required. 

374 """ 

375 if isinstance(credentials, Scoped) and credentials.requires_scopes: 

376 return credentials.with_scopes(scopes, default_scopes=default_scopes) 

377 else: 

378 return credentials 

379 

380 

381class Signing(metaclass=abc.ABCMeta): 

382 """Interface for credentials that can cryptographically sign messages.""" 

383 

384 @abc.abstractmethod 

385 def sign_bytes(self, message): 

386 """Signs the given message. 

387 

388 Args: 

389 message (bytes): The message to sign. 

390 

391 Returns: 

392 bytes: The message's cryptographic signature. 

393 """ 

394 # pylint: disable=missing-raises-doc,redundant-returns-doc 

395 # (pylint doesn't recognize that this is abstract) 

396 raise NotImplementedError("Sign bytes must be implemented.") 

397 

398 @abc.abstractproperty 

399 def signer_email(self): 

400 """Optional[str]: An email address that identifies the signer.""" 

401 # pylint: disable=missing-raises-doc 

402 # (pylint doesn't recognize that this is abstract) 

403 raise NotImplementedError("Signer email must be implemented.") 

404 

405 @abc.abstractproperty 

406 def signer(self): 

407 """google.auth.crypt.Signer: The signer used to sign bytes.""" 

408 # pylint: disable=missing-raises-doc 

409 # (pylint doesn't recognize that this is abstract) 

410 raise NotImplementedError("Signer must be implemented.")