summaryrefslogtreecommitdiffstats
path: root/contrib/python/pyOpenSSL/py3
diff options
context:
space:
mode:
Diffstat (limited to 'contrib/python/pyOpenSSL/py3')
-rw-r--r--contrib/python/pyOpenSSL/py3/.dist-info/METADATA97
-rw-r--r--contrib/python/pyOpenSSL/py3/OpenSSL/SSL.py182
-rw-r--r--contrib/python/pyOpenSSL/py3/OpenSSL/__init__.py1
-rw-r--r--contrib/python/pyOpenSSL/py3/OpenSSL/crypto.py408
-rw-r--r--contrib/python/pyOpenSSL/py3/OpenSSL/debug.py4
-rw-r--r--contrib/python/pyOpenSSL/py3/OpenSSL/version.py4
-rw-r--r--contrib/python/pyOpenSSL/py3/tests/conftest.py3
-rw-r--r--contrib/python/pyOpenSSL/py3/tests/memdbg.py5
-rw-r--r--contrib/python/pyOpenSSL/py3/tests/test_crypto.py579
-rw-r--r--contrib/python/pyOpenSSL/py3/tests/test_ssl.py231
-rw-r--r--contrib/python/pyOpenSSL/py3/ya.make2
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)