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 2020 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"""Helper functions for getting mTLS cert and key."""
17import contextlib
18import json
19import logging
20import os
21import re
22import subprocess
23import sys
24import tempfile
25from os import environ, getenv, path
26from typing import Generator, List, Optional, Tuple, Union, cast
27from urllib.parse import urlsplit
29from google.auth import _agent_identity_utils, _cloud_sdk, environment_vars, exceptions
31CONTEXT_AWARE_METADATA_PATH = "~/.secureConnect/context_aware_metadata.json"
33# Default gcloud config path, to be used with path.expanduser for cross-platform compatibility.
34CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
35_GKE_CREDENTIAL_BUNDLE_PATH = "/var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem"
36_CERT_PROVIDER_COMMAND = "cert_provider_command"
37_CERT_REGEX = re.compile(
38 b"-----BEGIN CERTIFICATE-----.+?-----END CERTIFICATE-----\r?\n?", re.DOTALL
39)
41# support various format of key files, e.g.
42# "-----BEGIN PRIVATE KEY-----...",
43# "-----BEGIN EC PRIVATE KEY-----...",
44# "-----BEGIN RSA PRIVATE KEY-----..."
45# "-----BEGIN ENCRYPTED PRIVATE KEY-----"
46_KEY_REGEX = re.compile(
47 b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+?-----END [A-Z ]*PRIVATE KEY-----\r?\n?",
48 re.DOTALL,
49)
51_LOGGER = logging.getLogger(__name__)
54_PASSPHRASE_REGEX = re.compile(
55 b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
56)
59class _MemfdCreationError(OSError):
60 """Raised when Linux in-memory virtual file creation (memfd) fails."""
62 pass
65def _can_read(path: Optional[str]) -> bool:
66 if path is None:
67 return True
68 try:
69 with open(path, "rb"):
70 pass
71 return True
72 except OSError:
73 return False
76@contextlib.contextmanager
77def secure_cert_key_paths(
78 cert: Union[bytes, str, None],
79 key: Union[bytes, str, None],
80 passphrase: Optional[bytes] = None,
81) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
82 """Provides secure file paths for certificate and key.
84 This function is implemented as a context manager generator to ensure that
85 any temporary resources (such as in-memory virtual files or encrypted physical
86 temp files) are automatically cleaned up and securely wiped when the context exits.
88 It supports mixed inputs (e.g. passing one as a string path and the other as bytes).
89 If a parameter is already a string path or None, it is passed through as-is, and
90 only raw bytes are written to temporary storage.
92 Args:
93 cert (Union[str, bytes, None]): Certificate path, raw PEM content bytes, or None.
94 key (Union[str, bytes, None]): Private key path, raw PEM content bytes, or None.
95 passphrase (Optional[bytes]): Optional passphrase for the private key.
97 Yields:
98 Tuple[str, str, Optional[bytes]]: The certificate path, key path, and
99 the passphrase needed to load the key (either the user's original,
100 or the newly generated one if Tier 3 had to encrypt the key).
102 Raises:
103 OSError: If temporary file creation or writing fails during the Tier 3 fallback.
104 """
105 # Normalize PEM strings to bytes so they are written to temporary storage.
106 # We check for "-----BEGIN " to distinguish between file paths and PEM payloads.
107 if isinstance(cert, str) and "-----BEGIN " in cert:
108 cert = cert.encode("utf-8")
109 if isinstance(key, str) and "-----BEGIN " in key:
110 key = key.encode("utf-8")
112 # Tier 1: Pass-through (No-op). If the caller already provided file paths,
113 # we yield them directly to avoid any unnecessary file creation.
114 if isinstance(cert, str) and isinstance(key, str):
115 yield cert, key, passphrase
116 return
118 # If a value is a string path, it is passed through. If bytes, we will write
119 # it to temporary storage. None values are also passed through as-is.
120 cert_bytes = cert if isinstance(cert, bytes) else None
121 key_bytes = key if isinstance(key, bytes) else None
123 # Tier 2: Linux RAM-backed virtual files. If supported by the OS, we write
124 # the bytes to anonymous in-memory files using memfd_create. This yields
125 # /proc/self/fd/... paths, keeping the private key entirely in memory.
126 if sys.platform == "linux" and hasattr(os, "memfd_create"):
127 try:
128 with _memfd_cert_key_paths(cert_bytes, key_bytes) as (cert_path, key_path):
129 # Handle cases where path exists but might be restricted.
130 if (cert_path is None or os.path.exists(cert_path)) and (
131 key_path is None or os.path.exists(key_path)
132 ):
133 if _can_read(cert_path) and _can_read(key_path):
134 yield (
135 cast(str, cert_path or cert),
136 cast(str, key_path or key),
137 passphrase,
138 )
139 return
140 except _MemfdCreationError:
141 pass # Fallback to Tier 3 on failure.
143 # Tier 3: Fallback Encrypted Temp Files. If in-memory files are not supported
144 # (macOS/Windows), we write to disk. To protect the key, we encrypt plaintext
145 # keys on-the-fly and securely wipe the files with null bytes during cleanup.
146 with _tempfile_cert_key_paths(cert_bytes, key_bytes, passphrase) as (
147 cert_path,
148 key_path,
149 new_passphrase,
150 ):
151 yield cast(str, cert_path or cert), cast(str, key_path or key), new_passphrase
154def _encrypt_key_if_plaintext(
155 key_bytes: bytes, passphrase: Optional[bytes]
156) -> Tuple[bytes, Optional[bytes]]:
157 """Encrypts a plaintext PEM key if necessary, returning the bytes and passphrase.
159 If the key is already encrypted, or if parsing/encryption fails, the key is
160 returned as-is (plaintext) as a fallback. This allows the caller (underlying SSL
161 context) to attempt loading the key directly and handle any failures.
162 """
163 import secrets
165 import cryptography
166 from cryptography.hazmat.primitives import serialization
168 try:
169 pkey = serialization.load_pem_private_key(key_bytes, password=None)
170 # It's plaintext, encrypt it.
171 target_passphrase = passphrase
172 if target_passphrase is None:
173 target_passphrase = secrets.token_hex(32).encode("utf-8")
174 elif isinstance(target_passphrase, str):
175 target_passphrase = target_passphrase.encode("utf-8")
177 encrypted_content = pkey.private_bytes(
178 encoding=serialization.Encoding.PEM,
179 format=serialization.PrivateFormat.PKCS8,
180 encryption_algorithm=serialization.BestAvailableEncryption(
181 target_passphrase
182 ),
183 )
184 return encrypted_content, target_passphrase
185 except (ValueError, TypeError, cryptography.exceptions.UnsupportedAlgorithm):
186 # Likely already encrypted, invalid, or unsupported algorithm, return as-is.
187 return key_bytes, passphrase
190def _secure_wipe_and_remove(file_path: str):
191 """Overwrites a file with null bytes before deleting it.
193 This is an extra security measure to make file recovery harder. However, on modern
194 solid-state drives (SSDs), the hardware optimizes where data is written, meaning
195 the original private key bytes might still physically remain on the storage chips
196 until the drive cleans them up.
197 """
198 if not os.path.exists(file_path):
199 return
200 try:
201 size = os.path.getsize(file_path)
202 with open(file_path, "r+b") as f:
203 f.write(b"\0" * size)
204 f.flush()
205 os.fsync(f.fileno())
206 except OSError:
207 pass # Ignore permission/lock errors during cleanup.
208 finally:
209 try:
210 os.remove(file_path)
211 except OSError:
212 pass
215@contextlib.contextmanager
216def _memfd_cert_key_paths(
217 cert_bytes: Optional[bytes], key_bytes: Optional[bytes]
218) -> Generator[Tuple[Optional[str], Optional[str]], None, None]:
219 """Creates secure, in-memory virtual files on Linux using memfd_create.
221 Yields:
222 Tuple[Optional[str], Optional[str]]: In-memory file paths pointing to
223 the active descriptors (e.g., '/proc/self/fd/3').
224 """
225 cleanup_fds = []
226 paths: List[Optional[str]] = []
228 try:
229 try:
230 for data, name in [(cert_bytes, "mtls_cert"), (key_bytes, "mtls_key")]:
231 if data is not None:
232 # MFD_CLOEXEC prevents FD leaks to spawned subprocesses.
233 fd = os.memfd_create(name, os.MFD_CLOEXEC) # type: ignore[attr-defined]
234 cleanup_fds.append(fd)
235 with os.fdopen(fd, "wb", closefd=False) as f:
236 f.write(data)
237 paths.append(f"/proc/self/fd/{fd}")
238 else:
239 paths.append(None)
240 except (OSError, AttributeError) as exc:
241 raise _MemfdCreationError(
242 "Failed to create in-memory virtual files"
243 ) from exc
245 cert_path, key_path = paths
246 yield cert_path, key_path
247 finally:
248 # Closing the descriptors automatically frees the RAM allocation.
249 for fd in cleanup_fds:
250 try:
251 os.close(fd)
252 except OSError:
253 pass
256def _write_secure_tempfile(fd: int, data: bytes) -> None:
257 """Writes data to a file descriptor, securely flushes to disk, and closes it."""
258 try:
259 f = os.fdopen(fd, "wb")
260 except BaseException:
261 try:
262 os.close(fd)
263 except OSError:
264 pass
265 raise
267 with f:
268 f.write(data)
269 f.flush()
270 try:
271 os.fsync(f.fileno())
272 except OSError:
273 pass
276@contextlib.contextmanager
277def _tempfile_cert_key_paths(
278 cert_bytes: Optional[bytes],
279 key_bytes: Optional[bytes],
280 passphrase: Optional[bytes],
281) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
282 """Creates secure temporary file paths on disk, encrypting private keys.
284 Yields:
285 Tuple[Optional[str], Optional[str], Optional[bytes]]: The temporary file
286 paths and the passphrase needed to load the key.
287 """
288 # Prioritize RAM-backed /dev/shm to avoid writing secrets to physical storage.
289 tmp_dir = (
290 "/dev/shm"
291 if os.path.isdir("/dev/shm") and os.access("/dev/shm", os.W_OK)
292 else None
293 )
294 cleanup_files: List[Optional[str]] = [None, None]
295 new_passphrase = passphrase
296 cert_data = cert_bytes
297 key_data = None
298 if key_bytes is not None:
299 key_data, new_passphrase = _encrypt_key_if_plaintext(key_bytes, passphrase)
301 try:
302 for i, data in enumerate([cert_data, key_data]):
303 if data is not None:
304 try:
305 fd, path = tempfile.mkstemp(dir=tmp_dir)
306 except OSError:
307 fd, path = tempfile.mkstemp(dir=None)
309 cleanup_files[i] = path
310 _write_secure_tempfile(fd, data)
312 yield cleanup_files[0], cleanup_files[1], new_passphrase
313 finally:
314 cert_cleanup_path = cleanup_files[0]
315 key_cleanup_path = cleanup_files[1]
317 try:
318 if key_cleanup_path:
319 _secure_wipe_and_remove(key_cleanup_path)
320 except Exception:
321 pass
322 finally:
323 if cert_cleanup_path:
324 try:
325 if os.path.exists(cert_cleanup_path):
326 os.remove(cert_cleanup_path)
327 except OSError:
328 pass
331def _check_config_path(config_path):
332 """Checks for config file path. If it exists, returns the absolute path with user expansion;
333 otherwise returns None.
335 Args:
336 config_path (str): The config file path for either context_aware_metadata.json or certificate_config.json for example
338 Returns:
339 str: absolute path if exists and None otherwise.
340 """
341 config_path = path.expanduser(config_path)
342 if not path.exists(config_path):
343 _LOGGER.debug("%s is not found.", config_path)
344 return None
345 return config_path
348def _load_json_file(path):
349 """Reads and loads JSON from the given path. Used to read both X509 workload certificate and
350 secure connect configurations.
352 Args:
353 path (str): the path to read from.
355 Returns:
356 Dict[str, str]: The JSON stored at the file.
358 Raises:
359 google.auth.exceptions.ClientCertError: If failed to parse the file as JSON.
360 """
361 try:
362 with open(path) as f:
363 json_data = json.load(f)
364 except ValueError as caught_exc:
365 new_exc = exceptions.ClientCertError(caught_exc)
366 raise new_exc from caught_exc
368 return json_data
371def _has_explicit_cert_config_env():
372 """Returns True if an explicit certificate config environment variable is set."""
373 return bool(
374 environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG)
375 or environ.get(
376 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH
377 )
378 )
381def _should_use_gke_credential_bundle(config_file_path=None):
382 """Returns True if no certificate config is present and the GKE credential bundle file is present."""
383 if config_file_path is not None or _has_explicit_cert_config_env():
384 return False
385 try:
386 return _agent_identity_utils._is_certificate_file_ready(
387 _GKE_CREDENTIAL_BUNDLE_PATH
388 )
389 except PermissionError:
390 return False
393def _get_workload_cert_and_key(
394 certificate_config_path=None, include_context_aware=True
395):
396 """Read the workload identity cert and key files specified in the certificate config provided.
397 If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG"
398 first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json",
399 and finally fall back to the GKE workload credential bundle at
400 "/var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem"
401 when no explicit or implicit certificate_config.json is present.
403 Args:
404 certificate_config_path (string): The certificate config path. If no path is provided,
405 the environment variable will be checked first, then the well known gcloud location.
406 include_context_aware (bool): If context aware metadata path should be checked for the
407 SecureConnect mTLS configuration.
409 Returns:
410 Tuple[Optional[bytes], Optional[bytes]]: client certificate bytes in PEM format and key
411 bytes in PEM format.
413 Raises:
414 google.auth.exceptions.ClientCertError: if problems occurs when retrieving
415 the certificate or key information.
416 """
418 cert_path, key_path, config_file_path = _get_workload_cert_and_key_paths(
419 certificate_config_path, include_context_aware
420 )
422 if cert_path is None and key_path is None:
423 if certificate_config_path is None and _should_use_gke_credential_bundle(
424 config_file_path
425 ):
426 return _read_credential_bundle_file(_GKE_CREDENTIAL_BUNDLE_PATH)
427 return None, None
429 return _read_cert_and_key_files(cert_path, key_path)
432def _get_cert_config_path(certificate_config_path=None, include_context_aware=True):
433 """Get the certificate configuration path based on the following order:
435 1: Explicit override, if set
436 2: Environment variable, if set
437 3: Well-known location
439 Returns "None" if the selected config file does not exist.
441 Args:
442 certificate_config_path (string): The certificate config path. If provided, the well known
443 location and environment variable will be ignored.
444 include_context_aware (bool): If context aware metadata path should be checked for the
445 SecureConnect mTLS configuration.
447 Returns:
448 The absolute path of the certificate config file, and None if the file does not exist.
449 """
451 source = "function argument"
452 is_explicit = True
453 if certificate_config_path is None:
454 env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None)
455 if env_path is not None and env_path != "":
456 certificate_config_path = env_path
457 source = (
458 f"environment variable {environment_vars.GOOGLE_API_CERTIFICATE_CONFIG}"
459 )
460 else:
461 env_path = environ.get(
462 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
463 None,
464 )
465 if include_context_aware and env_path is not None and env_path != "":
466 certificate_config_path = env_path
467 source = f"environment variable {environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH}"
468 else:
469 certificate_config_path = os.path.join(
470 _cloud_sdk.get_config_path(), "certificate_config.json"
471 )
472 is_explicit = False
474 certificate_config_path = path.expanduser(certificate_config_path)
475 if not path.exists(certificate_config_path):
476 if is_explicit:
477 _LOGGER.debug(
478 "Certificate configuration file explicitly specified via %s at %s does not exist",
479 source,
480 certificate_config_path,
481 )
482 return None
483 return certificate_config_path
486def _get_workload_cert_and_key_paths(config_path=None, include_context_aware=True):
487 absolute_path = _get_cert_config_path(config_path, include_context_aware)
488 if absolute_path is None:
489 return None, None, None
491 data = _load_json_file(absolute_path)
493 if (
494 not isinstance(data, dict)
495 or "cert_configs" not in data
496 or not isinstance(data["cert_configs"], dict)
497 ):
498 raise exceptions.ClientCertError(
499 'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
500 absolute_path
501 )
502 )
503 cert_configs = data["cert_configs"]
505 # We return None, None if the expected workload fields are not present.
506 # The certificate config might be present for other types of connections (e.g. gECC),
507 # and we want to gracefully fallback to testing other mTLS configurations
508 # like SecureConnect instead of throwing an exception.
510 if (
511 not isinstance(cert_configs, dict) or "workload" not in cert_configs
512 ) and config_path is None:
513 default_home_path = path.expanduser(
514 os.path.join(
515 _cloud_sdk.get_config_path(),
516 "certificate_config.json",
517 )
518 )
519 if path.exists(default_home_path) and os.path.normpath(
520 default_home_path
521 ) != os.path.normpath(absolute_path):
522 try:
523 home_data = _load_json_file(default_home_path)
524 if isinstance(home_data, dict):
525 home_cert_configs = home_data.get("cert_configs")
526 if (
527 isinstance(home_cert_configs, dict)
528 and "workload" in home_cert_configs
529 ):
530 cert_configs = home_cert_configs
531 absolute_path = default_home_path
532 except (exceptions.ClientCertError, OSError):
533 pass
535 if not isinstance(cert_configs, dict) or "workload" not in cert_configs:
536 return None, None, absolute_path
537 workload = cert_configs["workload"]
539 if (
540 not isinstance(workload, dict)
541 or not isinstance(workload.get("cert_path"), str)
542 or not workload.get("cert_path")
543 or not isinstance(workload.get("key_path"), str)
544 or not workload.get("key_path")
545 ):
546 raise exceptions.ClientCertError(
547 'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
548 absolute_path
549 )
550 )
551 cert_path = workload["cert_path"]
552 key_path = workload["key_path"]
554 return cert_path, key_path, absolute_path
557def _read_credential_bundle_file(bundle_path):
558 """Reads a combined PEM credential bundle containing certificate(s) and a private key."""
559 # Read the bundle once so a certificate rotation on disk cannot pair an
560 # old certificate chain with a new private key.
561 with open(bundle_path, "rb") as bundle_file:
562 bundle_data = bundle_file.read()
564 return (
565 _extract_cert_chain(bundle_data, bundle_path),
566 _extract_private_key(bundle_data, bundle_path),
567 )
570def _read_cert_and_key_files(cert_path, key_path):
571 cert_data = _read_cert_file(cert_path)
572 key_data = _read_key_file(key_path)
574 return cert_data, key_data
577def _join_cert_chain(cert_match):
578 return b"".join(
579 m if m.endswith(b"\n") or i == len(cert_match) - 1 else m + b"\n"
580 for i, m in enumerate(cert_match)
581 )
584def _has_unmatched_pem_markers(pem_bytes, cert_blocks):
585 """Checks that every BEGIN/END CERTIFICATE marker belongs to a complete cert block."""
586 return len(cert_blocks) != pem_bytes.count(b"-----BEGIN CERTIFICATE-----") or len(
587 cert_blocks
588 ) != pem_bytes.count(b"-----END CERTIFICATE-----")
591def _extract_cert_chain(cert_data, cert_path):
592 cert_match = re.findall(_CERT_REGEX, cert_data)
593 if not cert_match or _has_unmatched_pem_markers(cert_data, cert_match):
594 raise exceptions.ClientCertError(
595 "Certificate file {} is in an invalid format, at least one PEM formatted certificate is expected".format(
596 cert_path
597 )
598 )
599 return _join_cert_chain(cert_match)
602def _extract_private_key(key_data, key_path):
603 key_match = re.findall(_KEY_REGEX, key_data)
604 if len(key_match) != 1:
605 raise exceptions.ClientCertError(
606 "Private key file {} is in an invalid format, a single PEM formatted private key is expected".format(
607 key_path
608 )
609 )
611 return key_match[0]
614def _read_cert_file(cert_path):
615 with open(cert_path, "rb") as cert_file:
616 cert_data = cert_file.read()
618 return _extract_cert_chain(cert_data, cert_path)
621def _read_key_file(key_path):
622 with open(key_path, "rb") as key_file:
623 key_data = key_file.read()
625 return _extract_private_key(key_data, key_path)
628def _run_cert_provider_command(command, expect_encrypted_key=False):
629 """Run the provided command, and return client side mTLS cert, key and
630 passphrase.
632 Args:
633 command (List[str]): cert provider command.
634 expect_encrypted_key (bool): If encrypted private key is expected.
636 Returns:
637 Tuple[bytes, bytes, bytes]: client certificate bytes in PEM format, key
638 bytes in PEM format and passphrase bytes.
640 Raises:
641 google.auth.exceptions.ClientCertError: if problems occurs when running
642 the cert provider command or generating cert, key and passphrase.
643 """
644 try:
645 process = subprocess.Popen(
646 command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
647 )
648 stdout, stderr = process.communicate()
649 except OSError as caught_exc:
650 new_exc = exceptions.ClientCertError(caught_exc)
651 raise new_exc from caught_exc
653 # Check cert provider command execution error.
654 if process.returncode != 0:
655 raise exceptions.ClientCertError(
656 "Cert provider command returns non-zero status code %s" % process.returncode
657 )
659 # Extract certificate (chain), key and passphrase.
660 cert_match = re.findall(_CERT_REGEX, stdout)
661 if not cert_match or _has_unmatched_pem_markers(stdout, cert_match):
662 raise exceptions.ClientCertError("Client SSL certificate is missing or invalid")
663 cert_chain = _join_cert_chain(cert_match)
664 key_match = re.findall(_KEY_REGEX, stdout)
665 if len(key_match) != 1:
666 raise exceptions.ClientCertError("Client SSL key is missing or invalid")
667 passphrase_match = re.findall(_PASSPHRASE_REGEX, stdout)
669 if expect_encrypted_key:
670 if len(passphrase_match) != 1:
671 raise exceptions.ClientCertError("Passphrase is missing or invalid")
672 if b"ENCRYPTED" not in key_match[0]:
673 raise exceptions.ClientCertError("Encrypted private key is expected")
674 return cert_chain, key_match[0], passphrase_match[0].strip()
676 if b"ENCRYPTED" in key_match[0]:
677 raise exceptions.ClientCertError("Encrypted private key is not expected")
678 if len(passphrase_match) > 0:
679 raise exceptions.ClientCertError("Passphrase is not expected")
680 return cert_chain, key_match[0], None
683def get_client_ssl_credentials(
684 generate_encrypted_key=False,
685 context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
686 certificate_config_path=None,
687):
688 """Returns the client side certificate, private key and passphrase.
690 We look for certificates and keys with the following order of priority:
691 1. Certificate and key specified by certificate_config.json.
692 Currently, only X.509 workload certificates are supported.
693 2. Certificate and key specified by context aware metadata (i.e. SecureConnect).
695 Args:
696 generate_encrypted_key (bool): If set to True, encrypted private key
697 and passphrase will be generated; otherwise, unencrypted private key
698 will be generated and passphrase will be None. This option only
699 affects keys obtained via context_aware_metadata.json.
700 context_aware_metadata_path (str): The context_aware_metadata.json file path.
701 certificate_config_path (str): The certificate_config.json file path.
703 Returns:
704 Tuple[bool, bytes, bytes, bytes]:
705 A boolean indicating if cert, key and passphrase are obtained, the
706 cert bytes and key bytes both in PEM format, and passphrase bytes.
708 Raises:
709 google.auth.exceptions.ClientCertError: if problems occurs when getting
710 the cert, key and passphrase.
711 """
713 # 1. Attempt to retrieve X.509 Workload cert and key.
714 cert, key = _get_workload_cert_and_key(certificate_config_path)
715 if cert and key:
716 return True, cert, key, None
718 # 2. Check for context aware metadata json
719 metadata_path = _check_config_path(context_aware_metadata_path)
721 if metadata_path:
722 metadata_json = _load_json_file(metadata_path)
724 if _CERT_PROVIDER_COMMAND not in metadata_json:
725 raise exceptions.ClientCertError("Cert provider command is not found")
727 command = metadata_json[_CERT_PROVIDER_COMMAND]
729 if generate_encrypted_key and "--with_passphrase" not in command:
730 command.append("--with_passphrase")
732 # Execute the command.
733 cert, key, passphrase = _run_cert_provider_command(
734 command, expect_encrypted_key=generate_encrypted_key
735 )
736 return True, cert, key, passphrase
738 return False, None, None, None
741def get_client_cert_and_key(client_cert_callback=None):
742 """Returns the client side certificate and private key. The function first
743 tries to get certificate and key from client_cert_callback; if the callback
744 is None or doesn't provide certificate and key, the function tries application
745 default SSL credentials.
747 Args:
748 client_cert_callback (Optional[Callable[[], (bytes, bytes)]]): An
749 optional callback which returns client certificate bytes and private
750 key bytes both in PEM format.
752 Returns:
753 Tuple[bool, bytes, bytes]:
754 A boolean indicating if cert and key are obtained, the cert bytes
755 and key bytes both in PEM format.
757 Raises:
758 google.auth.exceptions.ClientCertError: if problems occurs when getting
759 the cert and key.
760 """
761 if client_cert_callback:
762 cert, key = client_cert_callback()
763 return True, cert, key
765 has_cert, cert, key, _ = get_client_ssl_credentials(generate_encrypted_key=False)
766 return has_cert, cert, key
769def decrypt_private_key(key, passphrase):
770 """A helper function to decrypt the private key with the given passphrase.
771 google-auth library doesn't support passphrase protected private key for
772 mutual TLS channel. This helper function can be used to decrypt the
773 passphrase protected private key in order to estalish mutual TLS channel.
775 For example, if you have a function which produces client cert, passphrase
776 protected private key and passphrase, you can convert it to a client cert
777 callback function accepted by google-auth::
779 from google.auth.transport import _mtls_helper
781 def your_client_cert_function():
782 return cert, encrypted_key, passphrase
784 # callback accepted by google-auth for mutual TLS channel.
785 def client_cert_callback():
786 cert, encrypted_key, passphrase = your_client_cert_function()
787 decrypted_key = _mtls_helper.decrypt_private_key(encrypted_key,
788 passphrase)
789 return cert, decrypted_key
791 Args:
792 key (bytes): The private key bytes in PEM format.
793 passphrase (bytes): The passphrase bytes.
795 Returns:
796 bytes: The decrypted private key in PEM format.
798 Raises:
799 ValueError: If there is any problem decrypting the private key.
800 """
801 if isinstance(key, str):
802 key = key.encode("utf-8")
803 if isinstance(passphrase, str):
804 passphrase = passphrase.encode("utf-8")
806 from cryptography.hazmat.primitives import serialization
808 # First convert encrypted_key_bytes to PKey object
809 pkey = serialization.load_pem_private_key(key, password=passphrase)
811 # Then dump the decrypted key bytes
812 return pkey.private_bytes(
813 encoding=serialization.Encoding.PEM,
814 format=serialization.PrivateFormat.PKCS8,
815 encryption_algorithm=serialization.NoEncryption(),
816 )
819def _check_use_client_cert_env():
820 use_client_cert = getenv(
821 environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE
822 ) or getenv(environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE)
824 if use_client_cert:
825 return use_client_cert.lower() == "true"
826 return None
829def check_use_client_cert():
830 """Returns boolean for whether the client certificate should be used for mTLS.
832 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
833 bool value will be returned. If the value is set to an unexpected string, it
834 will default to False.
835 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
836 as True (auto-enabled) if a workload config file exists (pointed at by
837 GOOGLE_API_CERTIFICATE_CONFIG or CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
838 or the default path like ~/.config/gcloud/certificate_config.json)
839 containing a "workload" section, or if no certificate config file is
840 configured/present and the GKE credential bundle exists at
841 /var/run/secrets/workload-spiffe-credentials/x509.credential-bundle.private-key.pem.
842 Otherwise, it returns False.
844 Returns:
845 bool: Whether the client certificate should be used for mTLS connection.
846 """
847 env_override = _check_use_client_cert_env()
848 if env_override is not None:
849 return env_override
851 # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
853 # Check if a workload config file exists.
854 cert_path = _get_cert_config_path(include_context_aware=True)
856 if cert_path:
857 try:
858 with open(cert_path, "r") as f:
859 content = json.load(f)
860 except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
861 _LOGGER.debug(
862 "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
863 cert_path,
864 e,
865 )
866 return False
868 # Structural validation
869 if isinstance(content, dict):
870 cert_configs = content.get("cert_configs")
871 if isinstance(cert_configs, dict) and "workload" in cert_configs:
872 return True
874 # If we got here, the file exists but the expected structure is missing
875 _LOGGER.debug(
876 "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
877 cert_path,
878 )
879 return False
881 return _should_use_gke_credential_bundle(cert_path)
884def check_parameters_for_unauthorized_response(cached_cert):
885 """Returns the cached and current cert fingerprint for reconfiguring mTLS.
887 Args:
888 cached_cert (Optional[bytes]): The cached client certificate.
890 Returns:
891 Tuple[Optional[bytes], Optional[bytes], Optional[str], Optional[str]]:
892 The client callback cert bytes, client callback key bytes,
893 base64-encoded SHA256 cached fingerprint, and base64-encoded SHA256
894 current cert fingerprint.
895 """
896 call_cert_bytes, call_key_bytes = call_client_cert_callback()
897 if not call_cert_bytes:
898 return None, None, None, None
899 cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
900 current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
901 cert_obj
902 )
903 if cached_cert:
904 cached_fingerprint = _agent_identity_utils.get_cached_cert_fingerprint(
905 cached_cert
906 )
907 else:
908 cached_fingerprint = current_cert_fingerprint
909 return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
912def call_client_cert_callback():
913 """Calls the client cert callback and returns the certificate and key."""
914 _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
915 generate_encrypted_key=True
916 )
917 return cert_bytes, key_bytes
920_MTLS_HOST_SUFFIXES = (
921 ".mtls.googleapis.com",
922 ".mtls.sandbox.googleapis.com",
923 ".p.googleapis.com",
924 ".mtls.run.app",
925)
926_MTLS_EXACT_HOSTS = (
927 "mtls.googleapis.com",
928 "mtls.sandbox.googleapis.com",
929 "p.googleapis.com",
930 "mtls.run.app",
931)
934def is_mtls_endpoint(url: Optional[Union[str, bytes, object]]) -> bool:
935 """Checks if the given URL corresponds to an mTLS (Google APIs or Cloud Run)
936 or Private Service Connect (PSC) endpoint.
938 Args:
939 url (Optional[Union[str, bytes, object]]): The request URL.
941 Returns:
942 bool: True if the URL targets an mTLS or PSC endpoint, False otherwise.
943 """
944 if not url:
945 return False
946 if hasattr(url, "url") and isinstance(url.url, (str, bytes)):
947 url = url.url
948 if isinstance(url, bytes):
949 try:
950 url = url.decode("utf-8")
951 except (UnicodeDecodeError, AttributeError):
952 return False
953 elif not isinstance(url, str):
954 url = str(url)
955 try:
956 hostname = urlsplit(url).hostname
957 except (ValueError, TypeError, AttributeError):
958 return False
960 if not hostname:
961 return False
963 hostname = hostname.rstrip(".").lower()
964 if not hostname:
965 return False
967 return hostname in _MTLS_EXACT_HOSTS or hostname.endswith(_MTLS_HOST_SUFFIXES)