Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 8 additions & 0 deletions Doc/library/ssl.rst
Original file line number Diff line number Diff line change
Expand Up @@ -1256,13 +1256,21 @@ SSL sockets also have the following additional methods and attributes:
If certificate verification was disabled method acts the same as
:meth:`~SSLSocket.get_unverified_chain`.

The verified chain is not stored with a session, so the list is empty
when the session is resumed (see :attr:`~SSLSocket.session_reused`).

.. versionadded:: 3.13

.. method:: SSLSocket.get_unverified_chain()

Returns raw certificate chain provided by the other
end of the SSL channel as a list of DER-encoded bytes.

No certificates are exchanged when a session is resumed (see
:attr:`~SSLSocket.session_reused`); the list then holds what was stored
with the session, which for a session resumed from a ticket on the
server side is the peer's own certificate only.

.. versionadded:: 3.13

.. method:: SSLSocket.cipher()
Expand Down
76 changes: 72 additions & 4 deletions Lib/test/test_ssl.py
Original file line number Diff line number Diff line change
Expand Up @@ -2862,11 +2862,19 @@ def run(self):
cert = self.sslconn.getpeercert()
self.write(repr(cert).encode("us-ascii") + b"\n")
elif stripped == b'VERIFIEDCHAIN':
certs = self.sslconn._sslobj.get_verified_chain()
self.write(len(certs).to_bytes(1, "big") + b"\n")
try:
certs = self.sslconn.get_verified_chain()
except Exception as e:
self.write(repr(e).encode("us-ascii") + b"\n")
else:
self.write(len(certs).to_bytes(1, "big") + b"\n")
elif stripped == b'UNVERIFIEDCHAIN':
certs = self.sslconn._sslobj.get_unverified_chain()
self.write(len(certs).to_bytes(1, "big") + b"\n")
try:
certs = self.sslconn.get_unverified_chain()
except Exception as e:
self.write(repr(e).encode("us-ascii") + b"\n")
else:
self.write(len(certs).to_bytes(1, "big") + b"\n")
else:
if (support.verbose and
self.server.connectionchatty):
Expand Down Expand Up @@ -5346,6 +5354,66 @@ def test_internal_chain_server(self):
res = s.recv(1024)
self.assertEqual(res, b'\x02\n')

def test_internal_chain_server_resumed_session(self):
# A session resumed from a ticket stores the client's certificate
# but not the chain it was sent with.
for version in ssl.TLSVersion.TLSv1_2, ssl.TLSVersion.TLSv1_3:
if not has_tls_version(version):
continue
with self.subTest(version=version.name):
self.check_chain_server_resumed_session(version)

def check_chain_server_resumed_session(self, version):
client_context, server_context, hostname = testing_context()
client_context.load_cert_chain(SIGNED_CERTFILE)
server_context.verify_mode = ssl.CERT_REQUIRED
client_context.minimum_version = version
client_context.maximum_version = version

server = ThreadedEchoServer(context=server_context, chatty=False)
with server:
with client_context.wrap_socket(
socket.socket(),
server_hostname=hostname
) as s:
s.connect((HOST, server.port))
s.write(b'UNVERIFIEDCHAIN\n')
res = s.recv(1024)
self.assertEqual(res, b'\x02\n')
# TLS 1.3 sends the ticket after the handshake, so only
# get the session once data was received.
session = s.session
self.assertTrue(session.has_ticket)
with client_context.wrap_socket(
socket.socket(),
server_hostname=hostname,
session=session
) as s:
s.connect((HOST, server.port))
self.assertTrue(s.session_reused)
s.write(b'UNVERIFIEDCHAIN\n')
res = s.recv(1024)
self.assertEqual(res, b'\x01\n')
# The verified chain is not stored with the session at all.
s.write(b'VERIFIEDCHAIN\n')
res = s.recv(1024)
self.assertEqual(res, b'\x00\n')

def test_internal_chain_server_no_client_cert(self):
client_context, server_context, hostname = testing_context()
server_context.verify_mode = ssl.CERT_OPTIONAL

server = ThreadedEchoServer(context=server_context, chatty=False)
with server:
with client_context.wrap_socket(
socket.socket(),
server_hostname=hostname
) as s:
s.connect((HOST, server.port))
s.write(b'UNVERIFIEDCHAIN\n')
res = s.recv(1024)
self.assertEqual(res, b'\x00\n')


class TestSSLDebug(unittest.TestCase):

Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,4 @@
On the server side, :meth:`ssl.SSLSocket.get_unverified_chain` now returns
the client's certificate when the session was resumed from a session ticket,
instead of an empty list, and returns an empty list instead of raising
:exc:`AttributeError` when the client sent no certificate.
42 changes: 24 additions & 18 deletions Modules/_ssl.c
Original file line number Diff line number Diff line change
Expand Up @@ -1954,31 +1954,37 @@ _ssl__SSLSocket_get_unverified_chain_impl(PySSLSocket *self)
/*[clinic end generated code: output=5acdae414e13f913 input=079f8ff5c205cb3b]*/
{
PyObject *retval;
X509 *peer = NULL;
/* borrowed reference */
/* TODO: include SSL_get_peer_certificate() for server-side sockets */
STACK_OF(X509) *chain = SSL_get_peer_cert_chain(self->ssl);
if (chain == NULL) {
/* OpenSSL does not include the peer cert in the chain for server side
* connections, so the chain alone does not tell whether the client sent
* a certificate: the chain is empty when it did not, and NULL when the
* session was resumed from a ticket (which stores the peer cert only). */
if (self->socket_type == PY_SSL_SERVER) {
peer = SSL_get_peer_certificate(self->ssl);
if (peer == NULL) {
Py_RETURN_NONE;
}
}
else if (chain == NULL) {
Py_RETURN_NONE;
}
retval = _PySSL_CertificateFromX509Stack(self->ctx->state, chain, 1);
retval = chain == NULL
? PyList_New(0)
: _PySSL_CertificateFromX509Stack(self->ctx->state, chain, 1);
if (retval == NULL) {
X509_free(peer);
return NULL;
}
/* OpenSSL does not include peer cert for server side connections */
if (self->socket_type == PY_SSL_SERVER) {
PyObject *peerobj = NULL;
X509 *peer = SSL_get_peer_certificate(self->ssl);

if (peer == NULL) {
peerobj = Py_NewRef(Py_None);
} else {
/* consume X509 reference on success */
peerobj = _PySSL_CertificateFromX509(self->ctx->state, peer, 0);
if (peerobj == NULL) {
X509_free(peer);
Py_DECREF(retval);
return NULL;
}
if (peer != NULL) {
/* consume X509 reference on success */
PyObject *peerobj = _PySSL_CertificateFromX509(self->ctx->state,
peer, 0);
if (peerobj == NULL) {
X509_free(peer);
Py_DECREF(retval);
return NULL;
}
int res = PyList_Insert(retval, 0, peerobj);
Py_DECREF(peerobj);
Expand Down
Loading