Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/_agent_identity_utils.py: 19%

1# Copyright 2025 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"""Helpers for Agent Identity credentials.""" 

16 

17import base64 

18import hashlib 

19import os 

20import re 

21import stat 

22import time 

23from urllib.parse import quote, urlparse 

24import warnings 

25 

26from google.auth import environment_vars, exceptions 

27 

28CRYPTOGRAPHY_NOT_FOUND_ERROR = ( 

29 "The cryptography library is required for certificate-based authentication." 

30 "Please install it with `pip install google-auth[cryptography]`." 

31) 

32 

33# SPIFFE trust domain patterns for Agent Identities. 

34_AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS = [ 

35 r"^agents\.global\.org-\d+\.system\.id\.goog$", 

36 r"^agents\.global\.proj-\d+\.system\.id\.goog$", 

37 r"^agents-nonprod\.global\.org-\d+\.system\.id\.goog$", 

38 r"^agents-nonprod\.global\.proj-\d+\.system\.id\.goog$", 

39] 

40 

41_WELL_KNOWN_CERT_PATH = "/var/run/secrets/workload-spiffe-credentials/certificates.pem" 

42 

43# Constants for polling the certificate file. 

44_FAST_POLL_CYCLES = 50 

45_FAST_POLL_INTERVAL = 0.1 # 100ms 

46_SLOW_POLL_INTERVAL = 0.5 # 500ms 

47_TOTAL_TIMEOUT = 30 # seconds 

48 

49# Calculate the number of slow poll cycles based on the total timeout. 

50_SLOW_POLL_CYCLES = int( 

51 (_TOTAL_TIMEOUT - (_FAST_POLL_CYCLES * _FAST_POLL_INTERVAL)) / _SLOW_POLL_INTERVAL 

52) 

53 

54_POLLING_INTERVALS = ([_FAST_POLL_INTERVAL] * _FAST_POLL_CYCLES) + ( 

55 [_SLOW_POLL_INTERVAL] * _SLOW_POLL_CYCLES 

56) 

57 

58 

59def _is_certificate_file_ready(path): 

60 """Checks if a file exists, is a regular file, and is not empty.""" 

61 if not path: 

62 return False 

63 try: 

64 # Check if the path points to a regular file and is not empty. 

65 # stat.S_ISREG is used instead of os.path.isfile to avoid swallowing 

66 # PermissionError exceptions, which the caller needs to propagate. 

67 st = os.stat(path) 

68 return stat.S_ISREG(st.st_mode) and st.st_size > 0 

69 except PermissionError: 

70 # Propagate PermissionError to let caller handle it (fail-fast or fallback) 

71 raise 

72 except OSError: 

73 return False 

74 

75 

76def get_agent_identity_certificate_path(): 

77 """Gets the certificate path from the certificate config file. 

78 

79 The path to the certificate config file is read from the 

80 GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function 

81 implements a retry mechanism to handle cases where the environment 

82 variable is set before the files are available on the filesystem. 

83 

84 Returns: 

85 str: The path to the leaf certificate file. 

86 

87 Raises: 

88 google.auth.exceptions.RefreshError: If the certificate config file 

89 or the certificate file cannot be found after retries. 

90 """ 

91 import json 

92 

93 cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) 

94 

95 # Check if the well-known workload directory is mounted. 

96 well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH) 

97 has_well_known_dir = os.path.exists(well_known_dir) 

98 

99 # If we have neither a config path nor a well-known mount directory, exit immediately. 

100 if not cert_config_path and not has_well_known_dir: 

101 return None 

102 

103 # If ECP config path is specified but does not exist, and we are on a workstation, fail-fast immediately. 

104 if ( 

105 cert_config_path 

106 and not has_well_known_dir 

107 and not os.path.exists(cert_config_path) 

108 ): 

109 return None 

110 

111 has_logged_config_warning = False 

112 has_logged_cert_warning = False 

113 

114 for interval in _POLLING_INTERVALS: 

115 try: 

116 # Path A: Config file is explicitly set 

117 if cert_config_path: 

118 with open(cert_config_path, "r") as f: 

119 cert_config = json.load(f) 

120 

121 cert_configs = ( 

122 cert_config.get("cert_configs") 

123 if isinstance(cert_config, dict) 

124 else None 

125 ) 

