WolfSSL
/
wolfcrypt-py
mirror of https://github.com/wolfSSL/wolfcrypt-py.git
1
0
Fork 0
Code Issues Releases Wiki Activity

Add support for AES GCM streaming

pull/38/head
Andrew Hutchings 2022-01-28 15:09:03 +00:00 committed by Daniele Lacamera
parent cbbe08b256
commit 16334bf5e6
7 changed files with 212 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
.gitignore vendored
View File

@ -6,6 +6,7 @@ __pycache__/
# C extensions
*.so
*.a
wolfcrypt/_ffi.*
# Distribution / packaging
.Python

1
docs/index.rst
View File

@ -11,5 +11,6 @@ Summary
digest
mac
random
streaming
.. include:: ../LICENSING.rst

50
docs/streaming.rst 100644
View File

@ -0,0 +1,50 @@
Streaming Encryption Algorithms
===============================
.. module:: wolfcrypt.ciphers
Steaming Encryption Classes
---------------------------
Interface
~~~~~~~~~
AesGcmStreamEncrypt
~~~~~~~~~~~~~~~~~~~
.. autoclass:: AesGcmStreamEncrypt
:members:
:inherited-members:
**Example:**
.. doctest::
>>> from wolfcrypt.ciphers import AesGcmStreamEncrypt
>>> from binascii import hexlify as b2h
>>> gcm = AesGcmStreamEncrypt(b'fedcba9876543210', b'0123456789abcdef')
>>> buf = gcm.update("hello world")
>>> authTag = gcm.final()
>>> b2h(buf)
b'5ba7d42e1bf01d7998e932'
>>> b2h(authTag)
b'cef91ba0c8c6431c7e19f64c9d9e371b'
AesGcmStreamDecrypt
~~~~~~~~~~~~~~~~~~~
.. autoclass:: AesGcmStreamDecrypt
:members:
:inherited-members:
**Example:**
.. doctest::
>>> from wolfcrypt.ciphers import AesGcmStreamDecrypt, t2b
>>> from binascii import unhexlify as h2b
>>> gcm = AesGcmStreamDecrypt(b'fedcba9876543210', b'0123456789abcdef')
>>> buf = gcm.update(h2b(b'5ba7d42e1bf01d7998e932'))
>>> gcm.final(h2b(b'cef91ba0c8c6431c7e19f64c9d9e371b'))
>>> t2b(buf)
b'hello world'

58
tests/test_aesgcmstream.py 100644
View File

@ -0,0 +1,58 @@
# test_hashes.py
#
# Copyright (C) 2006-2022 wolfSSL Inc.
#
# This file is part of wolfSSL. (formerly known as CyaSSL)
#
# wolfSSL is free software; you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation; either version 2 of the License, or
# (at your option) any later version.
#
# wolfSSL is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA
# pylint: disable=redefined-outer-name
from collections import namedtuple
import pytest
from wolfcrypt._ffi import ffi as _ffi
from wolfcrypt._ffi import lib as _lib
from wolfcrypt.utils import t2b
from binascii import hexlify as b2h, unhexlify as h2b
from wolfcrypt.ciphers import AesGcmStreamEncrypt, AesGcmStreamDecrypt
def test_encrypt():
key = "fedcba9876543210"
iv = "0123456789abcdef"
gcm = AesGcmStreamEncrypt(key, iv)
buf = gcm.update("hello world")
authTag = gcm.final()
assert b2h(authTag) == bytes('cef91ba0c8c6431c7e19f64c9d9e371b', 'utf-8')
assert b2h(buf) == bytes('5ba7d42e1bf01d7998e932', "utf-8")
gcmdec = AesGcmStreamDecrypt(key, iv)
bufdec = gcmdec.update(buf)
gcmdec.final(authTag)
assert bufdec == t2b("hello world")
def test_multipart():
key = "fedcba9876543210"
iv = "0123456789abcdef"
gcm = AesGcmStreamEncrypt(key, iv)
buf = gcm.update("hello")
buf += gcm.update(" world")
authTag = gcm.final()
assert b2h(authTag) == bytes('6862647a27c7b6aa0a6882b3e117e944', 'utf-8')
assert b2h(buf) == bytes('5ba7d42e1bf01d7998e932', "utf-8")
gcmdec = AesGcmStreamDecrypt(key, iv)
bufdec = gcmdec.update(buf[:5])
bufdec += gcmdec.update(buf[5:])
gcmdec.final(authTag)
assert bufdec == t2b("hello world")

