Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
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.
15"""Helpers for Agent Identity credentials."""
17import base64
18import hashlib
19import json
20import os
21import re
22import stat
23import time
24import warnings
25from urllib.parse import quote, urlparse
27from google.auth import environment_vars, exceptions
29CRYPTOGRAPHY_NOT_FOUND_ERROR = (
30 "The cryptography library is required for certificate-based authentication."
31 "Please install it with `pip install google-auth[cryptography]`."
32)
34# SPIFFE trust domain patterns for Agent Identities.
35_AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS = [
36 r"^agents\.global\.org-\d+\.system\.id\.goog$",
37 r"^agents\.global\.proj-\d+\.system\.id\.goog$",
38 r"^agents-nonprod\.global\.org-\d+\.system\.id\.goog$",
39 r"^agents-nonprod\.global\.proj-\d+\.system\.id\.goog$",
40]
42_WELL_KNOWN_CERT_PATH = "/var/run/secrets/workload-spiffe-credentials/certificates.pem"
44# Constants for polling the certificate file.
45_FAST_POLL_CYCLES = 50
46_FAST_POLL_INTERVAL = 0.1 # 100ms
47_SLOW_POLL_INTERVAL = 0.5 # 500ms
48_TOTAL_TIMEOUT = 30 # seconds
50# Calculate the number of slow poll cycles based on the total timeout.
51_SLOW_POLL_CYCLES = int(
52 (_TOTAL_TIMEOUT - (_FAST_POLL_CYCLES * _FAST_POLL_INTERVAL)) / _SLOW_POLL_INTERVAL
53)
55_POLLING_INTERVALS = ([_FAST_POLL_INTERVAL] * _FAST_POLL_CYCLES) + (
56 [_SLOW_POLL_INTERVAL] * _SLOW_POLL_CYCLES
57)
60def _is_certificate_file_ready(path):
61 """Checks if a file exists, is a regular file, and is not empty."""
62 if not path:
63 return False
64 try:
65 # Check if the path points to a regular file and is not empty.
66 # stat.S_ISREG is used instead of os.path.isfile to avoid swallowing
67 # PermissionError exceptions, which the caller needs to propagate.
68 st = os.stat(path)
69 return stat.S_ISREG(st.st_mode) and st.st_size > 0
70 except PermissionError:
71 # Propagate PermissionError to let caller handle it (e.g., return early or fallback)
72 raise
73 except OSError:
74 return False
77def _is_in_well_known_dir(path):
78 """Checks if the given path is inside the well-known Agent Identity directory."""
79 if not path:
80 return False
81 well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH)
82 try:
83 real_path = os.path.realpath(path)
84 real_well_known_dir = os.path.realpath(well_known_dir)
85 return (
86 os.path.commonpath([real_well_known_dir, real_path]) == real_well_known_dir
87 )
88 except ValueError:
89 return False
92def get_agent_identity_certificate_path():
93 """Gets the agent certificate path from the certificate config file or GKE bundle.
95 The path to the certificate config file is read from the
96 GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function
97 can optionally trigger polling to handle cases where the environment
98 variable is set before the files are available on the filesystem.
99 When GOOGLE_API_CERTIFICATE_CONFIG and
100 CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH are unset and no
101 implicit certificate_config.json is present, it falls back to checking
102 the GKE workload credential bundle path without polling.
104 Returns:
105 Optional[str]: The path to the agent's certificate file, or None if unavailable.
107 Raises:
108 google.auth.exceptions.RefreshError: If the certificate config file
109 or the certificate file cannot be found after retries.
110 """
111 from google.auth.transport import _mtls_helper
113 cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
115 if not cert_config_path:
116 if (
117 not _mtls_helper._has_explicit_cert_config_env()
118 and _mtls_helper._should_use_gke_credential_bundle(
119 _mtls_helper._get_cert_config_path()
120 )
121 ):
122 return _mtls_helper._GKE_CREDENTIAL_BUNDLE_PATH
123 return None
125 # We trigger polling only if the config path points to the well-known directory.
126 # Cloud Run dynamically generates these files in this directory, and both the
127 # config file and the certificate file may experience a brief startup latency.
128 # For all other paths, we return early to avoid introducing unnecessary startup
129 # delays.
130 should_poll = _is_in_well_known_dir(cert_config_path)
132 return _get_cert_path_with_optional_polling(cert_config_path, should_poll)
135def _get_cert_path_with_optional_polling(cert_config_path, should_poll):
136 """Gets the certificate path, optionally polling until it is ready.
138 Args:
139 cert_config_path (str): The path to the certificate configuration file.
140 should_poll (bool): If True, the function will poll for the file and
141 certificate to be ready. If False, it will check only once and
142 return early if they are not immediately available.
144 Returns:
145 str: The path to the certificate file, or None if unavailable.
147 Raises:
148 google.auth.exceptions.RefreshError: If the certificate config file
149 or the certificate file cannot be found after retries.
150 """
151 has_logged_config_warning = False
152 has_logged_cert_warning = False
154 for interval in _POLLING_INTERVALS:
155 try:
156 cert_path = _parse_cert_path_from_config(cert_config_path)
158 if cert_path is None:
159 return None
161 if _is_certificate_file_ready(cert_path):
162 return cert_path
164 # The config was parsed, but the cert file is not ready yet.
165 # Only poll if both the config path and cert path are in the well-known directory.
166 if not (should_poll and _is_in_well_known_dir(cert_path)):
167 return None
169 if not has_logged_cert_warning:
170 warnings.warn(
171 f"Certificate file not ready at {cert_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
172 )
173 has_logged_cert_warning = True
175 except PermissionError as e:
176 warnings.warn(
177 f"Permission denied when accessing certificate config or certificate file: {e}. "
178 "Token binding protection cannot be enabled. Falling back to unbound tokens."
179 )
180 return None
181 except (IOError, ValueError, KeyError) as e:
182 if os.path.exists(cert_config_path):
183 # If the file exists but has invalid JSON or is unreadable,
184 # we assume it is in its final format and return early (returning None).
185 return None
187 if not should_poll:
188 # If polling is disabled, return early if the file doesn't exist.
189 return None
191 if not has_logged_config_warning:
192 warnings.warn(
193 f"Certificate config file not found or incomplete: {e} (from "
194 f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). "
195 f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
196 )
197 has_logged_config_warning = True
199 # Sleep before the next polling attempt.
200 time.sleep(interval)
202 raise exceptions.RefreshError(
203 "Certificate config or certificate file not found after multiple retries. "
204 f"Token binding protection is failing. You can turn off this protection by setting "
205 f"{environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN} to false "
206 "to fall back to unbound tokens."
207 )
210def _parse_cert_path_from_config(cert_config_path):
211 """Reads the cert config file and returns the cert_path.
213 Args:
214 cert_config_path (str): The path to the certificate configuration file.
216 Returns:
217 Optional[str]: The path to the certificate file, or None if not found
218 in the config.
220 Raises:
221 IOError: If the certificate config file cannot be read.
222 ValueError: If the certificate config file contains invalid JSON.
223 KeyError: If the certificate config file does not contain the
224 expected structure.
225 """
226 with open(cert_config_path, "r", encoding="utf-8") as f:
227 cert_config = json.load(f)
229 cert_configs = (
230 cert_config.get("cert_configs") if isinstance(cert_config, dict) else None
231 )
232 workload_config = (
233 cert_configs.get("workload") if isinstance(cert_configs, dict) else None
234 )
236 if not isinstance(workload_config, dict) or "cert_path" not in workload_config:
237 return None
239 return workload_config["cert_path"]
242def _is_bound_token_opted_out():
243 """Returns True only if bound tokens are explicitly disabled via env vars."""
244 val = os.environ.get(
245 environment_vars.GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN, ""
246 ).strip()
247 if not val:
248 # Fall back to the deprecated env var for backward compatibility
249 val = os.environ.get(
250 environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
251 "true",
252 ).strip()
253 return val.lower() == "false"
256def get_agent_identity_certificate_and_bytes():
257 """Gets and parses the agent identity certificate if not opted out.
259 Checks if the user has opted out of certificate-bound tokens. If not,
260 it gets the certificate path, reads the file, and parses it.
262 Returns:
263 Tuple[Optional[cryptography.x509.Certificate], Optional[bytes]]: A tuple
264 of (parsed certificate object, certificate bytes) if found and not
265 opted out, otherwise (None, None).
266 """
267 from google.auth.transport import _mtls_helper
269 # If the user has opted out of cert bound tokens, there is no need to
270 # look up the certificate.
271 if _is_bound_token_opted_out():
272 return None, None
274 # Respect explicit opt-out of mTLS / client certs
275 env_override = _mtls_helper._check_use_client_cert_env()
276 if env_override is False:
277 return None, None
279 cert_path = get_agent_identity_certificate_path()
280 if not cert_path:
281 return None, None
283 try:
284 with open(cert_path, "rb") as cert_file:
285 raw_bytes = cert_file.read()
286 except OSError as e:
287 warnings.warn(
288 f"Failed to read agent identity certificate file at {cert_path}: {e}. "
289 "Token binding protection cannot be enabled. Falling back to unbound tokens."
290 )
291 return None, None
293 cert_blocks = _mtls_helper._CERT_REGEX.findall(raw_bytes)
294 if not cert_blocks:
295 warnings.warn(
296 f"No PEM certificate blocks found in {cert_path}. "
297 "Token binding protection cannot be enabled. Falling back to unbound tokens."
298 )
299 return None, None
301 cert_bytes = b"\n".join(block.strip() for block in cert_blocks) + b"\n"
302 try:
303 return parse_certificate(raw_bytes), cert_bytes
304 except (ValueError, ImportError) as e:
305 warnings.warn(
306 f"Failed to parse agent identity certificate at {cert_path}: {e}. "
307 "Token binding protection cannot be enabled. Falling back to unbound tokens."
308 )
309 return None, None
312def parse_certificate(cert_bytes):
313 """Parses a PEM-encoded certificate or certificate chain and returns the leaf certificate.
315 Args:
316 cert_bytes (bytes): The PEM-encoded certificate bytes.
318 Returns:
319 cryptography.x509.Certificate: The leaf (first) parsed certificate object.
321 Raises:
322 ValueError: If no certificates are found or any certificate in the chain
323 is malformed.
324 ImportError: If the cryptography library is not installed.
325 """
326 from google.auth.transport import _mtls_helper
328 if not cert_bytes:
329 raise ValueError("Certificate bytes cannot be empty or None.")
331 try:
332 from cryptography import x509
334 cert_blocks = _mtls_helper._CERT_REGEX.findall(cert_bytes)
335 if not cert_blocks:
336 return x509.load_pem_x509_certificate(cert_bytes)
337 if _mtls_helper._has_unmatched_pem_markers(cert_bytes, cert_blocks):
338 raise ValueError("Malformed or truncated PEM certificate chain.")
339 certs = [x509.load_pem_x509_certificate(block) for block in cert_blocks]
340 return certs[0]
341 except ImportError as e:
342 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
345def _is_agent_identity_certificate(cert):
346 """Checks if a certificate is an Agent Identity certificate.
348 This is determined by checking the Subject Alternative Name (SAN) for a
349 SPIFFE ID with a trust domain matching Agent Identity patterns.
351 Args:
352 cert (cryptography.x509.Certificate): The parsed certificate object.
354 Returns:
355 bool: True if the certificate is an Agent Identity certificate,
356 False otherwise.
357 """
358 try:
359 from cryptography import x509
360 from cryptography.x509.oid import ExtensionOID
362 try:
363 ext = cert.extensions.get_extension_for_oid(
364 ExtensionOID.SUBJECT_ALTERNATIVE_NAME
365 )
366 except x509.ExtensionNotFound:
367 return False
368 uris = ext.value.get_values_for_type(x509.UniformResourceIdentifier)
370 for uri in uris:
371 parsed_uri = urlparse(uri)
372 if parsed_uri.scheme == "spiffe":
373 trust_domain = parsed_uri.netloc
374 for pattern in _AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS:
375 if re.match(pattern, trust_domain):
376 return True
377 return False
378 except ImportError as e:
379 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
382def calculate_certificate_fingerprint(cert):
383 """Calculates the URL-encoded, unpadded, base64-encoded SHA256 hash of a
384 DER-encoded certificate.
386 Args:
387 cert (cryptography.x509.Certificate): The parsed certificate object.
389 Returns:
390 str: The URL-encoded, unpadded, base64-encoded SHA256 fingerprint.
391 """
392 try:
393 from cryptography.hazmat.primitives import serialization
395 der_cert = cert.public_bytes(serialization.Encoding.DER)
396 fingerprint = hashlib.sha256(der_cert).digest()
397 # The certificate fingerprint is generated in two steps to align with GFE's
398 # expectations and ensure proper URL transmission:
399 # 1. Standard base64 encoding is applied, and padding ('=') is removed.
400 # 2. The resulting string is then URL-encoded to handle special characters
401 # ('+', '/') that would otherwise be misinterpreted in URL parameters.
402 base64_fingerprint = base64.b64encode(fingerprint).decode("utf-8")
403 unpadded_base64_fingerprint = base64_fingerprint.rstrip("=")
404 return quote(unpadded_base64_fingerprint)
405 except ImportError as e:
406 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
409def should_request_bound_token(cert):
410 """Determines if a bound token should be requested.
412 This is based on the GOOGLE_API_ENABLE_RUNTIME_BOUND_TOKEN env var
413 (falls back to the deprecated GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
414 if unset or empty) and whether the certificate is an agent identity cert.
416 Args:
417 cert (cryptography.x509.Certificate): The parsed certificate object.
419 Returns:
420 bool: True if a bound token should be requested, False otherwise.
421 """
422 from google.auth.transport import _mtls_helper
424 if _is_bound_token_opted_out():
425 return False
427 # Respect explicit opt-out of mTLS / client certs
428 env_override = _mtls_helper._check_use_client_cert_env()
429 if env_override is False:
430 return False
432 return _is_agent_identity_certificate(cert)
435def get_cached_cert_fingerprint(cached_cert):
436 """Returns the fingerprint of the cached certificate."""
437 if cached_cert:
438 cert_obj = parse_certificate(cached_cert)
439 cached_cert_fingerprint = calculate_certificate_fingerprint(cert_obj)
440 else:
441 raise ValueError("mTLS connection is not configured.")
442 return cached_cert_fingerprint