Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/aio/credentials.py: 59%

1# Copyright 2024 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 asynchronous credentials.""" 

17 

18 

19from google.auth import _helpers 

20from google.auth import exceptions 

21from google.auth._credentials_base import _BaseCredentials 

22 

23 

24class Credentials(_BaseCredentials): 

25 """Base class for all asynchronous credentials. 

26 

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

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

29 no longer be valid. 

30 

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

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

33 :meth:`before_request`. 

34 

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

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

37 immutable. Various credentials will accept configuration such as private 

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

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

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

41 """ 

42 

43 def __init__(self): 

44 super(Credentials, self).__init__() 

45 

46 async def apply(self, headers, token=None): 

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

48 

49 Args: 

50 headers (Mapping): The HTTP request headers. 

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

52 token. 

53 """ 

54 self._apply(headers, token=token) 

55 

56 async def refresh(self, request): 

57 """Refreshes the access token. 

58 

59 Args: 

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

61 HTTP requests. 

62 

63 Raises: 

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

65 not be refreshed. 

66 """ 

67 raise NotImplementedError("Refresh must be implemented") 

68 

69 async def before_request(self, request, method, url, headers): 

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

71 

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

73 apply the token to the authentication header. 

74 

75 Args: 

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

77 HTTP requests. 

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

79 invoked. 

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

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

82 """ 

83 await self.apply(headers) 

84 

85 

86class StaticCredentials(Credentials): 

87 """Asynchronous Credentials representing an immutable access token. 

88 

89 The credentials are considered immutable except the tokens which can be 

90 configured in the constructor :: 

91 

92 credentials = StaticCredentials(token="token123") 

93 

94 StaticCredentials does not support :meth `refresh` and assumes that the configured 

95 token is valid and not expired. StaticCredentials will never attempt to 

96 refresh the token. 

97 """ 

98 

99 def __init__(self, token): 

100 """ 

101 Args: 

102 token (str): The access token. 

103 """ 

104 super(StaticCredentials, self).__init__() 

105 self.token = token 

106 

107 @_helpers.copy_docstring(Credentials) 

108 async def refresh(self, request): 

109 raise exceptions.InvalidOperation("Static credentials cannot be refreshed.") 

110 

111 # Note: before_request should never try to refresh access tokens. 

112 # StaticCredentials intentionally does not support it. 

113 @_helpers.copy_docstring(Credentials) 

114 async def before_request(self, request, method, url, headers): 

115 await self.apply(headers) 

116 

117 

118class AnonymousCredentials(Credentials): 

119 """Asynchronous Credentials that do not provide any authentication information. 

120 

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

122 local service emulators that do not use credentials. 

123 """ 

124 

125 async def refresh(self, request): 

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

127 refreshed.""" 

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

129 

130 async def apply(self, headers, token=None): 

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

132 

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

134 

135 Raises: 

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

137 """ 

138 if token is not None: 

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

140 

141 async def before_request(self, request, method, url, headers): 

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

143 pass