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