Merge lp:~mmzeeman/pyopenssl/pkcs7-extensions into lp:~exarkun/pyopenssl/trunk

[M-MZ]

[M-MZ (mmzeeman)]

Hello,

Over the past week I have been busy adding some missing functionality to the PKCS7 module. I have added functions to sign and verify pkcs7 structures.

Kind regards,

Maas-Maarten

[Ziga Seilnacht (zseil)]

Hello,

I'm not sure if I'm the right person for reviewing this, but
here it goes:

1. X509Obj_Sequence_2_Stack:
 - `stack` should be freed in the error branch of the for loop.
 - `sk_X509_push` can fail in low memory conditions. Its return
   value should be checked and appropriate action taken.

2. crypto_pkcs7_sign:
 - What happens if the PKey key contains only a public key or
   no key at all? It would be nice if this case was tested.

3. X509Stack_2_Sequence:
 - Both `X509_dup` and `crypto_X509_New` can fail in low memory
   conditions. An appropriate action should be taken in that case.

4. crypto_PKCS7_get0_signers:
 - `signer_certs` don't get freed if the list was created without
   errors. This looks wrong to me.
 - Why are `signer_certs` freed with `sk_X509_pop_free` instead
   of the more conventional `sk_X509_free`?
 - I don't like the name of this method. Is there a good reason
   for 0 in the name, apart from compatibility with the original
   OpenSSL name?

5. Documentation:
 - Docstrings should describe the required parameters and their
   types, preferably in epydoc format.
 - Latex documentation needs to be written.

6. Tests:
 - TestCase methods should have docstrings explaining what they
   are testing.
 - It would be good if there were some tests against "known good"
   serialized data, either stored as a string in the module or
   generated by spawning openssl.

I'm not qualified to say anything about the API itself, since I
find the win32 API intriguing :)

Thanks for working on this!

Regards,
Ziga

review: Needs Fixing

[M-MZ (mmzeeman)]

Thanks for reviewing.

>
> I'm not sure if I'm the right person for reviewing this, but
> here it goes:
>

Didn't hear anything from anybody else, so thanks.

> 1, 2, 3

You're right about the low memory stuff.

> 4. crypto_PKCS7_get0_signers:
> - `signer_certs` don't get freed if the list was created without
> errors. This looks wrong to me.
> - Why are `signer_certs` freed with `sk_X509_pop_free` instead
> of the more conventional `sk_X509_free`?
> - I don't like the name of this method. Is there a good reason
> for 0 in the name, apart from compatibility with the original
> OpenSSL name?

It's a long time ago since I programmed this. I will take a look into
it. Could be that it is related to the openssl api. Did't know that
sk_X509_free was the more conventional method.

I indeed took the name to follow the openssl api names. The 0 here
means the direct signers of the data.

> 5, 6

Yes indeed

> I'm not qualified to say anything about the API itself, since I
> find the win32 API intriguing :)

Who doesn't... ;) What I did is follow the openssl api closely as that
is what people might know. There is indeed some room for improvement
here (mildly stated).

Again, thanks for the review.

Maas

Unmerged revisions

127. By M-MZ on 2009-10-13

Added get_certificates call to pkcs7 objects

126. By M-MZ on 2009-10-09

Added pkcs7 get0_signers method

125. By M-MZ on 2009-10-09

Added verify method

124. By M-MZ on 2009-10-08

Added pkcs7 sign and pkcs7_get_data

123. By M-MZ on 2009-10-08

Added get_data, no tests yet

=== modified file 'src/crypto/crypto.c'
--- src/crypto/crypto.c	2009-07-17 17:50:12 +0000
+++ src/crypto/crypto.c	2009-10-13 12:45:19 +0000
@@ -483,6 +483,103 @@
     return (PyObject *)crypto_PKCS7_New(pkcs7, 1);
 }
 