126 workload_config = ( 

127 cert_configs.get("workload") 

128 if isinstance(cert_configs, dict) 

129 else None 

130 ) 

131 

132 if ( 

133 not isinstance(workload_config, dict) 

134 or "cert_path" not in workload_config 

135 ): 

136 return None 

137 

138 cert_path = workload_config["cert_path"] 

139 if _is_certificate_file_ready(cert_path): 

140 return cert_path 

141 

142 # The config was parsed, but the cert file is not ready yet 

143 target_path = cert_path 

144 

145 # Path B: Config is NOT set, fallback to the well-known path 

146 else: 

147 if _is_certificate_file_ready(_WELL_KNOWN_CERT_PATH): 

148 return _WELL_KNOWN_CERT_PATH 

149 

150 # The well-known cert file is not ready yet 

151 target_path = _WELL_KNOWN_CERT_PATH 

152 

153 # Log a warning on the first failed attempt to load the certificate file 

154 if not has_logged_cert_warning: 

155 warnings.warn( 

156 f"Certificate file not ready at {target_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..." 

157 ) 

158 has_logged_cert_warning = True 

159 

160 except PermissionError as e: 

161 warnings.warn( 

162 f"Permission denied when accessing certificate config or certificate file: {e}. " 

163 "Token binding protection cannot be enabled. Falling back to unbound tokens." 

164 ) 

165 return None 

166 except (IOError, ValueError, KeyError) as e: 

167 if cert_config_path and os.path.exists(cert_config_path): 

168 # If the file exists but has invalid JSON or is unreadable, 

169 # we assume it is in its final format and fail-fast by returning None. 

170 return None 

171 

172 if not has_logged_config_warning and cert_config_path: 

173 warnings.warn( 

174 f"Certificate config file not found or incomplete: {e} (from " 

175 f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). " 

176 f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..." 

177 ) 

178 has_logged_config_warning = True 

179 pass 

180 

181 # A sleep is required in two cases: 

182 # 1. The config file is not found (the except block). 

183 # 2. The config file/well-known path is found, but the certificate is not yet available. 

184 # In both cases, we need to poll, so we sleep on every iteration 

185 # that doesn't return a certificate. 

186 time.sleep(interval) 

187 

188 raise exceptions.RefreshError( 

189 "Certificate config or certificate file not found after multiple retries. " 

190 f"Token binding protection is failing. You can turn off this protection by setting " 

191 f"{environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES} to false " 

192 "to fall back to unbound tokens." 

193 ) 

194 

195 

196def get_and_parse_agent_identity_certificate(): 

197 """Gets and parses the agent identity certificate if not opted out. 

198 

199 Checks if the user has opted out of certificate-bound tokens. If not, 

200 it gets the certificate path, reads the file, and parses it. 

201 

202 Returns: 

203 The parsed certificate object if found and not opted out, otherwise None. 

204 """ 

205 # If the user has opted out of cert bound tokens, there is no need to 

206 # look up the certificate. 

207 is_opted_out = ( 

208 os.environ.get( 

209 environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, 

210 "true", 

211 ).lower() 

212 == "false" 

213 ) 

214 if is_opted_out: 

215 return None 

216 

217 # Respect explicit opt-out of mTLS / client certs 

218 from google.auth.transport import _mtls_helper 

219 

220 env_override = _mtls_helper._check_use_client_cert_env() 

221 if env_override is False: 

222 return None 

223 

224 cert_path = get_agent_identity_certificate_path() 

225 if not cert_path: 

226 return None 

227 

228 try: 

229 with open(cert_path, "rb") as cert_file: 

230 cert_bytes = cert_file.read() 

231 except PermissionError as e: 

232 warnings.warn( 

233 f"Failed to read agent identity certificate file at {cert_path}: {e}. " 

234 "Token binding protection cannot be enabled. Falling back to unbound tokens." 

235 ) 

236 return None 

237 

238 return parse_certificate(cert_bytes) 

239 

240 

241def parse_certificate(cert_bytes): 

242 """Parses a PEM-encoded certificate. 

243 

244 Args: 

245 cert_bytes (bytes): The PEM-encoded certificate bytes. 

246 

247 Returns: 

248 cryptography.x509.Certificate: The parsed certificate object. 

249 """ 

