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
21from os import environ, getenv, path
22import re
23import subprocess
24import sys
25import tempfile
26from typing import cast, Generator, List, Optional, Tuple, Union
27from urllib.parse import urlsplit
29from google.auth import _agent_identity_utils
30from google.auth import _cloud_sdk
31from google.auth import environment_vars
32from google.auth import exceptions
34CONTEXT_AWARE_METADATA_PATH = "~/.secureConnect/context_aware_metadata.json"
36# Default gcloud config path, to be used with path.expanduser for cross-platform compatibility.
37CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
38_CERT_PROVIDER_COMMAND = "cert_provider_command"
39_CERT_REGEX = re.compile(
40 b"-----BEGIN CERTIFICATE-----.+-----END CERTIFICATE-----\r?\n?", re.DOTALL
41)
43# support various format of key files, e.g.
44# "-----BEGIN PRIVATE KEY-----...",
45# "-----BEGIN EC PRIVATE KEY-----...",
46# "-----BEGIN RSA PRIVATE KEY-----..."
47# "-----BEGIN ENCRYPTED PRIVATE KEY-----"
48_KEY_REGEX = re.compile(
49 b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+-----END [A-Z ]*PRIVATE KEY-----\r?\n?",
50 re.DOTALL,
51)
53_LOGGER = logging.getLogger(__name__)
56_PASSPHRASE_REGEX = re.compile(
57 b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
58)
61class _MemfdCreationError(OSError):
62 """Raised when Linux in-memory virtual file creation (memfd) fails."""
64 pass
67def _can_read(path: Optional[str]) -> bool:
68 if path is None:
69 return True
70 try:
71 with open(path, "rb"):
72 pass
73 return True
74 except OSError:
75 return False
78@contextlib.contextmanager
79def secure_cert_key_paths(
80 cert: Union[bytes, str, None],
81 key: Union[bytes, str, None],
82 passphrase: Optional[bytes] = None,
83) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
84 """Provides secure file paths for certificate and key.
86 This function is implemented as a context manager generator to ensure that
87 any temporary resources (such as in-memory virtual files or encrypted physical
88 temp files) are automatically cleaned up and securely wiped when the context exits.
90 It supports mixed inputs (e.g. passing one as a string path and the other as bytes).
91 If a parameter is already a string path or None, it is passed through as-is, and
92 only raw bytes are written to temporary storage.
94 Args:
95 cert (Union[str, bytes, None]): Certificate path, raw PEM content bytes, or None.
96 key (Union[str, bytes, None]): Private key path, raw PEM content bytes, or None.
97 passphrase (Optional[bytes]): Optional passphrase for the private key.
99 Yields:
100 Tuple[str, str, Optional[bytes]]: The certificate path, key path, and
101 the passphrase needed to load the key (either the user's original,
102 or the newly generated one if Tier 3 had to encrypt the key).
104 Raises:
105 OSError: If temporary file creation or writing fails during the Tier 3 fallback.
106 """
107 # Normalize PEM strings to bytes so they are written to temporary storage.
108 # We check for "-----BEGIN " to distinguish between file paths and PEM payloads.
109 if isinstance(cert, str) and "-----BEGIN " in cert:
110 cert = cert.encode("utf-8")
111 if isinstance(key, str) and "-----BEGIN " in key:
112 key = key.encode("utf-8")
114 # Tier 1: Pass-through (No-op). If the caller already provided file paths,
115 # we yield them directly to avoid any unnecessary file creation.
116 if isinstance(cert, str) and isinstance(key, str):
117 yield cert, key, passphrase
118 return
120 # If a value is a string path, it is passed through. If bytes, we will write
121 # it to temporary storage. None values are also passed through as-is.
122 cert_bytes = cert if isinstance(cert, bytes) else None
123 key_bytes = key if isinstance(key, bytes) else None
125 # Tier 2: Linux RAM-backed virtual files. If supported by the OS, we write
126 # the bytes to anonymous in-memory files using memfd_create. This yields
127 # /proc/self/fd/... paths, keeping the private key entirely in memory.
128 if sys.platform == "linux" and hasattr(os, "memfd_create"):
129 try:
130 with _memfd_cert_key_paths(cert_bytes, key_bytes) as (cert_path, key_path):
131 # Handle cases where path exists but might be restricted.
132 if (cert_path is None or os.path.exists(cert_path)) and (
133 key_path is None or os.path.exists(key_path)
134 ):
135 if _can_read(cert_path) and _can_read(key_path):
136 yield cast(str, cert_path or cert), cast(
137 str, key_path or key
138 ), passphrase
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 cryptography
164 from cryptography.hazmat.primitives import serialization
165 import secrets
167 try:
168 pkey = serialization.load_pem_private_key(key_bytes, password=None)
169 # It's plaintext, encrypt it.
170 target_passphrase = passphrase
171 if target_passphrase is None:
172 target_passphrase = secrets.token_hex(32).encode("utf-8")
173 elif isinstance(target_passphrase, str):
174 target_passphrase = target_passphrase.encode("utf-8")
176 encrypted_content = pkey.private_bytes(
177 encoding=serialization.Encoding.PEM,
178 format=serialization.PrivateFormat.PKCS8,
179 encryption_algorithm=serialization.BestAvailableEncryption(
180 target_passphrase
181 ),
182 )
183 return encrypted_content, target_passphrase
184 except (ValueError, TypeError, cryptography.exceptions.UnsupportedAlgorithm):
185 # Likely already encrypted, invalid, or unsupported algorithm, return as-is.
186 return key_bytes, passphrase
189def _secure_wipe_and_remove(file_path: str):
190 """Overwrites a file with null bytes before deleting it.
192 This is an extra security measure to make file recovery harder. However, on modern
193 solid-state drives (SSDs), the hardware optimizes where data is written, meaning
194 the original private key bytes might still physically remain on the storage chips
195 until the drive cleans them up.
196 """
197 if not os.path.exists(file_path):
198 return
199 try:
200 size = os.path.getsize(file_path)
201 with open(file_path, "r+b") as f:
202 f.write(b"\0" * size)
203 f.flush()
204 os.fsync(f.fileno())
205 except OSError:
206 pass # Ignore permission/lock errors during cleanup.
207 finally:
208 try:
209 os.remove(file_path)
210 except OSError:
211 pass
214@contextlib.contextmanager
215def _memfd_cert_key_paths(
216 cert_bytes: Optional[bytes], key_bytes: Optional[bytes]
217) -> Generator[Tuple[Optional[str], Optional[str]], None, None]:
218 """Creates secure, in-memory virtual files on Linux using memfd_create.
220 Yields:
221 Tuple[Optional[str], Optional[str]]: In-memory file paths pointing to
222 the active descriptors (e.g., '/proc/self/fd/3').
223 """
224 cleanup_fds = []
225 paths: List[Optional[str]] = []
227 try:
228 try:
229 for data, name in [(cert_bytes, "mtls_cert"), (key_bytes, "mtls_key")]:
230 if data is not None:
231 # MFD_CLOEXEC prevents FD leaks to spawned subprocesses.
232 fd = os.memfd_create(name, os.MFD_CLOEXEC) # type: ignore[attr-defined]
233 cleanup_fds.append(fd)
234 with os.fdopen(fd, "wb", closefd=False) as f:
235 f.write(data)
236 paths.append(f"/proc/self/fd/{fd}")
237 else:
238 paths.append(None)
239 except (OSError, AttributeError) as exc:
240 raise _MemfdCreationError(
241 "Failed to create in-memory virtual files"
242 ) from exc
244 cert_path, key_path = paths
245 yield cert_path, key_path
246 finally:
247 # Closing the descriptors automatically frees the RAM allocation.
248 for fd in cleanup_fds:
249 try:
250 os.close(fd)
251 except OSError:
252 pass
255def _write_secure_tempfile(fd: int, data: bytes) -> None:
256 """Writes data to a file descriptor, securely flushes to disk, and closes it."""
257 try:
258 f = os.fdopen(fd, "wb")
259 except BaseException:
260 try:
261 os.close(fd)
262 except OSError:
263 pass
264 raise
266 with f:
267 f.write(data)
268 f.flush()
269 try:
270 os.fsync(f.fileno())
271 except OSError:
272 pass
275@contextlib.contextmanager
276def _tempfile_cert_key_paths(
277 cert_bytes: Optional[bytes],
278 key_bytes: Optional[bytes],
279 passphrase: Optional[bytes],
280) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
281 """Creates secure temporary file paths on disk, encrypting private keys.
283 Yields:
284 Tuple[Optional[str], Optional[str], Optional[bytes]]: The temporary file
285 paths and the passphrase needed to load the key.
286 """
287 # Prioritize RAM-backed /dev/shm to avoid writing secrets to physical storage.
288 tmp_dir = (
289 "/dev/shm"
290 if os.path.isdir("/dev/shm") and os.access("/dev/shm", os.W_OK)
291 else None
292 )
293 cleanup_files: List[Optional[str]] = [None, None]
294 new_passphrase = passphrase
295 cert_data = cert_bytes
296 key_data = None
297 if key_bytes is not None:
298 key_data, new_passphrase = _encrypt_key_if_plaintext(key_bytes, passphrase)
300 try:
301 for i, data in enumerate([cert_data, key_data]):
302 if data is not None:
303 try:
304 fd, path = tempfile.mkstemp(dir=tmp_dir)
305 except OSError:
306 fd, path = tempfile.mkstemp(dir=None)
308 cleanup_files[i] = path
309 _write_secure_tempfile(fd, data)
311 yield cleanup_files[0], cleanup_files[1], new_passphrase
312 finally:
313 cert_cleanup_path = cleanup_files[0]
314 key_cleanup_path = cleanup_files[1]
316 try:
317 if key_cleanup_path:
318 _secure_wipe_and_remove(key_cleanup_path)
319 except Exception:
320 pass
321 finally:
322 if cert_cleanup_path:
323 try:
324 if os.path.exists(cert_cleanup_path):
325 os.remove(cert_cleanup_path)
326 except OSError:
327 pass
330def _check_config_path(config_path):
331 """Checks for config file path. If it exists, returns the absolute path with user expansion;
332 otherwise returns None.
334 Args:
335 config_path (str): The config file path for either context_aware_metadata.json or certificate_config.json for example
337 Returns:
338 str: absolute path if exists and None otherwise.
339 """
340 config_path = path.expanduser(config_path)
341 if not path.exists(config_path):
342 _LOGGER.debug("%s is not found.", config_path)
343 return None
344 return config_path
347def _load_json_file(path):
348 """Reads and loads JSON from the given path. Used to read both X509 workload certificate and
349 secure connect configurations.
351 Args:
352 path (str): the path to read from.
354 Returns:
355 Dict[str, str]: The JSON stored at the file.
357 Raises:
358 google.auth.exceptions.ClientCertError: If failed to parse the file as JSON.
359 """
360 try:
361 with open(path) as f:
362 json_data = json.load(f)
363 except ValueError as caught_exc:
364 new_exc = exceptions.ClientCertError(caught_exc)
365 raise new_exc from caught_exc
367 return json_data
370def _get_workload_cert_and_key(
371 certificate_config_path=None, include_context_aware=True
372):
373 """Read the workload identity cert and key files specified in the certificate config provided.
374 If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG"
375 first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json".
377 Args:
378 certificate_config_path (string): The certificate config path. If no path is provided,
379 the environment variable will be checked first, then the well known gcloud location.
380 include_context_aware (bool): If context aware metadata path should be checked for the
381 SecureConnect mTLS configuration.
383 Returns:
384 Tuple[Optional[bytes], Optional[bytes]]: client certificate bytes in PEM format and key
385 bytes in PEM format.
387 Raises:
388 google.auth.exceptions.ClientCertError: if problems occurs when retrieving
389 the certificate or key information.
390 """
392 cert_path, key_path = _get_workload_cert_and_key_paths(
393 certificate_config_path, include_context_aware
394 )
396 if cert_path is None and key_path is None:
397 return None, None
399 return _read_cert_and_key_files(cert_path, key_path)
402def _get_cert_config_path(certificate_config_path=None, include_context_aware=True):
403 """Get the certificate configuration path based on the following order:
405 1: Explicit override, if set
406 2: Environment variable, if set
407 3: Well-known location
409 Returns "None" if the selected config file does not exist.
411 Args:
412 certificate_config_path (string): The certificate config path. If provided, the well known
413 location and environment variable will be ignored.
414 include_context_aware (bool): If context aware metadata path should be checked for the
415 SecureConnect mTLS configuration.
417 Returns:
418 The absolute path of the certificate config file, and None if the file does not exist.
419 """
421 source = "function argument"
422 is_explicit = True
423 if certificate_config_path is None:
424 env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None)
425 if env_path is not None and env_path != "":
426 certificate_config_path = env_path
427 source = (
428 f"environment variable {environment_vars.GOOGLE_API_CERTIFICATE_CONFIG}"
429 )
430 else:
431 env_path = environ.get(
432 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
433 None,
434 )
435 if include_context_aware and env_path is not None and env_path != "":
436 certificate_config_path = env_path
437 source = f"environment variable {environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH}"
438 else:
439 certificate_config_path = os.path.join(
440 _cloud_sdk.get_config_path(), "certificate_config.json"
441 )
442 is_explicit = False
444 certificate_config_path = path.expanduser(certificate_config_path)
445 if not path.exists(certificate_config_path):
446 if is_explicit:
447 _LOGGER.debug(
448 "Certificate configuration file explicitly specified via %s at %s does not exist",
449 source,
450 certificate_config_path,
451 )
452 return None
453 return certificate_config_path
456def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
457 absolute_path = _get_cert_config_path(config_path, include_context_aware)
458 if absolute_path is None:
459 return None, None
461 data = _load_json_file(absolute_path)
463 if (
464 not isinstance(data, dict)
465 or "cert_configs" not in data
466 or not isinstance(data["cert_configs"], dict)
467 ):
468 raise exceptions.ClientCertError(
469 'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
470 absolute_path
471 )
472 )
473 cert_configs = data["cert_configs"]
475 # We return None, None if the expected workload fields are not present.
476 # The certificate config might be present for other types of connections (e.g. gECC),
477 # and we want to gracefully fallback to testing other mTLS configurations
478 # like SecureConnect instead of throwing an exception.
480 if (
481 not isinstance(cert_configs, dict) or "workload" not in cert_configs
482 ) and config_path is None:
483 default_home_path = path.expanduser(
484 os.path.join(
485 _cloud_sdk.get_config_path(),
486 "certificate_config.json",
487 )
488 )
489 if path.exists(default_home_path) and os.path.normpath(
490 default_home_path
491 ) != os.path.normpath(absolute_path):
492 try:
493 home_data = _load_json_file(default_home_path)
494 if isinstance(home_data, dict):
495 home_cert_configs = home_data.get("cert_configs")
496 if (
497 isinstance(home_cert_configs, dict)
498 and "workload" in home_cert_configs
499 ):
500 cert_configs = home_cert_configs
501 absolute_path = default_home_path
502 except (exceptions.ClientCertError, OSError):
503 pass
505 if not isinstance(cert_configs, dict) or "workload" not in cert_configs:
506 return None, None
507 workload = cert_configs["workload"]
509 if (
510 not isinstance(workload, dict)
511 or "cert_path" not in workload
512 or "key_path" not in workload
513 ):
514 raise exceptions.ClientCertError(
515 'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
516 absolute_path
517 )
518 )
519 cert_path = workload["cert_path"]
520 key_path = workload["key_path"]
522 return cert_path, key_path
525def _read_cert_and_key_files(cert_path, key_path):
526 cert_data = _read_cert_file(cert_path)
527 key_data = _read_key_file(key_path)
529 return cert_data, key_data
532def _read_cert_file(cert_path):
533 with open(cert_path, "rb") as cert_file:
534 cert_data = cert_file.read()
536 cert_match = re.findall(_CERT_REGEX, cert_data)
537 if len(cert_match) != 1:
538 raise exceptions.ClientCertError(
539 "Certificate file {} is in an invalid format, a single PEM formatted certificate is expected".format(
540 cert_path
541 )
542 )
543 return cert_match[0]
546def _read_key_file(key_path):
547 with open(key_path, "rb") as key_file:
548 key_data = key_file.read()
550 key_match = re.findall(_KEY_REGEX, key_data)
551 if len(key_match) != 1:
552 raise exceptions.ClientCertError(
553 "Private key file {} is in an invalid format, a single PEM formatted private key is expected".format(
554 key_path
555 )
556 )
558 return key_match[0]
561def _run_cert_provider_command(command, expect_encrypted_key=False):
562 """Run the provided command, and return client side mTLS cert, key and
563 passphrase.
565 Args:
566 command (List[str]): cert provider command.
567 expect_encrypted_key (bool): If encrypted private key is expected.
569 Returns:
570 Tuple[bytes, bytes, bytes]: client certificate bytes in PEM format, key
571 bytes in PEM format and passphrase bytes.
573 Raises:
574 google.auth.exceptions.ClientCertError: if problems occurs when running
575 the cert provider command or generating cert, key and passphrase.
576 """
577 try:
578 process = subprocess.Popen(
579 command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
580 )
581 stdout, stderr = process.communicate()
582 except OSError as caught_exc:
583 new_exc = exceptions.ClientCertError(caught_exc)
584 raise new_exc from caught_exc
586 # Check cert provider command execution error.
587 if process.returncode != 0:
588 raise exceptions.ClientCertError(
589 "Cert provider command returns non-zero status code %s" % process.returncode
590 )
592 # Extract certificate (chain), key and passphrase.
593 cert_match = re.findall(_CERT_REGEX, stdout)
594 if len(cert_match) != 1:
595 raise exceptions.ClientCertError("Client SSL certificate is missing or invalid")
596 key_match = re.findall(_KEY_REGEX, stdout)
597 if len(key_match) != 1:
598 raise exceptions.ClientCertError("Client SSL key is missing or invalid")
599 passphrase_match = re.findall(_PASSPHRASE_REGEX, stdout)
601 if expect_encrypted_key:
602 if len(passphrase_match) != 1:
603 raise exceptions.ClientCertError("Passphrase is missing or invalid")
604 if b"ENCRYPTED" not in key_match[0]:
605 raise exceptions.ClientCertError("Encrypted private key is expected")
606 return cert_match[0], key_match[0], passphrase_match[0].strip()
608 if b"ENCRYPTED" in key_match[0]:
609 raise exceptions.ClientCertError("Encrypted private key is not expected")
610 if len(passphrase_match) > 0:
611 raise exceptions.ClientCertError("Passphrase is not expected")
612 return cert_match[0], key_match[0], None
615def get_client_ssl_credentials(
616 generate_encrypted_key=False,
617 context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
618 certificate_config_path=None,
619):
620 """Returns the client side certificate, private key and passphrase.
622 We look for certificates and keys with the following order of priority:
623 1. Certificate and key specified by certificate_config.json.
624 Currently, only X.509 workload certificates are supported.
625 2. Certificate and key specified by context aware metadata (i.e. SecureConnect).
627 Args:
628 generate_encrypted_key (bool): If set to True, encrypted private key
629 and passphrase will be generated; otherwise, unencrypted private key
630 will be generated and passphrase will be None. This option only
631 affects keys obtained via context_aware_metadata.json.
632 context_aware_metadata_path (str): The context_aware_metadata.json file path.
633 certificate_config_path (str): The certificate_config.json file path.
635 Returns:
636 Tuple[bool, bytes, bytes, bytes]:
637 A boolean indicating if cert, key and passphrase are obtained, the
638 cert bytes and key bytes both in PEM format, and passphrase bytes.
640 Raises:
641 google.auth.exceptions.ClientCertError: if problems occurs when getting
642 the cert, key and passphrase.
643 """
645 # 1. Attempt to retrieve X.509 Workload cert and key.
646 cert, key = _get_workload_cert_and_key(certificate_config_path)
647 if cert and key:
648 return True, cert, key, None
650 # 2. Check for context aware metadata json
651 metadata_path = _check_config_path(context_aware_metadata_path)
653 if metadata_path:
654 metadata_json = _load_json_file(metadata_path)
656 if _CERT_PROVIDER_COMMAND not in metadata_json:
657 raise exceptions.ClientCertError("Cert provider command is not found")
659 command = metadata_json[_CERT_PROVIDER_COMMAND]
661 if generate_encrypted_key and "--with_passphrase" not in command:
662 command.append("--with_passphrase")
664 # Execute the command.
665 cert, key, passphrase = _run_cert_provider_command(
666 command, expect_encrypted_key=generate_encrypted_key
667 )
668 return True, cert, key, passphrase
670 return False, None, None, None
673def get_client_cert_and_key(client_cert_callback=None):
674 """Returns the client side certificate and private key. The function first
675 tries to get certificate and key from client_cert_callback; if the callback
676 is None or doesn't provide certificate and key, the function tries application
677 default SSL credentials.
679 Args:
680 client_cert_callback (Optional[Callable[[], (bytes, bytes)]]): An
681 optional callback which returns client certificate bytes and private
682 key bytes both in PEM format.
684 Returns:
685 Tuple[bool, bytes, bytes]:
686 A boolean indicating if cert and key are obtained, the cert bytes
687 and key bytes both in PEM format.
689 Raises:
690 google.auth.exceptions.ClientCertError: if problems occurs when getting
691 the cert and key.
692 """
693 if client_cert_callback:
694 cert, key = client_cert_callback()
695 return True, cert, key
697 has_cert, cert, key, _ = get_client_ssl_credentials(generate_encrypted_key=False)
698 return has_cert, cert, key
701def decrypt_private_key(key, passphrase):
702 """A helper function to decrypt the private key with the given passphrase.
703 google-auth library doesn't support passphrase protected private key for
704 mutual TLS channel. This helper function can be used to decrypt the
705 passphrase protected private key in order to estalish mutual TLS channel.
707 For example, if you have a function which produces client cert, passphrase
708 protected private key and passphrase, you can convert it to a client cert
709 callback function accepted by google-auth::
711 from google.auth.transport import _mtls_helper
713 def your_client_cert_function():
714 return cert, encrypted_key, passphrase
716 # callback accepted by google-auth for mutual TLS channel.
717 def client_cert_callback():
718 cert, encrypted_key, passphrase = your_client_cert_function()
719 decrypted_key = _mtls_helper.decrypt_private_key(encrypted_key,
720 passphrase)
721 return cert, decrypted_key
723 Args:
724 key (bytes): The private key bytes in PEM format.
725 passphrase (bytes): The passphrase bytes.
727 Returns:
728 bytes: The decrypted private key in PEM format.
730 Raises:
731 ValueError: If there is any problem decrypting the private key.
732 """
733 if isinstance(key, str):
734 key = key.encode("utf-8")
735 if isinstance(passphrase, str):
736 passphrase = passphrase.encode("utf-8")
738 from cryptography.hazmat.primitives import serialization
740 # First convert encrypted_key_bytes to PKey object
741 pkey = serialization.load_pem_private_key(key, password=passphrase)
743 # Then dump the decrypted key bytes
744 return pkey.private_bytes(
745 encoding=serialization.Encoding.PEM,
746 format=serialization.PrivateFormat.PKCS8,
747 encryption_algorithm=serialization.NoEncryption(),
748 )
751def _check_use_client_cert_env():
752 use_client_cert = getenv(
753 environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE
754 ) or getenv(environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE)
756 if use_client_cert:
757 return use_client_cert.lower() == "true"
758 return None
761def check_use_client_cert():
762 """Returns boolean for whether the client certificate should be used for mTLS.
764 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
765 bool value will be returned. If the value is set to an unexpected string, it
766 will default to False.
767 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
768 as True (auto-enabled) if a workload config file exists (pointed at by
769 GOOGLE_API_CERTIFICATE_CONFIG or CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
770 or the default path like ~/.config/gcloud/certificate_config.json)
771 containing a "workload" section.
772 Otherwise, it returns False.
774 Returns:
775 bool: Whether the client certificate should be used for mTLS connection.
776 """
777 env_override = _check_use_client_cert_env()
778 if env_override is not None:
779 return env_override
781 # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
783 # Check if a workload config file exists.
784 cert_path = _get_cert_config_path(include_context_aware=True)
786 if cert_path:
787 try:
788 with open(cert_path, "r") as f:
789 content = json.load(f)
790 except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
791 _LOGGER.debug(
792 "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
793 cert_path,
794 e,
795 )
796 return False
798 # Structural validation
799 if isinstance(content, dict):
800 cert_configs = content.get("cert_configs")
801 if isinstance(cert_configs, dict) and "workload" in cert_configs:
802 return True
804 # If we got here, the file exists but the expected structure is missing
805 _LOGGER.debug(
806 "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
807 cert_path,
808 )
809 return False
812def check_parameters_for_unauthorized_response(cached_cert):
813 """Returns the cached and current cert fingerprint for reconfiguring mTLS.
815 Args:
816 cached_cert(bytes): The cached client certificate.
818 Returns:
819 bytes: The client callback cert bytes.
820 bytes: The client callback key bytes.
821 str: The base64-encoded SHA256 cached fingerprint.
822 str: The base64-encoded SHA256 current cert fingerprint.
823 """
824 call_cert_bytes, call_key_bytes = call_client_cert_callback()
825 cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
826 current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
827 cert_obj
828 )
829 if cached_cert:
830 cached_fingerprint = _agent_identity_utils.get_cached_cert_fingerprint(
831 cached_cert
832 )
833 else:
834 cached_fingerprint = current_cert_fingerprint
835 return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
838def call_client_cert_callback():
839 """Calls the client cert callback and returns the certificate and key."""
840 _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
841 generate_encrypted_key=True
842 )
843 return cert_bytes, key_bytes
846_MTLS_HOST_SUFFIXES = (
847 ".mtls.googleapis.com",
848 ".mtls.sandbox.googleapis.com",
849 ".p.googleapis.com",
850)
851_MTLS_EXACT_HOSTS = (
852 "mtls.googleapis.com",
853 "mtls.sandbox.googleapis.com",
854 "p.googleapis.com",
855)
858def is_mtls_endpoint(url: Optional[Union[str, bytes, object]]) -> bool:
859 """Checks if the given URL corresponds to an mTLS or Private Service Connect (PSC) endpoint.
861 Args:
862 url (Optional[Union[str, bytes, object]]): The request URL.
864 Returns:
865 bool: True if the URL targets an mTLS or PSC endpoint, False otherwise.
866 """
867 if not url:
868 return False
869 if hasattr(url, "url") and isinstance(url.url, (str, bytes)):
870 url = url.url
871 if isinstance(url, bytes):
872 try:
873 url = url.decode("utf-8")
874 except (UnicodeDecodeError, AttributeError):
875 return False
876 elif not isinstance(url, str):
877 url = str(url)
878 try:
879 hostname = urlsplit(url).hostname
880 except (ValueError, TypeError, AttributeError):
881 return False
883 if not hostname:
884 return False
886 hostname = hostname.rstrip(".").lower()
887 if not hostname:
888 return False
890 return hostname in _MTLS_EXACT_HOSTS or hostname.endswith(_MTLS_HOST_SUFFIXES)