mtls.py 3.7 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105
  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.
  14. """Utilites for mutual TLS."""
  15. import six
  16. from google.auth import exceptions
  17. from google.auth.transport import _mtls_helper
  18. def has_default_client_cert_source():
  19. """Check if default client SSL credentials exists on the device.
  20. Returns:
  21. bool: indicating if the default client cert source exists.
  22. """
  23. metadata_path = _mtls_helper._check_dca_metadata_path(
  24. _mtls_helper.CONTEXT_AWARE_METADATA_PATH
  25. )
  26. return metadata_path is not None
  27. def default_client_cert_source():
  28. """Get a callback which returns the default client SSL credentials.
  29. Returns:
  30. Callable[[], [bytes, bytes]]: A callback which returns the default
  31. client certificate bytes and private key bytes, both in PEM format.
  32. Raises:
  33. google.auth.exceptions.DefaultClientCertSourceError: If the default
  34. client SSL credentials don't exist or are malformed.
  35. """
  36. if not has_default_client_cert_source():
  37. raise exceptions.MutualTLSChannelError(
  38. "Default client cert source doesn't exist"
  39. )
  40. def callback():
  41. try:
  42. _, cert_bytes, key_bytes = _mtls_helper.get_client_cert_and_key()
  43. except (OSError, RuntimeError, ValueError) as caught_exc:
  44. new_exc = exceptions.MutualTLSChannelError(caught_exc)
  45. six.raise_from(new_exc, caught_exc)
  46. return cert_bytes, key_bytes
  47. return callback
  48. def default_client_encrypted_cert_source(cert_path, key_path):
  49. """Get a callback which returns the default encrpyted client SSL credentials.
  50. Args:
  51. cert_path (str): The cert file path. The default client certificate will
  52. be written to this file when the returned callback is called.
  53. key_path (str): The key file path. The default encrypted client key will
  54. be written to this file when the returned callback is called.
  55. Returns:
  56. Callable[[], [str, str, bytes]]: A callback which generates the default
  57. client certificate, encrpyted private key and passphrase. It writes
  58. the certificate and private key into the cert_path and key_path, and
  59. returns the cert_path, key_path and passphrase bytes.
  60. Raises:
  61. google.auth.exceptions.DefaultClientCertSourceError: If any problem
  62. occurs when loading or saving the client certificate and key.
  63. """
  64. if not has_default_client_cert_source():
  65. raise exceptions.MutualTLSChannelError(
  66. "Default client encrypted cert source doesn't exist"
  67. )
  68. def callback():
  69. try:
  70. (
  71. _,
  72. cert_bytes,
  73. key_bytes,
  74. passphrase_bytes,
  75. ) = _mtls_helper.get_client_ssl_credentials(generate_encrypted_key=True)
  76. with open(cert_path, "wb") as cert_file:
  77. cert_file.write(cert_bytes)
  78. with open(key_path, "wb") as key_file:
  79. key_file.write(key_bytes)
  80. except (exceptions.ClientCertError, OSError) as caught_exc:
  81. new_exc = exceptions.MutualTLSChannelError(caught_exc)
  82. six.raise_from(new_exc, caught_exc)
  83. return cert_path, key_path, passphrase_bytes
  84. return callback