250 try: 

251 from cryptography import x509 

252 

253 return x509.load_pem_x509_certificate(cert_bytes) 

254 except ImportError as e: 

255 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e 

256 

257 

258def _is_agent_identity_certificate(cert): 

259 """Checks if a certificate is an Agent Identity certificate. 

260 

261 This is determined by checking the Subject Alternative Name (SAN) for a 

262 SPIFFE ID with a trust domain matching Agent Identity patterns. 

263 

264 Args: 

265 cert (cryptography.x509.Certificate): The parsed certificate object. 

266 

267 Returns: 

268 bool: True if the certificate is an Agent Identity certificate, 

269 False otherwise. 

270 """ 

271 try: 

272 from cryptography import x509 

273 from cryptography.x509.oid import ExtensionOID 

274 

275 try: 

276 ext = cert.extensions.get_extension_for_oid( 

277 ExtensionOID.SUBJECT_ALTERNATIVE_NAME 

278 ) 

279 except x509.ExtensionNotFound: 

280 return False 

281 uris = ext.value.get_values_for_type(x509.UniformResourceIdentifier) 

282 

283 for uri in uris: 

284 parsed_uri = urlparse(uri) 

285 if parsed_uri.scheme == "spiffe": 

286 trust_domain = parsed_uri.netloc 

287 for pattern in _AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS: 

288 if re.match(pattern, trust_domain): 

289 return True 

290 return False 

291 except ImportError as e: 

292 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e 

293 

294 

295def calculate_certificate_fingerprint(cert): 

296 """Calculates the URL-encoded, unpadded, base64-encoded SHA256 hash of a 

297 DER-encoded certificate. 

298 

299 Args: 

300 cert (cryptography.x509.Certificate): The parsed certificate object. 

301 

302 Returns: 

303 str: The URL-encoded, unpadded, base64-encoded SHA256 fingerprint. 

304 """ 

305 try: 

306 from cryptography.hazmat.primitives import serialization 

307 

308 der_cert = cert.public_bytes(serialization.Encoding.DER) 

309 fingerprint = hashlib.sha256(der_cert).digest() 

310 # The certificate fingerprint is generated in two steps to align with GFE's 

311 # expectations and ensure proper URL transmission: 

312 # 1. Standard base64 encoding is applied, and padding ('=') is removed. 

313 # 2. The resulting string is then URL-encoded to handle special characters 

314 # ('+', '/') that would otherwise be misinterpreted in URL parameters. 

315 base64_fingerprint = base64.b64encode(fingerprint).decode("utf-8") 

316 unpadded_base64_fingerprint = base64_fingerprint.rstrip("=") 

317 return quote(unpadded_base64_fingerprint) 

318 except ImportError as e: 

319 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e 

320 

321 

322def should_request_bound_token(cert): 

323 """Determines if a bound token should be requested. 

324 

325 This is based on the GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES 

326 environment variable and whether the certificate is an agent identity cert. 

327 

328 Args: 

329 cert (cryptography.x509.Certificate): The parsed certificate object. 

330 

331 Returns: 

332 bool: True if a bound token should be requested, False otherwise. 

333 """ 

334 is_agent_cert = _is_agent_identity_certificate(cert) 

335 is_opted_in = ( 

336 os.environ.get( 

337 environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES, 

338 "true", 

339 ).lower() 

340 == "true" 

341 ) 

342 if not (is_agent_cert and is_opted_in): 

343 return False 

344 

345 # Respect explicit opt-out of mTLS / client certs 

346 from google.auth.transport import _mtls_helper 

347 

348 env_override = _mtls_helper._check_use_client_cert_env() 

349 if env_override is False: 

350 return False 

351 

352 return True 

353 

354 

355def get_cached_cert_fingerprint(cached_cert): 

356 """Returns the fingerprint of the cached certificate.""" 

357 if cached_cert: 

358 cert_obj = parse_certificate(cached_cert) 

359 cached_cert_fingerprint = calculate_certificate_fingerprint(cert_obj) 

360 else: 

361 raise ValueError("mTLS connection is not configured.") 

362 return cached_cert_fingerprint