+static STACK_OF(X509) *
+X509Obj_Sequence_2_Stack(PyObject *sequence)
+{
+    PyObject *e, *item;
+    int i, nr_of_extensions;
+    STACK_OF(X509) *stack;
+
+    /* Check and create a tuple to get the items */
+    if(!(e = PySequence_Fast(sequence, "Expected a sequence"))) {
+         return NULL;
+    }
+
+    /* Make a STACK_OF(X509_EXTENSION) from sequence */
+    if(!(stack = sk_X509_new_null())) {
+        exception_from_error_queue(crypto_Error);
+        return NULL;
+    }
+
+    /* Put the extensions in a stack */
+    nr_of_extensions = PySequence_Length(e);
+
+    for(i = 0; i < nr_of_extensions; i++) {
+        item = PySequence_Fast_GET_ITEM(e, i);
+        if(!(crypto_X509_Check(item))) {
+            PyErr_SetString(PyExc_ValueError,
+                            "One of the elements is not a X509 object");
+            Py_DECREF(e);
+            return NULL;
+        }
+        sk_X509_push(stack, ((crypto_X509Obj *)item)->x509);
+    }
+    Py_DECREF(e);
+
+    return stack;
+}
+
+
+static char crypto_pkcs7_sign_doc[] = "\n\
+Sign data and embed it in a pkcs7 object\n\
+\n\
+@return: The PKCS7 object\n\
+";
+static PyObject *
+crypto_pkcs7_sign(PyObject *spam, PyObject *args)
+{
+    PKCS7 *p7;
+    crypto_PKeyObj *key;
+    crypto_X509Obj *cert;
+    PyObject *chain = NULL;
+    STACK_OF(X509) *stack = NULL;
+    char *buffer;
+    int len;
+    int flags = -1;
+    BIO *bio;
+
+    if(!PyArg_ParseTuple(args, "O!O!s#|Oi:pkcs7_sign",
+                         &crypto_X509_Type, &cert,
+                         &crypto_PKey_Type, &key,
+                         &buffer, &len,
+                         &chain, &flags))
+        return NULL;
+
+    /* Set the chain to NULL if None is given */
+    if(chain == Py_None)
+        chain = NULL;
+
+    if(chain) {
+        if(!(stack = X509Obj_Sequence_2_Stack(chain)))
+            return NULL;
+    }
+
+    if(flags == -1) {
+        flags = PKCS7_BINARY | PKCS7_NOSMIMECAP;
+    }
+
+    if(!(bio = BIO_new_mem_buf(buffer, len))) {
+        PyErr_SetString(PyExc_MemoryError, "Could not allocate bio.");
+        return NULL;
+    }
+
+    if(!(p7 = PKCS7_sign(cert->x509, key->pkey, stack, bio, flags))) {
+        exception_from_error_queue(crypto_Error);
+        if(stack)
+             sk_X509_free(stack);
+        BIO_free(bio);
+        return NULL;
+    }
+   
+    if(stack)
+         sk_X509_free(stack);
+
+    BIO_free(bio);
+
+    return (PyObject *)crypto_PKCS7_New(p7, 1);
+}
+
+
 static char crypto_load_pkcs12_doc[] = "\n\
 Load a PKCS12 object from a buffer\n\
 \n\
@@ -556,6 +653,7 @@
     { "load_certificate_request", (PyCFunction)crypto_load_certificate_request, METH_VARARGS, crypto_load_certificate_request_doc },
     { "dump_certificate_request", (PyCFunction)crypto_dump_certificate_request, METH_VARARGS, crypto_dump_certificate_request_doc },
     { "load_pkcs7_data", (PyCFunction)crypto_load_pkcs7_data, METH_VARARGS, crypto_load_pkcs7_data_doc },
