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"""Utilites for mutual TLS."""
17import enum
18import logging
19import ssl
20from os import getenv
21from typing import Optional
23from google.auth import environment_vars, exceptions
24from google.auth.transport import _mtls_helper
26_LOGGER = logging.getLogger(__name__)
29class UseMtlsEndpointMode(enum.Enum):
30 ALWAYS = "always"
31 NEVER = "never"
32 AUTO = "auto"
35def has_default_client_cert_source(include_context_aware=True):
36 """Check if default client SSL credentials exists on the device.
38 Args:
39 include_context_aware (bool): include_context_aware indicates if context_aware
40 path location will be checked or should it be skipped.
42 Returns:
43 bool: indicating if the default client cert source exists.
44 """
45 cert_path = _mtls_helper._get_cert_config_path(
46 include_context_aware=include_context_aware
47 )
48 if cert_path is not None:
49 return True
50 if (
51 include_context_aware
52 and _mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH)
53 is not None
54 ):
55 return True
57 return False
60def default_client_cert_source():
61 """Get a callback which returns the default client SSL credentials.
63 Returns:
64 Callable[[], [bytes, bytes]]: A callback which returns the default
65 client certificate bytes and private key bytes, both in PEM format.
67 Raises:
68 google.auth.exceptions.MutualTLSChannelError: If the default
69 client SSL credentials don't exist or are malformed.
70 """
71 if not has_default_client_cert_source(include_context_aware=True):
72 raise exceptions.MutualTLSChannelError(
73 "Default client cert source doesn't exist"
74 )
76 def callback():
77 try:
78 _, cert_bytes, key_bytes = _mtls_helper.get_client_cert_and_key()
79 except (OSError, RuntimeError, ValueError) as caught_exc:
80 new_exc = exceptions.MutualTLSChannelError(caught_exc)
81 raise new_exc from caught_exc
83 return cert_bytes, key_bytes
85 return callback
88def default_client_encrypted_cert_source(cert_path, key_path):
89 """Get a callback which returns the default encrpyted client SSL credentials.
91 Args:
92 cert_path (str): The cert file path. The default client certificate will
93 be written to this file when the returned callback is called.
94 key_path (str): The key file path. The default encrypted client key will
95 be written to this file when the returned callback is called.
97 Returns:
98 Callable[[], [str, str, bytes]]: A callback which generates the default
99 client certificate, encrpyted private key and passphrase. It writes
100 the certificate and private key into the cert_path and key_path, and
101 returns the cert_path, key_path and passphrase bytes.
103 Raises:
104 google.auth.exceptions.MutualTLSChannelError: If any problem
105 occurs when loading or saving the client certificate and key.
106 """
107 if not has_default_client_cert_source(include_context_aware=True):
108 raise exceptions.MutualTLSChannelError(
109 "Default client encrypted cert source doesn't exist"
110 )
112 def callback():
113 try:
114 (
115 _,
116 cert_bytes,
117 key_bytes,
118 passphrase_bytes,
119 ) = _mtls_helper.get_client_ssl_credentials(generate_encrypted_key=True)
120 with open(cert_path, "wb") as cert_file:
121 cert_file.write(cert_bytes)
122 with open(key_path, "wb") as key_file:
123 key_file.write(key_bytes)
124 except (exceptions.ClientCertError, OSError) as caught_exc:
125 new_exc = exceptions.MutualTLSChannelError(caught_exc)
126 raise new_exc from caught_exc
128 return cert_path, key_path, passphrase_bytes
130 return callback
133def should_use_client_cert():
134 """Returns boolean for whether the client certificate should be used for mTLS.
136 This is a wrapper around _mtls_helper.check_use_client_cert().
137 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
138 bool value will be returned
139 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred by
140 reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG or
141 CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, or the default path
142 like ~/.config/gcloud/certificate_config.json, and verifying it
143 contains a "workload" section. If so, the function will return True,
144 otherwise False.
146 Returns:
147 bool: indicating whether the client certificate should be used for mTLS.
148 """
149 return _mtls_helper.check_use_client_cert()
152def _load_client_cert_into_context(
153 ctx: ssl.SSLContext,
154 cert_bytes: bytes,
155 key_bytes: bytes,
156 passphrase: Optional[bytes] = None,
157) -> None:
158 """Load a client certificate and key into an SSL context.
160 Args:
161 ctx (ssl.SSLContext): The SSL context to load the certificate and key into.
162 cert_bytes (bytes): The client certificate bytes in PEM format.
163 key_bytes (bytes): The client private key bytes in PEM format.
164 passphrase (Optional[bytes]): The passphrase for the client private key.
166 Raises:
167 google.auth.exceptions.MutualTLSChannelError: If the SSL context is invalid,
168 or if loading the certificate and key fails.
169 """
170 if not isinstance(ctx, ssl.SSLContext):
171 raise exceptions.MutualTLSChannelError(
172 "Failed to load client certificate and key for mTLS. The provided context "
173 "object is invalid or does not support loading certificate chains."
174 )
176 try:
177 with _mtls_helper.secure_cert_key_paths(
178 cert_bytes, key_bytes, passphrase=passphrase
179 ) as (
180 cert_path,
181 key_path,
182 passphrase_val,
183 ):
184 if cert_path is None or key_path is None:
185 raise exceptions.MutualTLSChannelError(
186 "Failed to generate temporary file paths for the client certificate and key."
187 )
188 ctx.load_cert_chain(
189 certfile=cert_path, keyfile=key_path, password=passphrase_val
190 )
191 except (
192 ssl.SSLError,
193 OSError,
194 ValueError,
195 RuntimeError,
196 TypeError,
197 ) as caught_exc:
198 new_exc = exceptions.MutualTLSChannelError(caught_exc)
199 raise new_exc from caught_exc
202def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
203 """Load the default client certificate and key into an SSL context if configured.
205 If client certificates are enabled and a default client certificate source is
206 found, the certificate and key are loaded into the SSL context.
208 Args:
209 ctx (ssl.SSLContext): The SSL context to load the default client certificate
210 and key into.
212 Returns:
213 bool: True if client certificates are enabled and the default client
214 certificate was successfully loaded. False if client certificates
215 are disabled or if no default certificate source is configured.
217 Raises:
218 google.auth.exceptions.MutualTLSChannelError: If the default client certificate
219 or key is malformed.
220 """
221 if not should_use_client_cert() or not has_default_client_cert_source():
222 return False
223 try:
224 (
225 has_cert,
226 cert_bytes,
227 key_bytes,
228 passphrase,
229 ) = _mtls_helper.get_client_ssl_credentials()
230 except (
231 exceptions.ClientCertError,
232 OSError,
233 RuntimeError,
234 ValueError,
235 ) as caught_exc:
236 new_exc = exceptions.MutualTLSChannelError(caught_exc)
237 raise new_exc from caught_exc
238 else:
239 if not has_cert:
240 return False
241 _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
242 return True
245def get_default_ssl_context() -> Optional[ssl.SSLContext]:
246 """Get a default SSL context loaded with the default client certificate.
248 Returns:
249 ssl.SSLContext: An SSL context loaded with the default client
250 certificate, or None if client certificates are not configured
251 or available.
253 Raises:
254 google.auth.exceptions.MutualTLSChannelError: If the default client certificate
255 or key is malformed.
256 """
257 if not should_use_client_cert() or not has_default_client_cert_source():
258 return None
260 ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
261 return ctx if load_default_client_cert(ctx) else None
264def should_use_mtls_endpoint(
265 client_cert_available: Optional[bool] = None,
266) -> bool:
267 """Determine whether to use an mTLS endpoint.
269 This relies on the GOOGLE_API_USE_MTLS_ENDPOINT environment variable. If set to
270 "always", returns True. If set to "never", returns False. If set to "auto"
271 or unset, returns whether a client certificate is available.
273 Args:
274 client_cert_available (Optional[bool]): indicating if a client certificate
275 is available. If None, this is determined by checking if client
276 certificates are enabled using :func:`should_use_client_cert`.
278 Returns:
279 bool: indicating if an mTLS endpoint should be used.
280 """
281 if client_cert_available is None:
282 client_cert_available = should_use_client_cert()
284 use_mtls_endpoint = getenv(environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT)
285 use_mtls_endpoint = (use_mtls_endpoint or "auto").strip().lower()
286 try:
287 mode = UseMtlsEndpointMode(use_mtls_endpoint)
288 except ValueError:
289 raise exceptions.MutualTLSChannelError(
290 f"Unsupported {environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT} value "
291 f"'{use_mtls_endpoint}'. Accepted values: never, auto, always."
292 )
294 if mode == UseMtlsEndpointMode.ALWAYS:
295 return True
296 if mode == UseMtlsEndpointMode.NEVER:
297 return False
298 if mode == UseMtlsEndpointMode.AUTO:
299 return client_cert_available