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
28from google.auth import _agent_identity_utils
29from google.auth import environment_vars
30from google.auth import exceptions
32CONTEXT_AWARE_METADATA_PATH = "~/.secureConnect/context_aware_metadata.json"
34# Default gcloud config path, to be used with path.expanduser for cross-platform compatibility.
35CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
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# A flag to track if we have already logged a warning about mTLS auto-enablement failures.
55# This prevents log spam when client libraries create transports or session instances
56# frequently within a single process.
57_has_logged_mtls_warning = False
60_PASSPHRASE_REGEX = re.compile(
61 b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
62)
64# Temporary patch to accomodate incorrect cert config in Cloud Run prod environment.
65_WELL_KNOWN_CLOUD_RUN_CERT_PATH = (
66 "/var/run/secrets/workload-spiffe-credentials/certificates.pem"
67)
68_WELL_KNOWN_CLOUD_RUN_KEY_PATH = (
69 "/var/run/secrets/workload-spiffe-credentials/private_key.pem"
70)
71_INCORRECT_CLOUD_RUN_CERT_PATH = (
72 "/var/lib/volumes/certificate/workload-certificates/certificates.pem"
73)
74_INCORRECT_CLOUD_RUN_KEY_PATH = (
75 "/var/lib/volumes/certificate/workload-certificates/private_key.pem"
76)
79class _MemfdCreationError(OSError):
80 """Raised when Linux in-memory virtual file creation (memfd) fails."""
82 pass
85def _can_read(path: Optional[str]) -> bool:
86 if path is None:
87 return True
88 try:
89 with open(path, "rb"):
90 pass
91 return True
92 except OSError:
93 return False
96@contextlib.contextmanager
97def secure_cert_key_paths(
98 cert: Union[bytes, str, None],
99 key: Union[bytes, str, None],
100 passphrase: Optional[bytes] = None,
101) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
102 """Provides secure file paths for certificate and key.
104 This function is implemented as a context manager generator to ensure that
105 any temporary resources (such as in-memory virtual files or encrypted physical
106 temp files) are automatically cleaned up and securely wiped when the context exits.
108 It supports mixed inputs (e.g. passing one as a string path and the other as bytes).
109 If a parameter is already a string path or None, it is passed through as-is, and
110 only raw bytes are written to temporary storage.
112 Args:
113 cert (Union[str, bytes, None]): Certificate path, raw PEM content bytes, or None.
114 key (Union[str, bytes, None]): Private key path, raw PEM content bytes, or None.
115 passphrase (Optional[bytes]): Optional passphrase for the private key.
117 Yields:
118 Tuple[str, str, Optional[bytes]]: The certificate path, key path, and
119 the passphrase needed to load the key (either the user's original,
120 or the newly generated one if Tier 3 had to encrypt the key).
122 Raises:
123 OSError: If temporary file creation or writing fails during the Tier 3 fallback.
124 """
125 # Normalize PEM strings to bytes so they are written to temporary storage.
126 # We check for "-----BEGIN " to distinguish between file paths and PEM payloads.
127 if isinstance(cert, str) and "-----BEGIN " in cert:
128 cert = cert.encode("utf-8")
129 if isinstance(key, str) and "-----BEGIN " in key:
130 key = key.encode("utf-8")
132 # Tier 1: Pass-through (No-op). If the caller already provided file paths,
133 # we yield them directly to avoid any unnecessary file creation.
134 if isinstance(cert, str) and isinstance(key, str):
135 yield cert, key, passphrase
136 return
138 # If a value is a string path, it is passed through. If bytes, we will write
139 # it to temporary storage. None values are also passed through as-is.
140 cert_bytes = cert if isinstance(cert, bytes) else None
141 key_bytes = key if isinstance(key, bytes) else None
143 # Tier 2: Linux RAM-backed virtual files. If supported by the OS, we write
144 # the bytes to anonymous in-memory files using memfd_create. This yields
145 # /proc/self/fd/... paths, keeping the private key entirely in memory.
146 if sys.platform == "linux" and hasattr(os, "memfd_create"):
147 try:
148 with _memfd_cert_key_paths(cert_bytes, key_bytes) as (cert_path, key_path):
149 # Handle cases where path exists but might be restricted.
150 if (cert_path is None or os.path.exists(cert_path)) and (
151 key_path is None or os.path.exists(key_path)
152 ):
153 if _can_read(cert_path) and _can_read(key_path):
154 yield cast(str, cert_path or cert), cast(
155 str, key_path or key
156 ), passphrase
157 return
158 except _MemfdCreationError:
159 pass # Fallback to Tier 3 on failure.
161 # Tier 3: Fallback Encrypted Temp Files. If in-memory files are not supported
162 # (macOS/Windows), we write to disk. To protect the key, we encrypt plaintext
163 # keys on-the-fly and securely wipe the files with null bytes during cleanup.
164 with _tempfile_cert_key_paths(cert_bytes, key_bytes, passphrase) as (
165 cert_path,
166 key_path,
167 new_passphrase,
168 ):
169 yield cast(str, cert_path or cert), cast(str, key_path or key), new_passphrase
172def _encrypt_key_if_plaintext(
173 key_bytes: bytes, passphrase: Optional[bytes]
174) -> Tuple[bytes, Optional[bytes]]:
175 """Encrypts a plaintext PEM key if necessary, returning the bytes and passphrase.
177 If the key is already encrypted, or if parsing/encryption fails, the key is
178 returned as-is (plaintext) as a fallback. This allows the caller (underlying SSL
179 context) to attempt loading the key directly and handle any failures.
180 """
181 import cryptography
182 from cryptography.hazmat.primitives import serialization
183 import secrets
185 try:
186 pkey = serialization.load_pem_private_key(key_bytes, password=None)
187 # It's plaintext, encrypt it.
188 target_passphrase = passphrase
189 if target_passphrase is None:
190 target_passphrase = secrets.token_hex(32).encode("utf-8")
191 elif isinstance(target_passphrase, str):
192 target_passphrase = target_passphrase.encode("utf-8")
194 encrypted_content = pkey.private_bytes(
195 encoding=serialization.Encoding.PEM,
196 format=serialization.PrivateFormat.PKCS8,
197 encryption_algorithm=serialization.BestAvailableEncryption(
198 target_passphrase
199 ),
200 )
201 return encrypted_content, target_passphrase
202 except (ValueError, TypeError, cryptography.exceptions.UnsupportedAlgorithm):
203 # Likely already encrypted, invalid, or unsupported algorithm, return as-is.
204 return key_bytes, passphrase
207def _secure_wipe_and_remove(file_path: str):
208 """Overwrites a file with null bytes before deleting it.
210 This is an extra security measure to make file recovery harder. However, on modern
211 solid-state drives (SSDs), the hardware optimizes where data is written, meaning
212 the original private key bytes might still physically remain on the storage chips
213 until the drive cleans them up.
214 """
215 if not os.path.exists(file_path):
216 return
217 try:
218 size = os.path.getsize(file_path)
219 with open(file_path, "r+b") as f:
220 f.write(b"\0" * size)
221 f.flush()
222 os.fsync(f.fileno())
223 except OSError:
224 pass # Ignore permission/lock errors during cleanup.
225 finally:
226 try:
227 os.remove(file_path)
228 except OSError:
229 pass
232@contextlib.contextmanager
233def _memfd_cert_key_paths(
234 cert_bytes: Optional[bytes], key_bytes: Optional[bytes]
235) -> Generator[Tuple[Optional[str], Optional[str]], None, None]:
236 """Creates secure, in-memory virtual files on Linux using memfd_create.
238 Yields:
239 Tuple[Optional[str], Optional[str]]: In-memory file paths pointing to
240 the active descriptors (e.g., '/proc/self/fd/3').
241 """
242 cleanup_fds = []
243 paths: List[Optional[str]] = []
245 try:
246 try:
247 for data, name in [(cert_bytes, "mtls_cert"), (key_bytes, "mtls_key")]:
248 if data is not None:
249 # MFD_CLOEXEC prevents FD leaks to spawned subprocesses.
250 fd = os.memfd_create(name, os.MFD_CLOEXEC) # type: ignore[attr-defined]
251 cleanup_fds.append(fd)
252 with os.fdopen(fd, "wb", closefd=False) as f:
253 f.write(data)
254 paths.append(f"/proc/self/fd/{fd}")
255 else:
256 paths.append(None)
257 except (OSError, AttributeError) as exc:
258 raise _MemfdCreationError(
259 "Failed to create in-memory virtual files"
260 ) from exc
262 cert_path, key_path = paths
263 yield cert_path, key_path
264 finally:
265 # Closing the descriptors automatically frees the RAM allocation.
266 for fd in cleanup_fds:
267 try:
268 os.close(fd)
269 except OSError:
270 pass
273def _write_secure_tempfile(fd: int, data: bytes) -> None:
274 """Writes data to a file descriptor, securely flushes to disk, and closes it."""
275 try:
276 f = os.fdopen(fd, "wb")
277 except BaseException:
278 try:
279 os.close(fd)
280 except OSError:
281 pass
282 raise
284 with f:
285 f.write(data)
286 f.flush()
287 try:
288 os.fsync(f.fileno())
289 except OSError:
290 pass
293@contextlib.contextmanager
294def _tempfile_cert_key_paths(
295 cert_bytes: Optional[bytes],
296 key_bytes: Optional[bytes],
297 passphrase: Optional[bytes],
298) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
299 """Creates secure temporary file paths on disk, encrypting private keys.
301 Yields:
302 Tuple[Optional[str], Optional[str], Optional[bytes]]: The temporary file
303 paths and the passphrase needed to load the key.
304 """
305 # Prioritize RAM-backed /dev/shm to avoid writing secrets to physical storage.
306 tmp_dir = (
307 "/dev/shm"
308 if os.path.isdir("/dev/shm") and os.access("/dev/shm", os.W_OK)
309 else None
310 )
311 cleanup_files: List[Optional[str]] = [None, None]
312 new_passphrase = passphrase
313 cert_data = cert_bytes
314 key_data = None
315 if key_bytes is not None:
316 key_data, new_passphrase = _encrypt_key_if_plaintext(key_bytes, passphrase)
318 try:
319 for i, data in enumerate([cert_data, key_data]):
320 if data is not None:
321 try:
322 fd, path = tempfile.mkstemp(dir=tmp_dir)
323 except OSError:
324 fd, path = tempfile.mkstemp(dir=None)
326 cleanup_files[i] = path
327 _write_secure_tempfile(fd, data)
329 yield cleanup_files[0], cleanup_files[1], new_passphrase
330 finally:
331 cert_cleanup_path = cleanup_files[0]
332 key_cleanup_path = cleanup_files[1]
334 try:
335 if key_cleanup_path:
336 _secure_wipe_and_remove(key_cleanup_path)
337 except Exception:
338 pass
339 finally:
340 if cert_cleanup_path:
341 try:
342 if os.path.exists(cert_cleanup_path):
343 os.remove(cert_cleanup_path)
344 except OSError:
345 pass
348def _check_config_path(config_path):
349 """Checks for config file path. If it exists, returns the absolute path with user expansion;
350 otherwise returns None.
352 Args:
353 config_path (str): The config file path for either context_aware_metadata.json or certificate_config.json for example
355 Returns:
356 str: absolute path if exists and None otherwise.
357 """
358 config_path = path.expanduser(config_path)
359 if not path.exists(config_path):
360 _LOGGER.debug("%s is not found.", config_path)
361 return None
362 return config_path
365def _load_json_file(path):
366 """Reads and loads JSON from the given path. Used to read both X509 workload certificate and
367 secure connect configurations.
369 Args:
370 path (str): the path to read from.
372 Returns:
373 Dict[str, str]: The JSON stored at the file.
375 Raises:
376 google.auth.exceptions.ClientCertError: If failed to parse the file as JSON.
377 """
378 try:
379 with open(path) as f:
380 json_data = json.load(f)
381 except ValueError as caught_exc:
382 new_exc = exceptions.ClientCertError(caught_exc)
383 raise new_exc from caught_exc
385 return json_data
388def _get_workload_cert_and_key(
389 certificate_config_path=None, include_context_aware=True
390):
391 """Read the workload identity cert and key files specified in the certificate config provided.
392 If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG"
393 first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json".
395 Args:
396 certificate_config_path (string): The certificate config path. If no path is provided,
397 the environment variable will be checked first, then the well known gcloud location.
398 include_context_aware (bool): If context aware metadata path should be checked for the
399 SecureConnect mTLS configuration.
401 Returns:
402 Tuple[Optional[bytes], Optional[bytes]]: client certificate bytes in PEM format and key
403 bytes in PEM format.
405 Raises:
406 google.auth.exceptions.ClientCertError: if problems occurs when retrieving
407 the certificate or key information.
408 """
410 cert_path, key_path = _get_workload_cert_and_key_paths(
411 certificate_config_path, include_context_aware
412 )
414 if cert_path is None and key_path is None:
415 return None, None
417 return _read_cert_and_key_files(cert_path, key_path)
420def _get_cert_config_path(certificate_config_path=None, include_context_aware=True):
421 """Get the certificate configuration path based on the following order:
423 1: Explicit override, if set
424 2: Environment variable, if set
425 3: Well-known location
427 Returns "None" if the selected config file does not exist.
429 Args:
430 certificate_config_path (string): The certificate config path. If provided, the well known
431 location and environment variable will be ignored.
432 include_context_aware (bool): If context aware metadata path should be checked for the
433 SecureConnect mTLS configuration.
435 Returns:
436 The absolute path of the certificate config file, and None if the file does not exist.
437 """
439 if certificate_config_path is None:
440 env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None)
441 if env_path is not None and env_path != "":
442 certificate_config_path = env_path
443 else:
444 env_path = environ.get(
445 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
446 None,
447 )
448 if include_context_aware and env_path is not None and env_path != "":
449 certificate_config_path = env_path
450 else:
451 certificate_config_path = CERTIFICATE_CONFIGURATION_DEFAULT_PATH
453 certificate_config_path = path.expanduser(certificate_config_path)
454 if not path.exists(certificate_config_path):
455 return None
456 return certificate_config_path
459def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
460 absolute_path = _get_cert_config_path(config_path, include_context_aware)
461 if absolute_path is None:
462 return None, None
464 data = _load_json_file(absolute_path)
466 if "cert_configs" not in data:
467 raise exceptions.ClientCertError(
468 'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
469 absolute_path
470 )
471 )
472 cert_configs = data["cert_configs"]
474 # We return None, None if the expected workload fields are not present.
475 # The certificate config might be present for other types of connections (e.g. gECC),
476 # and we want to gracefully fallback to testing other mTLS configurations
477 # like SecureConnect instead of throwing an exception.
479 if "workload" not in cert_configs:
480 return None, None
481 workload = cert_configs["workload"]
483 if "cert_path" not in workload or "key_path" not in workload:
484 raise exceptions.ClientCertError(
485 'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
486 absolute_path
487 )
488 )
489 cert_path = workload["cert_path"]
490 key_path = workload["key_path"]
492 # == BEGIN Temporary Cloud Run PATCH ==
493 # See https://github.com/googleapis/google-auth-library-python/issues/1881
494 if (cert_path == _INCORRECT_CLOUD_RUN_CERT_PATH) and (
495 key_path == _INCORRECT_CLOUD_RUN_KEY_PATH
496 ):
497 if not path.exists(cert_path) and not path.exists(key_path):
498 _LOGGER.debug(
499 "Applying Cloud Run certificate path patch. "
500 "Configured paths not found: %s, %s. "
501 "Using well-known paths: %s, %s",
502 cert_path,
503 key_path,
504 _WELL_KNOWN_CLOUD_RUN_CERT_PATH,
505 _WELL_KNOWN_CLOUD_RUN_KEY_PATH,
506 )
507 cert_path = _WELL_KNOWN_CLOUD_RUN_CERT_PATH
508 key_path = _WELL_KNOWN_CLOUD_RUN_KEY_PATH
509 # == END Temporary Cloud Run PATCH ==
511 return cert_path, key_path
514def _read_cert_and_key_files(cert_path, key_path):
515 cert_data = _read_cert_file(cert_path)
516 key_data = _read_key_file(key_path)
518 return cert_data, key_data
521def _read_cert_file(cert_path):
522 with open(cert_path, "rb") as cert_file:
523 cert_data = cert_file.read()
525 cert_match = re.findall(_CERT_REGEX, cert_data)
526 if len(cert_match) != 1:
527 raise exceptions.ClientCertError(
528 "Certificate file {} is in an invalid format, a single PEM formatted certificate is expected".format(
529 cert_path
530 )
531 )
532 return cert_match[0]
535def _read_key_file(key_path):
536 with open(key_path, "rb") as key_file:
537 key_data = key_file.read()
539 key_match = re.findall(_KEY_REGEX, key_data)
540 if len(key_match) != 1:
541 raise exceptions.ClientCertError(
542 "Private key file {} is in an invalid format, a single PEM formatted private key is expected".format(
543 key_path
544 )
545 )
547 return key_match[0]
550def _run_cert_provider_command(command, expect_encrypted_key=False):
551 """Run the provided command, and return client side mTLS cert, key and
552 passphrase.
554 Args:
555 command (List[str]): cert provider command.
556 expect_encrypted_key (bool): If encrypted private key is expected.
558 Returns:
559 Tuple[bytes, bytes, bytes]: client certificate bytes in PEM format, key
560 bytes in PEM format and passphrase bytes.
562 Raises:
563 google.auth.exceptions.ClientCertError: if problems occurs when running
564 the cert provider command or generating cert, key and passphrase.
565 """
566 try:
567 process = subprocess.Popen(
568 command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
569 )
570 stdout, stderr = process.communicate()
571 except OSError as caught_exc:
572 new_exc = exceptions.ClientCertError(caught_exc)
573 raise new_exc from caught_exc
575 # Check cert provider command execution error.
576 if process.returncode != 0:
577 raise exceptions.ClientCertError(
578 "Cert provider command returns non-zero status code %s" % process.returncode
579 )
581 # Extract certificate (chain), key and passphrase.
582 cert_match = re.findall(_CERT_REGEX, stdout)
583 if len(cert_match) != 1:
584 raise exceptions.ClientCertError("Client SSL certificate is missing or invalid")
585 key_match = re.findall(_KEY_REGEX, stdout)
586 if len(key_match) != 1:
587 raise exceptions.ClientCertError("Client SSL key is missing or invalid")
588 passphrase_match = re.findall(_PASSPHRASE_REGEX, stdout)
590 if expect_encrypted_key:
591 if len(passphrase_match) != 1:
592 raise exceptions.ClientCertError("Passphrase is missing or invalid")
593 if b"ENCRYPTED" not in key_match[0]:
594 raise exceptions.ClientCertError("Encrypted private key is expected")
595 return cert_match[0], key_match[0], passphrase_match[0].strip()
597 if b"ENCRYPTED" in key_match[0]:
598 raise exceptions.ClientCertError("Encrypted private key is not expected")
599 if len(passphrase_match) > 0:
600 raise exceptions.ClientCertError("Passphrase is not expected")
601 return cert_match[0], key_match[0], None
604def get_client_ssl_credentials(
605 generate_encrypted_key=False,
606 context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
607 certificate_config_path=None,
608):
609 """Returns the client side certificate, private key and passphrase.
611 We look for certificates and keys with the following order of priority:
612 1. Certificate and key specified by certificate_config.json.
613 Currently, only X.509 workload certificates are supported.
614 2. Certificate and key specified by context aware metadata (i.e. SecureConnect).
616 Args:
617 generate_encrypted_key (bool): If set to True, encrypted private key
618 and passphrase will be generated; otherwise, unencrypted private key
619 will be generated and passphrase will be None. This option only
620 affects keys obtained via context_aware_metadata.json.
621 context_aware_metadata_path (str): The context_aware_metadata.json file path.
622 certificate_config_path (str): The certificate_config.json file path.
624 Returns:
625 Tuple[bool, bytes, bytes, bytes]:
626 A boolean indicating if cert, key and passphrase are obtained, the
627 cert bytes and key bytes both in PEM format, and passphrase bytes.
629 Raises:
630 google.auth.exceptions.ClientCertError: if problems occurs when getting
631 the cert, key and passphrase.
632 """
634 # 1. Attempt to retrieve X.509 Workload cert and key.
635 cert, key = _get_workload_cert_and_key(certificate_config_path)
636 if cert and key:
637 return True, cert, key, None
639 # 2. Check for context aware metadata json
640 metadata_path = _check_config_path(context_aware_metadata_path)
642 if metadata_path:
643 metadata_json = _load_json_file(metadata_path)
645 if _CERT_PROVIDER_COMMAND not in metadata_json:
646 raise exceptions.ClientCertError("Cert provider command is not found")
648 command = metadata_json[_CERT_PROVIDER_COMMAND]
650 if generate_encrypted_key and "--with_passphrase" not in command:
651 command.append("--with_passphrase")
653 # Execute the command.
654 cert, key, passphrase = _run_cert_provider_command(
655 command, expect_encrypted_key=generate_encrypted_key
656 )
657 return True, cert, key, passphrase
659 return False, None, None, None
662def get_client_cert_and_key(client_cert_callback=None):
663 """Returns the client side certificate and private key. The function first
664 tries to get certificate and key from client_cert_callback; if the callback
665 is None or doesn't provide certificate and key, the function tries application
666 default SSL credentials.
668 Args:
669 client_cert_callback (Optional[Callable[[], (bytes, bytes)]]): An
670 optional callback which returns client certificate bytes and private
671 key bytes both in PEM format.
673 Returns:
674 Tuple[bool, bytes, bytes]:
675 A boolean indicating if cert and key are obtained, the cert bytes
676 and key bytes both in PEM format.
678 Raises:
679 google.auth.exceptions.ClientCertError: if problems occurs when getting
680 the cert and key.
681 """
682 if client_cert_callback:
683 cert, key = client_cert_callback()
684 return True, cert, key
686 has_cert, cert, key, _ = get_client_ssl_credentials(generate_encrypted_key=False)
687 return has_cert, cert, key
690def decrypt_private_key(key, passphrase):
691 """A helper function to decrypt the private key with the given passphrase.
692 google-auth library doesn't support passphrase protected private key for
693 mutual TLS channel. This helper function can be used to decrypt the
694 passphrase protected private key in order to estalish mutual TLS channel.
696 For example, if you have a function which produces client cert, passphrase
697 protected private key and passphrase, you can convert it to a client cert
698 callback function accepted by google-auth::
700 from google.auth.transport import _mtls_helper
702 def your_client_cert_function():
703 return cert, encrypted_key, passphrase
705 # callback accepted by google-auth for mutual TLS channel.
706 def client_cert_callback():
707 cert, encrypted_key, passphrase = your_client_cert_function()
708 decrypted_key = _mtls_helper.decrypt_private_key(encrypted_key,
709 passphrase)
710 return cert, decrypted_key
712 Args:
713 key (bytes): The private key bytes in PEM format.
714 passphrase (bytes): The passphrase bytes.
716 Returns:
717 bytes: The decrypted private key in PEM format.
719 Raises:
720 ValueError: If there is any problem decrypting the private key.
721 """
722 if isinstance(key, str):
723 key = key.encode("utf-8")
724 if isinstance(passphrase, str):
725 passphrase = passphrase.encode("utf-8")
727 from cryptography.hazmat.primitives import serialization
729 # First convert encrypted_key_bytes to PKey object
730 pkey = serialization.load_pem_private_key(key, password=passphrase)
732 # Then dump the decrypted key bytes
733 return pkey.private_bytes(
734 encoding=serialization.Encoding.PEM,
735 format=serialization.PrivateFormat.PKCS8,
736 encryption_algorithm=serialization.NoEncryption(),
737 )
740def _check_use_client_cert_env():
741 use_client_cert = getenv(
742 environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE
743 ) or getenv(environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE)
745 if use_client_cert:
746 return use_client_cert.lower() == "true"
747 return None
750def check_use_client_cert():
751 """Returns boolean for whether the client certificate should be used for mTLS.
753 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
754 bool value will be returned. If the value is set to an unexpected string, it
755 will default to False.
756 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
757 as True (auto-enabled) if a workload config file exists (pointed at by
758 GOOGLE_API_CERTIFICATE_CONFIG) containing a "workload" section.
759 Otherwise, it returns False.
761 Returns:
762 bool: Whether the client certificate should be used for mTLS connection.
763 """
764 global _has_logged_mtls_warning
765 env_override = _check_use_client_cert_env()
766 if env_override is not None:
767 return env_override
769 # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
771 # Check if the value of GOOGLE_API_CERTIFICATE_CONFIG is set.
772 cert_path = getenv(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG) or getenv(
773 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH
774 )
776 if cert_path:
777 try:
778 with open(cert_path, "r") as f:
779 content = json.load(f)
780 except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
781 if not _has_logged_mtls_warning:
782 _LOGGER.warning(
783 "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
784 cert_path,
785 e,
786 )
787 _has_logged_mtls_warning = True
788 return False
790 # Structural validation
791 if isinstance(content, dict):
792 cert_configs = content.get("cert_configs")
793 if isinstance(cert_configs, dict) and "workload" in cert_configs:
794 return True
796 # If we got here, the file exists but the expected structure is missing
797 if not _has_logged_mtls_warning:
798 _LOGGER.warning(
799 "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
800 cert_path,
801 )
802 _has_logged_mtls_warning = True
803 return False
806def check_parameters_for_unauthorized_response(cached_cert):
807 """Returns the cached and current cert fingerprint for reconfiguring mTLS.
809 Args:
810 cached_cert(bytes): The cached client certificate.
812 Returns:
813 bytes: The client callback cert bytes.
814 bytes: The client callback key bytes.
815 str: The base64-encoded SHA256 cached fingerprint.
816 str: The base64-encoded SHA256 current cert fingerprint.
817 """
818 call_cert_bytes, call_key_bytes = call_client_cert_callback()
819 cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
820 current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
821 cert_obj
822 )
823 if cached_cert:
824 cached_fingerprint = _agent_identity_utils.get_cached_cert_fingerprint(
825 cached_cert
826 )
827 else:
828 cached_fingerprint = current_cert_fingerprint
829 return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
832def call_client_cert_callback():
833 """Calls the client cert callback and returns the certificate and key."""
834 _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
835 generate_encrypted_key=True
836 )
837 return cert_bytes, key_bytes