+    { "pkcs7_sign", (PyCFunction)crypto_pkcs7_sign, METH_VARARGS, crypto_pkcs7_sign_doc },
     { "load_pkcs12", (PyCFunction)crypto_load_pkcs12, METH_VARARGS, crypto_load_pkcs12_doc },
     { "X509_verify_cert_error_string", (PyCFunction)crypto_X509_verify_cert_error_string, METH_VARARGS, crypto_X509_verify_cert_error_string_doc },
     { "_exception_from_error_queue", (PyCFunction)crypto_exception_from_error_queue, METH_NOARGS, crypto_exception_from_error_queue_doc },
@@ -671,6 +769,11 @@
     PyModule_AddIntConstant(module, "FILETYPE_ASN1", X509_FILETYPE_ASN1);
     PyModule_AddIntConstant(module, "FILETYPE_TEXT", X509_FILETYPE_TEXT);
 
+    PyModule_AddIntConstant(module, "PKCS7_DETACHED", PKCS7_DETACHED);
+    PyModule_AddIntConstant(module, "PKCS7_BINARY", PKCS7_BINARY);
+    PyModule_AddIntConstant(module, "PKCS7_NOSMIMECAP", PKCS7_NOSMIMECAP);
+    PyModule_AddIntConstant(module, "PKCS7_NOCERTS", PKCS7_NOCERTS);
+
     PyModule_AddIntConstant(module, "TYPE_RSA", crypto_TYPE_RSA);
     PyModule_AddIntConstant(module, "TYPE_DSA", crypto_TYPE_DSA);
 

=== modified file 'src/crypto/pkcs7.c'
--- src/crypto/pkcs7.c	2009-06-27 18:32:07 +0000
+++ src/crypto/pkcs7.c	2009-10-13 12:45:19 +0000
@@ -83,6 +83,40 @@
         return PyInt_FromLong(0L);
 }
 
+static PyObject *
+X509Stack_2_Sequence(STACK_OF(X509) *stack)
+{
+    PyObject *list;
+    X509 *cert;
+    crypto_X509Obj *wrapped_cert;
+    int i;
+
+    if(!(list = PyList_New(sk_X509_num(stack)))) {
+        PyErr_SetString(PyExc_MemoryError,
+                        "Could not allocate memory for list");
+        return NULL;
+    }
+
+    /* 
+     * Convert the stack to a sequence with crypto_X509Obj's
+     */
+    for(i = 0; i < sk_X509_num(stack); i++) {
+        cert = sk_X509_value(stack, i);
+
+        /* Wrap the "bare" certificate, a copy will be made, this
+         * will make it possible to destroy the pkcs7 structure
+         * without influencing the wrapped certificates.
+         */
+        wrapped_cert = crypto_X509_New(X509_dup(cert), 1);
+
+        /* And insert it into the list 
+         */
+        PyList_SetItem(list, i, (PyObject *)wrapped_cert);
+    }
+
+    return list;
+}
+
 static char crypto_PKCS7_get_type_name_doc[] = "\n\
 Returns the type name of the PKCS7 structure\n\
 \n\
@@ -101,6 +135,141 @@
     return PyString_FromString(OBJ_nid2sn(OBJ_obj2nid(self->pkcs7->type)));
 }
 