25
wolfcrypt/_build_ffi.py
View File

@ -102,6 +102,7 @@ FIPS_VERSION = 0
ERROR_STRINGS_ENABLED = 1
ASN_ENABLED = 1
WC_RNG_SEED_CB_ENABLED = 0
AESGCM_STREAM = 1
# detect native features based on options.h defines
if featureDetection:
@ -126,6 +127,7 @@ if featureDetection:
ERROR_STRINGS_ENABLED = 0 if '#define NO_ERROR_STRINGS' in optionsHeaderStr else 1
ASN_ENABLED = 0 if '#define NO_ASN' in optionsHeaderStr else 1
WC_RNG_SEED_CB_ENABLED = 1 if '#define WC_RNG_SEED_CB' in optionsHeaderStr else 0
AESGCM_STREAM = 1 if '#define WOLFSSL_AESGCM_STREAM' in optionsHeaderStr else 0
if '#define HAVE_FIPS' in optionsHeaderStr:
FIPS_ENABLED = 1
@ -202,6 +204,7 @@ extern "C" {
int FIPS_VERSION = """ + str(FIPS_VERSION) + """;
int ASN_ENABLED = """ + str(ASN_ENABLED) + """;
int WC_RNG_SEED_CB_ENABLED = """ + str(WC_RNG_SEED_CB_ENABLED) + """;
int AESGCM_STREAM = """ + str(AESGCM_STREAM) + """;
""",
include_dirs=[wolfssl_inc_path()],
library_dirs=[wolfssl_lib_path()],
@ -231,6 +234,7 @@ _cdef = """
extern int FIPS_VERSION;
extern int ASN_ENABLED;
extern int WC_RNG_SEED_CB_ENABLED;
extern int AESGCM_STREAM;
typedef unsigned char byte;
typedef unsigned int word32;
@ -322,6 +326,27 @@ if AES_ENABLED:
int wc_AesCbcDecrypt(Aes*, byte*, const byte*, word32);
"""
if AES_ENABLED and AESGCM_STREAM:
_cdef += """
int wc_AesInit(Aes* aes, void* heap, int devId);
int wc_AesGcmInit(Aes* aes, const byte* key, word32 len,
const byte* iv, word32 ivSz);
int wc_AesGcmEncryptInit(Aes* aes, const byte* key, word32 len,
const byte* iv, word32 ivSz);
int wc_AesGcmEncryptInit_ex(Aes* aes, const byte* key, word32 len,
byte* ivOut, word32 ivOutSz);
int wc_AesGcmEncryptUpdate(Aes* aes, byte* out, const byte* in,
word32 sz, const byte* authIn, word32 authInSz);
int wc_AesGcmEncryptFinal(Aes* aes, byte* authTag,
word32 authTagSz);
int wc_AesGcmDecryptInit(Aes* aes, const byte* key, word32 len,
const byte* iv, word32 ivSz);
int wc_AesGcmDecryptUpdate(Aes* aes, byte* out, const byte* in,
word32 sz, const byte* authIn, word32 authInSz);
int wc_AesGcmDecryptFinal(Aes* aes, const byte* authTag,
word32 authTagSz);
"""
if CHACHA_ENABLED:
_cdef += """
typedef struct { ...; } ChaCha;

14
wolfcrypt/_build_wolfssl.py
View File

@ -143,7 +143,7 @@ def make_flags(prefix):
flags.append("-DWOLFSSL_AES=yes")
flags.append("-DWOLFSSL_DES3=yes")
flags.append("-DWOLFSSL_CHACHA=yes")
flags.append("-DWOLFSSL_AESGCM=no")
flags.append("-DWOLFSSL_AESGCM=yes")
flags.append("-DWOLFSSL_SHA=yes")
flags.append("-DWOLFSSL_SHA384=yes")
flags.append("-DWOLFSSL_SHA512=yes")
@ -165,7 +165,7 @@ def make_flags(prefix):
flags.append("-DWOLFSSL_EXTENDED_MASTER=no")
flags.append("-DWOLFSSL_ERROR_STRINGS=no")
# Part of hack for missing CMake option
flags.append("-DCMAKE_C_FLAGS=\"/DWOLFSSL_KEY_GEN=1 /DWOLFCRYPT_ONLY=1\"")
flags.append("-DCMAKE_C_FLAGS=\"/DWOLFSSL_KEY_GEN=1 /DWOLFCRYPT_ONLY=1 /DWOLFSSL_AESGCM_STREAM=1\"")
return " ".join(flags)
else:
@ -187,7 +187,9 @@ def make_flags(prefix):
flags.append("--enable-des3")
flags.append("--enable-chacha")
flags.append("--disable-aesgcm")
flags.append("--enable-aesgcm-stream")
flags.append("--enable-aesgcm")
# hashes and MACs
flags.append("--enable-sha")
@ -233,6 +235,8 @@ def cmake_hack():
contents.insert(27, "#define WOLFSSL_KEY_GEN\n")
contents.insert(28, "#undef WOLFCRYPT_ONLY\n")
contents.insert(29, "#define WOLFCRYPT_ONLY\n")
contents.insert(30, "#undef WOLFSSL_AESGCM_STREAM\n")
contents.insert(31, "#define WOLFSSL_AESGCM_STREAM\n")
with open(options_file, "w") as f:
contents = "".join(contents)
@ -273,9 +277,9 @@ def build_wolfssl(version="master"):
else:
libfile = os.path.join(prefix, 'lib/libwolfssl.la')
rebuild = ensure_wolfssl_src(version)
ensure_wolfssl_src(version)
if rebuild or not os.path.isfile(libfile):
if not os.path.isfile(libfile):
make(make_flags(prefix))
if __name__ == "__main__":

68
wolfcrypt/ciphers.py
View File

@ -230,6 +230,74 @@ if _lib.AES_ENABLED:
return _lib.wc_AesCbcDecrypt(self._dec, destination,
source, len(source))
if _lib.AESGCM_STREAM:
class _AesGcmStream(object):
"""
AES GCM Stream
"""
block_size = 16
_key_sizes = [16, 24, 32]
_native_type = "Aes *"
def __init__(self, key, IV):
key = t2b(key)
IV = t2b(IV)
if len(key) not in self._key_sizes:
raise ValueError("key must be %s in length, not %d" %
(self._key_sizes, len(key)))
self._native_object = _ffi.new(self._native_type)
_lib.wc_AesInit(self._native_object, _ffi.NULL, -2)
self._authIn = _ffi.new("byte[%d]" % self.block_size)
ret = _lib.wc_AesGcmInit(self._native_object, key, len(key), IV, len(IV))
if ret < 0:
raise WolfCryptError("Init error (%d)" % ret)
def update(self, data):
"""
Updates the stream with another segment of data.
"""
ret = 0
data = t2b(data)
self._buf = _ffi.new("byte[%d]" % (len(data)))
ret = self._update(data)
if ret < 0:
raise WolfCryptError("Decryption error (%d)" % ret)
return bytes(self._buf)
class AesGcmStreamEncrypt(_AesGcmStream):
"""
AES GCM Streaming Encryption
"""
def _update(self, data):
return _lib.wc_AesGcmEncryptUpdate(self._native_object, self._buf, data, len(data), self._authIn, self.block_size)
def final(self):
"""
Finalize the stream and return an authentication tag for the stream.
"""
authTag = _ffi.new("byte[%d]" % self.block_size)
ret = _lib.wc_AesGcmEncryptFinal(self._native_object, authTag, self.block_size)
if ret < 0:
raise WolfCryptError("Encryption error (%d)" % ret)
return _ffi.buffer(authTag)[:]
class AesGcmStreamDecrypt(_AesGcmStream):
"""
AES GCM Streaming Decryption
"""
def _update(self, data):
return _lib.wc_AesGcmDecryptUpdate(self._native_object, self._buf, data, len(data), self._authIn, self.block_size)
def final(self, authTag):
"""
Finalize the stream and verify using the provided authentication tag.
"""
authTag = t2b(authTag)
ret = _lib.wc_AesGcmDecryptFinal(self._native_object, authTag, self.block_size)
if ret < 0:
raise WolfCryptError("Decryption error (%d)" % ret)
if _lib.CHACHA_ENABLED:
class ChaCha(_Cipher):
"""