rsa bindings

awscrt/crypto.py

@@ -2,6 +2,8 @@
 # SPDX-License-Identifier: Apache-2.0.
 
 import _awscrt
+from typing import Union
+from enum import IntEnum
 
 
 class Hash:
@@ -59,3 +61,91 @@ def update(self, to_hmac):
 
     def digest(self, truncate_to=0):
         return _awscrt.hmac_digest(self._hmac, truncate_to)
+
+
+class RSAEncryptionAlgorithmType(IntEnum):
+    """RSA Encryption Algorithm"""
+
+    PKCS1_5 = 0
+    """
+    PKCSv1.5 padding
+    """
+
+    OAEP_SHA256 = 1
+    """
+    OAEP padding with sha256 hash function
+    """
+
+    OAEP_SHA512 = 2
+    """
+    OAEP padding with sha512 hash function
+    """
+
+
+class RSASignatureAlgorithmType(IntEnum):
+    """RSA Encryption Algorithm"""
+
+    PKCS1_5_SHA256 = 0
+    """
+    PKCSv1.5 padding with sha256 hash function
+    """
+
+    PSS_SHA256 = 1
+    """
+    PSS padding with sha256 hash function
+    """
+
+
+class RSA:
+    def __init__(self, native_handle):
+        """
+        don't call me, I'm private
+        """
+        self._rsa = native_handle
+
+    @staticmethod
+    def rsa_private_key_from_pem_data(pem_data: Union[str, bytes, bytearray, memoryview]):
+        """
+        Creates a new instance of private RSA key pair from pem data.
+        Raises ValueError if pem does not have private key object.
+        """
+        return RSA(native_handle=_awscrt.rsa_private_key_from_pem_data(pem_data))
+
+    @staticmethod
+    def rsa_public_key_from_pem_data(pem_data: Union[str, bytes, bytearray, memoryview]):
+        """
+        Creates a new instance of public RSA key pair from pem data.
+        Raises ValueError if pem does not have public key object.
+        """
+        return RSA(native_handle=_awscrt.rsa_public_key_from_pem_data(pem_data))
+
+    def encrypt(self, encryption_algorithm: RSAEncryptionAlgorithmType,
+                plaintext: Union[str, bytes, bytearray, memoryview]) -> bytes:
+        """
+        Encrypts data using a given algorithm.
+        """
+        return _awscrt.rsa_encrypt(self._rsa, encryption_algorithm, plaintext)
+
+    def decrypt(self, encryption_algorithm: RSAEncryptionAlgorithmType,
+                ciphertext: Union[str, bytes, bytearray, memoryview]) -> bytes:
+        """
+        Decrypts data using a given algorithm.
+        """
+        return _awscrt.rsa_decrypt(self._rsa, encryption_algorithm, ciphertext)
+
+    def sign(self, encryption_algorithm: RSASignatureAlgorithmType,
+             digest: Union[str, bytes, bytearray, memoryview]) -> bytes:
+        """
+        Signs data using a given algorithm.
+        Note: function expects digest of the message, ex sha256
+        """
+        return _awscrt.rsa_sign(self._rsa, encryption_algorithm, digest)
+
+    def verify(self, encryption_algorithm: RSASignatureAlgorithmType,
+               digest: Union[str, bytes, bytearray, memoryview],
+               signature: Union[str, bytes, bytearray, memoryview]) -> bool:
+        """
+        Verifies signature against digest.
+        Returns True if signature matches and False if not.
+        """
+        return _awscrt.rsa_verify(self._rsa, encryption_algorithm, digest, signature)

requirements-dev.txt

@@ -3,3 +3,4 @@ setuptools # for installing from source
 sphinx # for building docs
 websockets # for tests
 wheel # for building wheels
+pytest-subtests # for tests

source/crypto.c

@@ -7,9 +7,12 @@
 
 #include "aws/cal/hash.h"
 #include "aws/cal/hmac.h"
+#include "aws/cal/rsa.h"
+#include "aws/io/pem.h"
 
 const char *s_capsule_name_hash = "aws_hash";
 const char *s_capsule_name_hmac = "aws_hmac";
