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 try:
46 (
47 cert_path,
48 key_path,
49 config_file_path,
50 ) = _mtls_helper._get_workload_cert_and_key_paths(
51 None, include_context_aware=include_context_aware
52 )
53 except (exceptions.ClientCertError, OSError):
54 # Config exists, but is malformed or unreadable. Let caller surface error.
55 return True
57 if cert_path is not None and key_path is not None:
58 return True
59 if _mtls_helper._should_use_gke_credential_bundle(config_file_path):
60 return True
61 if (
62 include_context_aware
63 and _mtls_helper._check_config_path(_mtls_helper.CONTEXT_AWARE_METADATA_PATH)
64 is not None
65 ):
66 return True
68 return False
71def default_client_cert_source():
72 """Get a callback which returns the default client SSL credentials.
74 Returns:
75 Callable[[], [bytes, bytes]]: A callback which returns the default
76 client certificate bytes and private key bytes, both in PEM format.
78 Raises:
79 google.auth.exceptions.MutualTLSChannelError: If the default
80 client SSL credentials don't exist or are malformed.
81 """
82 if not has_default_client_cert_source(include_context_aware=True):
83 raise exceptions.MutualTLSChannelError(
84 "Default client cert source doesn't exist"
85 )
87 def callback():
88 try:
89 _, cert_bytes, key_bytes = _mtls_helper.get_client_cert_and_key()
90 except (OSError, RuntimeError, ValueError) as caught_exc:
91 new_exc = exceptions.MutualTLSChannelError(caught_exc)
92 raise new_exc from caught_exc
94 return cert_bytes, key_bytes
96 return callback
99def default_client_encrypted_cert_source(cert_path, key_path):
100 """Get a callback which returns the default encrpyted client SSL credentials.
102 Args:
103 cert_path (str): The cert file path. The default client certificate will
104 be written to this file when the returned callback is called.
105 key_path (str): The key file path. The default encrypted client key will
106 be written to this file when the returned callback is called.
108 Returns:
109 Callable[[], [str, str, bytes]]: A callback which generates the default
110 client certificate, encrpyted private key and passphrase. It writes
111 the certificate and private key into the cert_path and key_path, and
112 returns the cert_path, key_path and passphrase bytes.
114 Raises:
115 google.auth.exceptions.MutualTLSChannelError: If any problem
116 occurs when loading or saving the client certificate and key.
117 """
118 if not has_default_client_cert_source(include_context_aware=True):
119 raise exceptions.MutualTLSChannelError(
120 "Default client encrypted cert source doesn't exist"
121 )
123 def callback():
124 try:
125 (
126 _,
127 cert_bytes,
128 key_bytes,
129 passphrase_bytes,
130 ) = _mtls_helper.get_client_ssl_credentials(generate_encrypted_key=True)
131 with open(cert_path, "wb") as cert_file:
132 cert_file.write(cert_bytes)
133 with open(key_path, "wb") as key_file:
134 key_file.write(key_bytes)
135 except (exceptions.ClientCertError, OSError) as caught_exc:
136 new_exc = exceptions.MutualTLSChannelError(caught_exc)
137 raise new_exc from caught_exc
139 return cert_path, key_path, passphrase_bytes
141 return callback
144def should_use_client_cert():
145 """Returns boolean for whether the client certificate should be used for mTLS.
147 This is a wrapper around _mtls_helper.check_use_client_cert().
148 If GOOGLE_API_USE_CLIENT_CERTIFICATE is set to true or false, a corresponding
149 bool value will be returned
150 If GOOGLE_API_USE_CLIENT_CERTIFICATE is unset, the value will be inferred by
151 reading a file pointed at by GOOGLE_API_CERTIFICATE_CONFIG or
152 CLOUDSDK_CONTEXT_AWARE_CERTIFICATE_CONFIG_FILE_PATH, or the default path
153 like ~/.config/gcloud/certificate_config.json, and verifying it
154 contains a "workload" section. If so, the function will return True,
155 otherwise False.
157 Returns:
158 bool: indicating whether the client certificate should be used for mTLS.
159 """
160 return _mtls_helper.check_use_client_cert()
163def _load_client_cert_into_context(
164 ctx: ssl.SSLContext,
165 cert_bytes: bytes,
166 key_bytes: bytes,
167 passphrase: Optional[bytes] = None,
168) -> None:
169 """Load a client certificate and key into an SSL context.
171 Args:
172 ctx (ssl.SSLContext): The SSL context to load the certificate and key into.
173 cert_bytes (bytes): The client certificate bytes in PEM format.
174 key_bytes (bytes): The client private key bytes in PEM format.
175 passphrase (Optional[bytes]): The passphrase for the client private key.
177 Raises:
178 google.auth.exceptions.MutualTLSChannelError: If the SSL context is invalid,
179 or if loading the certificate and key fails.
180 """
181 if not isinstance(ctx, ssl.SSLContext):
182 raise exceptions.MutualTLSChannelError(
183 "Failed to load client certificate and key for mTLS. The provided context "
184 "object is invalid or does not support loading certificate chains."
185 )
187 try:
188 with _mtls_helper.secure_cert_key_paths(
189 cert_bytes, key_bytes, passphrase=passphrase
190 ) as (
191 cert_path,
192 key_path,
193 passphrase_val,
194 ):
195 if cert_path is None or key_path is None:
196 raise exceptions.MutualTLSChannelError(
197 "Failed to generate temporary file paths for the client certificate and key."
198 )
199 ctx.load_cert_chain(
200 certfile=cert_path, keyfile=key_path, password=passphrase_val
201 )
202 except (
203 ssl.SSLError,
204 OSError,
205 ValueError,
206 RuntimeError,
207 TypeError,
208 ) as caught_exc:
209 new_exc = exceptions.MutualTLSChannelError(caught_exc)
210 raise new_exc from caught_exc
213def load_default_client_cert(ctx: ssl.SSLContext) -> bool:
214 """Load the default client certificate and key into an SSL context if configured.
216 If client certificates are enabled and a default client certificate source is
217 found, the certificate and key are loaded into the SSL context.
219 Args:
220 ctx (ssl.SSLContext): The SSL context to load the default client certificate
221 and key into.
223 Returns:
224 bool: True if client certificates are enabled and the default client
225 certificate was successfully loaded. False if client certificates
226 are disabled or if no default certificate source is configured.
228 Raises:
229 google.auth.exceptions.MutualTLSChannelError: If the default client certificate
230 or key is malformed.
231 """
232 if not should_use_client_cert() or not has_default_client_cert_source():
233 return False
234 try:
235 (
236 has_cert,
237 cert_bytes,
238 key_bytes,
239 passphrase,
240 ) = _mtls_helper.get_client_ssl_credentials()
241 except (
242 exceptions.ClientCertError,
243 OSError,
244 RuntimeError,
245 ValueError,
246 ) as caught_exc:
247 new_exc = exceptions.MutualTLSChannelError(caught_exc)
248 raise new_exc from caught_exc
249 else:
250 if not has_cert:
251 return False
252 _load_client_cert_into_context(ctx, cert_bytes, key_bytes, passphrase)
253 return True
256def get_default_ssl_context() -> Optional[ssl.SSLContext]:
257 """Get a default SSL context loaded with the default client certificate.
259 Returns:
260 ssl.SSLContext: An SSL context loaded with the default client
261 certificate, or None if client certificates are not configured
262 or available.
264 Raises:
265 google.auth.exceptions.MutualTLSChannelError: If the default client certificate
266 or key is malformed.
267 """
268 if not should_use_client_cert() or not has_default_client_cert_source():
269 return None
271 ctx = ssl.create_default_context(ssl.Purpose.SERVER_AUTH)
272 return ctx if load_default_client_cert(ctx) else None
275def should_use_mtls_endpoint(
276 client_cert_available: Optional[bool] = None,
277) -> bool:
278 """Determine whether to use an mTLS endpoint.
280 This relies on the GOOGLE_API_USE_MTLS_ENDPOINT environment variable. If set to
281 "always", returns True. If set to "never", returns False. If set to "auto"
282 or unset, returns whether a client certificate is available.
284 Args:
285 client_cert_available (Optional[bool]): indicating if a client certificate
286 is available. If None, this is determined by checking if client
287 certificates are enabled using :func:`should_use_client_cert`.
289 Returns:
290 bool: indicating if an mTLS endpoint should be used.
291 """
292 if client_cert_available is None:
293 client_cert_available = should_use_client_cert()
295 use_mtls_endpoint = getenv(environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT)
296 use_mtls_endpoint = (use_mtls_endpoint or "auto").strip().lower()
297 try:
298 mode = UseMtlsEndpointMode(use_mtls_endpoint)
299 except ValueError:
300 raise exceptions.MutualTLSChannelError(
301 f"Unsupported {environment_vars.GOOGLE_API_USE_MTLS_ENDPOINT} value "
302 f"'{use_mtls_endpoint}'. Accepted values: never, auto, always."
303 )
305 if mode == UseMtlsEndpointMode.ALWAYS:
306 return True
307 if mode == UseMtlsEndpointMode.NEVER:
308 return False
309 if mode == UseMtlsEndpointMode.AUTO:
310 return client_cert_available