binding.py 3.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107
  1. # This file is dual licensed under the terms of the Apache License, Version
  2. # 2.0, and the BSD License. See the LICENSE file in the root of this repository
  3. # for complete details.
  4. from __future__ import annotations
  5. import threading
  6. import types
  7. import typing
  8. from collections.abc import Callable, Mapping
  9. import cryptography
  10. from cryptography.exceptions import InternalError
  11. from cryptography.hazmat.bindings._rust import _openssl, openssl
  12. from cryptography.hazmat.bindings.openssl._conditional import CONDITIONAL_NAMES
  13. def _openssl_assert(ok: bool) -> None:
  14. if not ok:
  15. errors = openssl.capture_error_stack()
  16. raise InternalError(
  17. "Unknown OpenSSL error. This error is commonly encountered when "
  18. "another library is not cleaning up the OpenSSL error stack. If "
  19. "you are using cryptography with another library that uses "
  20. "OpenSSL try disabling it before reporting a bug. Otherwise "
  21. "please file an issue at https://github.com/pyca/cryptography/"
  22. "issues with information on how to reproduce "
  23. f"this. ({errors!r})",
  24. errors,
  25. )
  26. def build_conditional_library(
  27. lib: typing.Any,
  28. conditional_names: Mapping[str, Callable[[], list[str]]],
  29. ) -> typing.Any:
  30. conditional_lib = types.ModuleType("lib")
  31. conditional_lib._original_lib = lib # type: ignore[attr-defined]
  32. excluded_names = set()
  33. for condition, names_cb in conditional_names.items():
  34. if not getattr(lib, condition):
  35. excluded_names.update(names_cb())
  36. for attr in dir(lib):
  37. if attr not in excluded_names:
  38. setattr(conditional_lib, attr, getattr(lib, attr))
  39. return conditional_lib
  40. class Binding:
  41. """
  42. OpenSSL API wrapper.
  43. """
  44. lib: typing.ClassVar[typing.Any] = None
  45. ffi: typing.Any = _openssl.ffi
  46. _lib_loaded = False
  47. _init_lock = threading.Lock()
  48. def __init__(self) -> None:
  49. self._ensure_ffi_initialized()
  50. @classmethod
  51. def _ensure_ffi_initialized(cls) -> None:
  52. with cls._init_lock:
  53. if not cls._lib_loaded:
  54. cls.lib = build_conditional_library(
  55. _openssl.lib, CONDITIONAL_NAMES
  56. )
  57. cls._lib_loaded = True
  58. @classmethod
  59. def init_static_locks(cls) -> None:
  60. cls._ensure_ffi_initialized()
  61. def _verify_package_version(version: str) -> None:
  62. # Occasionally we run into situations where the version of the Python
  63. # package does not match the version of the shared object that is loaded.
  64. # This may occur in environments where multiple versions of cryptography
  65. # are installed and available in the python path. To avoid errors cropping
  66. # up later this code checks that the currently imported package and the
  67. # shared object that were loaded have the same version and raise an
  68. # ImportError if they do not
  69. so_package_version = _openssl.ffi.string(
  70. _openssl.lib.CRYPTOGRAPHY_PACKAGE_VERSION
  71. )
  72. if version.encode("ascii") != so_package_version:
  73. raise ImportError(
  74. "The version of cryptography does not match the loaded "
  75. "shared object. This can happen if you have multiple copies of "
  76. "cryptography installed in your Python path. Please try creating "
  77. "a new virtual environment to resolve this issue. "
  78. f"Loaded python version: {version}, "
  79. f"shared object version: {so_package_version}"
  80. )
  81. _openssl_assert(
  82. _openssl.lib.OpenSSL_version_num() == openssl.openssl_version(),
  83. )
  84. _verify_package_version(cryptography.__version__)
  85. Binding.init_static_locks()