add ECDSA support to coro.ssl.openssl

coro/ssl/openssl.pxi

@@ -154,6 +154,22 @@ cdef extern from "openssl/evp.h":
     int  EVP_VerifyFinal               (EVP_MD_CTX *, char *, int, EVP_PKEY *)
     int EVP_MAX_MD_SIZE
 
+cdef extern from "openssl/ec.h":
+    ctypedef struct EC_KEY
+    ctypedef struct EC_POINT
+    EC_KEY * EC_KEY_new_by_curve_name    (int)
+    int      EC_KEY_generate_key         (EC_KEY *)
+    void     EC_KEY_free                 (EC_KEY *)
+    EC_KEY * d2i_ECPrivateKey            (EC_KEY **, const unsigned char **, long)
+    EC_KEY * o2i_ECPublicKey             (EC_KEY **, const unsigned char **, long)
+    int      i2d_ECPrivateKey            (EC_KEY *, unsigned char **)
+    int      i2o_ECPublicKey             (EC_KEY *, unsigned char **)
+
+cdef extern from "openssl/ecdsa.h":
+    int            ECDSA_size     (const EC_KEY * eckey)
+    int            ECDSA_sign     (int type, const unsigned char *dgst, int dgstlen, unsigned char *sig, unsigned int *siglen, EC_KEY *eckey)
+    int            ECDSA_verify   (int type, const unsigned char *dgst, int dgstlen, const unsigned char *sig, int siglen, EC_KEY *eckey)
+
 cdef extern from "openssl/x509.h":
     ctypedef struct X509
     X509 *X509_new ()
@@ -346,6 +362,7 @@ cdef extern from "openssl/objects.h":
     int OBJ_obj2nid (ASN1_OBJECT *)
     char *OBJ_nid2sn (int)
     char *OBJ_nid2ln (int)
+    int OBJ_sn2nid (char *)
 
 cdef extern from "openssl/pkcs12.h":
     ctypedef struct PKCS12

coro/ssl/openssl.pyx

@@ -1114,7 +1114,7 @@ cdef class ssl_ctx:
             for proto in protos:
                 r.append (chr (len (proto)))
                 r.append (proto)
-            self.next_protos = <bytes> (''.join (r))
+            self.next_protos = ''.join (r)
             SSL_CTX_set_next_protos_advertised_cb (self.ctx, next_protos_server_callback, <void*>self)
             SSL_CTX_set_next_proto_select_cb (self.ctx, next_protos_client_callback, <void*>self)
 
@@ -1277,6 +1277,94 @@ cdef class digest:
 
 # ================================================================================
 
+cdef class ecdsa:
+
+    cdef EC_KEY * key
+
+    def __init__ (self, object curve):
+        cdef int nid
+        if type(curve) is int:
+            nid = curve
+        else:
+            nid = OBJ_sn2nid (curve)
+        if nid == 0:
+            raise_ssl_error()
+        else:
+            self.key = EC_KEY_new_by_curve_name (nid)
+            if self.key is NULL:
+                raise_ssl_error()
+
+    def __dealloc__ (self):
+        if self.key is not NULL:
+            EC_KEY_free (self.key)
+
+    def generate (self):
+        cdef int status = EC_KEY_generate_key (self.key)
+        if status != 1:
+            raise_ssl_error()
+
+    def set_privkey (self, bytes pkey):
+        cdef const unsigned char * p = pkey
+        cdef EC_KEY * result = d2i_ECPrivateKey (&self.key, &p, len(pkey))
+        if result is NULL:
+            raise_ssl_error()
+
+    def set_pubkey (self, key):
+        cdef const unsigned char * p = key
+        cdef EC_KEY * result = o2i_ECPublicKey (&self.key, &p, len (key))
+        if result is NULL:
+            raise_ssl_error()
+
+    def get_privkey (self):
+        cdef int r = 0
+        cdef int size = i2d_ECPrivateKey (self.key, NULL)
+        cdef bytes result
+        cdef unsigned char * p
+        if size == 0:
+            raise_ssl_error()
+        else:
+            result = PyBytes_FromStringAndSize (NULL, size)
+            p = result
+            r = i2d_ECPrivateKey (self.key, &p)
+            if r == 0:
+                raise_ssl_error()
+            else:
+                return result
+
+    def get_pubkey (self):
+        cdef int r = 0
+        cdef int size = i2o_ECPublicKey (self.key, NULL)
+        cdef bytes result
+        cdef unsigned char * p
+        if size == 0:
+            raise_ssl_error()
+        else:
+            result = PyBytes_FromStringAndSize (NULL, size)
+            p = result
+            r = i2o_ECPublicKey (self.key, &p)
+            if r == 0:
+                raise_ssl_error()
+            else:
+                return result
+
+    def sign (self, bytes data):
+        cdef unsigned int sig_size = ECDSA_size (self.key)
+        cdef bytes sig = PyBytes_FromStringAndSize (NULL, sig_size)
+        cdef int result = ECDSA_sign (0, data, len(data), sig, &sig_size, self.key)
+        if result != 1:
+            raise_ssl_error()
+        else:
+            return sig[:sig_size]
+
+    def verify (self, bytes data, bytes sig):
+        cdef int result = ECDSA_verify (0, data, len(data), sig, len(sig), self.key)
+        if result == -1:
+            raise_ssl_error()
+        else:
+            return result
+
+# ================================================================================
+
 def random_status():
     return RAND_status()
 

test/test_ecdsa.py

+# -*- Mode: Python -*-
+
+import hashlib
+import unittest
+
+import coro.ssl.openssl
+import coro_unittest
+
+class Test (unittest.TestCase):
+
+    def test_gen_key (self):
+        e = coro.ssl.openssl.ecdsa ("secp256k1")
+        e.generate()
+        h = hashlib.new ('sha256')
+        h.update ('asdfasdfasdfasdfasdf')
+        d = h.digest()
+        sig = e.sign (d)
+        self.assertEqual (e.verify (d, sig), 1)
+
+    def test_get_keys (self):
+        e = coro.ssl.openssl.ecdsa ("secp256k1")
+        e.generate()
+        k = e.get_privkey()
+        p = e.get_pubkey()
+
+    def test_pub_sig (self):
+        e0 = coro.ssl.openssl.ecdsa ("secp256k1")
+        e0.generate()
+        k0 = e0.get_privkey()
+        p0 = e0.get_pubkey()
+        h = hashlib.new ('sha256')
+        h.update ('asdfasdfasdfasdfasdf')
+        d = h.digest()
+        sig = e0.sign (d)
+        self.assertEqual (e0.verify (d, sig), 1)
+        e1 = coro.ssl.openssl.ecdsa ("secp256k1")
+        e1.set_pubkey (p0)
+        self.assertEqual (e1.verify (d, sig), 1)
+
+if __name__ == '__main__':
+    coro_unittest.run_tests()