Add a Connection method for inspecting the certificate chain.
diff --git a/ChangeLog b/ChangeLog
index 5f60a16..2a4c6ec 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -1,3 +1,9 @@
+2011-05-20 Jean-Paul Calderone <exarkun@twistedmatrix.com>
+
+ * OpenSSL/ssl/connection.c, OpenSSL/test/test_ssl.py: Add a new
+ method to the Connection type, get_peer_cert_chain, for retrieving
+ the peer's certificate chain.
+
2011-05-19 Jean-Paul Calderone <exarkun@twistedmatrix.com>
* OpenSSL/crypto/x509.c, OpenSSL/test/test_crypto.py: Add a new
diff --git a/OpenSSL/ssl/connection.c b/OpenSSL/ssl/connection.c
index 5b304b1..a3ec0f0 100755
--- a/OpenSSL/ssl/connection.c
+++ b/OpenSSL/ssl/connection.c
@@ -1098,6 +1098,44 @@
}
}
+static char ssl_Connection_get_peer_cert_chain_doc[] = "\n\
+Retrieve the other side's certificate (if any)\n\
+\n\
+@return: A list of X509 instances giving the peer's certificate chain,\n\
+ or None if it does not have one.\n\
+";
+static PyObject *
+ssl_Connection_get_peer_cert_chain(ssl_ConnectionObj *self, PyObject *args) {
+ STACK_OF(X509) *sk;
+ PyObject *chain;
+ crypto_X509Obj *cert;
+ Py_ssize_t i;
+
+ if (!PyArg_ParseTuple(args, ":get_peer_cert_chain")) {
+ return NULL;
+ }
+
+ sk = SSL_get_peer_cert_chain(self->ssl);
+ if (sk != NULL) {
+ chain = PyList_New(sk_X509_num(sk));
+ for (i = 0; i < sk_X509_num(sk); i++) {
+ cert = new_x509(sk_X509_value(sk, i), 1);
+ if (!cert) {
+ /* XXX Untested */
+ Py_DECREF(chain);
+ return NULL;
+ }
+ CRYPTO_add(&cert->x509->references, 1, CRYPTO_LOCK_X509);
+ PyList_SET_ITEM(chain, i, (PyObject *)cert);
+ }
+ return chain;
+ } else {
+ Py_INCREF(Py_None);
+ return Py_None;
+ }
+
+}
+
static char ssl_Connection_want_read_doc[] = "\n\
Checks if more data has to be read from the transport layer to complete an\n\
operation.\n\
@@ -1175,6 +1213,7 @@
ADD_METHOD(master_key),
ADD_METHOD(sock_shutdown),
ADD_METHOD(get_peer_certificate),
+ ADD_METHOD(get_peer_cert_chain),
ADD_METHOD(want_read),
ADD_METHOD(want_write),
ADD_METHOD(set_accept_state),
diff --git a/OpenSSL/test/test_ssl.py b/OpenSSL/test/test_ssl.py
index ff2e725..2761cec 100644
--- a/OpenSSL/test/test_ssl.py
+++ b/OpenSSL/test/test_ssl.py
@@ -12,7 +12,7 @@
from os.path import join
from unittest import main
-from OpenSSL.crypto import TYPE_RSA, FILETYPE_PEM, FILETYPE_ASN1
+from OpenSSL.crypto import TYPE_RSA, FILETYPE_PEM
from OpenSSL.crypto import PKey, X509, X509Extension
from OpenSSL.crypto import dump_privatekey, load_privatekey
from OpenSSL.crypto import dump_certificate, load_certificate
@@ -22,14 +22,19 @@
from OpenSSL.SSL import SENT_SHUTDOWN, RECEIVED_SHUTDOWN
from OpenSSL.SSL import SSLv2_METHOD, SSLv3_METHOD, SSLv23_METHOD, TLSv1_METHOD
from OpenSSL.SSL import OP_NO_SSLv2, OP_NO_SSLv3, OP_SINGLE_DH_USE
-from OpenSSL.SSL import VERIFY_PEER, VERIFY_FAIL_IF_NO_PEER_CERT, VERIFY_CLIENT_ONCE
-from OpenSSL.SSL import Error, SysCallError, WantReadError, ZeroReturnError, SSLeay_version
+from OpenSSL.SSL import (
+ VERIFY_PEER, VERIFY_FAIL_IF_NO_PEER_CERT, VERIFY_CLIENT_ONCE, VERIFY_NONE)
+from OpenSSL.SSL import (
+ Error, SysCallError, WantReadError, ZeroReturnError, SSLeay_version)
from OpenSSL.SSL import Context, ContextType, Connection, ConnectionType
from OpenSSL.test.util import TestCase, bytes, b
-from OpenSSL.test.test_crypto import cleartextCertificatePEM, cleartextPrivateKeyPEM
-from OpenSSL.test.test_crypto import client_cert_pem, client_key_pem
-from OpenSSL.test.test_crypto import server_cert_pem, server_key_pem, root_cert_pem
+from OpenSSL.test.test_crypto import (
+ cleartextCertificatePEM, cleartextPrivateKeyPEM)
+from OpenSSL.test.test_crypto import (
+ client_cert_pem, client_key_pem, server_cert_pem, server_key_pem,
+ root_cert_pem)
+
try:
from OpenSSL.SSL import OP_NO_QUERY_MTU
except ImportError:
@@ -63,6 +68,7 @@
def verify_cb(conn, cert, errnum, depth, ok):
return ok
+
def socket_pair():
"""
Establish and return a pair of network sockets connected to each other.
@@ -105,6 +111,60 @@
conns.remove(conn)
+def _create_certificate_chain():
+ """
+ Construct and return a chain of certificates.
+
+ 1. A new self-signed certificate authority certificate (cacert)
+ 2. A new intermediate certificate signed by cacert (icert)
+ 3. A new server certificate signed by icert (scert)
+ """
+ caext = X509Extension(b('basicConstraints'), False, b('CA:true'))
+
+ # Step 1
+ cakey = PKey()
+ cakey.generate_key(TYPE_RSA, 512)
+ cacert = X509()
+ cacert.get_subject().commonName = "Authority Certificate"
+ cacert.set_issuer(cacert.get_subject())
+ cacert.set_pubkey(cakey)
+ cacert.set_notBefore(b("20000101000000Z"))
+ cacert.set_notAfter(b("20200101000000Z"))
+ cacert.add_extensions([caext])
+ cacert.set_serial_number(0)
+ cacert.sign(cakey, "sha1")
+
+ # Step 2
+ ikey = PKey()
+ ikey.generate_key(TYPE_RSA, 512)
+ icert = X509()
+ icert.get_subject().commonName = "Intermediate Certificate"
+ icert.set_issuer(cacert.get_subject())
+ icert.set_pubkey(ikey)
+ icert.set_notBefore(b("20000101000000Z"))
+ icert.set_notAfter(b("20200101000000Z"))
+ icert.add_extensions([caext])
+ icert.set_serial_number(0)
+ icert.sign(cakey, "sha1")
+
+ # Step 3
+ skey = PKey()
+ skey.generate_key(TYPE_RSA, 512)
+ scert = X509()
+ scert.get_subject().commonName = "Server Certificate"
+ scert.set_issuer(icert.get_subject())
+ scert.set_pubkey(skey)
+ scert.set_notBefore(b("20000101000000Z"))
+ scert.set_notAfter(b("20200101000000Z"))
+ scert.add_extensions([
+ X509Extension(b('basicConstraints'), True, b('CA:false'))])
+ scert.set_serial_number(0)
+ scert.sign(ikey, "sha1")
+
+ return [(cakey, cacert), (ikey, icert), (skey, scert)]
+
+
+
class _LoopbackMixin:
"""
Helper mixin which defines methods for creating a connected socket pair and
@@ -150,7 +210,7 @@
# Give the side a chance to generate some more bytes, or
# succeed.
try:
- bytes = read.recv(2 ** 16)
+ data = read.recv(2 ** 16)
except WantReadError:
# It didn't succeed, so we'll hope it generated some
# output.
@@ -158,7 +218,7 @@
else:
# It did succeed, so we'll stop now and let the caller deal
# with it.
- return (read, bytes)
+ return (read, data)
while True:
# Keep copying as long as there's more stuff there.
@@ -175,6 +235,7 @@
write.bio_write(dirty)
+
class VersionTests(TestCase):
"""
Tests for version information exposed by
@@ -638,59 +699,6 @@
self.assertRaises(TypeError, context.add_extra_chain_cert, object(), object())
- def _create_certificate_chain(self):
- """
- Construct and return a chain of certificates.
-
- 1. A new self-signed certificate authority certificate (cacert)
- 2. A new intermediate certificate signed by cacert (icert)
- 3. A new server certificate signed by icert (scert)
- """
- caext = X509Extension(b('basicConstraints'), False, b('CA:true'))
-
- # Step 1
- cakey = PKey()
- cakey.generate_key(TYPE_RSA, 512)
- cacert = X509()
- cacert.get_subject().commonName = "Authority Certificate"
- cacert.set_issuer(cacert.get_subject())
- cacert.set_pubkey(cakey)
- cacert.set_notBefore(b("20000101000000Z"))
- cacert.set_notAfter(b("20200101000000Z"))
- cacert.add_extensions([caext])
- cacert.set_serial_number(0)
- cacert.sign(cakey, "sha1")
-
- # Step 2
- ikey = PKey()
- ikey.generate_key(TYPE_RSA, 512)
- icert = X509()
- icert.get_subject().commonName = "Intermediate Certificate"
- icert.set_issuer(cacert.get_subject())
- icert.set_pubkey(ikey)
- icert.set_notBefore(b("20000101000000Z"))
- icert.set_notAfter(b("20200101000000Z"))
- icert.add_extensions([caext])
- icert.set_serial_number(0)
- icert.sign(cakey, "sha1")
-
- # Step 3
- skey = PKey()
- skey.generate_key(TYPE_RSA, 512)
- scert = X509()
- scert.get_subject().commonName = "Server Certificate"
- scert.set_issuer(icert.get_subject())
- scert.set_pubkey(skey)
- scert.set_notBefore(b("20000101000000Z"))
- scert.set_notAfter(b("20200101000000Z"))
- scert.add_extensions([
- X509Extension(b('basicConstraints'), True, b('CA:false'))])
- scert.set_serial_number(0)
- scert.sign(ikey, "sha1")
-
- return [(cakey, cacert), (ikey, icert), (skey, scert)]
-
-
def _handshake_test(self, serverContext, clientContext):
"""
Verify that a client and server created with the given contexts can
@@ -726,7 +734,7 @@
to it with a client which trusts cacert and requires verification to
succeed.
"""
- chain = self._create_certificate_chain()
+ chain = _create_certificate_chain()
[(cakey, cacert), (ikey, icert), (skey, scert)] = chain
# Dump the CA certificate to a file because that's the only way to load
@@ -767,7 +775,7 @@
to it with a client which trusts cacert and requires verification to
succeed.
"""
- chain = self._create_certificate_chain()
+ chain = _create_certificate_chain()
[(cakey, cacert), (ikey, icert), (skey, scert)] = chain
# Write out the chain file.
@@ -1095,6 +1103,68 @@
self.assertRaises(NotImplementedError, conn.makefile)
+ def test_get_peer_cert_chain_wrong_args(self):
+ """
+ L{Connection.get_peer_cert_chain} raises L{TypeError} if called with any
+ arguments.
+ """
+ conn = Connection(Context(TLSv1_METHOD), None)
+ self.assertRaises(TypeError, conn.get_peer_cert_chain, 1)
+ self.assertRaises(TypeError, conn.get_peer_cert_chain, "foo")
+ self.assertRaises(TypeError, conn.get_peer_cert_chain, object())
+ self.assertRaises(TypeError, conn.get_peer_cert_chain, [])
+
+
+ def test_get_peer_cert_chain(self):
+ """
+ L{Connection.get_peer_cert_chain} returns a list of certificates which
+ the connected server returned for the certification verification.
+ """
+ chain = _create_certificate_chain()
+ [(cakey, cacert), (ikey, icert), (skey, scert)] = chain
+
+ serverContext = Context(TLSv1_METHOD)
+ serverContext.use_privatekey(skey)
+ serverContext.use_certificate(scert)
+ serverContext.add_extra_chain_cert(icert)
+ serverContext.add_extra_chain_cert(cacert)
+ server = Connection(serverContext, None)
+ server.set_accept_state()
+
+ # Create the client
+ clientContext = Context(TLSv1_METHOD)
+ clientContext.set_verify(VERIFY_NONE, verify_cb)
+ client = Connection(clientContext, None)
+ client.set_connect_state()
+
+ self._interactInMemory(client, server)
+
+ chain = client.get_peer_cert_chain()
+ self.assertEqual(len(chain), 3)
+ self.assertEqual(
+ "Server Certificate", chain[0].get_subject().CN)
+ self.assertEqual(
+ "Intermediate Certificate", chain[1].get_subject().CN)
+ self.assertEqual(
+ "Authority Certificate", chain[2].get_subject().CN)
+
+
+ def test_get_peer_cert_chain_none(self):
+ """
+ L{Connection.get_peer_cert_chain} returns C{None} if the peer sends no
+ certificate chain.
+ """
+ ctx = Context(TLSv1_METHOD)
+ ctx.use_privatekey(load_privatekey(FILETYPE_PEM, server_key_pem))
+ ctx.use_certificate(load_certificate(FILETYPE_PEM, server_cert_pem))
+ server = Connection(ctx, None)
+ server.set_accept_state()
+ client = Connection(Context(TLSv1_METHOD), None)
+ client.set_connect_state()
+ self._interactInMemory(client, server)
+ self.assertIdentical(None, server.get_peer_cert_chain())
+
+
class ConnectionGetCipherListTests(TestCase):
"""
diff --git a/OpenSSL/tsafe.py b/OpenSSL/tsafe.py
index fe4b75f..9d7ad2f 100644
--- a/OpenSSL/tsafe.py
+++ b/OpenSSL/tsafe.py
@@ -16,7 +16,7 @@
'setblocking', 'fileno', 'shutdown', 'close', 'get_cipher_list',
'getpeername', 'getsockname', 'getsockopt', 'setsockopt',
'makefile', 'get_app_data', 'set_app_data', 'state_string',
- 'sock_shutdown', 'get_peer_certificate', 'want_read',
+ 'sock_shutdown', 'get_peer_certificate', 'get_peer_cert_chain', 'want_read',
'want_write', 'set_connect_state', 'set_accept_state',
'connect_ex', 'sendall'):
exec("""def %s(self, *args):
diff --git a/doc/pyOpenSSL.tex b/doc/pyOpenSSL.tex
index fa7638d..93213f9 100644
--- a/doc/pyOpenSSL.tex
+++ b/doc/pyOpenSSL.tex
@@ -1203,6 +1203,10 @@
Retrieve the other side's certificate (if any)
\end{methoddesc}
+\begin{methoddesc}[Connection]{get_peer_cert_chain}{}
+Retrieve the tuple of the other side's certificate chain (if any)
+\end{methoddesc}
+
\begin{methoddesc}[Connection]{getpeername}{}
Call the \method{getpeername} method of the underlying socket.
\end{methoddesc}