JacobBarthelmeh
GitHub
commit
2e77f306c2
|
|
@ -1,41 +1,42 @@
|
|||
Certificate Revocation List (CRL):
|
||||
Version 2 (0x1)
|
||||
Signature Algorithm: sha256WithRSAEncryption
|
||||
Issuer: C = US, ST = Montana, L = Bozeman, O = Sawtooth, OU = Consulting, CN = www.wolfssl.com, emailAddress = info@wolfssl.com
|
||||
Last Update: Feb 15 12:50:27 2022 GMT
|
||||
Next Update: Nov 11 12:50:27 2024 GMT
|
||||
Issuer: C=US, ST=Montana, L=Bozeman, O=Sawtooth, OU=Consulting, CN=www.wolfssl.com, emailAddress=info@wolfssl.com
|
||||
Last Update: Nov 13 20:41:50 2025 GMT
|
||||
Next Update: Aug 9 20:41:50 2028 GMT
|
||||
CRL extensions:
|
||||
X509v3 CRL Number:
|
||||
2
|
||||
Revoked Certificates:
|
||||
Serial Number: 02
|
||||
Revocation Date: Feb 15 12:50:27 2022 GMT
|
||||
Revocation Date: Nov 13 20:41:50 2025 GMT
|
||||
Signature Algorithm: sha256WithRSAEncryption
|
||||
43:e6:3b:30:0e:32:53:32:a4:08:3c:e5:d5:2e:f1:ce:e9:95:
|
||||
ff:ba:d6:fe:2e:59:80:f8:0a:2f:cf:1e:e0:37:fe:ca:cc:33:
|
||||
66:8b:ed:65:50:7d:44:92:d3:5c:52:9a:95:a5:9d:a5:4e:77:
|
||||
8b:b4:7f:59:c8:7a:e0:eb:34:32:ae:a1:03:99:d2:3c:c0:f4:
|
||||
7e:1c:87:4c:6c:5a:ba:0a:95:e8:a1:44:01:7b:8f:3e:a4:e3:
|
||||
e8:1e:07:19:f0:09:7a:85:8f:f3:82:62:f8:1e:08:51:a3:60:
|
||||
30:5b:06:c8:a2:b3:ff:aa:28:66:ad:fe:4b:81:49:30:ef:5f:
|
||||
5d:ac:d9:ad:17:9f:2a:b6:22:d6:35:cc:9f:d9:11:26:dd:7a:
|
||||
06:35:d0:d5:c7:41:6c:52:97:8c:aa:82:5a:e5:a8:58:d4:b7:
|
||||
2b:31:84:34:15:bd:08:e4:9e:71:9e:c5:40:f8:02:a3:a0:1e:
|
||||
4f:98:72:2b:eb:9e:8a:4e:01:83:88:e5:cb:6e:3b:52:e3:a9:
|
||||
34:a1:7c:e4:79:2c:d1:e0:0b:74:22:ba:6d:cb:c3:a1:56:f9:
|
||||
c9:f4:20:bf:00:49:df:6b:59:49:18:c7:75:27:8e:a1:5a:a6:
|
||||
ff:f2:be:34:4a:c9:6d:6e:24:a3:1f:15:7e:34:90:b6:81:bf:
|
||||
15:80:c3:ac
|
||||
Signature Value:
|
||||
b7:0d:1c:78:99:1c:e8:0b:d9:33:a2:95:01:ad:cf:35:e9:86:
|
||||
28:7f:49:6b:93:76:c1:70:08:61:aa:77:57:34:af:45:82:78:
|
||||
5d:3b:7b:67:ca:b4:fb:d1:68:13:be:34:94:84:2d:65:ad:97:
|
||||
52:69:1d:67:ea:8e:a7:ff:21:0f:21:6c:8c:75:7f:c7:50:c5:
|
||||
6b:a5:fd:cd:3f:91:64:7b:5e:0f:4a:9c:c8:cd:39:a0:30:ad:
|
||||
80:27:50:e0:a7:bf:19:68:cf:6b:26:75:51:14:77:5a:62:6d:
|
||||
bc:66:1a:90:f7:00:09:34:c7:d0:9d:81:f3:b5:9f:90:40:02:
|
||||
8d:3f:68:7f:0d:1d:c5:00:32:e5:cf:42:35:1c:b6:eb:02:a8:
|
||||
d7:2a:a7:f3:f1:10:e2:d5:9e:41:de:2f:78:7d:7f:ad:68:06:
|
||||
a0:6d:40:96:dd:35:59:4d:a0:d3:bd:2e:ba:b6:75:f8:1c:43:
|
||||
b9:c0:b7:75:c4:38:59:46:00:71:ab:5a:df:f5:62:e9:ac:2b:
|
||||
76:11:4f:1b:42:2c:dd:b2:38:6e:57:cf:c5:75:67:4c:3e:27:
|
||||
bb:4c:d5:09:2c:4a:13:3d:8b:9c:89:76:b7:bd:73:1b:64:50:
|
||||
ea:d5:13:0e:51:48:d8:43:08:93:00:85:8f:2f:08:ad:0d:aa:
|
||||
d6:6c:f8:3d
|
||||
-----BEGIN X509 CRL-----
|
||||
MIICBDCB7QIBATANBgkqhkiG9w0BAQsFADCBlDELMAkGA1UEBhMCVVMxEDAOBgNV
|
||||
BAgMB01vbnRhbmExEDAOBgNVBAcMB0JvemVtYW4xETAPBgNVBAoMCFNhd3Rvb3Ro
|
||||
MRMwEQYDVQQLDApDb25zdWx0aW5nMRgwFgYDVQQDDA93d3cud29sZnNzbC5jb20x
|
||||
HzAdBgkqhkiG9w0BCQEWEGluZm9Ad29sZnNzbC5jb20XDTIyMDIxNTEyNTAyN1oX
|
||||
DTI0MTExMTEyNTAyN1owFDASAgECFw0yMjAyMTUxMjUwMjdaoA4wDDAKBgNVHRQE
|
||||
AwIBAjANBgkqhkiG9w0BAQsFAAOCAQEAQ+Y7MA4yUzKkCDzl1S7xzumV/7rW/i5Z
|
||||
gPgKL88e4Df+yswzZovtZVB9RJLTXFKalaWdpU53i7R/Wch64Os0Mq6hA5nSPMD0
|
||||
fhyHTGxaugqV6KFEAXuPPqTj6B4HGfAJeoWP84Ji+B4IUaNgMFsGyKKz/6ooZq3+
|
||||
S4FJMO9fXazZrRefKrYi1jXMn9kRJt16BjXQ1cdBbFKXjKqCWuWoWNS3KzGENBW9
|
||||
COSecZ7FQPgCo6AeT5hyK+ueik4Bg4jly247UuOpNKF85Hks0eALdCK6bcvDoVb5
|
||||
yfQgvwBJ32tZSRjHdSeOoVqm//K+NErJbW4kox8VfjSQtoG/FYDDrA==
|
||||
HzAdBgkqhkiG9w0BCQEWEGluZm9Ad29sZnNzbC5jb20XDTI1MTExMzIwNDE1MFoX
|
||||
DTI4MDgwOTIwNDE1MFowFDASAgECFw0yNTExMTMyMDQxNTBaoA4wDDAKBgNVHRQE
|
||||
AwIBAjANBgkqhkiG9w0BAQsFAAOCAQEAtw0ceJkc6AvZM6KVAa3PNemGKH9Ja5N2
|
||||
wXAIYap3VzSvRYJ4XTt7Z8q0+9FoE740lIQtZa2XUmkdZ+qOp/8hDyFsjHV/x1DF
|
||||
a6X9zT+RZHteD0qcyM05oDCtgCdQ4Ke/GWjPayZ1URR3WmJtvGYakPcACTTH0J2B
|
||||
87WfkEACjT9ofw0dxQAy5c9CNRy26wKo1yqn8/EQ4tWeQd4veH1/rWgGoG1Alt01
|
||||
WU2g070uurZ1+BxDucC3dcQ4WUYAcata3/Vi6awrdhFPG0Is3bI4blfPxXVnTD4n
|
||||
u0zVCSxKEz2LnIl2t71zG2RQ6tUTDlFI2EMIkwCFjy8IrQ2q1mz4PQ==
|
||||
-----END X509 CRL-----
|
||||
|
|
|
|||
|
|
@ -89,6 +89,12 @@ def build_arg_parser():
|
|||
help="Disable client cert check"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"-n", action="store_true",
|
||||
help="Disable server hostname check "
|
||||
"(IP literal hosts are never hostname-checked)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"-g", action="store_true",
|
||||
help="Send server HTTP GET"
|
||||
|
|
@ -125,6 +131,54 @@ def get_DTLSmethod(index):
|
|||
wolfssl.PROTOCOL_DTLSv1_3
|
||||
)[index]
|
||||
|
||||
def is_ip_literal(host):
|
||||
# AI_NUMERICHOST never resolves, it only parses. Unlike
|
||||
# socket.inet_pton() it is available on every supported platform.
|
||||
try:
|
||||
socket.getaddrinfo(host, None, 0, 0, 0, socket.AI_NUMERICHOST)
|
||||
return True
|
||||
except socket.error:
|
||||
return False
|
||||
|
||||
|
||||
def configure_verification(context, args):
|
||||
"""
|
||||
Configure peer certificate and hostname verification on the context
|
||||
according to the parsed arguments. Returns the server_hostname to pass
|
||||
to wrap_socket() (None when no hostname check should be performed).
|
||||
|
||||
When certificate verification is enabled (the default), hostname
|
||||
verification is enabled too so that a CA-trusted certificate issued for
|
||||
a different host is rejected. Pass -n to opt out explicitly (e.g. when
|
||||
using test certificates).
|
||||
|
||||
IP literal hosts are not hostname-checked: wolfSSL_check_domain_name()
|
||||
only matches DNS names (iPAddress SANs are skipped on this path), and
|
||||
RFC 6066 forbids IP literals in SNI. Certificate verification against
|
||||
the CA still applies. Connect by DNS name to also verify the hostname.
|
||||
"""
|
||||
if args.d:
|
||||
context.verify_mode = wolfssl.CERT_NONE
|
||||
context.check_hostname = False
|
||||
return None
|
||||
|
||||
context.verify_mode = wolfssl.CERT_REQUIRED
|
||||
context.load_verify_locations(args.A)
|
||||
|
||||
if args.n:
|
||||
context.check_hostname = False
|
||||
return None
|
||||
|
||||
if is_ip_literal(args.h):
|
||||
print("Note: skipping hostname check for IP literal '{}'. "
|
||||
"Connect by DNS name to enable it.".format(args.h))
|
||||
context.check_hostname = False
|
||||
return None
|
||||
|
||||
context.check_hostname = True
|
||||
return args.h
|
||||
|
||||
|
||||
def main():
|
||||
args = build_arg_parser().parse_args()
|
||||
|
||||
|
|
@ -147,18 +201,15 @@ def main():
|
|||
|
||||
context.load_cert_chain(args.c, args.k)
|
||||
|
||||
if args.d:
|
||||
context.verify_mode = wolfssl.CERT_NONE
|
||||
else:
|
||||
context.verify_mode = wolfssl.CERT_REQUIRED
|
||||
context.load_verify_locations(args.A)
|
||||
server_hostname = configure_verification(context, args)
|
||||
|
||||
if args.l:
|
||||
context.set_ciphers(args.l)
|
||||
|
||||
secure_socket = None
|
||||
try:
|
||||
secure_socket = context.wrap_socket(bind_socket)
|
||||
secure_socket = context.wrap_socket(
|
||||
bind_socket, server_hostname=server_hostname)
|
||||
|
||||
if not args.C:
|
||||
secure_socket.enable_crl(1)
|
||||
|
|
|
|||
|
|
@ -115,6 +115,21 @@ def get_DTLSmethod(index):
|
|||
)[index]
|
||||
|
||||
|
||||
# Large enough to peek a DTLS ClientHello source address.
|
||||
PEEK_BUFSIZE = 1500
|
||||
|
||||
|
||||
def peek_peer_address(sock):
|
||||
"""
|
||||
Return the source address of the next pending datagram without removing
|
||||
it from the socket queue. MSG_PEEK leaves the datagram (the DTLS
|
||||
ClientHello) intact so wolfSSL_accept() can consume it during the
|
||||
handshake.
|
||||
"""
|
||||
_, from_addr = sock.recvfrom(PEEK_BUFSIZE, socket.MSG_PEEK)
|
||||
return from_addr
|
||||
|
||||
|
||||
def main():
|
||||
args = build_arg_parser().parse_args()
|
||||
# DTLS connection over UDP
|
||||
|
|
@ -124,7 +139,6 @@ def main():
|
|||
args.v = 1
|
||||
bind_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, 0)
|
||||
bind_socket.bind(("" if args.b else "localhost", args.p))
|
||||
data, from_addr = bind_socket.recvfrom(1)
|
||||
context = wolfssl.SSLContext(get_DTLSmethod(args.v), server_side=True)
|
||||
# SSL/TLS connection over TCP
|
||||
else:
|
||||
|
|
@ -156,6 +170,9 @@ def main():
|
|||
try:
|
||||
secure_socket = None
|
||||
if args.u:
|
||||
# Peek the client's address for this connection without
|
||||
# consuming the ClientHello datagram needed by the handshake.
|
||||
from_addr = peek_peer_address(bind_socket)
|
||||
secure_socket = context.wrap_socket(bind_socket)
|
||||
else:
|
||||
new_socket, from_addr = bind_socket.accept()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,171 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_client_example.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
|
||||
import os
|
||||
import socket
|
||||
import subprocess
|
||||
import sys
|
||||
import threading
|
||||
|
||||
import wolfssl
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "examples"))
|
||||
import client as client_example # noqa: E402
|
||||
|
||||
|
||||
def _args(argv):
|
||||
return client_example.build_arg_parser().parse_args(argv)
|
||||
|
||||
|
||||
def _ctx():
|
||||
return wolfssl.SSLContext(wolfssl.PROTOCOL_TLSv1_2)
|
||||
|
||||
|
||||
def test_verification_enables_hostname_check_by_default():
|
||||
"""
|
||||
F-5621: with cert verification on (the default), the client must also
|
||||
verify the peer's hostname and pass server_hostname to wrap_socket.
|
||||
"""
|
||||
args = _args(["-h", "example.com"])
|
||||
ctx = _ctx()
|
||||
|
||||
server_hostname = client_example.configure_verification(ctx, args)
|
||||
|
||||
assert ctx.verify_mode == wolfssl.CERT_REQUIRED
|
||||
assert ctx.check_hostname is True
|
||||
assert server_hostname == "example.com"
|
||||
|
||||
|
||||
def test_disable_cert_check_skips_hostname():
|
||||
args = _args(["-d"])
|
||||
ctx = _ctx()
|
||||
# Simulate a reused context that previously had hostname checking on:
|
||||
# -d must clear it, not leave it dangling against CERT_NONE.
|
||||
ctx.verify_mode = wolfssl.CERT_REQUIRED
|
||||
ctx.check_hostname = True
|
||||
|
||||
server_hostname = client_example.configure_verification(ctx, args)
|
||||
|
||||
assert ctx.verify_mode == wolfssl.CERT_NONE
|
||||
assert ctx.check_hostname is False
|
||||
assert server_hostname is None
|
||||
|
||||
|
||||
def test_hostname_check_can_be_opted_out():
|
||||
"""An explicit opt-out is provided for test certificates."""
|
||||
args = _args(["-n"])
|
||||
ctx = _ctx()
|
||||
# Reused context with hostname checking previously enabled: -n must
|
||||
# actively turn it back off.
|
||||
ctx.verify_mode = wolfssl.CERT_REQUIRED
|
||||
ctx.check_hostname = True
|
||||
|
||||
server_hostname = client_example.configure_verification(ctx, args)
|
||||
|
||||
assert ctx.verify_mode == wolfssl.CERT_REQUIRED
|
||||
assert ctx.check_hostname is False
|
||||
assert server_hostname is None
|
||||
|
||||
|
||||
def test_ip_literal_host_skips_hostname_check():
|
||||
"""
|
||||
The default host (127.0.0.1) is an IP literal. wolfSSL's
|
||||
check_domain_name() never matches iPAddress SANs, so the example must
|
||||
not hostname-check IP literals; certificate verification stays on.
|
||||
"""
|
||||
args = _args([])
|
||||
ctx = _ctx()
|
||||
ctx.verify_mode = wolfssl.CERT_REQUIRED
|
||||
ctx.check_hostname = True
|
||||
|
||||
server_hostname = client_example.configure_verification(ctx, args)
|
||||
|
||||
assert args.h == "127.0.0.1"
|
||||
assert ctx.verify_mode == wolfssl.CERT_REQUIRED
|
||||
assert ctx.check_hostname is False
|
||||
assert server_hostname is None
|
||||
|
||||
|
||||
def test_ipv6_literal_host_skips_hostname_check():
|
||||
args = _args(["-h", "::1"])
|
||||
ctx = _ctx()
|
||||
|
||||
server_hostname = client_example.configure_verification(ctx, args)
|
||||
|
||||
assert ctx.verify_mode == wolfssl.CERT_REQUIRED
|
||||
assert ctx.check_hostname is False
|
||||
assert server_hostname is None
|
||||
|
||||
|
||||
def _free_port():
|
||||
sock = socket.socket()
|
||||
sock.bind(("localhost", 0))
|
||||
port = sock.getsockname()[1]
|
||||
sock.close()
|
||||
return port
|
||||
|
||||
|
||||
def test_default_invocation_end_to_end():
|
||||
"""
|
||||
`python client.py` with the default host must complete a connection to
|
||||
`python server.py` using the bundled certificates: verification is on
|
||||
by default and the IP literal host must not trip the hostname check.
|
||||
"""
|
||||
root = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
|
||||
port = _free_port()
|
||||
|
||||
# The examples must import the same wolfssl as this test process, even
|
||||
# when the package is not installed (source-tree runs).
|
||||
env = dict(os.environ)
|
||||
pkg_root = os.path.dirname(os.path.dirname(os.path.abspath(
|
||||
wolfssl.__file__)))
|
||||
env["PYTHONPATH"] = os.pathsep.join(
|
||||
[pkg_root] + ([env["PYTHONPATH"]] if env.get("PYTHONPATH") else []))
|
||||
|
||||
server = subprocess.Popen(
|
||||
[sys.executable, "-u", os.path.join("examples", "server.py"),
|
||||
"-p", str(port)],
|
||||
cwd=root, env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
|
||||
try:
|
||||
line = server.stdout.readline().decode()
|
||||
assert "Server listening" in line, line
|
||||
|
||||
client = subprocess.Popen(
|
||||
[sys.executable, "-u", os.path.join("examples", "client.py"),
|
||||
"-p", str(port)],
|
||||
cwd=root, env=env, stdout=subprocess.PIPE,
|
||||
stderr=subprocess.STDOUT)
|
||||
# Watchdog instead of communicate(timeout=...), which is 3.3+.
|
||||
watchdog = threading.Timer(60, client.kill)
|
||||
watchdog.start()
|
||||
try:
|
||||
out, _ = client.communicate()
|
||||
finally:
|
||||
watchdog.cancel()
|
||||
|
||||
assert client.returncode == 0, out.decode()
|
||||
assert b"I hear you fa shizzle" in out
|
||||
finally:
|
||||
server.kill()
|
||||
server.wait()
|
||||
|
|
@ -0,0 +1,105 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_dtls_handshake_once.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
# pylint: disable=protected-access
|
||||
|
||||
"""
|
||||
F-4136: for DTLS, write()/read()/recv_into() used to call do_handshake() on
|
||||
every single call. Once the handshake has completed, re-running it is wasteful
|
||||
and, on a non-blocking socket, wolfSSL_accept/connect can raise
|
||||
SSLWantReadError and abort an otherwise valid I/O. The handshake must only be
|
||||
driven until it completes.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
import wolfssl
|
||||
|
||||
|
||||
class _OkLib:
|
||||
"""_lib stub whose I/O calls always succeed."""
|
||||
|
||||
def wolfSSL_write(self, ssl, data, length):
|
||||
return length
|
||||
|
||||
def wolfSSL_read(self, ssl, data, length):
|
||||
return length
|
||||
|
||||
def wolfSSL_get_error(self, ssl, ret): # pragma: no cover
|
||||
return 0
|
||||
|
||||
|
||||
def _make_dtls_socket(handshake_complete):
|
||||
sock = wolfssl.SSLSocket.__new__(wolfssl.SSLSocket)
|
||||
sock.native_object = object()
|
||||
sock._connected = True
|
||||
sock._server_side = True
|
||||
sock._context = SimpleNamespace(protocol=wolfssl.PROTOCOL_DTLSv1_2)
|
||||
sock._handshake_complete = handshake_complete
|
||||
sock._release_native_object = lambda: None
|
||||
return sock
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def spy_handshake(monkeypatch):
|
||||
monkeypatch.setattr(wolfssl, "_lib", _OkLib())
|
||||
calls = []
|
||||
|
||||
def _record(sock):
|
||||
calls.append(True)
|
||||
sock._handshake_complete = True
|
||||
|
||||
return calls, _record
|
||||
|
||||
|
||||
@pytest.mark.parametrize("op", ["write", "read", "recv_into"])
|
||||
def test_dtls_io_does_not_redrive_completed_handshake(spy_handshake, op):
|
||||
calls, record = spy_handshake
|
||||
sock = _make_dtls_socket(handshake_complete=True)
|
||||
sock.do_handshake = lambda block=False: record(sock)
|
||||
|
||||
if op == "write":
|
||||
sock.write(b"payload")
|
||||
elif op == "read":
|
||||
sock.read(8)
|
||||
else:
|
||||
sock.recv_into(bytearray(8))
|
||||
|
||||
assert calls == [], "do_handshake() must not run once the handshake is done"
|
||||
|
||||
|
||||
@pytest.mark.parametrize("op", ["write", "read", "recv_into"])
|
||||
def test_dtls_io_drives_handshake_until_complete(spy_handshake, op):
|
||||
calls, record = spy_handshake
|
||||
sock = _make_dtls_socket(handshake_complete=False)
|
||||
sock.do_handshake = lambda block=False: record(sock)
|
||||
|
||||
if op == "write":
|
||||
sock.write(b"payload")
|
||||
elif op == "read":
|
||||
sock.read(8)
|
||||
else:
|
||||
sock.recv_into(bytearray(8))
|
||||
|
||||
assert calls == [True], "first DTLS I/O must drive the handshake once"
|
||||
|
|
@ -0,0 +1,73 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_dtls_server_example.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
|
||||
import os
|
||||
import sys
|
||||
import socket
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "examples"))
|
||||
import server as server_example # noqa: E402
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def udp_pair():
|
||||
srv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
srv.bind(("localhost", 0))
|
||||
cli = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
||||
try:
|
||||
yield srv, cli, srv.getsockname()
|
||||
finally:
|
||||
srv.close()
|
||||
cli.close()
|
||||
|
||||
|
||||
def test_peek_peer_address_returns_source(udp_pair):
|
||||
srv, cli, srv_addr = udp_pair
|
||||
cli.bind(("localhost", 0))
|
||||
cli.sendto(b"clienthello-payload", srv_addr)
|
||||
|
||||
addr = server_example.peek_peer_address(srv)
|
||||
|
||||
assert addr == cli.getsockname()
|
||||
|
||||
|
||||
def test_peek_peer_address_does_not_consume_datagram(udp_pair):
|
||||
"""
|
||||
Regression test for F-3481: peeking the client's address before the
|
||||
DTLS handshake must leave the ClientHello datagram intact. The previous
|
||||
example used recvfrom(1), which consumed the datagram and discarded
|
||||
everything past the first byte, breaking the handshake.
|
||||
"""
|
||||
srv, cli, srv_addr = udp_pair
|
||||
payload = b"X" * 256 # stand-in for a DTLS ClientHello record
|
||||
cli.sendto(payload, srv_addr)
|
||||
|
||||
server_example.peek_peer_address(srv)
|
||||
|
||||
# The datagram must still be fully available for wolfSSL_accept().
|
||||
srv.settimeout(2)
|
||||
data, _ = srv.recvfrom(4096)
|
||||
assert data == payload
|
||||
|
|
@ -0,0 +1,131 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_getpeercert.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
# pylint: disable=protected-access
|
||||
|
||||
import socket
|
||||
from contextlib import contextmanager
|
||||
from threading import Thread
|
||||
|
||||
import pytest
|
||||
|
||||
import wolfssl
|
||||
|
||||
|
||||
@contextmanager
|
||||
def _client_server_session():
|
||||
"""
|
||||
Establish a real TLS connection to a local server that does NOT request a
|
||||
client certificate. Yields (client_socket, server_result); server_result
|
||||
is populated (after the block exits) with the server's view of the peer:
|
||||
{"x509", "cert"} on success or {"error"} if a call raised.
|
||||
"""
|
||||
result = {}
|
||||
|
||||
server_ctx = wolfssl.SSLContext(wolfssl.PROTOCOL_TLS, server_side=True)
|
||||
server_ctx.verify_mode = wolfssl.CERT_NONE
|
||||
server_ctx.load_cert_chain("certs/server-cert.pem", "certs/server-key.pem")
|
||||
|
||||
listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||||
listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||||
listener.bind(("localhost", 0))
|
||||
listener.listen(1)
|
||||
port = listener.getsockname()[1]
|
||||
|
||||
def serve():
|
||||
conn, _ = listener.accept()
|
||||
ssock = server_ctx.wrap_socket(conn, server_side=True)
|
||||
try:
|
||||
ssock.read(1024)
|
||||
# Client sent no certificate: these must not raise.
|
||||
result["x509"] = ssock.get_peer_x509()
|
||||
result["cert"] = ssock.getpeercert()
|
||||
ssock.write(b"ok")
|
||||
except Exception as exc: # pylint: disable=broad-except
|
||||
result["error"] = exc
|
||||
finally:
|
||||
ssock.close()
|
||||
|
||||
server_thread = Thread(target=serve, daemon=True)
|
||||
server_thread.start()
|
||||
|
||||
client_ctx = wolfssl.SSLContext(wolfssl.PROTOCOL_TLS)
|
||||
client_ctx.verify_mode = wolfssl.CERT_NONE
|
||||
client = client_ctx.wrap_socket(
|
||||
socket.socket(socket.AF_INET, socket.SOCK_STREAM))
|
||||
client.connect(("localhost", port))
|
||||
client.write(b"hi")
|
||||
try:
|
||||
yield client, result
|
||||
finally:
|
||||
try:
|
||||
client.read(1024)
|
||||
except Exception: # pylint: disable=broad-except
|
||||
pass
|
||||
client.close()
|
||||
server_thread.join(timeout=10)
|
||||
listener.close()
|
||||
|
||||
|
||||
def test_getpeercert_returns_none_without_peer_cert():
|
||||
"""
|
||||
F-5623: on a valid TLS connection where the peer presented no
|
||||
certificate (here, a server that does not request a client cert),
|
||||
getpeercert()/get_peer_x509() must return None instead of raising.
|
||||
"""
|
||||
with _client_server_session() as (client, result):
|
||||
# The peer (server) always presents a certificate.
|
||||
server_cert = client.getpeercert()
|
||||
|
||||
assert "error" not in result, "getpeercert raised: %r" % result.get("error")
|
||||
assert result["x509"] is None
|
||||
assert result["cert"] is None
|
||||
# Positive path: the server's certificate is still returned to the client.
|
||||
assert server_cert is not None
|
||||
|
||||
|
||||
def test_wolfsslx509_accepts_session_for_backward_compat():
|
||||
"""
|
||||
WolfSSLX509 historically accepted a WOLFSSL* session and fetched the peer
|
||||
certificate itself. That constructor form must keep working alongside the
|
||||
new WOLFSSL_X509* form used by get_peer_x509().
|
||||
"""
|
||||
with _client_server_session() as (client, _result):
|
||||
from_session = wolfssl.WolfSSLX509(client.native_object)
|
||||
from_helper = client.get_peer_x509()
|
||||
|
||||
# Both forms resolve to the same server certificate.
|
||||
assert from_session.get_subject_cn() != ""
|
||||
assert from_session.get_subject_cn() == from_helper.get_subject_cn()
|
||||
|
||||
|
||||
def test_wolfsslx509_rejects_unexpected_types():
|
||||
"""
|
||||
WolfSSLX509 discriminates WOLFSSL* from WOLFSSL_X509* by cffi type.
|
||||
Anything else must raise TypeError instead of being treated as a
|
||||
certificate pointer.
|
||||
"""
|
||||
with pytest.raises(TypeError):
|
||||
wolfssl.WolfSSLX509(object())
|
||||
with pytest.raises(TypeError):
|
||||
wolfssl.WolfSSLX509(wolfssl._ffi.new("int *"))
|
||||
|
|
@ -0,0 +1,119 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_io_error_mapping.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
# pylint: disable=protected-access
|
||||
|
||||
"""
|
||||
These tests exercise the error-code-to-exception mapping in SSLSocket's
|
||||
read/write/recv_into. wolfSSL_write can return WANT_READ and wolfSSL_read can
|
||||
return WANT_WRITE during a renegotiation; non-blocking callers rely on these
|
||||
being surfaced as SSLWantReadError / SSLWantWriteError (matching the stdlib
|
||||
ssl module) rather than a generic SSLError.
|
||||
|
||||
The renegotiation conditions are awkward to force over a real socket, so the
|
||||
native wolfSSL_read/wolfSSL_write/wolfSSL_get_error functions are stubbed to
|
||||
return the relevant codes and the Python-level mapping is verified directly.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
import wolfssl
|
||||
|
||||
|
||||
class _FakeLib:
|
||||
"""Stand-in for wolfssl._lib that forces a given I/O return / error."""
|
||||
|
||||
def __init__(self, io_ret, err):
|
||||
self._io_ret = io_ret
|
||||
self._err = err
|
||||
|
||||
def wolfSSL_write(self, ssl, data, length):
|
||||
return self._io_ret
|
||||
|
||||
def wolfSSL_read(self, ssl, data, length):
|
||||
return self._io_ret
|
||||
|
||||
def wolfSSL_get_error(self, ssl, ret):
|
||||
return self._err
|
||||
|
||||
|
||||
def _make_socket():
|
||||
"""A minimal, non-DTLS SSLSocket that skips __init__/native setup."""
|
||||
sock = wolfssl.SSLSocket.__new__(wolfssl.SSLSocket)
|
||||
sock.native_object = object() # non-NULL so _check_closed passes
|
||||
sock._connected = True # so _check_connected is a no-op
|
||||
sock._context = SimpleNamespace(protocol=wolfssl.PROTOCOL_TLS)
|
||||
# The dummy native_object isn't a real cdata pointer, so make __del__
|
||||
# a no-op to avoid wolfSSL_free() choking on it during GC.
|
||||
sock._release_native_object = lambda: None
|
||||
return sock
|
||||
|
||||
|
||||
def _patch_lib(monkeypatch, io_ret, err):
|
||||
monkeypatch.setattr(wolfssl, "_lib", _FakeLib(io_ret, err))
|
||||
|
||||
|
||||
def test_write_want_read_raises_wantread(monkeypatch):
|
||||
"""F-3905: wolfSSL_write returning WANT_READ -> SSLWantReadError."""
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_READ)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantReadError):
|
||||
sock.write(b"data")
|
||||
|
||||
|
||||
def test_write_want_write_still_raises_wantwrite(monkeypatch):
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_WRITE)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantWriteError):
|
||||
sock.write(b"data")
|
||||
|
||||
|
||||
def test_read_want_write_raises_wantwrite(monkeypatch):
|
||||
"""F-3906: wolfSSL_read returning WANT_WRITE -> SSLWantWriteError."""
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_WRITE)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantWriteError):
|
||||
sock.read(16)
|
||||
|
||||
|
||||
def test_read_want_read_still_raises_wantread(monkeypatch):
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_READ)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantReadError):
|
||||
sock.read(16)
|
||||
|
||||
|
||||
def test_recv_into_want_write_raises_wantwrite(monkeypatch):
|
||||
"""F-3907: wolfSSL_read in recv_into returning WANT_WRITE."""
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_WRITE)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantWriteError):
|
||||
sock.recv_into(bytearray(16))
|
||||
|
||||
|
||||
def test_recv_into_want_read_still_raises_wantread(monkeypatch):
|
||||
_patch_lib(monkeypatch, -1, wolfssl._SSL_ERROR_WANT_READ)
|
||||
sock = _make_socket()
|
||||
with pytest.raises(wolfssl.SSLWantReadError):
|
||||
sock.recv_into(bytearray(16))
|
||||
|
|
@ -0,0 +1,85 @@
|
|||
# -*- coding: utf-8 -*-
|
||||
#
|
||||
# test_write_bytes.py
|
||||
#
|
||||
# Copyright (C) 2006-2020 wolfSSL Inc.
|
||||
#
|
||||
# This file is part of wolfSSL. (formerly known as CyaSSL)
|
||||
#
|
||||
# wolfSSL is free software; you can redistribute it and/or modify
|
||||
# it under the terms of the GNU General Public License as published by
|
||||
# the Free Software Foundation; either version 2 of the License, or
|
||||
# (at your option) any later version.
|
||||
#
|
||||
# wolfSSL is distributed in the hope that it will be useful,
|
||||
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
# GNU General Public License for more details.
|
||||
#
|
||||
# You should have received a copy of the GNU General Public License
|
||||
# along with this program; if not, write to the Free Software
|
||||
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
|
||||
|
||||
# pylint: disable=missing-docstring, invalid-name, import-error
|
||||
# pylint: disable=protected-access
|
||||
|
||||
"""
|
||||
F-5622: SSLSocket.write() ran data through t2b(), which str()-encodes anything
|
||||
that is not already bytes. Valid bytes-like inputs (bytearray, memoryview)
|
||||
were therefore serialized as their Python repr ("bytearray(b'...')",
|
||||
"<memory at 0x...>") instead of their actual contents.
|
||||
"""
|
||||
|
||||
from types import SimpleNamespace
|
||||
|
||||
import wolfssl
|
||||
|
||||
|
||||
class _CaptureLib:
|
||||
def __init__(self):
|
||||
self.written = None
|
||||
|
||||
def wolfSSL_write(self, ssl, data, length):
|
||||
self.written = bytes(data[:length])
|
||||
return length
|
||||
|
||||
def wolfSSL_get_error(self, ssl, ret): # pragma: no cover
|
||||
return 0
|
||||
|
||||
|
||||
def _make_socket(monkeypatch):
|
||||
lib = _CaptureLib()
|
||||
monkeypatch.setattr(wolfssl, "_lib", lib)
|
||||
sock = wolfssl.SSLSocket.__new__(wolfssl.SSLSocket)
|
||||
sock.native_object = object()
|
||||
sock._connected = True
|
||||
sock._context = SimpleNamespace(protocol=wolfssl.PROTOCOL_TLS)
|
||||
sock._release_native_object = lambda: None
|
||||
return sock, lib
|
||||
|
||||
|
||||
def test_write_bytes_unchanged(monkeypatch):
|
||||
sock, lib = _make_socket(monkeypatch)
|
||||
sock.write(b"hello")
|
||||
assert lib.written == b"hello"
|
||||
|
||||
|
||||
def test_write_bytearray_sends_contents(monkeypatch):
|
||||
sock, lib = _make_socket(monkeypatch)
|
||||
sock.write(bytearray(b"hello"))
|
||||
assert lib.written == b"hello"
|
||||
|
||||
|
||||
def test_write_memoryview_sends_contents(monkeypatch):
|
||||
sock, lib = _make_socket(monkeypatch)
|
||||
sock.write(memoryview(b"hello"))
|
||||
assert lib.written == b"hello"
|
||||
|
||||
|
||||
def test_write_str_is_utf8_encoded(monkeypatch):
|
||||
# Backward compatibility: str is UTF-8 encoded (historical t2b()
|
||||
# behavior), not rejected. The \u00e9 escape keeps the source 7-bit
|
||||
# ASCII while still exercising a multi-byte UTF-8 encoding.
|
||||
sock, lib = _make_socket(monkeypatch)
|
||||
sock.write("h\u00e9llo")
|
||||
assert lib.written == "h\u00e9llo".encode("utf-8")
|
||||
|
|
@ -97,7 +97,22 @@ class WolfSSLX509(object):
|
|||
"""
|
||||
|
||||
def __init__(self, session):
|
||||
self.native_object = _lib.wolfSSL_get_peer_certificate(session)
|
||||
# `session` kept as the original public parameter name. Accept a
|
||||
# WOLFSSL* session (fetch the peer cert here) or an already-obtained
|
||||
# WOLFSSL_X509* (used by SSLSocket.get_peer_x509()).
|
||||
# Compare cffi type objects, not type name strings: typeof()
|
||||
# results are interned per FFI instance, so this is exact and
|
||||
# does not depend on how cffi renders the name.
|
||||
ctype = _ffi.typeof(session)
|
||||
if ctype is _ffi.typeof("WOLFSSL *"):
|
||||
x509 = _lib.wolfSSL_get_peer_certificate(session)
|
||||
elif ctype is _ffi.typeof("WOLFSSL_X509 *"):
|
||||
x509 = session
|
||||
else:
|
||||
raise TypeError("session must be a WOLFSSL* or a WOLFSSL_X509*, "
|
||||
"got %s" % ctype)
|
||||
|
||||
self.native_object = x509
|
||||
|
||||
if self.native_object == _ffi.NULL:
|
||||
raise SSLError("Unable to get internal WOLFSSL_X509 from wolfSSL")
|
||||
|
|
@ -460,6 +475,9 @@ class SSLSocket(object):
|
|||
|
||||
self._closed = False
|
||||
self._connected = connected
|
||||
# Tracks whether the (DTLS) handshake has completed so I/O methods
|
||||
# don't re-drive it on every call.
|
||||
self._handshake_complete = False
|
||||
|
||||
# create the SSL object
|
||||
self.native_object = _lib.wolfSSL_new(self.context.native_object)
|
||||
|
|
@ -574,14 +592,20 @@ class SSLSocket(object):
|
|||
Returns number of bytes of DATA actually transmitted.
|
||||
"""
|
||||
self._check_closed("write")
|
||||
# Check connected if not DTLS
|
||||
# Check connected if not DTLS
|
||||
if self._context.protocol < PROTOCOL_DTLSv1:
|
||||
self._check_connected()
|
||||
# Complete handshake if DTLS connection
|
||||
else:
|
||||
# Drive the DTLS handshake only until it has completed.
|
||||
elif not self._handshake_complete:
|
||||
self.do_handshake()
|
||||
|
||||
data = t2b(data)
|
||||
# Send bytes-like objects verbatim; fall back to t2b() for other
|
||||
# types (e.g. str) to preserve backward compatibility.
|
||||
if not isinstance(data, bytes):
|
||||
try:
|
||||
data = bytes(memoryview(data))
|
||||
except TypeError:
|
||||
data = t2b(data)
|
||||
|
||||
ret = _lib.wolfSSL_write(
|
||||
self.native_object, data, len(data))
|
||||
|
|
@ -590,6 +614,9 @@ class SSLSocket(object):
|
|||
self.native_object, 0)
|
||||
if err == _SSL_ERROR_WANT_WRITE:
|
||||
raise SSLWantWriteError()
|
||||
elif err == _SSL_ERROR_WANT_READ:
|
||||
# wolfSSL_write can require a read first (e.g. renegotiation).
|
||||
raise SSLWantReadError()
|
||||
else:
|
||||
raise SSLError(
|
||||
"wolfSSL_write error (%d)" % err)
|
||||
|
|
@ -640,8 +667,8 @@ class SSLSocket(object):
|
|||
# Check connected if not DTLS
|
||||
if self._context.protocol < PROTOCOL_DTLSv1:
|
||||
self._check_connected()
|
||||
# Complete handshake if DTLS connection
|
||||
else:
|
||||
# Drive the DTLS handshake only until it has completed.
|
||||
elif not self._handshake_complete:
|
||||
self.do_handshake()
|
||||
|
||||
if buffer is not None:
|
||||
|
|
@ -655,6 +682,9 @@ class SSLSocket(object):
|
|||
err = _lib.wolfSSL_get_error(self.native_object, 0)
|
||||
if err == _SSL_ERROR_WANT_READ:
|
||||
raise SSLWantReadError()
|
||||
elif err == _SSL_ERROR_WANT_WRITE:
|
||||
# wolfSSL_read can require a write first (e.g. renegotiation).
|
||||
raise SSLWantWriteError()
|
||||
else:
|
||||
raise SSLError("wolfSSL_read error (%d)" % err)
|
||||
|
||||
|
|
@ -675,7 +705,8 @@ class SSLSocket(object):
|
|||
self._check_closed("read")
|
||||
if self._context.protocol < PROTOCOL_DTLSv1:
|
||||
self._check_connected()
|
||||
else:
|
||||
# Drive the DTLS handshake only until it has completed.
|
||||
elif not self._handshake_complete:
|
||||
self.do_handshake()
|
||||
|
||||
if buffer is None:
|
||||
|
|
@ -696,6 +727,9 @@ class SSLSocket(object):
|
|||
err = _lib.wolfSSL_get_error(self.native_object, 0)
|
||||
if err == _SSL_ERROR_WANT_READ:
|
||||
raise SSLWantReadError()
|
||||
elif err == _SSL_ERROR_WANT_WRITE:
|
||||
# wolfSSL_read can require a write first (e.g. renegotiation).
|
||||
raise SSLWantWriteError()
|
||||
else:
|
||||
raise SSLError("wolfSSL_read error (%d)" % err)
|
||||
|
||||
|
|
@ -817,6 +851,9 @@ class SSLSocket(object):
|
|||
raise SSLError("do_handshake failed with error %d: %s" %
|
||||
(err, eStr))
|
||||
|
||||
# Reached only on success (every failure path above raises).
|
||||
self._handshake_complete = True
|
||||
|
||||
def _real_connect(self, addr, connect_ex):
|
||||
if self._server_side:
|
||||
raise ValueError("can't connect in server-side mode")
|
||||
|
|
@ -877,13 +914,17 @@ class SSLSocket(object):
|
|||
|
||||
def get_peer_x509(self):
|
||||
"""
|
||||
Returns WolfSSLX509 object representing the peer's certificate,
|
||||
after making a successful SSL/TLS connection.
|
||||
Returns a WolfSSLX509 object representing the peer's certificate,
|
||||
or None if the peer did not present one (or there is no session).
|
||||
"""
|
||||
if self.native_object == _ffi.NULL:
|
||||
return None
|
||||
|
||||
return WolfSSLX509(self.native_object)
|
||||
x509 = _lib.wolfSSL_get_peer_certificate(self.native_object)
|
||||
if x509 == _ffi.NULL:
|
||||
return None
|
||||
|
||||
return WolfSSLX509(x509)
|
||||
|
||||
def getpeercert(self, binary_form=False):
|
||||
"""
|
||||
|
|
|
|||
Loading…
Reference in New Issue