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 _cloud_sdk
30from google.auth import environment_vars
31from google.auth import exceptions
33CONTEXT_AWARE_METADATA_PATH = "~/.secureConnect/context_aware_metadata.json"
35# Default gcloud config path, to be used with path.expanduser for cross-platform compatibility.
36CERTIFICATE_CONFIGURATION_DEFAULT_PATH = "~/.config/gcloud/certificate_config.json"
37_CERT_PROVIDER_COMMAND = "cert_provider_command"
38_CERT_REGEX = re.compile(
39 b"-----BEGIN CERTIFICATE-----.+-----END CERTIFICATE-----\r?\n?", re.DOTALL
40)
42# support various format of key files, e.g.
43# "-----BEGIN PRIVATE KEY-----...",
44# "-----BEGIN EC PRIVATE KEY-----...",
45# "-----BEGIN RSA PRIVATE KEY-----..."
46# "-----BEGIN ENCRYPTED PRIVATE KEY-----"
47_KEY_REGEX = re.compile(
48 b"-----BEGIN [A-Z ]*PRIVATE KEY-----.+-----END [A-Z ]*PRIVATE KEY-----\r?\n?",
49 re.DOTALL,
50)
52_LOGGER = logging.getLogger(__name__)
55_PASSPHRASE_REGEX = re.compile(
56 b"-----BEGIN PASSPHRASE-----(.+)-----END PASSPHRASE-----", re.DOTALL
57)
60class _MemfdCreationError(OSError):
61 """Raised when Linux in-memory virtual file creation (memfd) fails."""
63 pass
66def _can_read(path: Optional[str]) -> bool:
67 if path is None:
68 return True
69 try:
70 with open(path, "rb"):
71 pass
72 return True
73 except OSError:
74 return False
77@contextlib.contextmanager
78def secure_cert_key_paths(
79 cert: Union[bytes, str, None],
80 key: Union[bytes, str, None],
81 passphrase: Optional[bytes] = None,
82) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
83 """Provides secure file paths for certificate and key.
85 This function is implemented as a context manager generator to ensure that
86 any temporary resources (such as in-memory virtual files or encrypted physical
87 temp files) are automatically cleaned up and securely wiped when the context exits.
89 It supports mixed inputs (e.g. passing one as a string path and the other as bytes).
90 If a parameter is already a string path or None, it is passed through as-is, and
91 only raw bytes are written to temporary storage.
93 Args:
94 cert (Union[str, bytes, None]): Certificate path, raw PEM content bytes, or None.
95 key (Union[str, bytes, None]): Private key path, raw PEM content bytes, or None.
96 passphrase (Optional[bytes]): Optional passphrase for the private key.
98 Yields:
99 Tuple[str, str, Optional[bytes]]: The certificate path, key path, and
100 the passphrase needed to load the key (either the user's original,
101 or the newly generated one if Tier 3 had to encrypt the key).
103 Raises:
104 OSError: If temporary file creation or writing fails during the Tier 3 fallback.
105 """
106 # Normalize PEM strings to bytes so they are written to temporary storage.
107 # We check for "-----BEGIN " to distinguish between file paths and PEM payloads.
108 if isinstance(cert, str) and "-----BEGIN " in cert:
109 cert = cert.encode("utf-8")
110 if isinstance(key, str) and "-----BEGIN " in key:
111 key = key.encode("utf-8")
113 # Tier 1: Pass-through (No-op). If the caller already provided file paths,
114 # we yield them directly to avoid any unnecessary file creation.
115 if isinstance(cert, str) and isinstance(key, str):
116 yield cert, key, passphrase
117 return
119 # If a value is a string path, it is passed through. If bytes, we will write
120 # it to temporary storage. None values are also passed through as-is.
121 cert_bytes = cert if isinstance(cert, bytes) else None
122 key_bytes = key if isinstance(key, bytes) else None
124 # Tier 2: Linux RAM-backed virtual files. If supported by the OS, we write
125 # the bytes to anonymous in-memory files using memfd_create. This yields
126 # /proc/self/fd/... paths, keeping the private key entirely in memory.
127 if sys.platform == "linux" and hasattr(os, "memfd_create"):
128 try:
129 with _memfd_cert_key_paths(cert_bytes, key_bytes) as (cert_path, key_path):
130 # Handle cases where path exists but might be restricted.
131 if (cert_path is None or os.path.exists(cert_path)) and (
132 key_path is None or os.path.exists(key_path)
133 ):
134 if _can_read(cert_path) and _can_read(key_path):
135 yield cast(str, cert_path or cert), cast(
136 str, key_path or key
137 ), passphrase
138 return
139 except _MemfdCreationError:
140 pass # Fallback to Tier 3 on failure.
142 # Tier 3: Fallback Encrypted Temp Files. If in-memory files are not supported
143 # (macOS/Windows), we write to disk. To protect the key, we encrypt plaintext
144 # keys on-the-fly and securely wipe the files with null bytes during cleanup.
145 with _tempfile_cert_key_paths(cert_bytes, key_bytes, passphrase) as (
146 cert_path,
147 key_path,
148 new_passphrase,
149 ):
150 yield cast(str, cert_path or cert), cast(str, key_path or key), new_passphrase
153def _encrypt_key_if_plaintext(
154 key_bytes: bytes, passphrase: Optional[bytes]
155) -> Tuple[bytes, Optional[bytes]]:
156 """Encrypts a plaintext PEM key if necessary, returning the bytes and passphrase.
158 If the key is already encrypted, or if parsing/encryption fails, the key is
159 returned as-is (plaintext) as a fallback. This allows the caller (underlying SSL
160 context) to attempt loading the key directly and handle any failures.
161 """
162 import cryptography
163 from cryptography.hazmat.primitives import serialization
164 import secrets
166 try:
167 pkey = serialization.load_pem_private_key(key_bytes, password=None)
168 # It's plaintext, encrypt it.
169 target_passphrase = passphrase
170 if target_passphrase is None:
171 target_passphrase = secrets.token_hex(32).encode("utf-8")
172 elif isinstance(target_passphrase, str):
173 target_passphrase = target_passphrase.encode("utf-8")
175 encrypted_content = pkey.private_bytes(
176 encoding=serialization.Encoding.PEM,
177 format=serialization.PrivateFormat.PKCS8,
178 encryption_algorithm=serialization.BestAvailableEncryption(
179 target_passphrase
180 ),
181 )
182 return encrypted_content, target_passphrase
183 except (ValueError, TypeError, cryptography.exceptions.UnsupportedAlgorithm):
184 # Likely already encrypted, invalid, or unsupported algorithm, return as-is.
185 return key_bytes, passphrase
188def _secure_wipe_and_remove(file_path: str):
189 """Overwrites a file with null bytes before deleting it.
191 This is an extra security measure to make file recovery harder. However, on modern
192 solid-state drives (SSDs), the hardware optimizes where data is written, meaning
193 the original private key bytes might still physically remain on the storage chips
194 until the drive cleans them up.
195 """
196 if not os.path.exists(file_path):
197 return
198 try:
199 size = os.path.getsize(file_path)
200 with open(file_path, "r+b") as f:
201 f.write(b"\0" * size)
202 f.flush()
203 os.fsync(f.fileno())
204 except OSError:
205 pass # Ignore permission/lock errors during cleanup.
206 finally:
207 try:
208 os.remove(file_path)
209 except OSError:
210 pass
213@contextlib.contextmanager
214def _memfd_cert_key_paths(
215 cert_bytes: Optional[bytes], key_bytes: Optional[bytes]
216) -> Generator[Tuple[Optional[str], Optional[str]], None, None]:
217 """Creates secure, in-memory virtual files on Linux using memfd_create.
219 Yields:
220 Tuple[Optional[str], Optional[str]]: In-memory file paths pointing to
221 the active descriptors (e.g., '/proc/self/fd/3').
222 """
223 cleanup_fds = []
224 paths: List[Optional[str]] = []
226 try:
227 try:
228 for data, name in [(cert_bytes, "mtls_cert"), (key_bytes, "mtls_key")]:
229 if data is not None:
230 # MFD_CLOEXEC prevents FD leaks to spawned subprocesses.
231 fd = os.memfd_create(name, os.MFD_CLOEXEC) # type: ignore[attr-defined]
232 cleanup_fds.append(fd)
233 with os.fdopen(fd, "wb", closefd=False) as f:
234 f.write(data)
235 paths.append(f"/proc/self/fd/{fd}")
236 else:
237 paths.append(None)
238 except (OSError, AttributeError) as exc:
239 raise _MemfdCreationError(
240 "Failed to create in-memory virtual files"
241 ) from exc
243 cert_path, key_path = paths
244 yield cert_path, key_path
245 finally:
246 # Closing the descriptors automatically frees the RAM allocation.
247 for fd in cleanup_fds:
248 try:
249 os.close(fd)
250 except OSError:
251 pass
254def _write_secure_tempfile(fd: int, data: bytes) -> None:
255 """Writes data to a file descriptor, securely flushes to disk, and closes it."""
256 try:
257 f = os.fdopen(fd, "wb")
258 except BaseException:
259 try:
260 os.close(fd)
261 except OSError:
262 pass
263 raise
265 with f:
266 f.write(data)
267 f.flush()
268 try:
269 os.fsync(f.fileno())
270 except OSError:
271 pass
274@contextlib.contextmanager
275def _tempfile_cert_key_paths(
276 cert_bytes: Optional[bytes],
277 key_bytes: Optional[bytes],
278 passphrase: Optional[bytes],
279) -> Generator[Tuple[Optional[str], Optional[str], Optional[bytes]], None, None]:
280 """Creates secure temporary file paths on disk, encrypting private keys.
282 Yields:
283 Tuple[Optional[str], Optional[str], Optional[bytes]]: The temporary file
284 paths and the passphrase needed to load the key.
285 """
286 # Prioritize RAM-backed /dev/shm to avoid writing secrets to physical storage.
287 tmp_dir = (
288 "/dev/shm"
289 if os.path.isdir("/dev/shm") and os.access("/dev/shm", os.W_OK)
290 else None
291 )
292 cleanup_files: List[Optional[str]] = [None, None]
293 new_passphrase = passphrase
294 cert_data = cert_bytes
295 key_data = None
296 if key_bytes is not None:
297 key_data, new_passphrase = _encrypt_key_if_plaintext(key_bytes, passphrase)
299 try:
300 for i, data in enumerate([cert_data, key_data]):
301 if data is not None:
302 try:
303 fd, path = tempfile.mkstemp(dir=tmp_dir)
304 except OSError:
305 fd, path = tempfile.mkstemp(dir=None)
307 cleanup_files[i] = path
308 _write_secure_tempfile(fd, data)
310 yield cleanup_files[0], cleanup_files[1], new_passphrase
311 finally:
312 cert_cleanup_path = cleanup_files[0]
313 key_cleanup_path = cleanup_files[1]
315 try:
316 if key_cleanup_path:
317 _secure_wipe_and_remove(key_cleanup_path)
318 except Exception:
319 pass
320 finally:
321 if cert_cleanup_path:
322 try:
323 if os.path.exists(cert_cleanup_path):
324 os.remove(cert_cleanup_path)
325 except OSError:
326 pass
329def _check_config_path(config_path):
330 """Checks for config file path. If it exists, returns the absolute path with user expansion;
331 otherwise returns None.
333 Args:
334 config_path (str): The config file path for either context_aware_metadata.json or certificate_config.json for example
336 Returns:
337 str: absolute path if exists and None otherwise.
338 """
339 config_path = path.expanduser(config_path)
340 if not path.exists(config_path):
341 _LOGGER.debug("%s is not found.", config_path)
342 return None
343 return config_path
346def _load_json_file(path):
347 """Reads and loads JSON from the given path. Used to read both X509 workload certificate and
348 secure connect configurations.
350 Args:
351 path (str): the path to read from.
353 Returns:
354 Dict[str, str]: The JSON stored at the file.
356 Raises:
357 google.auth.exceptions.ClientCertError: If failed to parse the file as JSON.
358 """
359 try:
360 with open(path) as f:
361 json_data = json.load(f)
362 except ValueError as caught_exc:
363 new_exc = exceptions.ClientCertError(caught_exc)
364 raise new_exc from caught_exc
366 return json_data
369def _get_workload_cert_and_key(
370 certificate_config_path=None, include_context_aware=True
371):
372 """Read the workload identity cert and key files specified in the certificate config provided.
373 If no config path is provided, check the environment variable: "GOOGLE_API_CERTIFICATE_CONFIG"
374 first, then the well known gcloud location: "~/.config/gcloud/certificate_config.json".
376 Args:
377 certificate_config_path (string): The certificate config path. If no path is provided,
378 the environment variable will be checked first, then the well known gcloud location.
379 include_context_aware (bool): If context aware metadata path should be checked for the
380 SecureConnect mTLS configuration.
382 Returns:
383 Tuple[Optional[bytes], Optional[bytes]]: client certificate bytes in PEM format and key
384 bytes in PEM format.
386 Raises:
387 google.auth.exceptions.ClientCertError: if problems occurs when retrieving
388 the certificate or key information.
389 """
391 cert_path, key_path = _get_workload_cert_and_key_paths(
392 certificate_config_path, include_context_aware
393 )
395 if cert_path is None and key_path is None:
396 return None, None
398 return _read_cert_and_key_files(cert_path, key_path)
401def _get_cert_config_path(certificate_config_path=None, include_context_aware=True):
402 """Get the certificate configuration path based on the following order:
404 1: Explicit override, if set
405 2: Environment variable, if set
406 3: Well-known location
408 Returns "None" if the selected config file does not exist.
410 Args:
411 certificate_config_path (string): The certificate config path. If provided, the well known
412 location and environment variable will be ignored.
413 include_context_aware (bool): If context aware metadata path should be checked for the
414 SecureConnect mTLS configuration.
416 Returns:
417 The absolute path of the certificate config file, and None if the file does not exist.
418 """
420 source = "function argument"
421 is_explicit = True
422 if certificate_config_path is None:
423 env_path = environ.get(environment_vars.GOOGLE_API_CERTIFICATE_CONFIG, None)
424 if env_path is not None and env_path != "":
425 certificate_config_path = env_path
426 source = (
427 f"environment variable {environment_vars.GOOGLE_API_CERTIFICATE_CONFIG}"
428 )
429 else:
430 env_path = environ.get(
431 environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
432 None,
433 )
434 if include_context_aware and env_path is not None and env_path != "":
435 certificate_config_path = env_path
436 source = f"environment variable {environment_vars.CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH}"
437 else:
438 certificate_config_path = os.path.join(
439 _cloud_sdk.get_config_path(), "certificate_config.json"
440 )
441 is_explicit = False
443 certificate_config_path = path.expanduser(certificate_config_path)
444 if not path.exists(certificate_config_path):
445 if is_explicit:
446 _LOGGER.debug(
447 "Certificate configuration file explicitly specified via %s at %s does not exist",
448 source,
449 certificate_config_path,
450 )
451 return None
452 return certificate_config_path
455def _get_workload_cert_and_key_paths(config_path, include_context_aware=True):
456 absolute_path = _get_cert_config_path(config_path, include_context_aware)
457 if absolute_path is None:
458 return None, None
460 data = _load_json_file(absolute_path)
462 if "cert_configs" not in data:
463 raise exceptions.ClientCertError(
464 'Certificate config file {} is in an invalid format, a "cert configs" object is expected'.format(
465 absolute_path
466 )
467 )
468 cert_configs = data["cert_configs"]
470 # We return None, None if the expected workload fields are not present.
471 # The certificate config might be present for other types of connections (e.g. gECC),
472 # and we want to gracefully fallback to testing other mTLS configurations
473 # like SecureConnect instead of throwing an exception.
475 if "workload" not in cert_configs:
476 return None, None
477 workload = cert_configs["workload"]
479 if "cert_path" not in workload or "key_path" not in workload:
480 raise exceptions.ClientCertError(
481 'Workload certificate configuration is missing "cert_path" or "key_path" in {}'.format(
482 absolute_path
483 )
484 )
485 cert_path = workload["cert_path"]
486 key_path = workload["key_path"]
488 return cert_path, key_path
491def _read_cert_and_key_files(cert_path, key_path):
492 cert_data = _read_cert_file(cert_path)
493 key_data = _read_key_file(key_path)
495 return cert_data, key_data
498def _read_cert_file(cert_path):
499 with open(cert_path, "rb") as cert_file:
500 cert_data = cert_file.read()
502 cert_match = re.findall(_CERT_REGEX, cert_data)
503 if len(cert_match) != 1:
504 raise exceptions.ClientCertError(
505 "Certificate file {} is in an invalid format, a single PEM formatted certificate is expected".format(
506 cert_path
507 )
508 )
509 return cert_match[0]
512def _read_key_file(key_path):
513 with open(key_path, "rb") as key_file:
514 key_data = key_file.read()
516 key_match = re.findall(_KEY_REGEX, key_data)
517 if len(key_match) != 1:
518 raise exceptions.ClientCertError(
519 "Private key file {} is in an invalid format, a single PEM formatted private key is expected".format(
520 key_path
521 )
522 )
524 return key_match[0]
527def _run_cert_provider_command(command, expect_encrypted_key=False):
528 """Run the provided command, and return client side mTLS cert, key and
529 passphrase.
531 Args:
532 command (List[str]): cert provider command.
533 expect_encrypted_key (bool): If encrypted private key is expected.
535 Returns:
536 Tuple[bytes, bytes, bytes]: client certificate bytes in PEM format, key
537 bytes in PEM format and passphrase bytes.
539 Raises:
540 google.auth.exceptions.ClientCertError: if problems occurs when running
541 the cert provider command or generating cert, key and passphrase.
542 """
543 try:
544 process = subprocess.Popen(
545 command, stdout=subprocess.PIPE, stderr=subprocess.PIPE
546 )
547 stdout, stderr = process.communicate()
548 except OSError as caught_exc:
549 new_exc = exceptions.ClientCertError(caught_exc)
550 raise new_exc from caught_exc
552 # Check cert provider command execution error.
553 if process.returncode != 0:
554 raise exceptions.ClientCertError(
555 "Cert provider command returns non-zero status code %s" % process.returncode
556 )
558 # Extract certificate (chain), key and passphrase.
559 cert_match = re.findall(_CERT_REGEX, stdout)
560 if len(cert_match) != 1:
561 raise exceptions.ClientCertError("Client SSL certificate is missing or invalid")
562 key_match = re.findall(_KEY_REGEX, stdout)
563 if len(key_match) != 1:
564 raise exceptions.ClientCertError("Client SSL key is missing or invalid")
565 passphrase_match = re.findall(_PASSPHRASE_REGEX, stdout)
567 if expect_encrypted_key:
568 if len(passphrase_match) != 1:
569 raise exceptions.ClientCertError("Passphrase is missing or invalid")
570 if b"ENCRYPTED" not in key_match[0]:
571 raise exceptions.ClientCertError("Encrypted private key is expected")
572 return cert_match[0], key_match[0], passphrase_match[0].strip()
574 if b"ENCRYPTED" in key_match[0]:
575 raise exceptions.ClientCertError("Encrypted private key is not expected")
576 if len(passphrase_match) > 0:
577 raise exceptions.ClientCertError("Passphrase is not expected")
578 return cert_match[0], key_match[0], None
581def get_client_ssl_credentials(
582 generate_encrypted_key=False,
583 context_aware_metadata_path=CONTEXT_AWARE_METADATA_PATH,
584 certificate_config_path=None,
585):
586 """Returns the client side certificate, private key and passphrase.
588 We look for certificates and keys with the following order of priority:
589 1. Certificate and key specified by certificate_config.json.
590 Currently, only X.509 workload certificates are supported.
591 2. Certificate and key specified by context aware metadata (i.e. SecureConnect).
593 Args:
594 generate_encrypted_key (bool): If set to True, encrypted private key
595 and passphrase will be generated; otherwise, unencrypted private key
596 will be generated and passphrase will be None. This option only
597 affects keys obtained via context_aware_metadata.json.
598 context_aware_metadata_path (str): The context_aware_metadata.json file path.
599 certificate_config_path (str): The certificate_config.json file path.
601 Returns:
602 Tuple[bool, bytes, bytes, bytes]:
603 A boolean indicating if cert, key and passphrase are obtained, the
604 cert bytes and key bytes both in PEM format, and passphrase bytes.
606 Raises:
607 google.auth.exceptions.ClientCertError: if problems occurs when getting
608 the cert, key and passphrase.
609 """
611 # 1. Attempt to retrieve X.509 Workload cert and key.
612 cert, key = _get_workload_cert_and_key(certificate_config_path)
613 if cert and key:
614 return True, cert, key, None
616 # 2. Check for context aware metadata json
617 metadata_path = _check_config_path(context_aware_metadata_path)
619 if metadata_path:
620 metadata_json = _load_json_file(metadata_path)
622 if _CERT_PROVIDER_COMMAND not in metadata_json:
623 raise exceptions.ClientCertError("Cert provider command is not found")
625 command = metadata_json[_CERT_PROVIDER_COMMAND]
627 if generate_encrypted_key and "--with_passphrase" not in command:
628 command.append("--with_passphrase")
630 # Execute the command.
631 cert, key, passphrase = _run_cert_provider_command(
632 command, expect_encrypted_key=generate_encrypted_key
633 )
634 return True, cert, key, passphrase
636 return False, None, None, None
639def get_client_cert_and_key(client_cert_callback=None):
640 """Returns the client side certificate and private key. The function first
641 tries to get certificate and key from client_cert_callback; if the callback
642 is None or doesn't provide certificate and key, the function tries application
643 default SSL credentials.
645 Args:
646 client_cert_callback (Optional[Callable[[], (bytes, bytes)]]): An
647 optional callback which returns client certificate bytes and private
648 key bytes both in PEM format.
650 Returns:
651 Tuple[bool, bytes, bytes]:
652 A boolean indicating if cert and key are obtained, the cert bytes
653 and key bytes both in PEM format.
655 Raises:
656 google.auth.exceptions.ClientCertError: if problems occurs when getting
657 the cert and key.
658 """
659 if client_cert_callback:
660 cert, key = client_cert_callback()
661 return True, cert, key
663 has_cert, cert, key, _ = get_client_ssl_credentials(generate_encrypted_key=False)
664 return has_cert, cert, key
667def decrypt_private_key(key, passphrase):
668 """A helper function to decrypt the private key with the given passphrase.
669 google-auth library doesn't support passphrase protected private key for
670 mutual TLS channel. This helper function can be used to decrypt the
671 passphrase protected private key in order to estalish mutual TLS channel.
673 For example, if you have a function which produces client cert, passphrase
674 protected private key and passphrase, you can convert it to a client cert
675 callback function accepted by google-auth::
677 from google.auth.transport import _mtls_helper
679 def your_client_cert_function():
680 return cert, encrypted_key, passphrase
682 # callback accepted by google-auth for mutual TLS channel.
683 def client_cert_callback():
684 cert, encrypted_key, passphrase = your_client_cert_function()
685 decrypted_key = _mtls_helper.decrypt_private_key(encrypted_key,
686 passphrase)
687 return cert, decrypted_key
689 Args:
690 key (bytes): The private key bytes in PEM format.
691 passphrase (bytes): The passphrase bytes.
693 Returns:
694 bytes: The decrypted private key in PEM format.
696 Raises:
697 ValueError: If there is any problem decrypting the private key.
698 """
699 if isinstance(key, str):
700 key = key.encode("utf-8")
701 if isinstance(passphrase, str):
702 passphrase = passphrase.encode("utf-8")
704 from cryptography.hazmat.primitives import serialization
706 # First convert encrypted_key_bytes to PKey object
707 pkey = serialization.load_pem_private_key(key, password=passphrase)
709 # Then dump the decrypted key bytes
710 return pkey.private_bytes(
711 encoding=serialization.Encoding.PEM,
712 format=serialization.PrivateFormat.PKCS8,
713 encryption_algorithm=serialization.NoEncryption(),
714 )
717def _check_use_client_cert_env():
718 use_client_cert = getenv(
719 environment_vars.GOOGLE_API_USE_CLIENT_CERTIFICATE
720 ) or getenv(environment_vars.CLOUDSDK_CONTEXT_AWARE_USE_CLIENT_CERTIFICATE)
722 if use_client_cert:
723 return use_client_cert.lower() == "true"
724 return None
727def check_use_client_cert():
728 """Returns boolean for whether the client certificate should be used for mTLS.
730 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
731 bool value will be returned. If the value is set to an unexpected string, it
732 will default to False.
733 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred
734 as True (auto-enabled) if a workload config file exists (pointed at by
735 GOOGLE_API_CERTIFICATE_CONFIG or CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH,
736 or the default path like ~/.config/gcloud/certificate_config.json)
737 containing a "workload" section.
738 Otherwise, it returns False.
740 Returns:
741 bool: Whether the client certificate should be used for mTLS connection.
742 """
743 env_override = _check_use_client_cert_env()
744 if env_override is not None:
745 return env_override
747 # Auto-enablement checks (when GOOGLE_API_USE_CLIENT_CERTIFICATE is not set)
749 # Check if a workload config file exists.
750 cert_path = _get_cert_config_path(include_context_aware=True)
752 if cert_path:
753 try:
754 with open(cert_path, "r") as f:
755 content = json.load(f)
756 except (FileNotFoundError, OSError, json.JSONDecodeError) as e:
757 _LOGGER.debug(
758 "mTLS auto-enablement failed: Could not read/parse certificate file at %s. Error: %s",
759 cert_path,
760 e,
761 )
762 return False
764 # Structural validation
765 if isinstance(content, dict):
766 cert_configs = content.get("cert_configs")
767 if isinstance(cert_configs, dict) and "workload" in cert_configs:
768 return True
770 # If we got here, the file exists but the expected structure is missing
771 _LOGGER.debug(
772 "mTLS auto-enablement failed: Certificate configuration file at %s is missing the required ['cert_configs']['workload'] section.",
773 cert_path,
774 )
775 return False
778def check_parameters_for_unauthorized_response(cached_cert):
779 """Returns the cached and current cert fingerprint for reconfiguring mTLS.
781 Args:
782 cached_cert(bytes): The cached client certificate.
784 Returns:
785 bytes: The client callback cert bytes.
786 bytes: The client callback key bytes.
787 str: The base64-encoded SHA256 cached fingerprint.
788 str: The base64-encoded SHA256 current cert fingerprint.
789 """
790 call_cert_bytes, call_key_bytes = call_client_cert_callback()
791 cert_obj = _agent_identity_utils.parse_certificate(call_cert_bytes)
792 current_cert_fingerprint = _agent_identity_utils.calculate_certificate_fingerprint(
793 cert_obj
794 )
795 if cached_cert:
796 cached_fingerprint = _agent_identity_utils.get_cached_cert_fingerprint(
797 cached_cert
798 )
799 else:
800 cached_fingerprint = current_cert_fingerprint
801 return call_cert_bytes, call_key_bytes, cached_fingerprint, current_cert_fingerprint
804def call_client_cert_callback():
805 """Calls the client cert callback and returns the certificate and key."""
806 _, cert_bytes, key_bytes, passphrase = get_client_ssl_credentials(
807 generate_encrypted_key=True
808 )
809 return cert_bytes, key_bytes