Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/cloud/storage/hmac_key.py: 44%

1# Copyright 2019 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"""Configure HMAC keys that can be used to authenticate requests to Google Cloud Storage. 

16 

17See [HMAC keys documentation](https://cloud.google.com/storage/docs/authentication/hmackeys) 

18""" 

19 

20from google.cloud.exceptions import NotFound 

21from google.cloud._helpers import _rfc3339_nanos_to_datetime 

22 

23from google.cloud.storage._opentelemetry_tracing import create_trace_span 

24from google.cloud.storage.constants import _DEFAULT_TIMEOUT 

25from google.cloud.storage.retry import DEFAULT_RETRY 

26from google.cloud.storage.retry import DEFAULT_RETRY_IF_ETAG_IN_JSON 

27 

28 

29class HMACKeyMetadata(object): 

30 """Metadata about an HMAC service account key withn Cloud Storage. 

31 

32 :type client: :class:`~google.cloud.stoage.client.Client` 

33 :param client: client associated with the key metadata. 

34 

35 :type access_id: str 

36 :param access_id: (Optional) Unique ID of an existing key. 

37 

38 :type project_id: str 

39 :param project_id: (Optional) Project ID of an existing key. 

40 Defaults to client's project. 

41 

42 :type user_project: str 

43 :param user_project: (Optional) This parameter is currently ignored. 

44 """ 

45 

46 ACTIVE_STATE = "ACTIVE" 

47 """Key is active, and may be used to sign requests.""" 

48 INACTIVE_STATE = "INACTIVE" 

49 """Key is inactive, and may not be used to sign requests. 

50 

51 It can be re-activated via :meth:`update`. 

52 """ 

53 DELETED_STATE = "DELETED" 

54 """Key is deleted. It cannot be re-activated.""" 

55 

56 _SETTABLE_STATES = (ACTIVE_STATE, INACTIVE_STATE) 

57 

58 def __init__(self, client, access_id=None, project_id=None, user_project=None): 

59 self._client = client 

60 self._properties = {} 

61 

62 if access_id is not None: 

63 self._properties["accessId"] = access_id 

64 

65 if project_id is not None: 

66 self._properties["projectId"] = project_id 

67 

68 self._user_project = user_project 

69 

70 def __eq__(self, other): 

71 if not isinstance(other, self.__class__): 

72 return NotImplemented 

73 

74 return self._client == other._client and self.access_id == other.access_id 

75 

76 def __hash__(self): 

77 return hash(self._client) + hash(self.access_id) 

78 

79 @property 

80 def access_id(self): 

81 """Access ID of the key. 

82 

83 :rtype: str or None 

84 :returns: unique identifier of the key within a project. 

85 """ 

86 return self._properties.get("accessId") 

87 

88 @property 

89 def etag(self): 

90 """ETag identifying the version of the key metadata. 

91 

92 :rtype: str or None 

93 :returns: ETag for the version of the key's metadata. 

94 """ 

95 return self._properties.get("etag") 

96 

97 @property 

98 def id(self): 

99 """ID of the key, including the Project ID and the Access ID. 

100 

101 :rtype: str or None 

102 :returns: ID of the key. 

103 """ 

104 return self._properties.get("id") 

105 

106 @property 

107 def project(self): 

108 """Project ID associated with the key. 

109 

110 :rtype: str or None 

111 :returns: project identfier for the key. 

112 """ 

113 return self._properties.get("projectId") 

114 

115 @property 

116 def service_account_email(self): 

117 """Service account e-mail address associated with the key. 

118 

119 :rtype: str or None 

120 :returns: e-mail address for the service account which created the key. 

121 """ 

122 return self._properties.get("serviceAccountEmail") 

123 

124 @property 

125 def state(self): 

126 """Get / set key's state. 

127 

128 One of: 

129 - ``ACTIVE`` 

130 - ``INACTIVE`` 

131 - ``DELETED`` 

132 

133 :rtype: str or None 

134 :returns: key's current state. 

135 """ 

136 return self._properties.get("state") 

137 

138 @state.setter 

139 def state(self, value): 

140 self._properties["state"] = value 

141 

142 @property 

143 def time_created(self): 

144 """Retrieve the timestamp at which the HMAC key was created. 

145 

146 :rtype: :class:`datetime.datetime` or ``NoneType`` 

147 :returns: Datetime object parsed from RFC3339 valid timestamp, or 

148 ``None`` if the bucket's resource has not been loaded 

149 from the server. 

150 """ 

151 value = self._properties.get("timeCreated") 

152 if value is not None: 

153 return _rfc3339_nanos_to_datetime(value) 

154 

155 @property 

156 def updated(self): 

157 """Retrieve the timestamp at which the HMAC key was created. 

158 

159 :rtype: :class:`datetime.datetime` or ``NoneType`` 

160 :returns: Datetime object parsed from RFC3339 valid timestamp, or 

161 ``None`` if the bucket's resource has not been loaded 

162 from the server. 

163 """ 

164 value = self._properties.get("updated") 

165 if value is not None: 

166 return _rfc3339_nanos_to_datetime(value) 

167 

168 @property 

169 def path(self): 

170 """Resource path for the metadata's key.""" 

171 

172 if self.access_id is None: 

173 raise ValueError("No 'access_id' set.") 

174 

175 project = self.project 

176 if project is None: 

177 project = self._client.project 

178 

179 return f"/projects/{project}/hmacKeys/{self.access_id}" 

180 

181 @property 

182 def user_project(self): 

183 """Project ID to be billed for API requests made via this bucket. 

184 

185 This property is currently ignored by the server. 

186 

187 :rtype: str 

188 """ 

189 return self._user_project 

190 

191 def exists(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

192 """Determine whether or not the key for this metadata exists. 

193 

194 :type timeout: float or tuple 

195 :param timeout: 

196 (Optional) The amount of time, in seconds, to wait 

197 for the server response. See: :ref:`configuring_timeouts` 

198 

199 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

200 :param retry: 

201 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

202 

203 :rtype: bool 

204 :returns: True if the key exists in Cloud Storage. 

205 """ 

206 with create_trace_span(name="Storage.HmacKey.exists"): 

207 try: 

208 qs_params = {} 

209 

210 if self.user_project is not None: 

211 qs_params["userProject"] = self.user_project 

212 

213 self._client._get_resource( 

214 self.path, 

215 query_params=qs_params, 

216 timeout=timeout, 

217 retry=retry, 

218 ) 

219 except NotFound: 

220 return False 

221 else: 

222 return True 

223 

224 def reload(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

225 """Reload properties from Cloud Storage. 

226 

227 :type timeout: float or tuple 

228 :param timeout: 

229 (Optional) The amount of time, in seconds, to wait 

230 for the server response. See: :ref:`configuring_timeouts` 

231 

232 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

233 :param retry: 

234 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

235 

236 :raises :class:`~google.api_core.exceptions.NotFound`: 

237 if the key does not exist on the back-end. 

238 """ 

239 with create_trace_span(name="Storage.HmacKey.reload"): 

240 qs_params = {} 

241 

242 if self.user_project is not None: 

243 qs_params["userProject"] = self.user_project 

244 

245 self._properties = self._client._get_resource( 

246 self.path, 

247 query_params=qs_params, 

248 timeout=timeout, 

249 retry=retry, 

250 ) 

251 

252 def update(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY_IF_ETAG_IN_JSON): 

253 """Save writable properties to Cloud Storage. 

254 

255 :type timeout: float or tuple 

256 :param timeout: 

257 (Optional) The amount of time, in seconds, to wait 

258 for the server response. See: :ref:`configuring_timeouts` 

259 

260 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

261 :param retry: 

262 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

263 

264 :raises :class:`~google.api_core.exceptions.NotFound`: 

265 if the key does not exist on the back-end. 

266 """ 

267 with create_trace_span(name="Storage.HmacKey.update"): 

268 qs_params = {} 

269 if self.user_project is not None: 

270 qs_params["userProject"] = self.user_project 

271 

272 payload = {"state": self.state} 

273 self._properties = self._client._put_resource( 

274 self.path, 

275 payload, 

276 query_params=qs_params, 

277 timeout=timeout, 

278 retry=retry, 

279 ) 

280 

281 def delete(self, timeout=_DEFAULT_TIMEOUT, retry=DEFAULT_RETRY): 

282 """Delete the key from Cloud Storage. 

283 

284 :type timeout: float or tuple 

285 :param timeout: 

286 (Optional) The amount of time, in seconds, to wait 

287 for the server response. See: :ref:`configuring_timeouts` 

288 

289 :type retry: google.api_core.retry.Retry or google.cloud.storage.retry.ConditionalRetryPolicy 

290 :param retry: 

291 (Optional) How to retry the RPC. See: :ref:`configuring_retries` 

292 

293 :raises :class:`~google.api_core.exceptions.NotFound`: 

294 if the key does not exist on the back-end. 

295 """ 

296 with create_trace_span(name="Storage.HmacKey.delete"): 

297 qs_params = {} 

298 if self.user_project is not None: 

299 qs_params["userProject"] = self.user_project 

300 

301 self._client._delete_resource( 

302 self.path, 

303 query_params=qs_params, 

304 timeout=timeout, 

305 retry=retry, 

306 )