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