Allow setting and inspecting the preferred client certificate signer list.
This exposes SSL_CTX_add_client_CA, SSL_CTX_set_client_CA_list
and SSL_get_client_CA_list functions to python.
The other *client_CA* functions don't look so useful, so I'll leave them
to someone else.
diff --git a/src/ssl/connection.c b/src/ssl/connection.c
index 5d6d008..bd39b66 100755
--- a/src/ssl/connection.c
+++ b/src/ssl/connection.c
@@ -840,6 +840,58 @@
return lst;
}
+static char ssl_Connection_get_client_CA_list_doc[] = "\n\
+Get CAs whose certificates are suggested for client authentication.\n\
+\n\
+@return: A list of X509Names representing the acceptable CAs as set by\n\
+ ssl.Context.{set, add}_client_CA* if this is a server connection\n\
+ or as sent by the server if this is a client connection.\n\
+";
+
+static PyObject *
+ssl_Connection_get_client_CA_list(ssl_ConnectionObj *self, PyObject *args)
+{
+ STACK_OF(X509_NAME) *CANames;
+ PyObject *CAList;
+ int i, n;
+
+ if (!PyArg_ParseTuple(args, ":get_client_CA_list")) {
+ return NULL;
+ }
+ CANames = SSL_get_client_CA_list(self->ssl);
+ if (CANames == NULL) {
+ return PyList_New(0);
+ }
+ n = sk_X509_NAME_num(CANames);
+ CAList = PyList_New(n);
+ if (CAList == NULL) {
+ return NULL;
+ }
+ for (i = 0; i < n; i++) {
+ X509_NAME *CAName;
+ PyObject *CA;
+
+ CAName = X509_NAME_dup(sk_X509_NAME_value(CANames, i));
+ if (CAName == NULL) {
+ Py_DECREF(CAList);
+ exception_from_error_queue(ssl_Error);
+ return NULL;
+ }
+ CA = (PyObject *)crypto_X509Name_New(CAName, 1);
+ if (CA == NULL) {
+ X509_NAME_free(CAName);
+ Py_DECREF(CAList);
+ return NULL;
+ }
+ if (PyList_SetItem(CAList, i, CA)) {
+ Py_DECREF(CA);
+ Py_DECREF(CAList);
+ return NULL;
+ }
+ }
+ return CAList;
+}
+
static char ssl_Connection_makefile_doc[] = "\n\
The makefile() method is not implemented, since there is no dup semantics\n\
for SSL connections\n\
@@ -1098,6 +1150,7 @@
ADD_METHOD(bio_shutdown),
ADD_METHOD(shutdown),
ADD_METHOD(get_cipher_list),
+ ADD_METHOD(get_client_CA_list),
ADD_METHOD(makefile),
ADD_METHOD(get_app_data),
ADD_METHOD(set_app_data),
diff --git a/src/ssl/context.c b/src/ssl/context.c
index d1ffbe8..a3d798e 100644
--- a/src/ssl/context.c
+++ b/src/ssl/context.c
@@ -800,6 +800,111 @@
}
}
+static char ssl_Context_set_client_CA_list_doc[] = "\n\
+Set the list of preferred client certificate signers for this server context.\n\
+\n\
+This list of certificate authorities will be sent to the client when the\n\
+server requests a client certificate.\n\
+\n\
+@param certificate_authorities: a sequence of X509Names.\n\
+@return: None\n\
+";
+
+static PyObject *
+ssl_Context_set_client_CA_list(ssl_ContextObj *self, PyObject *args)
+{
+ static PyTypeObject *X509NameType;
+ PyObject *sequence, *tuple, *item;
+ crypto_X509NameObj *name;
+ X509_NAME *sslname;
+ STACK_OF(X509_NAME) *CANames;
+ Py_ssize_t length;
+ int i;
+
+ if (X509NameType == NULL) {
+ X509NameType = import_crypto_type("X509Name", sizeof(crypto_X509NameObj));
+ if (X509NameType == NULL) {
+ return NULL;
+ }
+ }
+ if (!PyArg_ParseTuple(args, "O:set_client_CA_list", &sequence)) {
+ return NULL;
+ }
+ tuple = PySequence_Tuple(sequence);
+ if (tuple == NULL) {
+ return NULL;
+ }
+ length = PyTuple_Size(tuple);
+ if (length >= INT_MAX) {
+ PyErr_SetString(PyExc_ValueError, "client CA list is too long");
+ Py_DECREF(tuple);
+ return NULL;
+ }
+ CANames = sk_X509_NAME_new_null();
+ if (CANames == NULL) {
+ Py_DECREF(tuple);
+ exception_from_error_queue(ssl_Error);
+ return NULL;
+ }
+ for (i = 0; i < length; i++) {
+ item = PyTuple_GetItem(tuple, i);
+ if (item->ob_type != X509NameType) {
+ PyErr_Format(PyExc_TypeError,
+ "client CAs must be X509Name objects, not %s objects",
+ item->ob_type->tp_name);
+ sk_X509_NAME_free(CANames);
+ Py_DECREF(tuple);
+ return NULL;
+ }
+ name = (crypto_X509NameObj *)item;
+ sslname = X509_NAME_dup(name->x509_name);
+ if (sslname == NULL) {
+ sk_X509_NAME_free(CANames);
+ Py_DECREF(tuple);
+ exception_from_error_queue(ssl_Error);
+ return NULL;
+ }
+ if (!sk_X509_NAME_push(CANames, sslname)) {
+ X509_NAME_free(sslname);
+ sk_X509_NAME_free(CANames);
+ Py_DECREF(tuple);
+ exception_from_error_queue(ssl_Error);
+ return NULL;
+ }
+ }
+ Py_DECREF(tuple);
+ SSL_CTX_set_client_CA_list(self->ctx, CANames);
+ Py_INCREF(Py_None);
+ return Py_None;
+}
+
+static char ssl_Context_add_client_CA_doc[] = "\n\
+Add the CA certificate to the list of preferred signers for this context.\n\
+\n\
+The list of certificate authorities will be sent to the client when the\n\
+server requests a client certificate.\n\
+\n\
+@param certificate_authority: certificate authority's X509 certificate.\n\
+@return: None\n\
+";
+
+static PyObject *
+ssl_Context_add_client_CA(ssl_ContextObj *self, PyObject *args)
+{
+ crypto_X509Obj *cert;
+
+ cert = parse_certificate_argument("O!:add_client_CA", args);
+ if (cert == NULL) {
+ return NULL;
+ }
+ if (!SSL_CTX_add_client_CA(self->ctx, cert->x509)) {
+ exception_from_error_queue(ssl_Error);
+ return NULL;
+ }
+ Py_INCREF(Py_None);
+ return Py_None;
+}
+
static char ssl_Context_set_timeout_doc[] = "\n\
Set session timeout\n\
\n\
@@ -971,6 +1076,8 @@
ADD_METHOD(get_verify_depth),
ADD_METHOD(load_tmp_dh),
ADD_METHOD(set_cipher_list),
+ ADD_METHOD(set_client_CA_list),
+ ADD_METHOD(add_client_CA),
ADD_METHOD(set_timeout),
ADD_METHOD(get_timeout),
ADD_METHOD(set_info_callback),
diff --git a/src/util.h b/src/util.h
index d9dc7d2..6724f5a 100644
--- a/src/util.h
+++ b/src/util.h
@@ -119,6 +119,10 @@
}
#endif
-
+#if !defined(PY_SSIZE_T_MIN)
+typedef int Py_ssize_t;
+#define PY_SSIZE_T_MAX INT_MAX
+#define PY_SSIZE_T_MIN INT_MIN
+#endif
#endif
diff --git a/test/test_ssl.py b/test/test_ssl.py
index 6698e02..850b6d2 100644
--- a/test/test_ssl.py
+++ b/test/test_ssl.py
@@ -584,6 +584,121 @@
self.assertEquals(e.__class__, Error)
+ def _check_client_CA_list(self, func):
+ server = self._server(None)
+ client = self._client(None)
+ ctx = server.get_context()
+ expected = func(ctx)
+ self.assertEqual(server.get_client_CA_list(), expected)
+ self._loopback(client, server)
+ self.assertEqual(client.get_client_CA_list(), expected)
+ self.assertEqual(server.get_client_CA_list(), expected)
+
+
+ def test_set_client_CA_list_basic(self):
+ """
+ Test paramater validation and return value for
+ L{Context.set_client_CA_list}.
+ """
+ ctx = Context(TLSv1_METHOD)
+ self.assertRaises(TypeError, ctx.set_client_CA_list, "spam")
+ self.assertRaises(TypeError, ctx.set_client_CA_list, ["spam"])
+ self.assertIdentical(ctx.set_client_CA_list([]), None)
+
+
+ def test_set_client_CA_list_functional(self):
+ """
+ The list of CAs set by L{Context.set_client_CA_list} and read by
+ L{Connection.get_client_CA_list} should match on server and
+ client side.
+ """
+ cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
+ secert = load_certificate(FILETYPE_PEM, server_cert_pem)
+ clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
+
+ cadesc = cacert.get_subject()
+ sedesc = secert.get_subject()
+ cldesc = clcert.get_subject()
+
+ def single_CA(ctx):
+ ctx.set_client_CA_list([cadesc])
+ return [cadesc]
+ self._check_client_CA_list(single_CA)
+
+ def no_CA(ctx):
+ ctx.set_client_CA_list([])
+ return []
+ self._check_client_CA_list(no_CA)
+
+ def multiple_CA(ctx):
+ L = [cadesc, sedesc, cldesc]
+ ctx.set_client_CA_list(L)
+ return L
+ self._check_client_CA_list(multiple_CA)
+
+ def changed_CA(ctx):
+ ctx.set_client_CA_list([sedesc, cldesc])
+ ctx.set_client_CA_list([cadesc])
+ return [cadesc]
+ self._check_client_CA_list(changed_CA)
+
+ def mutated_CA(ctx):
+ L = [cadesc]
+ ctx.set_client_CA_list([cadesc])
+ L.append(sedesc)
+ return [cadesc]
+ self._check_client_CA_list(mutated_CA)
+
+
+ def test_add_client_CA_basic(self):
+ """
+ Test paramater validation and return value for
+ L{Context.add_client_CA}.
+ """
+ ctx = Context(TLSv1_METHOD)
+ cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
+ self.assertRaises(TypeError, ctx.add_client_CA, "spam")
+ self.assertIdentical(ctx.add_client_CA(cacert), None)
+
+
+ def test_add_client_CA_functional(self):
+ """
+ The list of CAs set by L{Context.add_client_CA} and read by
+ L{Connection.get_client_CA_list} should match on server and
+ client side.
+ """
+ cacert = load_certificate(FILETYPE_PEM, root_cert_pem)
+ secert = load_certificate(FILETYPE_PEM, server_cert_pem)
+ clcert = load_certificate(FILETYPE_PEM, server_cert_pem)
+
+ cadesc = cacert.get_subject()
+ sedesc = secert.get_subject()
+ cldesc = clcert.get_subject()
+
+ def single_CA(ctx):
+ ctx.add_client_CA(cacert)
+ return [cadesc]
+ self._check_client_CA_list(single_CA)
+
+ def multiple_CA(ctx):
+ ctx.add_client_CA(cacert)
+ ctx.add_client_CA(secert)
+ return [cadesc, sedesc]
+ self._check_client_CA_list(multiple_CA)
+
+ def mixed_set_add_CA(ctx):
+ ctx.set_client_CA_list([cadesc, sedesc])
+ ctx.add_client_CA(clcert)
+ return [cadesc, sedesc, cldesc]
+ self._check_client_CA_list(mixed_set_add_CA)
+
+ def set_replaces_add_CA(ctx):
+ ctx.add_client_CA(clcert)
+ ctx.set_client_CA_list([cadesc])
+ ctx.add_client_CA(secert)
+ return [cadesc, sedesc]
+ self._check_client_CA_list(set_replaces_add_CA)
+
class ModuleTests(TestCase):
"""