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 os
20import re
21import stat
22import time
23from urllib.parse import quote, urlparse
24import warnings
26from google.auth import environment_vars, exceptions
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)
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]
41_WELL_KNOWN_CERT_PATH = "/var/run/secrets/workload-spiffe-credentials/certificates.pem"
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
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)
54_POLLING_INTERVALS = ([_FAST_POLL_INTERVAL] * _FAST_POLL_CYCLES) + (
55 [_SLOW_POLL_INTERVAL] * _SLOW_POLL_CYCLES
56)
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 (e.g., return early or fallback)
71 raise
72 except OSError:
73 return False
76def get_agent_identity_certificate_path():
77 """Gets the agent certificate path from the certificate config file.
79 The path to the certificate config file is read from the
80 GOOGLE_API_CERTIFICATE_CONFIG environment variable. This function
81 can optionally trigger polling to handle cases where the environment
82 variable is set before the files are available on the filesystem.
84 Returns:
85 Optional[str]: The path to the agent's certificate file, or None if unavailable.
87 Raises:
88 google.auth.exceptions.RefreshError: If the certificate config file
89 or the certificate file cannot be found after retries.
90 """
91 cert_config_path = os.environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
93 if not cert_config_path:
94 return None
96 # We trigger polling only if the config path points to the well-known directory.
97 # Cloud Run dynamically generates these files in this directory, and both the
98 # config file and the certificate file may experience a brief startup latency.
99 # For all other paths, we return early to avoid introducing unnecessary startup
100 # delays.
101 well_known_dir = os.path.dirname(_WELL_KNOWN_CERT_PATH)
102 try:
103 abs_cert_path = os.path.abspath(cert_config_path)
104 abs_well_known_dir = os.path.abspath(well_known_dir)
105 should_poll = (
106 os.path.commonpath([abs_well_known_dir, abs_cert_path])
107 == abs_well_known_dir
108 )
109 except ValueError:
110 should_poll = False
112 return _get_cert_path_with_optional_polling(cert_config_path, should_poll)
115def _get_cert_path_with_optional_polling(cert_config_path, should_poll):
116 """Gets the certificate path, optionally polling until it is ready.
118 Args:
119 cert_config_path (str): The path to the certificate configuration file.
120 should_poll (bool): If True, the function will poll for the file and
121 certificate to be ready. If False, it will check only once and
122 return early if they are not immediately available.
124 Returns:
125 str: The path to the certificate file, or None if unavailable.
127 Raises:
128 google.auth.exceptions.RefreshError: If the certificate config file
129 or the certificate file cannot be found after retries.
130 """
131 has_logged_config_warning = False
132 has_logged_cert_warning = False
134 for interval in _POLLING_INTERVALS:
135 try:
136 cert_path = _parse_cert_path_from_config(cert_config_path)
138 if cert_path is None:
139 return None
141 if _is_certificate_file_ready(cert_path):
142 return cert_path
144 # The config was parsed, but the cert file is not ready yet
145 if not should_poll:
146 # If polling is disabled, return early.
147 return None
149 if not has_logged_cert_warning:
150 warnings.warn(
151 f"Certificate file not ready at {cert_path}. Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
152 )
153 has_logged_cert_warning = True
155 except PermissionError as e:
156 warnings.warn(
157 f"Permission denied when accessing certificate config or certificate file: {e}. "
158 "Token binding protection cannot be enabled. Falling back to unbound tokens."
159 )
160 return None
161 except (IOError, ValueError, KeyError) as e:
162 if os.path.exists(cert_config_path):
163 # If the file exists but has invalid JSON or is unreadable,
164 # we assume it is in its final format and return early (returning None).
165 return None
167 if not should_poll:
168 # If polling is disabled, return early if the file doesn't exist.
169 return None
171 if not has_logged_config_warning:
172 warnings.warn(
173 f"Certificate config file not found or incomplete: {e} (from "
174 f"{environment_vars.GOOGLE_API_CERTIFICATE_CONFIG} environment variable). "
175 f"Retrying until startup timeout (up to {_TOTAL_TIMEOUT} seconds total)..."
176 )
177 has_logged_config_warning = True
179 # Sleep before the next polling attempt.
180 time.sleep(interval)
182 raise exceptions.RefreshError(
183 "Certificate config or certificate file not found after multiple retries. "
184 f"Token binding protection is failing. You can turn off this protection by setting "
185 f"{environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES} to false "
186 "to fall back to unbound tokens."
187 )
190def _parse_cert_path_from_config(cert_config_path):
191 """Reads the cert config file and returns the cert_path.
193 Args:
194 cert_config_path (str): The path to the certificate configuration file.
196 Returns:
197 Optional[str]: The path to the certificate file, or None if not found
198 in the config.
200 Raises:
201 IOError: If the certificate config file cannot be read.
202 ValueError: If the certificate config file contains invalid JSON.
203 KeyError: If the certificate config file does not contain the
204 expected structure.
205 """
206 import json
208 with open(cert_config_path, "r", encoding="utf-8") as f:
209 cert_config = json.load(f)
211 cert_configs = (
212 cert_config.get("cert_configs") if isinstance(cert_config, dict) else None
213 )
214 workload_config = (
215 cert_configs.get("workload") if isinstance(cert_configs, dict) else None
216 )
218 if not isinstance(workload_config, dict) or "cert_path" not in workload_config:
219 return None
221 return workload_config["cert_path"]
224def get_and_parse_agent_identity_certificate():
225 """Gets and parses the agent identity certificate if not opted out.
227 Checks if the user has opted out of certificate-bound tokens. If not,
228 it gets the certificate path, reads the file, and parses it.
230 Returns:
231 The parsed certificate object if found and not opted out, otherwise None.
232 """
233 # If the user has opted out of cert bound tokens, there is no need to
234 # look up the certificate.
235 is_opted_out = (
236 os.environ.get(
237 environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
238 "true",
239 ).lower()
240 == "false"
241 )
242 if is_opted_out:
243 return None
245 # Respect explicit opt-out of mTLS / client certs
246 from google.auth.transport import _mtls_helper
248 env_override = _mtls_helper._check_use_client_cert_env()
249 if env_override is False:
250 return None
252 cert_path = get_agent_identity_certificate_path()
253 if not cert_path:
254 return None
256 try:
257 with open(cert_path, "rb") as cert_file:
258 cert_bytes = cert_file.read()
259 except PermissionError as e:
260 warnings.warn(
261 f"Failed to read agent identity certificate file at {cert_path}: {e}. "
262 "Token binding protection cannot be enabled. Falling back to unbound tokens."
263 )
264 return None
266 return parse_certificate(cert_bytes)
269def parse_certificate(cert_bytes):
270 """Parses a PEM-encoded certificate.
272 Args:
273 cert_bytes (bytes): The PEM-encoded certificate bytes.
275 Returns:
276 cryptography.x509.Certificate: The parsed certificate object.
277 """
278 try:
279 from cryptography import x509
281 return x509.load_pem_x509_certificate(cert_bytes)
282 except ImportError as e:
283 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
286def _is_agent_identity_certificate(cert):
287 """Checks if a certificate is an Agent Identity certificate.
289 This is determined by checking the Subject Alternative Name (SAN) for a
290 SPIFFE ID with a trust domain matching Agent Identity patterns.
292 Args:
293 cert (cryptography.x509.Certificate): The parsed certificate object.
295 Returns:
296 bool: True if the certificate is an Agent Identity certificate,
297 False otherwise.
298 """
299 try:
300 from cryptography import x509
301 from cryptography.x509.oid import ExtensionOID
303 try:
304 ext = cert.extensions.get_extension_for_oid(
305 ExtensionOID.SUBJECT_ALTERNATIVE_NAME
306 )
307 except x509.ExtensionNotFound:
308 return False
309 uris = ext.value.get_values_for_type(x509.UniformResourceIdentifier)
311 for uri in uris:
312 parsed_uri = urlparse(uri)
313 if parsed_uri.scheme == "spiffe":
314 trust_domain = parsed_uri.netloc
315 for pattern in _AGENT_IDENTITY_SPIFFE_TRUST_DOMAIN_PATTERNS:
316 if re.match(pattern, trust_domain):
317 return True
318 return False
319 except ImportError as e:
320 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
323def calculate_certificate_fingerprint(cert):
324 """Calculates the URL-encoded, unpadded, base64-encoded SHA256 hash of a
325 DER-encoded certificate.
327 Args:
328 cert (cryptography.x509.Certificate): The parsed certificate object.
330 Returns:
331 str: The URL-encoded, unpadded, base64-encoded SHA256 fingerprint.
332 """
333 try:
334 from cryptography.hazmat.primitives import serialization
336 der_cert = cert.public_bytes(serialization.Encoding.DER)
337 fingerprint = hashlib.sha256(der_cert).digest()
338 # The certificate fingerprint is generated in two steps to align with GFE's
339 # expectations and ensure proper URL transmission:
340 # 1. Standard base64 encoding is applied, and padding ('=') is removed.
341 # 2. The resulting string is then URL-encoded to handle special characters
342 # ('+', '/') that would otherwise be misinterpreted in URL parameters.
343 base64_fingerprint = base64.b64encode(fingerprint).decode("utf-8")
344 unpadded_base64_fingerprint = base64_fingerprint.rstrip("=")
345 return quote(unpadded_base64_fingerprint)
346 except ImportError as e:
347 raise ImportError(CRYPTOGRAPHY_NOT_FOUND_ERROR) from e
350def should_request_bound_token(cert):
351 """Determines if a bound token should be requested.
353 This is based on the GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES
354 environment variable and whether the certificate is an agent identity cert.
356 Args:
357 cert (cryptography.x509.Certificate): The parsed certificate object.
359 Returns:
360 bool: True if a bound token should be requested, False otherwise.
361 """
362 is_agent_cert = _is_agent_identity_certificate(cert)
363 is_opted_in = (
364 os.environ.get(
365 environment_vars.GOOGLE_API_PREVENT_AGENT_TOKEN_SHARING_FOR_GCP_SERVICES,
366 "true",
367 ).lower()
368 == "true"
369 )
370 if not (is_agent_cert and is_opted_in):
371 return False
373 # Respect explicit opt-out of mTLS / client certs
374 from google.auth.transport import _mtls_helper
376 env_override = _mtls_helper._check_use_client_cert_env()
377 if env_override is False:
378 return False
380 return True
383def get_cached_cert_fingerprint(cached_cert):
384 """Returns the fingerprint of the cached certificate."""
385 if cached_cert:
386 cert_obj = parse_certificate(cached_cert)
387 cached_cert_fingerprint = calculate_certificate_fingerprint(cert_obj)
388 else:
389 raise ValueError("mTLS connection is not configured.")
390 return cached_cert_fingerprint