+const char *s_capsule_name_rsa = "aws_rsa";
 
 static void s_hash_destructor(PyObject *hash_capsule) {
     assert(PyCapsule_CheckExact(hash_capsule));
@@ -238,3 +241,236 @@ PyObject *aws_py_hmac_digest(PyObject *self, PyObject *args) {
 
     return PyBytes_FromStringAndSize((const char *)output, digest_buf.len);
 }
+
+static void s_rsa_destructor(PyObject *rsa_capsule) {
+    assert(PyCapsule_CheckExact(rsa_capsule));
+
+    struct aws_rsa_key_pair *key_pair = PyCapsule_GetPointer(rsa_capsule, s_capsule_name_rsa);
+    assert(key_pair);
+
+    aws_rsa_key_pair_release(key_pair);
+}
+
+struct aws_pem_object *s_find_pem_object(struct aws_array_list *pem_list, enum aws_pem_object_type pem_type) {
+    for (size_t i = 0; i < aws_array_list_length(pem_list); ++i) {
+        struct aws_pem_object *pem_object = NULL;
+        if (aws_array_list_get_at_ptr(pem_list, (void **)&pem_object, 0)) {
+            return NULL;
+        }
+
+        if (pem_object->type == pem_type) {
+            return pem_object;
+        }
+    }
+
+    return NULL;
+}
+
+PyObject *aws_py_rsa_private_key_from_pem_data(PyObject *self, PyObject *args) {
+    (void)self;
+
+    const char *pem_data_ptr;
+    Py_ssize_t pem_data_len;
+    if (!PyArg_ParseTuple(args, "s#", &pem_data_ptr, &pem_data_len)) {
+        return NULL;
+    }
+
+    PyObject *capsule = NULL;
+    struct aws_byte_cursor pem_data_cur = aws_byte_cursor_from_array(pem_data_ptr, pem_data_len);
+    struct aws_allocator *allocator = aws_py_get_allocator();
+    struct aws_array_list pem_list;
+    if (aws_pem_objects_init_from_file_contents(&pem_list, allocator, pem_data_cur)) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_pem_object *found_pem_object = s_find_pem_object(&pem_list, AWS_PEM_TYPE_PRIVATE_RSA_PKCS1);
+
+    if (found_pem_object == NULL) {
+        PyErr_SetString(PyExc_ValueError, "RSA private key not found in PEM.");
+        goto on_done;
+    }
+
+    struct aws_rsa_key_pair *key_pair =
+        aws_rsa_key_pair_new_from_private_key_pkcs1(allocator, aws_byte_cursor_from_buf(&found_pem_object->data));
+
+    if (key_pair == NULL) {
+        PyErr_AwsLastError();
+        goto on_done;
+    }
+
+    capsule = PyCapsule_New(key_pair, s_capsule_name_rsa, s_rsa_destructor);
+
+    if (capsule == NULL) {
+        aws_rsa_key_pair_release(key_pair);
+    }
+
+on_done:
+    aws_pem_objects_clean_up(&pem_list);
+    return capsule;
+}
+
+PyObject *aws_py_rsa_public_key_from_pem_data(PyObject *self, PyObject *args) {
+    (void)self;
+
+    const char *pem_data_ptr;
+    Py_ssize_t pem_data_len;
+    if (!PyArg_ParseTuple(args, "s#", &pem_data_ptr, &pem_data_len)) {
+        return NULL;
+    }
+
+    PyObject *capsule = NULL;
+    struct aws_byte_cursor pem_data_cur = aws_byte_cursor_from_array(pem_data_ptr, pem_data_len);
+    struct aws_allocator *allocator = aws_py_get_allocator();
+    struct aws_array_list pem_list;
+    if (aws_pem_objects_init_from_file_contents(&pem_list, allocator, pem_data_cur)) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_pem_object *found_pem_object = s_find_pem_object(&pem_list, AWS_PEM_TYPE_PUBLIC_RSA_PKCS1);
+
+    if (found_pem_object == NULL) {
+        PyErr_SetString(PyExc_ValueError, "RSA public key not found in PEM.");
+        goto on_done;
+    }
+
+    struct aws_rsa_key_pair *key_pair =
+        aws_rsa_key_pair_new_from_public_key_pkcs1(allocator, aws_byte_cursor_from_buf(&found_pem_object->data));
+
+    if (key_pair == NULL) {
+        PyErr_AwsLastError();
+        goto on_done;
+    }
+
+    capsule = PyCapsule_New(key_pair, s_capsule_name_rsa, s_rsa_destructor);
+
+    if (capsule == NULL) {
+        aws_rsa_key_pair_release(key_pair);
+    }
+
+on_done:
+    aws_pem_objects_clean_up(&pem_list);
+    return capsule;
+}
+
+PyObject *aws_py_rsa_encrypt(PyObject *self, PyObject *args) {
+    (void)self;
+
+    struct aws_allocator *allocator = aws_py_get_allocator();
+    PyObject *rsa_capsule = NULL;
+    int encrypt_algo = 0;
+    const char *plaintext_ptr;
+    Py_ssize_t plaintext_len;
+    if (!PyArg_ParseTuple(args, "Ois#", &rsa_capsule, &encrypt_algo, &plaintext_ptr, &plaintext_len)) {
+        return NULL;
+    }
+
+    struct aws_rsa_key_pair *rsa = PyCapsule_GetPointer(rsa_capsule, s_capsule_name_rsa);
+    if (rsa == NULL) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_byte_cursor plaintext_cur = aws_byte_cursor_from_array(plaintext_ptr, plaintext_len);
+    struct aws_byte_buf result_buf;
+    aws_byte_buf_init(&result_buf, allocator, aws_rsa_key_pair_block_length(rsa));
+
+    if (aws_rsa_key_pair_encrypt(rsa, encrypt_algo, plaintext_cur, &result_buf)) {
+        return PyErr_AwsLastError();
+    }
+
+    PyObject *ret = PyBytes_FromStringAndSize((const char *)result_buf.buffer, result_buf.len);
+    aws_byte_buf_clean_up_secure(&result_buf);
+    return ret;
+}
+
+PyObject *aws_py_rsa_decrypt(PyObject *self, PyObject *args) {
+    (void)self;
+
+    struct aws_allocator *allocator = aws_py_get_allocator();
+    PyObject *rsa_capsule = NULL;
+    int encrypt_algo = 0;
+    const char *ciphertext_ptr;
+    Py_ssize_t ciphertext_len;
+    if (!PyArg_ParseTuple(args, "Ois#", &rsa_capsule, &encrypt_algo, &ciphertext_ptr, &ciphertext_len)) {
+        return NULL;
+    }
+
+    struct aws_rsa_key_pair *rsa = PyCapsule_GetPointer(rsa_capsule, s_capsule_name_rsa);
+    if (rsa == NULL) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_byte_cursor ciphertext_cur = aws_byte_cursor_from_array(ciphertext_ptr, ciphertext_len);
+    struct aws_byte_buf result_buf;
+    aws_byte_buf_init(&result_buf, allocator, aws_rsa_key_pair_block_length(rsa));
+
+    if (aws_rsa_key_pair_decrypt(rsa, encrypt_algo, ciphertext_cur, &result_buf)) {
+        return PyErr_AwsLastError();
+    }
+
+    PyObject *ret = PyBytes_FromStringAndSize((const char *)result_buf.buffer, result_buf.len);
+    aws_byte_buf_clean_up_secure(&result_buf);
+    return ret;
+}
+
+PyObject *aws_py_rsa_sign(PyObject *self, PyObject *args) {
+    (void)self;
+
+    struct aws_allocator *allocator = aws_py_get_allocator();
+    PyObject *rsa_capsule = NULL;
+    int sign_algo = 0;
+    const char *digest_ptr;
+    Py_ssize_t digest_len;
+    if (!PyArg_ParseTuple(args, "Ois#", &rsa_capsule, &sign_algo, &digest_ptr, &digest_len)) {
+        return NULL;
+    }
+
+    struct aws_rsa_key_pair *rsa = PyCapsule_GetPointer(rsa_capsule, s_capsule_name_rsa);
+    if (rsa == NULL) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_byte_cursor digest_cur = aws_byte_cursor_from_array(digest_ptr, digest_len);
+    struct aws_byte_buf result_buf;
+    aws_byte_buf_init(&result_buf, allocator, aws_rsa_key_pair_signature_length(rsa));
+
+    if (aws_rsa_key_pair_sign_message(rsa, sign_algo, digest_cur, &result_buf)) {
+        return PyErr_AwsLastError();
+    }
+
+    PyObject *ret = PyBytes_FromStringAndSize((const char *)result_buf.buffer, result_buf.len);
+    aws_byte_buf_clean_up_secure(&result_buf);
+    return ret;
+}
+
+PyObject *aws_py_rsa_verify(PyObject *self, PyObject *args) {
+    (void)self;
+
+    PyObject *rsa_capsule = NULL;
+    int sign_algo = 0;
+    const char *digest_ptr;
+    Py_ssize_t digest_len;
+    const char *signature_ptr;
+    Py_ssize_t signature_len;
+    if (!PyArg_ParseTuple(
+            args, "Ois#s#", &rsa_capsule, &sign_algo, &digest_ptr, &digest_len, &signature_ptr, &signature_len)) {
+        return NULL;
+    }
+
+    struct aws_rsa_key_pair *rsa = PyCapsule_GetPointer(rsa_capsule, s_capsule_name_rsa);
+    if (rsa == NULL) {
+        return PyErr_AwsLastError();
+    }
+
+    struct aws_byte_cursor digest_cur = aws_byte_cursor_from_array(digest_ptr, digest_len);
+    struct aws_byte_cursor signature_cur = aws_byte_cursor_from_array(signature_ptr, signature_len);
+
+    if (aws_rsa_key_pair_verify_signature(rsa, sign_algo, digest_cur, signature_cur)) {
+        if (aws_last_error() == AWS_ERROR_CAL_SIGNATURE_VALIDATION_FAILED) {
+            aws_reset_error();
+            Py_RETURN_FALSE;
+        }
+        return PyErr_AwsLastError();
+    }
+
+    Py_RETURN_TRUE;
+}

source/crypto.h

@@ -10,6 +10,8 @@
 extern const char *s_capsule_name_hash;
 /** Name string for hmac capsule. */
 extern const char *s_capsule_name_hmac;
+/** Name string for rsa capsule. */
+extern const char *s_capsule_name_rsa;
 
 PyObject *aws_py_sha1_new(PyObject *self, PyObject *args);
 PyObject *aws_py_sha256_new(PyObject *self, PyObject *args);
@@ -27,4 +29,12 @@ PyObject *aws_py_sha256_compute(PyObject *self, PyObject *args);
 PyObject *aws_py_md5_compute(PyObject *self, PyObject *args);
 PyObject *aws_py_sha256_hmac_compute(PyObject *self, PyObject *args);
 
+PyObject *aws_py_rsa_private_key_from_pem_data(PyObject *self, PyObject *args);
+PyObject *aws_py_rsa_public_key_from_pem_data(PyObject *self, PyObject *args);
+
+PyObject *aws_py_rsa_encrypt(PyObject *self, PyObject *args);
+PyObject *aws_py_rsa_decrypt(PyObject *self, PyObject *args);
+PyObject *aws_py_rsa_sign(PyObject *self, PyObject *args);
+PyObject *aws_py_rsa_verify(PyObject *self, PyObject *args);
+
 #endif /* AWS_CRT_PYTHON_CRYPTO_H */

source/io.h

@@ -29,7 +29,7 @@ PyObject *aws_py_init_logging(PyObject *self, PyObject *args);
 PyObject *aws_py_is_alpn_available(PyObject *self, PyObject *args);
 
 /**
- * Returns True if the input TLS Cipher Preference Enum is suupported on the current platform. False otherwise.
+ * Returns True if the input TLS Cipher Preference Enum is supported on the current platform. False otherwise.
  */
 PyObject *aws_py_is_tls_cipher_supported(PyObject *self, PyObject *args);