+static char crypto_PKCS7_get0_signers_doc[] ="\n\
+Retrieve the certificates of the signers of this PKCS7 structure\n\
+\n\
+@return: A list with X509 certificates\n\
+";
+
+static PyObject *
+crypto_PKCS7_get0_signers(crypto_PKCS7Obj *self, PyObject *args)
+{
+    STACK_OF(X509) *signer_certs;
+    PyObject *list;
+
+    if(!PyArg_ParseTuple(args, ":get0_signers"))
+        return NULL;
+
+    if(!(signer_certs = PKCS7_get0_signers(self->pkcs7, NULL, 0))) {
+        exception_from_error_queue(crypto_Error);
+        return NULL;
+    }
+
+    list = X509Stack_2_Sequence(signer_certs);
+
+    if(!list) {
+        sk_X509_pop_free(signer_certs, X509_free);
+        return NULL;
+    }
+
+    return list;
+}
+
+static char crypto_PKCS7_get_certificates_doc[] = "\n\
+Retrieve all certificates from the PKCS7 structure\n\
+\n\
+@return: A list with X509 certificates\n\
+";
+
+static PyObject *
+crypto_PKCS7_get_certificates(crypto_PKCS7Obj *self, PyObject *args)
+{
+    STACK_OF(X509) *certs;
+
+    if(!PyArg_ParseTuple(args, ":get_certificates"))
+        return NULL;
+
+    switch(OBJ_obj2nid(self->pkcs7->type)) {
+    case NID_pkcs7_signed:
+        certs = self->pkcs7->d.sign->cert;
+        break;
+    case NID_pkcs7_signedAndEnveloped:
+        certs = self->pkcs7->d.signed_and_enveloped->cert;
+        break;
+    default:
+        PyErr_SetString(PyExc_ValueError,
+                        "It is only possible to get certificates from a signed pkcs7 object");
+        return NULL;
+    }
+
+    return X509Stack_2_Sequence(certs);
+} 
+
+static char crypto_PKCS7_verify_doc[] = "\n\
+Verifies the signature(s) in the PKCS7 structure\n\
+\n\
+@return True if the signatures are ok\n\
+";
+
+static PyObject *
+crypto_PKCS7_verify(crypto_PKCS7Obj *self, PyObject *args)
+{
+    int flags = 0;
+    int indata_len = 0;
+    unsigned char *indata = NULL;
+    BIO *inbio = NULL;
+    int vr;
+
+    if(!PyArg_ParseTuple(args, "|s#i:verify", &indata, &indata_len, &flags))
+        return NULL;
+
+    /* 
+     * Call openssl's verify, but without verifying the certificates.
+     * This can be done in python.
+     */
+    flags |= PKCS7_NOVERIFY;
+
+    if(indata_len > 0) {
+        if(!(inbio = BIO_new_mem_buf(indata, indata_len))) {
+            PyErr_SetString(PyExc_MemoryError, "Could not allocate bio.");
+            return NULL;
+        }
+    }
+
+    vr = PKCS7_verify(self->pkcs7, NULL, NULL, inbio, NULL, flags); 
+    if(inbio) {
+        BIO_free(inbio);
+    }
+
+    if(vr != 1) {
+        exception_from_error_queue(crypto_Error);
+        return NULL;
+    }
+
+    Py_INCREF(Py_True);
+    return Py_True;
+}
+
+static char crypto_PKCS7_get_data_doc[] = "\n\
+Returns the data embedded in the PKCS7 structure\n\
+\n\
+@return: A string with the embedded data\n\
+";
+static PyObject *
+crypto_PKCS7_get_data(crypto_PKCS7Obj *self, PyObject *args)
+{
+    BIO *bio;
+    char *buf = NULL;
+    int len = 0;
+    PyObject *data;
+
+    if(!PyArg_ParseTuple(args, ":get_data"))
+        return NULL;
+
+    bio = PKCS7_dataInit(self->pkcs7, NULL);
+    if(!bio) {
+         PyErr_SetString(PyExc_ValueError,
+                         "PKCS#7 structure does not have embedded data.");
+         return NULL;
+    }
+
+    len = BIO_get_mem_data(bio, &buf);
+    data = PyString_FromStringAndSize(buf, len);
+    BIO_free_all(bio);
+
+    return data;
+}
+
 /*
  * ADD_METHOD(name) expands to a correct PyMethodDef declaration
  *   {  'name', (PyCFunction)crypto_PKCS7_name, METH_VARARGS }
@@ -115,6 +284,10 @@
     ADD_METHOD(type_is_signedAndEnveloped),
     ADD_METHOD(type_is_data),
     ADD_METHOD(get_type_name),
+    ADD_METHOD(get0_signers),
+    ADD_METHOD(get_certificates),
+    ADD_METHOD(verify),
+    ADD_METHOD(get_data),
     { NULL, NULL }
 };
 #undef ADD_METHOD

=== modified file 'test/test_crypto.py'
--- test/test_crypto.py	2009-08-27 16:47:55 +0000
+++ test/test_crypto.py	2009-10-13 12:45:19 +0000
@@ -11,6 +11,7 @@
 from datetime import datetime, timedelta
 
 from OpenSSL.crypto import TYPE_RSA, TYPE_DSA, Error, PKey, PKeyType
+from OpenSSL.crypto import PKCS7_DETACHED
 from OpenSSL.crypto import X509, X509Type, X509Name, X509NameType
 from OpenSSL.crypto import X509Req, X509ReqType
 from OpenSSL.crypto import X509Extension, X509ExtensionType
@@ -18,7 +19,7 @@
 from OpenSSL.crypto import FILETYPE_PEM, FILETYPE_ASN1, FILETYPE_TEXT
 from OpenSSL.crypto import dump_certificate, load_certificate_request
 from OpenSSL.crypto import dump_certificate_request, dump_privatekey
-from OpenSSL.crypto import PKCS7Type, load_pkcs7_data
+from OpenSSL.crypto import PKCS7Type, load_pkcs7_data, pkcs7_sign
 from OpenSSL.crypto import PKCS12Type, load_pkcs12, PKCS12
 from OpenSSL.crypto import NetscapeSPKI, NetscapeSPKIType
 from OpenSSL.test.util import TestCase
@@ -1509,6 +1510,43 @@
         # XXX This doesn't currently work.
         # self.assertIdentical(PKCS7, PKCS7Type)
 
+    def test_pkcs7_sign_and_verify(self):
+        root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
+        root_key = load_privatekey(FILETYPE_PEM, root_key_pem)
+        pkcs7 = pkcs7_sign(root_cert, root_key, "Hello world") 
+        self.assertTrue(pkcs7)
+
+        self.assertEquals("Hello world", pkcs7.get_data())
+        self.assertEquals(True, pkcs7.verify())
+
+        # Too bad, no direct compare possible.
+        self.assertEquals(root_cert.get_serial_number(), 
+            pkcs7.get0_signers()[0].get_serial_number())
+
+    def test_pkcs7_sign_and_verify_detached(self):
+        root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
+        root_key = load_privatekey(FILETYPE_PEM, root_key_pem)
+        pkcs7 = pkcs7_sign(root_cert, root_key, "Hello world", [], PKCS7_DETACHED) 
+        self.assertEquals("", pkcs7.get_data())
+        self.assertEquals(True, pkcs7.verify("Hello world", PKCS7_DETACHED))
+        self.assertRaises(Error, pkcs7.verify, "Bad data", PKCS7_DETACHED)
+
+    def test_pkcs7_get_certificates(self):
+        root_cert = load_certificate(FILETYPE_PEM, root_cert_pem)
+        server_cert = load_certificate(FILETYPE_PEM, server_cert_pem)
+        server_key = load_privatekey(FILETYPE_PEM, server_key_pem)
+
+        pkcs7 = pkcs7_sign(server_cert, server_key, "Hello world", [], PKCS7_DETACHED)
+        self.assertEquals(1, len(pkcs7.get0_signers()))
+        self.assertEquals(1, len(pkcs7.get_certificates()))
+
+        pkcs7 = pkcs7_sign(server_cert, server_key, "Hello world", [root_cert])
+        self.assertEquals(1, len(pkcs7.get0_signers()))
+        self.assertEquals(2, len(pkcs7.get_certificates()))
+
+    def test_get_data(self):
+        pkcs7 = load_pkcs7_data(FILETYPE_PEM, pkcs7Data)
+        self.assertRaises(ValueError, pkcs7.get_data)
 
 
 class NetscapeSPKITests(TestCase):