diff options
Diffstat (limited to 'contrib/python/pyOpenSSL/py3')
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/.dist-info/METADATA | 97 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py | 182 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py | 1 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py | 408 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/OpenSSL/debug.py | 4 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/OpenSSL/version.py | 4 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/tests/conftest.py | 3 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/tests/memdbg.py | 5 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/tests/test_crypto.py | 579 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/tests/test_ssl.py | 231 | ||||
| -rw-r--r-- | contrib/python/pyOpenSSL/py3/ya.make | 2 |
11 files changed, 750 insertions, 766 deletions
diff --git a/contrib/python/pyOpenSSL/py3/.dist-info/METADATA b/contrib/python/pyOpenSSL/py3/.dist-info/METADATA index e4139673e38..5691fc9e4ee 100644 --- a/contrib/python/pyOpenSSL/py3/.dist-info/METADATA +++ b/contrib/python/pyOpenSSL/py3/.dist-info/METADATA @@ -1,6 +1,6 @@ Metadata-Version: 2.1 Name: pyOpenSSL -Version: 23.0.0 +Version: 23.3.0 Summary: Python wrapper module around the OpenSSL library Home-page: https://pyopenssl.org/ Author: The pyOpenSSL developers @@ -14,27 +14,27 @@ Classifier: Operating System :: MacOS :: MacOS X Classifier: Operating System :: Microsoft :: Windows Classifier: Operating System :: POSIX Classifier: Programming Language :: Python :: 3 -Classifier: Programming Language :: Python :: 3.6 Classifier: Programming Language :: Python :: 3.7 Classifier: Programming Language :: Python :: 3.8 Classifier: Programming Language :: Python :: 3.9 Classifier: Programming Language :: Python :: 3.10 Classifier: Programming Language :: Python :: 3.11 +Classifier: Programming Language :: Python :: 3.12 Classifier: Programming Language :: Python :: Implementation :: CPython Classifier: Programming Language :: Python :: Implementation :: PyPy Classifier: Topic :: Security :: Cryptography Classifier: Topic :: Software Development :: Libraries :: Python Modules Classifier: Topic :: System :: Networking -Requires-Python: >=3.6 +Requires-Python: >=3.7 License-File: LICENSE -Requires-Dist: cryptography (<40,>=38.0.0) +Requires-Dist: cryptography <42,>=41.0.5 Provides-Extra: docs -Requires-Dist: sphinx (!=5.2.0,!=5.2.0.post0) ; extra == 'docs' +Requires-Dist: sphinx !=5.2.0,!=5.2.0.post0,!=7.2.5 ; extra == 'docs' Requires-Dist: sphinx-rtd-theme ; extra == 'docs' Provides-Extra: test Requires-Dist: flaky ; extra == 'test' Requires-Dist: pretend ; extra == 'test' -Requires-Dist: pytest (>=3.0.1) ; extra == 'test' +Requires-Dist: pytest >=3.0.1 ; extra == 'test' ======================================================== pyOpenSSL -- A Python wrapper around the OpenSSL library @@ -87,6 +87,89 @@ You can also join ``#pyca`` on ``irc.libera.chat`` to ask questions or get invol Release Information =================== +23.3.0 (2023-10-25) +------------------- + +Backward-incompatible changes: +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +- Dropped support for Python 3.6. +- The minimum ``cryptography`` version is now 41.0.5. +- Removed ``OpenSSL.crypto.loads_pkcs7`` and ``OpenSSL.crypto.loads_pkcs12`` which had been deprecated for 3 years. +- Added ``OpenSSL.SSL.OP_LEGACY_SERVER_CONNECT`` to allow legacy insecure renegotiation between OpenSSL and unpatched servers. + `#1234 <https://github.com/pyca/pyopenssl/pull/1234>`_. + +Deprecations: +^^^^^^^^^^^^^ + +- Deprecated ``OpenSSL.crypto.PKCS12`` (which was intended to have been deprecated at the same time as ``OpenSSL.crypto.load_pkcs12``). +- Deprecated ``OpenSSL.crypto.NetscapeSPKI``. +- Deprecated ``OpenSSL.crypto.CRL`` +- Deprecated ``OpenSSL.crypto.Revoked`` +- Deprecated ``OpenSSL.crypto.load_crl`` and ``OpenSSL.crypto.dump_crl`` +- Deprecated ``OpenSSL.crypto.sign`` and ``OpenSSL.crypto.verify`` +- Deprecated ``OpenSSL.crypto.X509Extension`` + +Changes: +^^^^^^^^ + +- Changed ``OpenSSL.crypto.X509Store.add_crl`` to also accept + ``cryptography``'s ``x509.CertificateRevocationList`` arguments in addition + to the now deprecated ``OpenSSL.crypto.CRL`` arguments. +- Fixed ``test_set_default_verify_paths`` test so that it is skipped if no + network connection is available. + +23.2.0 (2023-05-30) +------------------- + +Backward-incompatible changes: +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +- Removed ``X509StoreFlags.NOTIFY_POLICY``. + `#1213 <https://github.com/pyca/pyopenssl/pull/1213>`_. + +Deprecations: +^^^^^^^^^^^^^ + +Changes: +^^^^^^^^ + +- ``cryptography`` maximum version has been increased to 41.0.x. +- Invalid versions are now rejected in ``OpenSSL.crypto.X509Req.set_version``. +- Added ``X509VerificationCodes`` to ``OpenSSL.SSL``. + `#1202 <https://github.com/pyca/pyopenssl/pull/1202>`_. + +23.1.1 (2023-03-28) +------------------- + +Backward-incompatible changes: +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Deprecations: +^^^^^^^^^^^^^ + +Changes: +^^^^^^^^ + +- Worked around an issue in OpenSSL 3.1.0 which caused `X509Extension.get_short_name` to raise an exception when no short name was known to OpenSSL. + `#1204 <https://github.com/pyca/pyopenssl/pull/1204>`_. + +23.1.0 (2023-03-24) +------------------- + +Backward-incompatible changes: +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ + +Deprecations: +^^^^^^^^^^^^^ + +Changes: +^^^^^^^^ + +- ``cryptography`` maximum version has been increased to 40.0.x. +- Add ``OpenSSL.SSL.Connection.DTLSv1_get_timeout`` and ``OpenSSL.SSL.Connection.DTLSv1_handle_timeout`` + to support DTLS timeouts `#1180 <https://github.com/pyca/pyopenssl/pull/1180>`_. + 23.0.0 (2023-01-01) ------------------- @@ -111,7 +194,7 @@ Backward-incompatible changes: ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ - Remove support for SSLv2 and SSLv3. -- The minimum ``cryptography`` version is now 38.0.x (and we now pin releases +- The minimum ``cryptography`` version is now 38.0.x (and we now pin releases against ``cryptography`` major versions to prevent future breakage) - The ``OpenSSL.crypto.X509StoreContextError`` exception has been refactored, changing its internal attributes. diff --git a/contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py b/contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py index dfbd1094e9d..ae07442f5d1 100644 --- a/contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py +++ b/contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py @@ -1,5 +1,6 @@ import os import socket +import typing from errno import errorcode from functools import partial, wraps from itertools import chain, count @@ -8,18 +9,32 @@ from weakref import WeakValueDictionary from OpenSSL._util import ( UNSPECIFIED as _UNSPECIFIED, +) +from OpenSSL._util import ( exception_from_error_queue as _exception_from_error_queue, +) +from OpenSSL._util import ( ffi as _ffi, +) +from OpenSSL._util import ( lib as _lib, +) +from OpenSSL._util import ( make_assert as _make_assert, +) +from OpenSSL._util import ( no_zero_allocator as _no_zero_allocator, +) +from OpenSSL._util import ( path_bytes as _path_bytes, +) +from OpenSSL._util import ( text_to_bytes_and_warn as _text_to_bytes_and_warn, ) from OpenSSL.crypto import ( FILETYPE_PEM, - PKey, X509, + PKey, X509Name, X509Store, _PassphraseHelper, @@ -123,6 +138,7 @@ __all__ = [ "Session", "Context", "Connection", + "X509VerificationCodes", ] @@ -216,6 +232,12 @@ try: except AttributeError: pass +try: + OP_LEGACY_SERVER_CONNECT = _lib.SSL_OP_LEGACY_SERVER_CONNECT + __all__.append("OP_LEGACY_SERVER_CONNECT") +except AttributeError: + pass + OP_ALL = _lib.SSL_OP_ALL VERIFY_PEER = _lib.SSL_VERIFY_PEER @@ -250,6 +272,113 @@ SSL_CB_CONNECT_EXIT = _lib.SSL_CB_CONNECT_EXIT SSL_CB_HANDSHAKE_START = _lib.SSL_CB_HANDSHAKE_START SSL_CB_HANDSHAKE_DONE = _lib.SSL_CB_HANDSHAKE_DONE + +class X509VerificationCodes: + """ + Success and error codes for X509 verification, as returned by the + underlying ``X509_STORE_CTX_get_error()`` function and passed by pyOpenSSL + to verification callback functions. + + See `OpenSSL Verification Errors + <https://www.openssl.org/docs/manmaster/man3/X509_verify_cert_error_string.html#ERROR-CODES>`_ + for details. + """ + + OK = _lib.X509_V_OK + ERR_UNABLE_TO_GET_ISSUER_CERT = _lib.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT + ERR_UNABLE_TO_GET_CRL = _lib.X509_V_ERR_UNABLE_TO_GET_CRL + ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE = ( + _lib.X509_V_ERR_UNABLE_TO_DECRYPT_CERT_SIGNATURE + ) + ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE = ( + _lib.X509_V_ERR_UNABLE_TO_DECRYPT_CRL_SIGNATURE + ) + ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY = ( + _lib.X509_V_ERR_UNABLE_TO_DECODE_ISSUER_PUBLIC_KEY + ) + ERR_CERT_SIGNATURE_FAILURE = _lib.X509_V_ERR_CERT_SIGNATURE_FAILURE + ERR_CRL_SIGNATURE_FAILURE = _lib.X509_V_ERR_CRL_SIGNATURE_FAILURE + ERR_CERT_NOT_YET_VALID = _lib.X509_V_ERR_CERT_NOT_YET_VALID + ERR_CERT_HAS_EXPIRED = _lib.X509_V_ERR_CERT_HAS_EXPIRED + ERR_CRL_NOT_YET_VALID = _lib.X509_V_ERR_CRL_NOT_YET_VALID + ERR_CRL_HAS_EXPIRED = _lib.X509_V_ERR_CRL_HAS_EXPIRED + ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD = ( + _lib.X509_V_ERR_ERROR_IN_CERT_NOT_BEFORE_FIELD + ) + ERR_ERROR_IN_CERT_NOT_AFTER_FIELD = ( + _lib.X509_V_ERR_ERROR_IN_CERT_NOT_AFTER_FIELD + ) + ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD = ( + _lib.X509_V_ERR_ERROR_IN_CRL_LAST_UPDATE_FIELD + ) + ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD = ( + _lib.X509_V_ERR_ERROR_IN_CRL_NEXT_UPDATE_FIELD + ) + ERR_OUT_OF_MEM = _lib.X509_V_ERR_OUT_OF_MEM + ERR_DEPTH_ZERO_SELF_SIGNED_CERT = ( + _lib.X509_V_ERR_DEPTH_ZERO_SELF_SIGNED_CERT + ) + ERR_SELF_SIGNED_CERT_IN_CHAIN = _lib.X509_V_ERR_SELF_SIGNED_CERT_IN_CHAIN + ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY = ( + _lib.X509_V_ERR_UNABLE_TO_GET_ISSUER_CERT_LOCALLY + ) + ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE = ( + _lib.X509_V_ERR_UNABLE_TO_VERIFY_LEAF_SIGNATURE + ) + ERR_CERT_CHAIN_TOO_LONG = _lib.X509_V_ERR_CERT_CHAIN_TOO_LONG + ERR_CERT_REVOKED = _lib.X509_V_ERR_CERT_REVOKED + ERR_INVALID_CA = _lib.X509_V_ERR_INVALID_CA + ERR_PATH_LENGTH_EXCEEDED = _lib.X509_V_ERR_PATH_LENGTH_EXCEEDED + ERR_INVALID_PURPOSE = _lib.X509_V_ERR_INVALID_PURPOSE + ERR_CERT_UNTRUSTED = _lib.X509_V_ERR_CERT_UNTRUSTED + ERR_CERT_REJECTED = _lib.X509_V_ERR_CERT_REJECTED + ERR_SUBJECT_ISSUER_MISMATCH = _lib.X509_V_ERR_SUBJECT_ISSUER_MISMATCH + ERR_AKID_SKID_MISMATCH = _lib.X509_V_ERR_AKID_SKID_MISMATCH + ERR_AKID_ISSUER_SERIAL_MISMATCH = ( + _lib.X509_V_ERR_AKID_ISSUER_SERIAL_MISMATCH + ) + ERR_KEYUSAGE_NO_CERTSIGN = _lib.X509_V_ERR_KEYUSAGE_NO_CERTSIGN + ERR_UNABLE_TO_GET_CRL_ISSUER = _lib.X509_V_ERR_UNABLE_TO_GET_CRL_ISSUER + ERR_UNHANDLED_CRITICAL_EXTENSION = ( + _lib.X509_V_ERR_UNHANDLED_CRITICAL_EXTENSION + ) + ERR_KEYUSAGE_NO_CRL_SIGN = _lib.X509_V_ERR_KEYUSAGE_NO_CRL_SIGN + ERR_UNHANDLED_CRITICAL_CRL_EXTENSION = ( + _lib.X509_V_ERR_UNHANDLED_CRITICAL_CRL_EXTENSION + ) + ERR_INVALID_NON_CA = _lib.X509_V_ERR_INVALID_NON_CA + ERR_PROXY_PATH_LENGTH_EXCEEDED = _lib.X509_V_ERR_PROXY_PATH_LENGTH_EXCEEDED + ERR_KEYUSAGE_NO_DIGITAL_SIGNATURE = ( + _lib.X509_V_ERR_KEYUSAGE_NO_DIGITAL_SIGNATURE + ) + ERR_PROXY_CERTIFICATES_NOT_ALLOWED = ( + _lib.X509_V_ERR_PROXY_CERTIFICATES_NOT_ALLOWED + ) + ERR_INVALID_EXTENSION = _lib.X509_V_ERR_INVALID_EXTENSION + ERR_INVALID_POLICY_EXTENSION = _lib.X509_V_ERR_INVALID_POLICY_EXTENSION + ERR_NO_EXPLICIT_POLICY = _lib.X509_V_ERR_NO_EXPLICIT_POLICY + ERR_DIFFERENT_CRL_SCOPE = _lib.X509_V_ERR_DIFFERENT_CRL_SCOPE + ERR_UNSUPPORTED_EXTENSION_FEATURE = ( + _lib.X509_V_ERR_UNSUPPORTED_EXTENSION_FEATURE + ) + ERR_UNNESTED_RESOURCE = _lib.X509_V_ERR_UNNESTED_RESOURCE + ERR_PERMITTED_VIOLATION = _lib.X509_V_ERR_PERMITTED_VIOLATION + ERR_EXCLUDED_VIOLATION = _lib.X509_V_ERR_EXCLUDED_VIOLATION + ERR_SUBTREE_MINMAX = _lib.X509_V_ERR_SUBTREE_MINMAX + ERR_UNSUPPORTED_CONSTRAINT_TYPE = ( + _lib.X509_V_ERR_UNSUPPORTED_CONSTRAINT_TYPE + ) + ERR_UNSUPPORTED_CONSTRAINT_SYNTAX = ( + _lib.X509_V_ERR_UNSUPPORTED_CONSTRAINT_SYNTAX + ) + ERR_UNSUPPORTED_NAME_SYNTAX = _lib.X509_V_ERR_UNSUPPORTED_NAME_SYNTAX + ERR_CRL_PATH_VALIDATION_ERROR = _lib.X509_V_ERR_CRL_PATH_VALIDATION_ERROR + ERR_HOSTNAME_MISMATCH = _lib.X509_V_ERR_HOSTNAME_MISMATCH + ERR_EMAIL_MISMATCH = _lib.X509_V_ERR_EMAIL_MISMATCH + ERR_IP_ADDRESS_MISMATCH = _lib.X509_V_ERR_IP_ADDRESS_MISMATCH + ERR_APPLICATION_VERIFICATION = _lib.X509_V_ERR_APPLICATION_VERIFICATION + + # Taken from https://golang.org/src/crypto/x509/root_linux.go _CERTIFICATE_FILE_LOCATIONS = [ "/etc/ssl/certs/ca-certificates.crt", # Debian/Ubuntu/Gentoo etc. @@ -689,7 +818,7 @@ class Context: not be used. """ - _methods = { + _methods: typing.ClassVar[typing.Dict] = { SSLv23_METHOD: (_lib.TLS_method, None), TLSv1_METHOD: (_lib.TLS_method, TLS1_VERSION), TLSv1_1_METHOD: (_lib.TLS_method, TLS1_1_VERSION), @@ -847,9 +976,9 @@ class Context: the ``[email protected]`` `Homebrew <https://brew.sh>`_ formula installed in the default location. * Windows will not work. - * manylinux1 cryptography wheels will work on most common Linux + * manylinux cryptography wheels will work on most common Linux distributions in pyOpenSSL 17.1.0 and above. pyOpenSSL detects the - manylinux1 wheel and attempts to load roots via a fallback path. + manylinux wheel and attempts to load roots via a fallback path. :return: None """ @@ -874,13 +1003,13 @@ class Context: default_dir = _ffi.string(_lib.X509_get_default_cert_dir()) default_file = _ffi.string(_lib.X509_get_default_cert_file()) # Now we check to see if the default_dir and default_file are set - # to the exact values we use in our manylinux1 builds. If they are + # to the exact values we use in our manylinux builds. If they are # then we know to load the fallbacks if ( default_dir == _CRYPTOGRAPHY_MANYLINUX_CA_DIR and default_file == _CRYPTOGRAPHY_MANYLINUX_CA_FILE ): - # This is manylinux1, let's load our fallback paths + # This is manylinux, let's load our fallback paths self._fallback_default_verify_paths( _CERTIFICATE_FILE_LOCATIONS, _CERTIFICATE_PATH_LOCATIONS ) @@ -899,7 +1028,7 @@ class Context: def _fallback_default_verify_paths(self, file_path, dir_path): """ Default verify paths are based on the compiled version of OpenSSL. - However, when pyca/cryptography is compiled as a manylinux1 wheel + However, when pyca/cryptography is compiled as a manylinux wheel that compiled location can potentially be wrong. So, like Go, we will try a predefined set of paths and attempt to load roots from there. @@ -1259,8 +1388,8 @@ class Context: for ca_name in certificate_authorities: if not isinstance(ca_name, X509Name): raise TypeError( - "client CAs must be X509Name objects, not %s " - "objects" % (type(ca_name).__name__,) + "client CAs must be X509Name objects, not {} " + "objects".format(type(ca_name).__name__) ) copy = _lib.X509_NAME_dup(ca_name._name) _openssl_assert(copy != _ffi.NULL) @@ -1663,8 +1792,7 @@ class Connection: """ if self._socket is None: raise AttributeError( - "'%s' object has no attribute '%s'" - % (self.__class__.__name__, name) + f"'{self.__class__.__name__}' object has no attribute '{name}'" ) else: return getattr(self._socket, name) @@ -1916,7 +2044,6 @@ class Connection: buf = _text_to_bytes_and_warn("buf", buf) with _ffi.from_buffer(buf) as data: - left_to_send = len(buf) total_sent = 0 @@ -2160,6 +2287,37 @@ class Connection: if result < 0: self._raise_ssl_error(self._ssl, result) + def DTLSv1_get_timeout(self): + """ + Determine when the DTLS SSL object next needs to perform internal + processing due to the passage of time. + + When the returned number of seconds have passed, the + :meth:`DTLSv1_handle_timeout` method needs to be called. + + :return: The time left in seconds before the next timeout or `None` + if no timeout is currently active. + """ + ptv_sec = _ffi.new("time_t *") + ptv_usec = _ffi.new("long *") + if _lib.Cryptography_DTLSv1_get_timeout(self._ssl, ptv_sec, ptv_usec): + return ptv_sec[0] + (ptv_usec[0] / 1000000) + else: + return None + + def DTLSv1_handle_timeout(self): + """ + Handles any timeout events which have become pending on a DTLS SSL + object. + + :return: `True` if there was a pending timeout, `False` otherwise. + """ + result = _lib.DTLSv1_handle_timeout(self._ssl) + if result < 0: + self._raise_ssl_error(self._ssl, result) + else: + return bool(result) + def bio_shutdown(self): """ If the Connection was created with a memory BIO, this method can be diff --git a/contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py b/contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py index 0af3acdb8fd..2442ee6f054 100644 --- a/contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py +++ b/contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py @@ -17,7 +17,6 @@ from OpenSSL.version import ( __version__, ) - __all__ = [ "SSL", "crypto", diff --git a/contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py b/contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py index a9b673c53f5..cccc216bbda 100644 --- a/contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py +++ b/contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py @@ -1,6 +1,7 @@ import calendar import datetime import functools +import typing from base64 import b16encode from functools import partial from os import PathLike @@ -22,23 +23,36 @@ from cryptography import utils, x509 from cryptography.hazmat.primitives.asymmetric import ( dsa, ec, - ed25519, ed448, + ed25519, rsa, ) from OpenSSL._util import ( UNSPECIFIED as _UNSPECIFIED, +) +from OpenSSL._util import ( byte_string as _byte_string, +) +from OpenSSL._util import ( exception_from_error_queue as _exception_from_error_queue, +) +from OpenSSL._util import ( ffi as _ffi, +) +from OpenSSL._util import ( lib as _lib, +) +from OpenSSL._util import ( make_assert as _make_assert, +) +from OpenSSL._util import ( path_bytes as _path_bytes, +) +from OpenSSL._util import ( text_to_bytes_and_warn as _text_to_bytes_and_warn, ) - __all__ = [ "FILETYPE_PEM", "FILETYPE_ASN1", @@ -63,7 +77,6 @@ __all__ = [ "dump_privatekey", "Revoked", "CRL", - "PKCS7", "PKCS12", "NetscapeSPKI", "load_publickey", @@ -74,8 +87,6 @@ __all__ = [ "verify", "dump_crl", "load_crl", - "load_pkcs7_data", - "load_pkcs12", ] @@ -114,7 +125,7 @@ def _untested_error(where: str) -> NoReturn: encountered isn't one that's exercised by the test suite so future behavior of pyOpenSSL is now somewhat less predictable. """ - raise RuntimeError("Unknown %s failure" % (where,)) + raise RuntimeError(f"Unknown {where} failure") def _new_mem_buf(buffer: Optional[bytes] = None) -> Any: @@ -451,7 +462,7 @@ class _EllipticCurve: circumstance. """ if isinstance(other, _EllipticCurve): - return super(_EllipticCurve, self).__ne__(other) + return super().__ne__(other) return NotImplemented @classmethod @@ -521,7 +532,7 @@ class _EllipticCurve: self.name = name def __repr__(self) -> str: - return "<Curve %r>" % (self.name,) + return f"<Curve {self.name!r}>" def _to_EC_KEY(self) -> Any: """ @@ -605,14 +616,15 @@ class X509Name: def __setattr__(self, name: str, value: Any) -> None: if name.startswith("_"): - return super(X509Name, self).__setattr__(name, value) + return super().__setattr__(name, value) # Note: we really do not want str subclasses here, so we do not use # isinstance. - if type(name) is not str: + if type(name) is not str: # noqa: E721 raise TypeError( - "attribute name must be string, not '%.200s'" - % (type(value).__name__,) + "attribute name must be string, not '{:.200}'".format( + type(value).__name__ + ) ) nid = _lib.OBJ_txt2nid(_byte_string(name)) @@ -704,7 +716,7 @@ class X509Name: ) _openssl_assert(format_result != _ffi.NULL) - return "<X509Name object '%s'>" % ( + return "<X509Name object '{}'>".format( _ffi.string(result_buffer).decode("utf-8"), ) @@ -842,7 +854,7 @@ class X509Extension: _lib.X509_EXTENSION_get_object(self._extension) ) - _prefixes = { + _prefixes: typing.ClassVar[typing.Dict[int, str]] = { _lib.GEN_EMAIL: "email", _lib.GEN_DNS: "DNS", _lib.GEN_URI: "URI", @@ -904,7 +916,14 @@ class X509Extension: """ obj = _lib.X509_EXTENSION_get_object(self._extension) nid = _lib.OBJ_obj2nid(obj) - return _ffi.string(_lib.OBJ_nid2sn(nid)) + # OpenSSL 3.1.0 has a bug where nid2sn returns NULL for NIDs that + # previously returned UNDEF. This is a workaround for that issue. + # https://github.com/openssl/openssl/commit/908ba3ed9adbb3df90f76 + buf = _lib.OBJ_nid2sn(nid) + if buf != _ffi.NULL: + return _ffi.string(buf) + else: + return b"UNDEF" def get_data(self) -> bytes: """ @@ -922,6 +941,19 @@ class X509Extension: return _ffi.buffer(char_result, result_length)[:] +_X509ExtensionInternal = X509Extension +utils.deprecated( + X509Extension, + __name__, + ( + "X509Extension support in pyOpenSSL is deprecated. You should use the " + "APIs in cryptography." + ), + DeprecationWarning, + name="X509Extension", +) + + class X509Req: """ An X.509 certificate signing requests. @@ -1003,6 +1035,12 @@ class X509Req: :param int version: The version number. :return: ``None`` """ + if not isinstance(version, int): + raise TypeError("version must be an int") + if version != 0: + raise ValueError( + "Invalid version. The only valid version for X509Req is 0." + ) set_result = _lib.X509_REQ_set_version(self._req, version) _openssl_assert(set_result == 1) @@ -1038,7 +1076,9 @@ class X509Req: return name - def add_extensions(self, extensions: Iterable[X509Extension]) -> None: + def add_extensions( + self, extensions: Iterable[_X509ExtensionInternal] + ) -> None: """ Add extensions to the certificate signing request. @@ -1052,7 +1092,7 @@ class X509Req: stack = _ffi.gc(stack, _lib.sk_X509_EXTENSION_free) for ext in extensions: - if not isinstance(ext, X509Extension): + if not isinstance(ext, _X509ExtensionInternal): raise ValueError("One of the elements is not an X509Extension") # TODO push can fail (here and elsewhere) @@ -1061,7 +1101,7 @@ class X509Req: add_result = _lib.X509_REQ_add_extensions(self._req, stack) _openssl_assert(add_result == 1) - def get_extensions(self) -> List[X509Extension]: + def get_extensions(self) -> List[_X509ExtensionInternal]: """ Get X.509 extensions in the certificate signing request. @@ -1081,7 +1121,7 @@ class X509Req: ) for i in range(_lib.sk_X509_EXTENSION_num(native_exts_obj)): - ext = X509Extension.__new__(X509Extension) + ext = _X509ExtensionInternal.__new__(_X509ExtensionInternal) extension = _lib.X509_EXTENSION_dup( _lib.sk_X509_EXTENSION_value(native_exts_obj, i) ) @@ -1282,8 +1322,10 @@ class X509: .. versionadded:: 0.13 """ - algor = _lib.X509_get0_tbs_sigalg(self._x509) - nid = _lib.OBJ_obj2nid(algor.algorithm) + sig_alg = _lib.X509_get0_tbs_sigalg(self._x509) + alg = _ffi.new("ASN1_OBJECT **") + _lib.X509_ALGOR_get0(alg, _ffi.NULL, _ffi.NULL, sig_alg) + nid = _lib.OBJ_obj2nid(alg[0]) if nid == _lib.NID_undef: raise ValueError("Undefined signature algorithm") return _ffi.string(_lib.OBJ_nid2ln(nid)) @@ -1427,7 +1469,9 @@ class X509: time_string = time_bytes.decode("utf-8") not_after = datetime.datetime.strptime(time_string, "%Y%m%d%H%M%SZ") - return not_after < datetime.datetime.utcnow() + UTC = datetime.timezone.utc + utcnow = datetime.datetime.now(UTC).replace(tzinfo=None) + return not_after < utcnow def _get_boundary_time(self, which: Any) -> Optional[bytes]: return _get_asn1_time(which(self._x509)) @@ -1573,7 +1617,9 @@ class X509: """ return _lib.X509_get_ext_count(self._x509) - def add_extensions(self, extensions: Iterable[X509Extension]) -> None: + def add_extensions( + self, extensions: Iterable[_X509ExtensionInternal] + ) -> None: """ Add extensions to the certificate. @@ -1582,14 +1628,14 @@ class X509: :return: ``None`` """ for ext in extensions: - if not isinstance(ext, X509Extension): + if not isinstance(ext, _X509ExtensionInternal): raise ValueError("One of the elements is not an X509Extension") add_result = _lib.X509_add_ext(self._x509, ext._extension, -1) if not add_result: _raise_current_error() - def get_extension(self, index: int) -> X509Extension: + def get_extension(self, index: int) -> _X509ExtensionInternal: """ Get a specific extension of the certificate by index. @@ -1603,7 +1649,7 @@ class X509: .. versionadded:: 0.12 """ - ext = X509Extension.__new__(X509Extension) + ext = _X509ExtensionInternal.__new__(_X509ExtensionInternal) ext._extension = _lib.X509_get_ext(self._x509, index) if ext._extension == _ffi.NULL: raise IndexError("extension index out of bounds") @@ -1632,7 +1678,6 @@ class X509StoreFlags: POLICY_CHECK: int = _lib.X509_V_FLAG_POLICY_CHECK EXPLICIT_POLICY: int = _lib.X509_V_FLAG_EXPLICIT_POLICY INHIBIT_MAP: int = _lib.X509_V_FLAG_INHIBIT_MAP - NOTIFY_POLICY: int = _lib.X509_V_FLAG_NOTIFY_POLICY CHECK_SS_SIGNATURE: int = _lib.X509_V_FLAG_CHECK_SS_SIGNATURE PARTIAL_CHAIN: int = _lib.X509_V_FLAG_PARTIAL_CHAIN @@ -1677,7 +1722,9 @@ class X509Store: res = _lib.X509_STORE_add_cert(self._store, cert._x509) _openssl_assert(res == 1) - def add_crl(self, crl: "CRL") -> None: + def add_crl( + self, crl: Union["_CRLInternal", x509.CertificateRevocationList] + ) -> None: """ Add a certificate revocation list to this store. @@ -1687,11 +1734,29 @@ class X509Store: .. versionadded:: 16.1.0 - :param CRL crl: The certificate revocation list to add to this store. + :param crl: The certificate revocation list to add to this store. + :type crl: ``Union[CRL, cryptography.x509.CertificateRevocationList]`` :return: ``None`` if the certificate revocation list was added successfully. """ - _openssl_assert(_lib.X509_STORE_add_crl(self._store, crl._crl) != 0) + if isinstance(crl, x509.CertificateRevocationList): + from cryptography.hazmat.primitives.serialization import Encoding + + bio = _new_mem_buf(crl.public_bytes(Encoding.DER)) + openssl_crl = _lib.d2i_X509_CRL_bio(bio, _ffi.NULL) + if openssl_crl == _ffi.NULL: + _raise_current_error() + + crl = _ffi.gc(openssl_crl, _lib.X509_CRL_free) + elif isinstance(crl, _CRLInternal): + crl = crl._crl + else: + raise TypeError( + "CRL must be of type OpenSSL.crypto.CRL or " + "cryptography.x509.CertificateRevocationList" + ) + + _openssl_assert(_lib.X509_STORE_add_crl(self._store, crl) != 0) def set_flags(self, flags: int) -> None: """ @@ -1803,7 +1868,7 @@ class X509StoreContextError(Exception): def __init__( self, message: str, errors: List[Any], certificate: X509 ) -> None: - super(X509StoreContextError, self).__init__(message) + super().__init__(message) self.errors = errors self.certificate = certificate @@ -2155,7 +2220,7 @@ class Revoked: # which differs from crl_reasons of crypto/x509v3/v3_enum.c that matches # OCSP_crl_reason_str. We use the latter, just like the command line # program. - _crl_reasons = [ + _crl_reasons: typing.ClassVar[typing.List[bytes]] = [ b"unspecified", b"keyCompromise", b"CACompromise", @@ -2323,6 +2388,19 @@ class Revoked: return _get_asn1_time(dt) +_RevokedInternal = Revoked +utils.deprecated( + Revoked, + __name__, + ( + "CRL support in pyOpenSSL is deprecated. You should use the APIs " + "in cryptography." + ), + DeprecationWarning, + name="Revoked", +) + + class CRL: """ A certificate revocation list. @@ -2342,13 +2420,13 @@ class CRL: """ from cryptography.x509 import load_der_x509_crl - der = dump_crl(FILETYPE_ASN1, self) + der = _dump_crl_internal(FILETYPE_ASN1, self) return load_der_x509_crl(der) @classmethod def from_cryptography( cls, crypto_crl: x509.CertificateRevocationList - ) -> "CRL": + ) -> "_CRLInternal": """ Construct based on a ``cryptography`` *crypto_crl*. @@ -2365,9 +2443,9 @@ class CRL: from cryptography.hazmat.primitives.serialization import Encoding der = crypto_crl.public_bytes(Encoding.DER) - return load_crl(FILETYPE_ASN1, der) + return _load_crl_internal(FILETYPE_ASN1, der) - def get_revoked(self) -> Optional[Tuple[Revoked, ...]]: + def get_revoked(self) -> Optional[Tuple[_RevokedInternal, ...]]: """ Return the revocations in this certificate revocation list. @@ -2382,14 +2460,14 @@ class CRL: for i in range(_lib.sk_X509_REVOKED_num(revoked_stack)): revoked = _lib.sk_X509_REVOKED_value(revoked_stack, i) revoked_copy = _lib.X509_REVOKED_dup(revoked) - pyrev = Revoked.__new__(Revoked) + pyrev = _RevokedInternal.__new__(_RevokedInternal) pyrev._revoked = _ffi.gc(revoked_copy, _lib.X509_REVOKED_free) results.append(pyrev) if results: return tuple(results) return None - def add_revoked(self, revoked: Revoked) -> None: + def add_revoked(self, revoked: _RevokedInternal) -> None: """ Add a revoked (by value not reference) to the CRL structure @@ -2552,54 +2630,20 @@ class CRL: if not sign_result: _raise_current_error() - return dump_crl(type, self) - - -class PKCS7: - - _pkcs7: Any - - def type_is_signed(self) -> bool: - """ - Check if this NID_pkcs7_signed object - - :return: True if the PKCS7 is of type signed - """ - return bool(_lib.PKCS7_type_is_signed(self._pkcs7)) - - def type_is_enveloped(self) -> bool: - """ - Check if this NID_pkcs7_enveloped object - - :returns: True if the PKCS7 is of type enveloped - """ - return bool(_lib.PKCS7_type_is_enveloped(self._pkcs7)) - - def type_is_signedAndEnveloped(self) -> bool: - """ - Check if this NID_pkcs7_signedAndEnveloped object - - :returns: True if the PKCS7 is of type signedAndEnveloped - """ - return bool(_lib.PKCS7_type_is_signedAndEnveloped(self._pkcs7)) - - def type_is_data(self) -> bool: - """ - Check if this NID_pkcs7_data object - - :return: True if the PKCS7 is of type data - """ - return bool(_lib.PKCS7_type_is_data(self._pkcs7)) + return _dump_crl_internal(type, self) - def get_type_name(self) -> str: - """ - Returns the type name of the PKCS7 structure - :return: A string with the typename - """ - nid = _lib.OBJ_obj2nid(self._pkcs7.type) - string_type = _lib.OBJ_nid2sn(nid) - return _ffi.string(string_type) +_CRLInternal = CRL +utils.deprecated( + CRL, + __name__, + ( + "CRL support in pyOpenSSL is deprecated. You should use the APIs " + "in cryptography." + ), + DeprecationWarning, + name="CRL", +) class PKCS12: @@ -2703,7 +2747,7 @@ class PKCS12: self._friendlyname = None elif not isinstance(name, bytes): raise TypeError( - "name must be a byte string or None (not %r)" % (name,) + f"name must be a byte string or None (not {name!r})" ) self._friendlyname = name @@ -2789,6 +2833,18 @@ class PKCS12: return _bio_to_string(bio) +utils.deprecated( + PKCS12, + __name__, + ( + "PKCS#12 support in pyOpenSSL is deprecated. You should use the APIs " + "in cryptography." + ), + DeprecationWarning, + name="PKCS12", +) + + class NetscapeSPKI: """ A Netscape SPKI object. @@ -2879,6 +2935,15 @@ class NetscapeSPKI: _openssl_assert(set_result == 1) +utils.deprecated( + NetscapeSPKI, + __name__, + "NetscapeSPKI support in pyOpenSSL is deprecated.", + DeprecationWarning, + name="NetscapeSPKI", +) + + class _PassphraseHelper: def __init__( self, @@ -2920,7 +2985,6 @@ class _PassphraseHelper: def raise_if_problem(self, exceptionType: Type[Exception] = Error) -> None: if self._problems: - # Flush the OpenSSL error queue try: _exception_from_error_queue(exceptionType) @@ -3124,6 +3188,15 @@ def sign(pkey: PKey, data: Union[str, bytes], digest: str) -> bytes: return _ffi.buffer(signature_buffer, signature_length[0])[:] +utils.deprecated( + sign, + __name__, + "sign() is deprecated. Use the equivilant APIs in cryptography.", + DeprecationWarning, + name="sign", +) + + def verify( cert: X509, signature: bytes, data: Union[str, bytes], digest: str ) -> None: @@ -3162,7 +3235,16 @@ def verify( _raise_current_error() -def dump_crl(type: int, crl: CRL) -> bytes: +utils.deprecated( + verify, + __name__, + "verify() is deprecated. Use the equivilant APIs in cryptography.", + DeprecationWarning, + name="verify", +) + + +def dump_crl(type: int, crl: _CRLInternal) -> bytes: """ Dump a certificate revocation list to a buffer. @@ -3191,7 +3273,20 @@ def dump_crl(type: int, crl: CRL) -> bytes: return _bio_to_string(bio) -def load_crl(type: int, buffer: Union[str, bytes]) -> CRL: +_dump_crl_internal = dump_crl +utils.deprecated( + dump_crl, + __name__, + ( + "CRL support in pyOpenSSL is deprecated. You should use the APIs " + "in cryptography." + ), + DeprecationWarning, + name="dump_crl", +) + + +def load_crl(type: int, buffer: Union[str, bytes]) -> _CRLInternal: """ Load Certificate Revocation List (CRL) data from a string *buffer*. *buffer* encoded with the type *type*. @@ -3216,146 +3311,19 @@ def load_crl(type: int, buffer: Union[str, bytes]) -> CRL: if crl == _ffi.NULL: _raise_current_error() - result = CRL.__new__(CRL) + result = _CRLInternal.__new__(_CRLInternal) result._crl = _ffi.gc(crl, _lib.X509_CRL_free) return result -def load_pkcs7_data(type: int, buffer: Union[str, bytes]) -> PKCS7: - """ - Load pkcs7 data from the string *buffer* encoded with the type - *type*. - - :param type: The file type (one of FILETYPE_PEM or FILETYPE_ASN1) - :param buffer: The buffer with the pkcs7 data. - :return: The PKCS7 object - """ - if isinstance(buffer, str): - buffer = buffer.encode("ascii") - - bio = _new_mem_buf(buffer) - - if type == FILETYPE_PEM: - pkcs7 = _lib.PEM_read_bio_PKCS7(bio, _ffi.NULL, _ffi.NULL, _ffi.NULL) - elif type == FILETYPE_ASN1: - pkcs7 = _lib.d2i_PKCS7_bio(bio, _ffi.NULL) - else: - raise ValueError("type argument must be FILETYPE_PEM or FILETYPE_ASN1") - - if pkcs7 == _ffi.NULL: - _raise_current_error() - - pypkcs7 = PKCS7.__new__(PKCS7) - pypkcs7._pkcs7 = _ffi.gc(pkcs7, _lib.PKCS7_free) - return pypkcs7 - - -utils.deprecated( - load_pkcs7_data, - __name__, - ( - "PKCS#7 support in pyOpenSSL is deprecated. You should use the APIs " - "in cryptography." - ), - DeprecationWarning, - name="load_pkcs7_data", -) - - -def load_pkcs12( - buffer: Union[str, bytes], passphrase: Optional[bytes] = None -) -> PKCS12: - """ - Load pkcs12 data from the string *buffer*. If the pkcs12 structure is - encrypted, a *passphrase* must be included. The MAC is always - checked and thus required. - - See also the man page for the C function :py:func:`PKCS12_parse`. - - :param buffer: The buffer the certificate is stored in - :param passphrase: (Optional) The password to decrypt the PKCS12 lump - :returns: The PKCS12 object - """ - passphrase = _text_to_bytes_and_warn("passphrase", passphrase) - - if isinstance(buffer, str): - buffer = buffer.encode("ascii") - - bio = _new_mem_buf(buffer) - - # Use null passphrase if passphrase is None or empty string. With PKCS#12 - # password based encryption no password and a zero length password are two - # different things, but OpenSSL implementation will try both to figure out - # which one works. - if not passphrase: - passphrase = _ffi.NULL - - p12 = _lib.d2i_PKCS12_bio(bio, _ffi.NULL) - if p12 == _ffi.NULL: - _raise_current_error() - p12 = _ffi.gc(p12, _lib.PKCS12_free) - - pkey = _ffi.new("EVP_PKEY**") - cert = _ffi.new("X509**") - cacerts = _ffi.new("Cryptography_STACK_OF_X509**") - - parse_result = _lib.PKCS12_parse(p12, passphrase, pkey, cert, cacerts) - if not parse_result: - _raise_current_error() - - cacerts = _ffi.gc(cacerts[0], _lib.sk_X509_free) - - # openssl 1.0.0 sometimes leaves an X509_check_private_key error in the - # queue for no particular reason. This error isn't interesting to anyone - # outside this function. It's not even interesting to us. Get rid of it. - try: - _raise_current_error() - except Error: - pass - - if pkey[0] == _ffi.NULL: - pykey = None - else: - pykey = PKey.__new__(PKey) - pykey._pkey = _ffi.gc(pkey[0], _lib.EVP_PKEY_free) - - if cert[0] == _ffi.NULL: - pycert = None - friendlyname = None - else: - pycert = X509._from_raw_x509_ptr(cert[0]) - - friendlyname_length = _ffi.new("int*") - friendlyname_buffer = _lib.X509_alias_get0( - cert[0], friendlyname_length - ) - friendlyname = _ffi.buffer( - friendlyname_buffer, friendlyname_length[0] - )[:] - if friendlyname_buffer == _ffi.NULL: - friendlyname = None - - pycacerts = [] - for i in range(_lib.sk_X509_num(cacerts)): - x509 = _lib.sk_X509_value(cacerts, i) - pycacert = X509._from_raw_x509_ptr(x509) - pycacerts.append(pycacert) - - pkcs12 = PKCS12.__new__(PKCS12) - pkcs12._pkey = pykey - pkcs12._cert = pycert - pkcs12._cacerts = pycacerts if pycacerts else None - pkcs12._friendlyname = friendlyname - return pkcs12 - - +_load_crl_internal = load_crl utils.deprecated( - load_pkcs12, + load_crl, __name__, ( - "PKCS#12 support in pyOpenSSL is deprecated. You should use the APIs " + "CRL support in pyOpenSSL is deprecated. You should use the APIs " "in cryptography." ), DeprecationWarning, - name="load_pkcs12", + name="load_crl", ) diff --git a/contrib/python/pyOpenSSL/py3/OpenSSL/debug.py b/contrib/python/pyOpenSSL/py3/OpenSSL/debug.py index e39b128a7e2..e0ed3f81d62 100644 --- a/contrib/python/pyOpenSSL/py3/OpenSSL/debug.py +++ b/contrib/python/pyOpenSSL/py3/OpenSSL/debug.py @@ -1,17 +1,13 @@ -from __future__ import print_function - import ssl import sys import cffi - import cryptography import OpenSSL.SSL from . import version - _env_info = """\ pyOpenSSL: {pyopenssl} cryptography: {cryptography} diff --git a/contrib/python/pyOpenSSL/py3/OpenSSL/version.py b/contrib/python/pyOpenSSL/py3/OpenSSL/version.py index 61ec17b2d8f..4120cff6dda 100644 --- a/contrib/python/pyOpenSSL/py3/OpenSSL/version.py +++ b/contrib/python/pyOpenSSL/py3/OpenSSL/version.py @@ -17,7 +17,7 @@ __all__ = [ "__version__", ] -__version__ = "23.0.0" +__version__ = "23.3.0" __title__ = "pyOpenSSL" __uri__ = "https://pyopenssl.org/" @@ -25,4 +25,4 @@ __summary__ = "Python wrapper module around the OpenSSL library" __author__ = "The pyOpenSSL developers" __email__ = "[email protected]" __license__ = "Apache License, Version 2.0" -__copyright__ = "Copyright 2001-2023 {0}".format(__author__) +__copyright__ = f"Copyright 2001-2023 {__author__}" diff --git a/contrib/python/pyOpenSSL/py3/tests/conftest.py b/contrib/python/pyOpenSSL/py3/tests/conftest.py index 5bae6b8f717..573b4b3d602 100644 --- a/contrib/python/pyOpenSSL/py3/tests/conftest.py +++ b/contrib/python/pyOpenSSL/py3/tests/conftest.py @@ -7,9 +7,10 @@ import pytest def pytest_report_header(config): - import OpenSSL.SSL import cryptography + import OpenSSL.SSL + return "OpenSSL: {openssl}\ncryptography: {cryptography}".format( openssl=OpenSSL.SSL.SSLeay_version(OpenSSL.SSL.SSLEAY_VERSION), cryptography=cryptography.__version__, diff --git a/contrib/python/pyOpenSSL/py3/tests/memdbg.py b/contrib/python/pyOpenSSL/py3/tests/memdbg.py index 0dd8c318e4e..5eed1ba8b0a 100644 --- a/contrib/python/pyOpenSSL/py3/tests/memdbg.py +++ b/contrib/python/pyOpenSSL/py3/tests/memdbg.py @@ -3,7 +3,6 @@ import traceback from cffi import api as _api - sys.modules["ssl"] = None sys.modules["_hashlib"] = None @@ -22,7 +21,7 @@ _ffi.cdef( char **backtrace_symbols(void *const *buffer, int size); void backtrace_symbols_fd(void *const *buffer, int size, int fd); """ -) # noqa +) _api = _ffi.verify( """ #include <openssl/crypto.h> @@ -79,7 +78,7 @@ def free(p): if p != _ffi.NULL: C.free(p) del heap[p] - log("free(0x%x)" % (int(_ffi.cast("int", p)),)) + log("free(0x{:x})".format(int(_ffi.cast("int", p)))) if _api.CRYPTO_set_mem_functions(malloc, realloc, free): diff --git a/contrib/python/pyOpenSSL/py3/tests/test_crypto.py b/contrib/python/pyOpenSSL/py3/tests/test_crypto.py index 44bbd0f2aba..6351f0d703c 100644 --- a/contrib/python/pyOpenSSL/py3/tests/test_crypto.py +++ b/contrib/python/pyOpenSSL/py3/tests/test_crypto.py @@ -6,34 +6,27 @@ Unit tests for :py:mod:`OpenSSL.crypto`. """ import base64 import sys -from datetime import datetime, timedelta +import warnings +from datetime import datetime, timedelta, timezone from subprocess import PIPE, Popen -from warnings import simplefilter - -from cryptography import x509 -from cryptography.hazmat.primitives import serialization -from cryptography.hazmat.primitives.asymmetric import ec, ed25519, ed448, rsa import flaky - import pytest +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec, ed448, ed25519, rsa -from OpenSSL._util import ffi as _ffi, lib as _lib +from OpenSSL._util import ffi as _ffi +from OpenSSL._util import lib as _lib from OpenSSL.crypto import ( - CRL, - Error, FILETYPE_ASN1, FILETYPE_PEM, FILETYPE_TEXT, - NetscapeSPKI, - PKCS12, - PKCS7, - PKey, - Revoked, TYPE_DSA, TYPE_RSA, X509, - X509Extension, + Error, + PKey, X509Name, X509Req, X509Store, @@ -42,26 +35,33 @@ from OpenSSL.crypto import ( X509StoreFlags, dump_certificate, dump_certificate_request, - dump_crl, dump_privatekey, dump_publickey, get_elliptic_curve, get_elliptic_curves, load_certificate, load_certificate_request, - load_crl, - load_pkcs12, - load_pkcs7_data, load_privatekey, load_publickey, sign, verify, ) +with pytest.warns(DeprecationWarning): + from OpenSSL.crypto import ( + CRL, + PKCS12, + NetscapeSPKI, + Revoked, + X509Extension, + dump_crl, + load_crl, + ) + from .util import ( - EqualityTestsMixin, NON_ASCII, WARNING_TYPE_EXPECTED, + EqualityTestsMixin, is_consistent_type, ) @@ -70,6 +70,10 @@ def normalize_privatekey_pem(pem): return dump_privatekey(FILETYPE_PEM, load_privatekey(FILETYPE_PEM, pem)) +def utcnow(): + return datetime.now(timezone.utc).replace(tzinfo=None) + + GOOD_CIPHER = "blowfish" BAD_CIPHER = "zippers" @@ -554,57 +558,6 @@ ywIDAQAB -----END PUBLIC KEY----- """ -# Some PKCS#7 stuff. Generated with the openssl command line: -# -# openssl crl2pkcs7 -inform pem -outform pem -certfile s.pem -nocrl -# -# with a certificate and key (but the key should be irrelevant) in s.pem -pkcs7Data = b"""\ ------BEGIN PKCS7----- -MIIDNwYJKoZIhvcNAQcCoIIDKDCCAyQCAQExADALBgkqhkiG9w0BBwGgggMKMIID -BjCCAm+gAwIBAgIBATANBgkqhkiG9w0BAQQFADB7MQswCQYDVQQGEwJTRzERMA8G -A1UEChMITTJDcnlwdG8xFDASBgNVBAsTC00yQ3J5cHRvIENBMSQwIgYDVQQDExtN -MkNyeXB0byBDZXJ0aWZpY2F0ZSBNYXN0ZXIxHTAbBgkqhkiG9w0BCQEWDm5ncHNA -cG9zdDEuY29tMB4XDTAwMDkxMDA5NTEzMFoXDTAyMDkxMDA5NTEzMFowUzELMAkG -A1UEBhMCU0cxETAPBgNVBAoTCE0yQ3J5cHRvMRIwEAYDVQQDEwlsb2NhbGhvc3Qx -HTAbBgkqhkiG9w0BCQEWDm5ncHNAcG9zdDEuY29tMFwwDQYJKoZIhvcNAQEBBQAD -SwAwSAJBAKy+e3dulvXzV7zoTZWc5TzgApr8DmeQHTYC8ydfzH7EECe4R1Xh5kwI -zOuuFfn178FBiS84gngaNcrFi0Z5fAkCAwEAAaOCAQQwggEAMAkGA1UdEwQCMAAw -LAYJYIZIAYb4QgENBB8WHU9wZW5TU0wgR2VuZXJhdGVkIENlcnRpZmljYXRlMB0G -A1UdDgQWBBTPhIKSvnsmYsBVNWjj0m3M2z0qVTCBpQYDVR0jBIGdMIGagBT7hyNp -65w6kxXlxb8pUU/+7Sg4AaF/pH0wezELMAkGA1UEBhMCU0cxETAPBgNVBAoTCE0y -Q3J5cHRvMRQwEgYDVQQLEwtNMkNyeXB0byBDQTEkMCIGA1UEAxMbTTJDcnlwdG8g -Q2VydGlmaWNhdGUgTWFzdGVyMR0wGwYJKoZIhvcNAQkBFg5uZ3BzQHBvc3QxLmNv -bYIBADANBgkqhkiG9w0BAQQFAAOBgQA7/CqT6PoHycTdhEStWNZde7M/2Yc6BoJu -VwnW8YxGO8Sn6UJ4FeffZNcYZddSDKosw8LtPOeWoK3JINjAk5jiPQ2cww++7QGG -/g5NDjxFZNDJP1dGiLAxPW6JXwov4v0FmdzfLOZ01jDcgQQZqEpYlgpuI5JEWUQ9 -Ho4EzbYCOaEAMQA= ------END PKCS7----- -""" - -pkcs7DataASN1 = base64.b64decode( - b""" -MIIDNwYJKoZIhvcNAQcCoIIDKDCCAyQCAQExADALBgkqhkiG9w0BBwGgggMKMIID -BjCCAm+gAwIBAgIBATANBgkqhkiG9w0BAQQFADB7MQswCQYDVQQGEwJTRzERMA8G -A1UEChMITTJDcnlwdG8xFDASBgNVBAsTC00yQ3J5cHRvIENBMSQwIgYDVQQDExtN -MkNyeXB0byBDZXJ0aWZpY2F0ZSBNYXN0ZXIxHTAbBgkqhkiG9w0BCQEWDm5ncHNA -cG9zdDEuY29tMB4XDTAwMDkxMDA5NTEzMFoXDTAyMDkxMDA5NTEzMFowUzELMAkG -A1UEBhMCU0cxETAPBgNVBAoTCE0yQ3J5cHRvMRIwEAYDVQQDEwlsb2NhbGhvc3Qx -HTAbBgkqhkiG9w0BCQEWDm5ncHNAcG9zdDEuY29tMFwwDQYJKoZIhvcNAQEBBQAD -SwAwSAJBAKy+e3dulvXzV7zoTZWc5TzgApr8DmeQHTYC8ydfzH7EECe4R1Xh5kwI -zOuuFfn178FBiS84gngaNcrFi0Z5fAkCAwEAAaOCAQQwggEAMAkGA1UdEwQCMAAw -LAYJYIZIAYb4QgENBB8WHU9wZW5TU0wgR2VuZXJhdGVkIENlcnRpZmljYXRlMB0G -A1UdDgQWBBTPhIKSvnsmYsBVNWjj0m3M2z0qVTCBpQYDVR0jBIGdMIGagBT7hyNp -65w6kxXlxb8pUU/+7Sg4AaF/pH0wezELMAkGA1UEBhMCU0cxETAPBgNVBAoTCE0y -Q3J5cHRvMRQwEgYDVQQLEwtNMkNyeXB0byBDQTEkMCIGA1UEAxMbTTJDcnlwdG8g -Q2VydGlmaWNhdGUgTWFzdGVyMR0wGwYJKoZIhvcNAQkBFg5uZ3BzQHBvc3QxLmNv -bYIBADANBgkqhkiG9w0BAQQFAAOBgQA7/CqT6PoHycTdhEStWNZde7M/2Yc6BoJu -VwnW8YxGO8Sn6UJ4FeffZNcYZddSDKosw8LtPOeWoK3JINjAk5jiPQ2cww++7QGG -/g5NDjxFZNDJP1dGiLAxPW6JXwov4v0FmdzfLOZ01jDcgQQZqEpYlgpuI5JEWUQ9 -Ho4EzbYCOaEAMQA= -""" -) - crlData = b"""\ -----BEGIN X509 CRL----- MIIBWzCBxTANBgkqhkiG9w0BAQQFADBYMQswCQYDVQQGEwJVUzELMAkGA1UECBMC @@ -1601,20 +1554,12 @@ class TestX509Req(_PKeyInteractionTestsMixin): """ `X509Req.set_version` sets the X.509 version of the certificate request. `X509Req.get_version` returns the X.509 version of the - certificate request. The only defined version is 0. Others may or - may not be supported depending on backend. + certificate request. The only defined version is 0. """ request = X509Req() assert request.get_version() == 0 request.set_version(0) assert request.get_version() == 0 - try: - request.set_version(1) - assert request.get_version() == 1 - request.set_version(3) - assert request.get_version() == 3 - except Error: - pass def test_version_wrong_args(self): """ @@ -1624,6 +1569,8 @@ class TestX509Req(_PKeyInteractionTestsMixin): request = X509Req() with pytest.raises(TypeError): request.set_version("foo") + with pytest.raises(ValueError): + request.set_version(2) def test_get_subject(self): """ @@ -1681,6 +1628,14 @@ class TestX509Req(_PKeyInteractionTestsMixin): exts = request.get_extensions() assert len(exts) == 2 + def test_undef_oid(self): + assert ( + X509Extension( + b"1.2.3.4.5.6.7", False, b"DER:05:00" + ).get_short_name() + == b"UNDEF" + ) + def test_add_extensions_wrong_args(self): """ `X509Req.add_extensions` raises `TypeError` if called with a @@ -1916,14 +1871,14 @@ class TestX509(_PKeyInteractionTestsMixin): current time plus the number of seconds passed in. """ cert = load_certificate(FILETYPE_PEM, self.pemData) - not_before_min = datetime.utcnow().replace(microsecond=0) + timedelta( + not_before_min = utcnow().replace(microsecond=0) + timedelta( seconds=100 ) cert.gmtime_adj_notBefore(100) not_before = datetime.strptime( cert.get_notBefore().decode(), "%Y%m%d%H%M%SZ" ) - not_before_max = datetime.utcnow() + timedelta(seconds=100) + not_before_max = utcnow() + timedelta(seconds=100) assert not_before_min <= not_before <= not_before_max def test_gmtime_adj_notAfter_wrong_args(self): @@ -1942,14 +1897,14 @@ class TestX509(_PKeyInteractionTestsMixin): to be the current time plus the number of seconds passed in. """ cert = load_certificate(FILETYPE_PEM, self.pemData) - not_after_min = datetime.utcnow().replace(microsecond=0) + timedelta( + not_after_min = utcnow().replace(microsecond=0) + timedelta( seconds=100 ) cert.gmtime_adj_notAfter(100) not_after = datetime.strptime( cert.get_notAfter().decode(), "%Y%m%d%H%M%SZ" ) - not_after_max = datetime.utcnow() + timedelta(seconds=100) + not_after_max = utcnow() + timedelta(seconds=100) assert not_after_min <= not_after <= not_after_max def test_has_expired(self): @@ -2384,7 +2339,7 @@ class TestX509Store: class TestPKCS12: """ - Test for `OpenSSL.crypto.PKCS12` and `OpenSSL.crypto.load_pkcs12`. + Test for `OpenSSL.crypto.PKCS12`. """ def test_type(self): @@ -2427,7 +2382,7 @@ class TestPKCS12: def test_key_only(self): """ A `PKCS12` with only a private key can be exported using - `PKCS12.export` and loaded again using `load_pkcs12`. + `PKCS12.export`. """ passwd = b"blah" p12 = PKCS12() @@ -2435,25 +2390,12 @@ class TestPKCS12: p12.set_privatekey(pkey) assert None is p12.get_certificate() assert pkey == p12.get_privatekey() - try: - dumped_p12 = p12.export(passphrase=passwd, iter=2, maciter=3) - except Error: - # Some versions of OpenSSL will throw an exception - # for this nearly useless PKCS12 we tried to generate: - # [('PKCS12 routines', 'PKCS12_create', 'invalid null argument')] - return - p12 = load_pkcs12(dumped_p12, passwd) - assert None is p12.get_ca_certificates() - assert None is p12.get_certificate() - - # OpenSSL fails to bring the key back to us. So sad. Perhaps in the - # future this will be improved. - assert isinstance(p12.get_privatekey(), (PKey, type(None))) + p12.export(passphrase=passwd, iter=2, maciter=3) def test_cert_only(self): """ A `PKCS12` with only a certificate can be exported using - `PKCS12.export` and loaded again using `load_pkcs12`. + `PKCS12.export`. """ passwd = b"blah" p12 = PKCS12() @@ -2461,51 +2403,26 @@ class TestPKCS12: p12.set_certificate(cert) assert cert == p12.get_certificate() assert None is p12.get_privatekey() - try: - dumped_p12 = p12.export(passphrase=passwd, iter=2, maciter=3) - except Error: - # Some versions of OpenSSL will throw an exception - # for this nearly useless PKCS12 we tried to generate: - # [('PKCS12 routines', 'PKCS12_create', 'invalid null argument')] - return - p12 = load_pkcs12(dumped_p12, passwd) - assert None is p12.get_privatekey() - - # OpenSSL fails to bring the cert back to us. Groany mcgroan. - assert isinstance(p12.get_certificate(), (X509, type(None))) - - # Oh ho. It puts the certificate into the ca certificates list, in - # fact. Totally bogus, I would think. Nevertheless, let's exploit - # that to check to see if it reconstructed the certificate we expected - # it to. At some point, hopefully this will change so that - # p12.get_certificate() is actually what returns the loaded - # certificate. - assert root_cert_pem == dump_certificate( - FILETYPE_PEM, p12.get_ca_certificates()[0] - ) + p12.export(passphrase=passwd, iter=2, maciter=3) - def gen_pkcs12( - self, cert_pem=None, key_pem=None, ca_pem=None, friendly_name=None - ): + def gen_pkcs12(self, cert_pem=None, key_pem=None, ca_pem=None): """ Generate a PKCS12 object with components from PEM. Verify that the set functions return None. """ p12 = PKCS12() - if cert_pem: - ret = p12.set_certificate(load_certificate(FILETYPE_PEM, cert_pem)) - assert ret is None - if key_pem: - ret = p12.set_privatekey(load_privatekey(FILETYPE_PEM, key_pem)) - assert ret is None + + ret = p12.set_certificate(load_certificate(FILETYPE_PEM, cert_pem)) + assert ret is None + + ret = p12.set_privatekey(load_privatekey(FILETYPE_PEM, key_pem)) + assert ret is None + if ca_pem: ret = p12.set_ca_certificates( (load_certificate(FILETYPE_PEM, ca_pem),) ) assert ret is None - if friendly_name: - ret = p12.set_friendlyname(friendly_name) - assert ret is None return p12 def check_recovery( @@ -2515,29 +2432,29 @@ class TestPKCS12: Use openssl program to confirm three components are recoverable from a PKCS12 string. """ - if key: - recovered_key = _runopenssl( - p12_str, - b"pkcs12", - b"-nocerts", - b"-nodes", - b"-passin", - b"pass:" + passwd, - *extra, - ) - assert recovered_key[-len(key) :] == key - if cert: - recovered_cert = _runopenssl( - p12_str, - b"pkcs12", - b"-clcerts", - b"-nodes", - b"-passin", - b"pass:" + passwd, - b"-nokeys", - *extra, - ) - assert recovered_cert[-len(cert) :] == cert + recovered_key = _runopenssl( + p12_str, + b"pkcs12", + b"-nocerts", + b"-nodes", + b"-passin", + b"pass:" + passwd, + *extra, + ).replace(b"\r\n", b"\n") + assert recovered_key[-len(key) :] == key + + recovered_cert = _runopenssl( + p12_str, + b"pkcs12", + b"-clcerts", + b"-nodes", + b"-passin", + b"pass:" + passwd, + b"-nokeys", + *extra, + ).replace(b"\r\n", b"\n") + assert recovered_cert[-len(cert) :] == cert + if ca: recovered_cert = _runopenssl( p12_str, @@ -2548,141 +2465,9 @@ class TestPKCS12: b"pass:" + passwd, b"-nokeys", *extra, - ) + ).replace(b"\r\n", b"\n") assert recovered_cert[-len(ca) :] == ca - def verify_pkcs12_container(self, p12): - """ - Verify that the PKCS#12 container contains the correct client - certificate and private key. - - :param p12: The PKCS12 instance to verify. - :type p12: `PKCS12` - """ - cert_pem = dump_certificate(FILETYPE_PEM, p12.get_certificate()) - key_pem = dump_privatekey(FILETYPE_PEM, p12.get_privatekey()) - assert (client_cert_pem, client_key_pem, None) == ( - cert_pem, - key_pem, - p12.get_ca_certificates(), - ) - - def test_load_pkcs12(self): - """ - A PKCS12 string generated using the openssl command line can be loaded - with `load_pkcs12` and its components extracted and examined. - """ - passwd = b"whatever" - pem = client_key_pem + client_cert_pem - p12_str = _runopenssl( - pem, - b"pkcs12", - b"-export", - b"-clcerts", - b"-passout", - b"pass:" + passwd, - ) - p12 = load_pkcs12(p12_str, passphrase=passwd) - self.verify_pkcs12_container(p12) - - def test_load_pkcs12_text_passphrase(self): - """ - A PKCS12 string generated using the openssl command line can be loaded - with `load_pkcs12` and its components extracted and examined. - Using text as passphrase instead of bytes. DeprecationWarning expected. - """ - pem = client_key_pem + client_cert_pem - passwd = b"whatever" - p12_str = _runopenssl( - pem, - b"pkcs12", - b"-export", - b"-clcerts", - b"-passout", - b"pass:" + passwd, - ) - with pytest.warns(DeprecationWarning) as w: - simplefilter("always") - p12 = load_pkcs12(p12_str, passphrase=b"whatever".decode("ascii")) - msg = "{0} for passphrase is no longer accepted, use bytes".format( - WARNING_TYPE_EXPECTED - ) - assert msg == str(w[-1].message) - - self.verify_pkcs12_container(p12) - - def test_load_pkcs12_no_passphrase(self): - """ - A PKCS12 string generated using openssl command line can be loaded with - `load_pkcs12` without a passphrase and its components extracted - and examined. - """ - pem = client_key_pem + client_cert_pem - p12_str = _runopenssl( - pem, b"pkcs12", b"-export", b"-clcerts", b"-passout", b"pass:" - ) - p12 = load_pkcs12(p12_str) - self.verify_pkcs12_container(p12) - - def _dump_and_load(self, dump_passphrase, load_passphrase): - """ - A helper method to dump and load a PKCS12 object. - """ - p12 = self.gen_pkcs12(client_cert_pem, client_key_pem) - dumped_p12 = p12.export(passphrase=dump_passphrase, iter=2, maciter=3) - return load_pkcs12(dumped_p12, passphrase=load_passphrase) - - def test_load_pkcs12_null_passphrase_load_empty(self): - """ - A PKCS12 string can be dumped with a null passphrase, loaded with an - empty passphrase with `load_pkcs12`, and its components - extracted and examined. - """ - self.verify_pkcs12_container( - self._dump_and_load(dump_passphrase=None, load_passphrase=b"") - ) - - def test_load_pkcs12_null_passphrase_load_null(self): - """ - A PKCS12 string can be dumped with a null passphrase, loaded with a - null passphrase with `load_pkcs12`, and its components - extracted and examined. - """ - self.verify_pkcs12_container( - self._dump_and_load(dump_passphrase=None, load_passphrase=None) - ) - - def test_load_pkcs12_empty_passphrase_load_empty(self): - """ - A PKCS12 string can be dumped with an empty passphrase, loaded with an - empty passphrase with `load_pkcs12`, and its components - extracted and examined. - """ - self.verify_pkcs12_container( - self._dump_and_load(dump_passphrase=b"", load_passphrase=b"") - ) - - def test_load_pkcs12_empty_passphrase_load_null(self): - """ - A PKCS12 string can be dumped with an empty passphrase, loaded with a - null passphrase with `load_pkcs12`, and its components - extracted and examined. - """ - self.verify_pkcs12_container( - self._dump_and_load(dump_passphrase=b"", load_passphrase=None) - ) - - def test_load_pkcs12_garbage(self): - """ - `load_pkcs12` raises `OpenSSL.crypto.Error` when passed - a string which is not a PKCS12 dump. - """ - passwd = b"whatever" - with pytest.raises(Error) as err: - load_pkcs12(b"fruit loops", passwd) - assert err.value.args[0][0][0] == "asn1 encoding routines" - assert len(err.value.args[0][0]) == 3 - def test_replace(self): """ `PKCS12.set_certificate` replaces the certificate in a PKCS12 @@ -2713,20 +2498,7 @@ class TestPKCS12: for friendly_name in [b"Serverlicious", None, b"###"]: p12.set_friendlyname(friendly_name) assert p12.get_friendlyname() == friendly_name - dumped_p12 = p12.export(passphrase=passwd, iter=2, maciter=3) - reloaded_p12 = load_pkcs12(dumped_p12, passwd) - assert p12.get_friendlyname() == reloaded_p12.get_friendlyname() - # We would use the openssl program to confirm the friendly - # name, but it is not possible. The pkcs12 command - # does not store the friendly name in the cert's - # alias, which we could then extract. - self.check_recovery( - dumped_p12, - key=server_key_pem, - cert=server_cert_pem, - ca=root_cert_pem, - passwd=passwd, - ) + p12.export(passphrase=passwd, iter=2, maciter=3) def test_various_empty_passphrases(self): """ @@ -2777,19 +2549,7 @@ class TestPKCS12: """ passwd = b"Lake Michigan" p12 = self.gen_pkcs12(server_cert_pem, server_key_pem, root_cert_pem) - dumped_p12 = p12.export(maciter=-1, passphrase=passwd, iter=2) - try: - recovered_p12 = load_pkcs12(dumped_p12, passwd) - # The person who generated this PCKS12 should be flogged, - # or better yet we should have a means to determine - # whether a PCKS12 had a MAC that was verified. - # Anyway, libopenssl chooses to allow it, so the - # pyopenssl binding does as well. - assert isinstance(recovered_p12, PKCS12) - except Error: - # Failing here with an exception is preferred as some openssl - # versions do. - pass + p12.export(maciter=-1, passphrase=passwd, iter=2) def test_zero_len_list_for_ca(self): """ @@ -2821,9 +2581,9 @@ class TestPKCS12: p12 = self.gen_pkcs12(server_cert_pem, server_key_pem, root_cert_pem) with pytest.warns(DeprecationWarning) as w: - simplefilter("always") + warnings.simplefilter("always") dumped_p12 = p12.export(passphrase=b"randomtext".decode("ascii")) - msg = "{0} for passphrase is no longer accepted, use bytes".format( + msg = "{} for passphrase is no longer accepted, use bytes".format( WARNING_TYPE_EXPECTED ) assert msg == str(w[-1].message) @@ -2849,7 +2609,7 @@ def _runopenssl(pem, *args): Run the command line openssl tool with the given arguments and write the given PEM to its stdin. Not safe for quotes. """ - proc = Popen([b"openssl"] + list(args), stdin=PIPE, stdout=PIPE) + proc = Popen([b"openssl", *list(args)], stdin=PIPE, stdout=PIPE) proc.stdin.write(pem) proc.stdin.close() output = proc.stdout.read() @@ -3288,36 +3048,6 @@ class TestFunction: with pytest.raises(Error): load_privatekey(FILETYPE_PEM, encrypted_key_pem, passphrase) - def test_load_pkcs7_data_pem(self): - """ - `load_pkcs7_data` accepts a PKCS#7 string and returns an instance of - `PKCS`. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert isinstance(pkcs7, PKCS7) - - def test_load_pkcs7_data_asn1(self): - """ - `load_pkcs7_data` accepts a bytes containing ASN1 data representing - PKCS#7 and returns an instance of `PKCS7`. - """ - pkcs7 = load_pkcs7_data(FILETYPE_ASN1, pkcs7DataASN1) - assert isinstance(pkcs7, PKCS7) - - def test_load_pkcs7_data_invalid(self): - """ - If the data passed to `load_pkcs7_data` is invalid, `Error` is raised. - """ - with pytest.raises(Error): - load_pkcs7_data(FILETYPE_PEM, b"foo") - - def test_load_pkcs7_type_invalid(self): - """ - If the type passed to `load_pkcs7_data`, `ValueError` is raised. - """ - with pytest.raises(ValueError): - load_pkcs7_data(object(), b"foo") - class TestLoadCertificate: """ @@ -3343,61 +3073,6 @@ class TestLoadCertificate: load_certificate(FILETYPE_ASN1, b"lol") -class TestPKCS7: - """ - Tests for `PKCS7`. - """ - - def test_type_is_signed(self): - """ - `PKCS7.type_is_signed` returns `True` if the PKCS7 object is of - the type *signed*. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert pkcs7.type_is_signed() - - def test_type_is_enveloped(self): - """ - `PKCS7.type_is_enveloped` returns `False` if the PKCS7 object is not - of the type *enveloped*. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert not pkcs7.type_is_enveloped() - - def test_type_is_signed_and_enveloped(self): - """ - `PKCS7.type_is_signedAndEnveloped` returns `False` - if the PKCS7 object is not of the type *signed and enveloped*. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert not pkcs7.type_is_signedAndEnveloped() - - def test_type_is_data(self): - """ - `PKCS7.type_is_data` returns `False` if the PKCS7 object is not of - the type data. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert not pkcs7.type_is_data() - - def test_get_type_name(self): - """ - `PKCS7.get_type_name` returns a `str` giving the - type name. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - assert pkcs7.get_type_name() == b"pkcs7-signedData" - - def test_attribute(self): - """ - If an attribute other than one of the methods tested here is accessed - on an instance of `PKCS7`, `AttributeError` is raised. - """ - pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data) - with pytest.raises(AttributeError): - pkcs7.foo - - class TestNetscapeSPKI(_PKeyInteractionTestsMixin): """ Tests for `OpenSSL.crypto.NetscapeSPKI`. @@ -3676,8 +3351,8 @@ class TestCRL: not emit a deprecation warning. """ crl = self._get_crl() - with pytest.warns(None) as catcher: - simplefilter("always") + with warnings.catch_warnings(record=True) as catcher: + warnings.simplefilter("always") assert 0 == len(catcher) dumped_crl = crl.export(self.cert, self.pkey, digest=b"md5") text = _runopenssl(dumped_crl, b"crl", b"-noout", b"-text") @@ -3830,7 +3505,8 @@ class TestCRL: buf = dump_crl(FILETYPE_PEM, crl) assert buf == crlData - def _make_test_crl(self, issuer_cert, issuer_key, certs=()): + @staticmethod + def _make_test_crl(issuer_cert, issuer_key, certs=()): """ Create a CRL. @@ -3856,7 +3532,55 @@ class TestCRL: crl.sign(issuer_cert, issuer_key, digest=b"sha512") return crl - def test_verify_with_revoked(self): + @staticmethod + def _make_test_crl_cryptography(issuer_cert, issuer_key, certs=()): + """ + Create a CRL using cryptography's API. + + :param list[X509] certs: A list of certificates to revoke. + :rtype: ``cryptography.x509.CertificateRevocationList`` + """ + from cryptography.x509.extensions import CRLReason, ReasonFlags + + builder = x509.CertificateRevocationListBuilder() + builder = builder.issuer_name( + X509.to_cryptography(issuer_cert).subject + ) + for cert in certs: + revoked = ( + x509.RevokedCertificateBuilder() + .serial_number(cert.get_serial_number()) + .revocation_date(datetime(2014, 6, 1, 0, 0, 0)) + .add_extension(CRLReason(ReasonFlags.unspecified), False) + .build() + ) + builder = builder.add_revoked_certificate(revoked) + + builder = builder.last_update(datetime(2014, 6, 1, 0, 0, 0)) + # The year 5000 is far into the future so that this CRL isn't + # considered to have expired. + builder = builder.next_update(datetime(5000, 6, 1, 0, 0, 0)) + + crl = builder.sign( + private_key=PKey.to_cryptography_key(issuer_key), + algorithm=hashes.SHA512(), + ) + return crl + + @pytest.mark.parametrize( + "create_crl", + [ + pytest.param( + _make_test_crl.__func__, + id="pyOpenSSL CRL", + ), + pytest.param( + _make_test_crl_cryptography.__func__, + id="cryptography CRL", + ), + ], + ) + def test_verify_with_revoked(self, create_crl): """ `verify_certificate` raises error when an intermediate certificate is revoked. @@ -3864,10 +3588,10 @@ class TestCRL: store = X509Store() store.add_cert(self.root_cert) store.add_cert(self.intermediate_cert) - root_crl = self._make_test_crl( + root_crl = create_crl( self.root_cert, self.root_key, certs=[self.intermediate_cert] ) - intermediate_crl = self._make_test_crl( + intermediate_crl = create_crl( self.intermediate_cert, self.intermediate_key, certs=[] ) store.add_crl(root_crl) @@ -3880,7 +3604,20 @@ class TestCRL: store_ctx.verify_certificate() assert str(err.value) == "certificate revoked" - def test_verify_with_missing_crl(self): + @pytest.mark.parametrize( + "create_crl", + [ + pytest.param( + _make_test_crl.__func__, + id="pyOpenSSL CRL", + ), + pytest.param( + _make_test_crl_cryptography.__func__, + id="cryptography CRL", + ), + ], + ) + def test_verify_with_missing_crl(self, create_crl): """ `verify_certificate` raises error when an intermediate certificate's CRL is missing. @@ -3888,7 +3625,7 @@ class TestCRL: store = X509Store() store.add_cert(self.root_cert) store.add_cert(self.intermediate_cert) - root_crl = self._make_test_crl( + root_crl = create_crl( self.root_cert, self.root_key, certs=[self.intermediate_cert] ) store.add_crl(root_crl) @@ -4235,7 +3972,7 @@ class TestX509StoreContext: @staticmethod def _create_ca_file(base_path, hash_directory, cacert): - ca_hash = "{:08x}.0".format(cacert.subject_name_hash()) + ca_hash = f"{cacert.subject_name_hash():08x}.0" cafile = base_path.join(hash_directory, ca_hash) cafile.write_binary( dump_certificate(FILETYPE_PEM, cacert), ensure=True @@ -4366,16 +4103,16 @@ class TestSignVerify: cert = load_certificate(FILETYPE_PEM, root_cert_pem) for digest in ["md5", "sha1", "sha256"]: with pytest.warns(DeprecationWarning) as w: - simplefilter("always") + warnings.simplefilter("always") sig = sign(priv_key, content, digest) - assert "{0} for data is no longer accepted, use bytes".format( + assert "{} for data is no longer accepted, use bytes".format( WARNING_TYPE_EXPECTED ) == str(w[-1].message) with pytest.warns(DeprecationWarning) as w: - simplefilter("always") + warnings.simplefilter("always") verify(cert, sig, content, digest) - assert "{0} for data is no longer accepted, use bytes".format( + assert "{} for data is no longer accepted, use bytes".format( WARNING_TYPE_EXPECTED ) == str(w[-1].message) @@ -4459,7 +4196,7 @@ class TestEllipticCurve: """ curves = get_elliptic_curves() curve = next(iter(curves)) - assert "<Curve %r>" % (curve.name,) == repr(curve) + assert f"<Curve {curve.name!r}>" == repr(curve) def test_to_EC_KEY(self): """ diff --git a/contrib/python/pyOpenSSL/py3/tests/test_ssl.py b/contrib/python/pyOpenSSL/py3/tests/test_ssl.py index fdedd7133a0..b5305eb9200 100644 --- a/contrib/python/pyOpenSSL/py3/tests/test_ssl.py +++ b/contrib/python/pyOpenSSL/py3/tests/test_ssl.py @@ -9,6 +9,7 @@ import datetime import gc import select import sys +import time import uuid from errno import ( EAFNOSUPPORT, @@ -26,42 +27,50 @@ from socket import ( AF_INET6, MSG_PEEK, SHUT_RDWR, - error, + gaierror, socket, ) from sys import getfilesystemencoding, platform from typing import Union -from warnings import simplefilter from weakref import ref +import flaky +import pytest from cryptography import x509 -from cryptography.hazmat.primitives import hashes -from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID - -import flaky - from pretend import raiser -import pytest - from OpenSSL import SSL +from OpenSSL._util import ffi as _ffi +from OpenSSL._util import lib as _lib +from OpenSSL.crypto import ( + FILETYPE_PEM, + TYPE_RSA, + X509, + PKey, + X509Store, + dump_certificate, + dump_privatekey, + get_elliptic_curves, + load_certificate, + load_privatekey, +) + +with pytest.warns(DeprecationWarning): + from OpenSSL.crypto import X509Extension + from OpenSSL.SSL import ( - Connection, - Context, DTLS_METHOD, - Error, MODE_RELEASE_BUFFERS, NO_OVERLAPPING_PROTOCOLS, - OPENSSL_VERSION_NUMBER, OP_COOKIE_EXCHANGE, OP_NO_COMPRESSION, OP_NO_QUERY_MTU, - OP_NO_SSLv2, - OP_NO_SSLv3, OP_NO_TICKET, OP_SINGLE_DH_USE, + OPENSSL_VERSION_NUMBER, RECEIVED_SHUTDOWN, SENT_SHUTDOWN, SESS_CACHE_BOTH, @@ -72,11 +81,6 @@ from OpenSSL.SSL import ( SESS_CACHE_NO_INTERNAL_STORE, SESS_CACHE_OFF, SESS_CACHE_SERVER, - SSLEAY_BUILT_ON, - SSLEAY_CFLAGS, - SSLEAY_DIR, - SSLEAY_PLATFORM, - SSLEAY_VERSION, SSL_CB_ACCEPT_EXIT, SSL_CB_ACCEPT_LOOP, SSL_CB_ALERT, @@ -93,45 +97,41 @@ from OpenSSL.SSL import ( SSL_ST_ACCEPT, SSL_ST_CONNECT, SSL_ST_MASK, - SSLeay_version, - SSLv23_METHOD, - Session, - SysCallError, + SSLEAY_BUILT_ON, + SSLEAY_CFLAGS, + SSLEAY_DIR, + SSLEAY_PLATFORM, + SSLEAY_VERSION, TLS1_1_VERSION, TLS1_2_VERSION, TLS1_3_VERSION, TLS_METHOD, - TLSv1_1_METHOD, - TLSv1_2_METHOD, - TLSv1_METHOD, VERIFY_CLIENT_ONCE, VERIFY_FAIL_IF_NO_PEER_CERT, VERIFY_NONE, VERIFY_PEER, + Connection, + Context, + Error, + OP_NO_SSLv2, + OP_NO_SSLv3, + Session, + SSLeay_version, + SSLv23_METHOD, + SysCallError, + TLSv1_1_METHOD, + TLSv1_2_METHOD, + TLSv1_METHOD, WantReadError, WantWriteError, ZeroReturnError, _make_requires, ) -from OpenSSL._util import ffi as _ffi, lib as _lib -from OpenSSL.crypto import ( - FILETYPE_PEM, - PKey, - TYPE_RSA, - X509, - X509Extension, - X509Store, - dump_certificate, - dump_privatekey, - get_elliptic_curves, - load_certificate, - load_privatekey, -) try: from OpenSSL.SSL import ( - SSL_ST_INIT, SSL_ST_BEFORE, + SSL_ST_INIT, SSL_ST_OK, SSL_ST_RENEGOTIATE, ) @@ -153,7 +153,6 @@ from .test_crypto import ( ) from .util import NON_ASCII, WARNING_TYPE_EXPECTED, is_consistent_type - # openssl dhparam 2048 -out dh-2048.pem dhparam = """\ -----BEGIN DH PARAMETERS----- @@ -170,7 +169,7 @@ i5s5yYK7a/0eWxxRr2qraYaUj8RwDpH9CwIBAg== def socket_any_family(): try: return socket(AF_INET) - except error as e: + except OSError as e: if e.errno == EAFNOSUPPORT: return socket(AF_INET6) raise @@ -359,11 +358,10 @@ def interact_in_memory(client_conn, server_conn): # Copy stuff from each side's send buffer to the other side's # receive buffer. - for (read, write) in [ + for read, write in [ (client_conn, server_conn), (server_conn, client_conn), ]: - # Give the side a chance to generate some more bytes, or succeed. try: data = read.recv(2**16) @@ -642,7 +640,7 @@ class TestContext: key = PKey() key.generate_key(TYPE_RSA, 1024) - with open(pemfile, "wt") as pem: + with open(pemfile, "w") as pem: pem.write(dump_privatekey(FILETYPE_PEM, key).decode("ascii")) ctx = Context(SSLv23_METHOD) @@ -1140,23 +1138,30 @@ class TestContext: self._load_verify_locations_test(None, capath) - def test_load_verify_directory_bytes_capath(self, tmpfile): + @pytest.mark.parametrize( + "pathtype", + [ + "ascii_path", + pytest.param( + "unicode_path", + marks=pytest.mark.skipif( + platform == "win32", + reason="Unicode paths not supported on Windows", + ), + ), + ], + ) + @pytest.mark.parametrize("argtype", ["bytes_arg", "unicode_arg"]) + def test_load_verify_directory_capath(self, pathtype, argtype, tmpfile): """ `Context.load_verify_locations` accepts a directory name as a `bytes` instance and uses the certificates within for verification purposes. """ - self._load_verify_directory_locations_capath( - tmpfile + NON_ASCII.encode(getfilesystemencoding()) - ) - - def test_load_verify_directory_unicode_capath(self, tmpfile): - """ - `Context.load_verify_locations` accepts a directory name as a `unicode` - instance and uses the certificates within for verification purposes. - """ - self._load_verify_directory_locations_capath( - tmpfile.decode(getfilesystemencoding()) + NON_ASCII - ) + if pathtype == "unicode_path": + tmpfile += NON_ASCII.encode(getfilesystemencoding()) + if argtype == "unicode_arg": + tmpfile = tmpfile.decode(getfilesystemencoding()) + self._load_verify_directory_locations_capath(tmpfile) def test_load_verify_locations_wrong_args(self): """ @@ -1172,7 +1177,7 @@ class TestContext: @pytest.mark.skipif( not platform.startswith("linux"), reason="Loading fallback paths is a linux-specific behavior to " - "accommodate pyca/cryptography manylinux1 wheels", + "accommodate pyca/cryptography manylinux wheels", ) def test_fallback_default_verify_paths(self, monkeypatch): """ @@ -1265,7 +1270,10 @@ class TestContext: ) client = socket_any_family() - client.connect(("encrypted.google.com", 443)) + try: + client.connect(("encrypted.google.com", 443)) + except gaierror: + pytest.skip("cannot connect to encrypted.google.com") clientSSL = Connection(context, client) clientSSL.set_connect_state() clientSSL.set_tlsext_host_name(b"encrypted.google.com") @@ -1717,7 +1725,7 @@ class TestContext: """ context = Context(SSLv23_METHOD) with pytest.raises(TypeError): - context.set_tlsext_use_srtp(str("SRTP_AES128_CM_SHA1_80")) + context.set_tlsext_use_srtp("SRTP_AES128_CM_SHA1_80") def test_set_tlsext_use_srtp_invalid_profile(self): """ @@ -1776,7 +1784,7 @@ class TestServerNameCallback: if callback is not None: referrers = get_referrers(callback) if len(referrers) > 1: # pragma: nocover - pytest.fail("Some references remain: %r" % (referrers,)) + pytest.fail(f"Some references remain: {referrers!r}") def test_no_servername(self): """ @@ -2364,7 +2372,7 @@ class TestConnection: # 2.6: https://github.com/pytest-dev/pytest/issues/988 try: clientSSL.connect((loopback_address(client), 1)) - except error as e: + except OSError as e: exc = e assert exc.args[0] == ECONNREFUSED @@ -2380,10 +2388,6 @@ class TestConnection: clientSSL.connect((loopback_address(port), port.getsockname()[1])) # XXX An assertion? Or something? - @pytest.mark.skipif( - platform == "darwin", - reason="connect_ex sometimes causes a kernel panic on OS X 10.6.4", - ) def test_connect_ex(self): """ If there is a connection error, `Connection.connect_ex` returns the @@ -2706,7 +2710,7 @@ class TestConnection: if callback is not None: # pragma: nocover referrers = get_referrers(callback) if len(referrers) > 1: - pytest.fail("Some references remain: %r" % (referrers,)) + pytest.fail(f"Some references remain: {referrers!r}") def test_get_session_unconnected(self): """ @@ -2762,7 +2766,7 @@ class TestConnection: ctx = Context(TLSv1_2_METHOD) ctx.use_privatekey(key) ctx.use_certificate(cert) - ctx.set_session_id("unity-test") + ctx.set_session_id(b"unity-test") def makeServer(socket): server = Connection(ctx, socket) @@ -2838,23 +2842,24 @@ class TestConnection: """ client_socket, server_socket = socket_pair() # Fill up the client's send buffer so Connection won't be able to write - # anything. Only write a single byte at a time so we can be sure we + # anything. Start by sending larger chunks (Windows Socket I/O is slow) + # and continue by writing a single byte at a time so we can be sure we # completely fill the buffer. Even though the socket API is allowed to # signal a short write via its return value it seems this doesn't # always happen on all platforms (FreeBSD and OS X particular) for the # very last bit of available buffer space. - msg = b"x" - for i in range(1024 * 1024 * 64): - try: - client_socket.send(msg) - except error as e: - if e.errno == EWOULDBLOCK: - break - raise - else: - pytest.fail( - "Failed to fill socket buffer, cannot test BIO want write" - ) + for msg in [b"x" * 65536, b"x"]: + for i in range(1024 * 1024 * 64): + try: + client_socket.send(msg) + except OSError as e: + if e.errno == EWOULDBLOCK: + break + raise # pragma: no cover + else: # pragma: no cover + pytest.fail( + "Failed to fill socket buffer, cannot test BIO want write" + ) ctx = Context(SSLv23_METHOD) conn = Connection(ctx, client_socket) @@ -3117,9 +3122,8 @@ class TestConnectionSend: """ server, client = loopback() with pytest.warns(DeprecationWarning) as w: - simplefilter("always") count = server.send(b"xy".decode("ascii")) - assert "{0} for buf is no longer accepted, use bytes".format( + assert "{} for buf is no longer accepted, use bytes".format( WARNING_TYPE_EXPECTED ) == str(w[-1].message) assert count == 2 @@ -3325,9 +3329,8 @@ class TestConnectionSendall: """ server, client = loopback() with pytest.warns(DeprecationWarning) as w: - simplefilter("always") server.sendall(b"x".decode("ascii")) - assert "{0} for buf is no longer accepted, use bytes".format( + assert "{} for buf is no longer accepted, use bytes".format( WARNING_TYPE_EXPECTED ) == str(w[-1].message) assert client.recv(1) == b"x" @@ -3753,13 +3756,16 @@ class TestMemoryBIO: """ If the connection is lost before an orderly SSL shutdown occurs, `OpenSSL.SSL.SysCallError` is raised with a message of - "Unexpected EOF". + "Unexpected EOF" (or WSAECONNRESET on Windows). """ server_conn, client_conn = loopback() client_conn.sock_shutdown(SHUT_RDWR) with pytest.raises(SysCallError) as err: server_conn.recv(1024) - assert err.value.args == (-1, "Unexpected EOF") + if platform == "win32": + assert err.value.args == (10054, "WSAECONNRESET") + else: + assert err.value.args == (-1, "Unexpected EOF") def _check_client_ca_list(self, func): """ @@ -4370,10 +4376,11 @@ class TestDTLS: # new versions of OpenSSL, this is unnecessary, but harmless, because the # DTLS state machine treats it like a network hiccup that duplicated a # packet, which DTLS is robust against. - def test_it_works_at_all(self): - # arbitrary number larger than any conceivable handshake volley - LARGE_BUFFER = 65536 + # Arbitrary number larger than any conceivable handshake volley. + LARGE_BUFFER = 65536 + + def test_it_works_at_all(self): s_ctx = Context(DTLS_METHOD) def generate_cookie(ssl): @@ -4404,7 +4411,7 @@ class TestDTLS: def pump_membio(label, source, sink): try: - chunk = source.bio_read(LARGE_BUFFER) + chunk = source.bio_read(self.LARGE_BUFFER) except WantReadError: return False # I'm not sure this check is needed, but I'm not sure it's *not* @@ -4484,3 +4491,39 @@ class TestDTLS: assert 0 < c.get_cleartext_mtu() < 500 except NotImplementedError: # OpenSSL 1.1.0 and earlier pass + + def test_timeout(self, monkeypatch): + c_ctx = Context(DTLS_METHOD) + c = Connection(c_ctx) + + # No timeout before the handshake starts. + assert c.DTLSv1_get_timeout() is None + assert c.DTLSv1_handle_timeout() is False + + # Start handshake and check there is data to send. + c.set_connect_state() + try: + c.do_handshake() + except SSL.WantReadError: + pass + assert c.bio_read(self.LARGE_BUFFER) + + # There should now be an active timeout. + seconds = c.DTLSv1_get_timeout() + assert seconds is not None + + # Handle the timeout and check there is data to send. + time.sleep(seconds) + assert c.DTLSv1_handle_timeout() is True + assert c.bio_read(self.LARGE_BUFFER) + + # After the maximum number of allowed timeouts is reached, + # DTLSv1_handle_timeout will return -1. + # + # Testing this directly is prohibitively time consuming as the timeout + # duration is doubled on each retry, so the best we can do is to mock + # this condition. + monkeypatch.setattr(_lib, "DTLSv1_handle_timeout", lambda x: -1) + + with pytest.raises(Error): + c.DTLSv1_handle_timeout() diff --git a/contrib/python/pyOpenSSL/py3/ya.make b/contrib/python/pyOpenSSL/py3/ya.make index 08bed6fbc0f..28f890402b4 100644 --- a/contrib/python/pyOpenSSL/py3/ya.make +++ b/contrib/python/pyOpenSSL/py3/ya.make @@ -4,7 +4,7 @@ PY3_LIBRARY() SUBSCRIBER(g:python-contrib) -VERSION(23.0.0) +VERSION(23.3.0) LICENSE(Apache-2.0) |
