a60d9a9892
Make `KAT(CCM)?Parser` into a context suite-capable object by implementing `__enter__` and `__exit__` methods which manage opening up the file descriptors and closing them on context exit. This implementation was decided over adding destructor logic to a `__del__` method, as there are a number of issues around object lifetimes when dealing with threading cleanup, atexit handlers, and a number of other less obvious edgecases. Plus, the architected solution is more pythonic and clean. Complete the iterator implementation by implementing a `__next__` method for both classes which handles iterating over the data using a generator pattern, and by changing `__iter__` to return the object instead of the data which it would iterate over. Alias the `__next__` method to `next` when working with python 2.x in order to maintain functional compatibility between the two major versions. As part of this work and to ensure readability, push the initialization of the parser objects up one layer and pass it down to a helper function. This could have been done via a decorator, but I was trying to keep it simple for other developers to make it easier to modify in the future. This fixes ResourceWarnings with python 3. PR: 237403 MFC after: 1 week Tested with: python 2.7.16 (amd64), python 3.6.8 (amd64)
507 lines
20 KiB
Python
507 lines
20 KiB
Python
#!/usr/local/bin/python2
|
|
#
|
|
# Copyright (c) 2014 The FreeBSD Foundation
|
|
# All rights reserved.
|
|
#
|
|
# This software was developed by John-Mark Gurney under
|
|
# the sponsorship from the FreeBSD Foundation.
|
|
# Redistribution and use in source and binary forms, with or without
|
|
# modification, are permitted provided that the following conditions
|
|
# are met:
|
|
# 1. Redistributions of source code must retain the above copyright
|
|
# notice, this list of conditions and the following disclaimer.
|
|
# 2. Redistributions in binary form must reproduce the above copyright
|
|
# notice, this list of conditions and the following disclaimer in the
|
|
# documentation and/or other materials provided with the distribution.
|
|
#
|
|
# THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
|
|
# ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
|
|
# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
|
|
# ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
|
|
# FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
|
|
# DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
|
|
# OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
|
|
# HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
|
|
# LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
|
|
# OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
|
|
# SUCH DAMAGE.
|
|
#
|
|
# $FreeBSD$
|
|
#
|
|
|
|
from __future__ import print_function
|
|
|
|
import binascii
|
|
import errno
|
|
import cryptodev
|
|
import itertools
|
|
import os
|
|
import struct
|
|
import unittest
|
|
from cryptodev import *
|
|
from glob import iglob
|
|
|
|
katdir = '/usr/local/share/nist-kat'
|
|
|
|
def katg(base, glob):
|
|
assert os.path.exists(katdir), "Please 'pkg install nist-kat'"
|
|
if not os.path.exists(os.path.join(katdir, base)):
|
|
raise unittest.SkipTest("Missing %s test vectors" % (base))
|
|
return iglob(os.path.join(katdir, base, glob))
|
|
|
|
aesmodules = [ 'cryptosoft0', 'aesni0', 'ccr0', 'ccp0' ]
|
|
desmodules = [ 'cryptosoft0', ]
|
|
shamodules = [ 'cryptosoft0', 'aesni0', 'ccr0', 'ccp0' ]
|
|
|
|
def GenTestCase(cname):
|
|
try:
|
|
crid = cryptodev.Crypto.findcrid(cname)
|
|
except IOError:
|
|
return None
|
|
|
|
class GendCryptoTestCase(unittest.TestCase):
|
|
###############
|
|
##### AES #####
|
|
###############
|
|
@unittest.skipIf(cname not in aesmodules, 'skipping AES-XTS on %s' % (cname))
|
|
def test_xts(self):
|
|
for i in katg('XTSTestVectors/format tweak value input - data unit seq no', '*.rsp'):
|
|
self.runXTS(i, cryptodev.CRYPTO_AES_XTS)
|
|
|
|
@unittest.skipIf(cname not in aesmodules, 'skipping AES-CBC on %s' % (cname))
|
|
def test_cbc(self):
|
|
for i in katg('KAT_AES', 'CBC[GKV]*.rsp'):
|
|
self.runCBC(i)
|
|
|
|
@unittest.skipIf(cname not in aesmodules, 'skipping AES-CCM on %s' % (cname))
|
|
def test_ccm(self):
|
|
for i in katg('ccmtestvectors', 'V*.rsp'):
|
|
self.runCCMEncrypt(i)
|
|
|
|
for i in katg('ccmtestvectors', 'D*.rsp'):
|
|
self.runCCMDecrypt(i)
|
|
|
|
@unittest.skipIf(cname not in aesmodules, 'skipping AES-GCM on %s' % (cname))
|
|
def test_gcm(self):
|
|
for i in katg('gcmtestvectors', 'gcmEncrypt*'):
|
|
self.runGCM(i, 'ENCRYPT')
|
|
|
|
for i in katg('gcmtestvectors', 'gcmDecrypt*'):
|
|
self.runGCM(i, 'DECRYPT')
|
|
|
|
_gmacsizes = { 32: cryptodev.CRYPTO_AES_256_NIST_GMAC,
|
|
24: cryptodev.CRYPTO_AES_192_NIST_GMAC,
|
|
16: cryptodev.CRYPTO_AES_128_NIST_GMAC,
|
|
}
|
|
def runGCM(self, fname, mode):
|
|
curfun = None
|
|
if mode == 'ENCRYPT':
|
|
swapptct = False
|
|
curfun = Crypto.encrypt
|
|
elif mode == 'DECRYPT':
|
|
swapptct = True
|
|
curfun = Crypto.decrypt
|
|
else:
|
|
raise RuntimeError('unknown mode: %r' % repr(mode))
|
|
|
|
columns = [ 'Count', 'Key', 'IV', 'CT', 'AAD', 'Tag', 'PT', ]
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runGCMWithParser(parser, mode)
|
|
|
|
def runGCMWithParser(self, parser, mode):
|
|
for _, lines in next(parser):
|
|
for data in lines:
|
|
curcnt = int(data['Count'])
|
|
cipherkey = binascii.unhexlify(data['Key'])
|
|
iv = binascii.unhexlify(data['IV'])
|
|
aad = binascii.unhexlify(data['AAD'])
|
|
tag = binascii.unhexlify(data['Tag'])
|
|
if 'FAIL' not in data:
|
|
pt = binascii.unhexlify(data['PT'])
|
|
ct = binascii.unhexlify(data['CT'])
|
|
|
|
if len(iv) != 12:
|
|
# XXX - isn't supported
|
|
continue
|
|
|
|
try:
|
|
c = Crypto(cryptodev.CRYPTO_AES_NIST_GCM_16,
|
|
cipherkey,
|
|
mac=self._gmacsizes[len(cipherkey)],
|
|
mackey=cipherkey, crid=crid,
|
|
maclen=16)
|
|
except EnvironmentError as e:
|
|
# Can't test algorithms the driver does not support.
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
|
|
if mode == 'ENCRYPT':
|
|
try:
|
|
rct, rtag = c.encrypt(pt, iv, aad)
|
|
except EnvironmentError as e:
|
|
# Can't test inputs the driver does not support.
|
|
if e.errno != errno.EINVAL:
|
|
raise
|
|
continue
|
|
rtag = rtag[:len(tag)]
|
|
data['rct'] = binascii.hexlify(rct)
|
|
data['rtag'] = binascii.hexlify(rtag)
|
|
self.assertEqual(rct, ct, repr(data))
|
|
self.assertEqual(rtag, tag, repr(data))
|
|
else:
|
|
if len(tag) != 16:
|
|
continue
|
|
args = (ct, iv, aad, tag)
|
|
if 'FAIL' in data:
|
|
self.assertRaises(IOError,
|
|
c.decrypt, *args)
|
|
else:
|
|
try:
|
|
rpt, rtag = c.decrypt(*args)
|
|
except EnvironmentError as e:
|
|
# Can't test inputs the driver does not support.
|
|
if e.errno != errno.EINVAL:
|
|
raise
|
|
continue
|
|
data['rpt'] = binascii.hexlify(rpt)
|
|
data['rtag'] = binascii.hexlify(rtag)
|
|
self.assertEqual(rpt, pt,
|
|
repr(data))
|
|
|
|
def runCBC(self, fname):
|
|
columns = [ 'COUNT', 'KEY', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runCBCWithParser(parser)
|
|
|
|
def runCBCWithParser(self, parser):
|
|
curfun = None
|
|
for mode, lines in next(parser):
|
|
if mode == 'ENCRYPT':
|
|
swapptct = False
|
|
curfun = Crypto.encrypt
|
|
elif mode == 'DECRYPT':
|
|
swapptct = True
|
|
curfun = Crypto.decrypt
|
|
else:
|
|
raise RuntimeError('unknown mode: %r' % repr(mode))
|
|
|
|
for data in lines:
|
|
curcnt = int(data['COUNT'])
|
|
cipherkey = binascii.unhexlify(data['KEY'])
|
|
iv = binascii.unhexlify(data['IV'])
|
|
pt = binascii.unhexlify(data['PLAINTEXT'])
|
|
ct = binascii.unhexlify(data['CIPHERTEXT'])
|
|
|
|
if swapptct:
|
|
pt, ct = ct, pt
|
|
# run the fun
|
|
c = Crypto(cryptodev.CRYPTO_AES_CBC, cipherkey, crid=crid)
|
|
r = curfun(c, pt, iv)
|
|
self.assertEqual(r, ct)
|
|
|
|
def runXTS(self, fname, meth):
|
|
columns = [ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT',
|
|
'CT']
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runXTSWithParser(parser, meth)
|
|
|
|
def runXTSWithParser(self, parser, meth):
|
|
curfun = None
|
|
for mode, lines in next(parser):
|
|
if mode == 'ENCRYPT':
|
|
swapptct = False
|
|
curfun = Crypto.encrypt
|
|
elif mode == 'DECRYPT':
|
|
swapptct = True
|
|
curfun = Crypto.decrypt
|
|
else:
|
|
raise RuntimeError('unknown mode: %r' % repr(mode))
|
|
|
|
for data in lines:
|
|
curcnt = int(data['COUNT'])
|
|
nbits = int(data['DataUnitLen'])
|
|
cipherkey = binascii.unhexlify(data['Key'])
|
|
iv = struct.pack('QQ', int(data['DataUnitSeqNumber']), 0)
|
|
pt = binascii.unhexlify(data['PT'])
|
|
ct = binascii.unhexlify(data['CT'])
|
|
|
|
if nbits % 128 != 0:
|
|
# XXX - mark as skipped
|
|
continue
|
|
if swapptct:
|
|
pt, ct = ct, pt
|
|
# run the fun
|
|
try:
|
|
c = Crypto(meth, cipherkey, crid=crid)
|
|
r = curfun(c, pt, iv)
|
|
except EnvironmentError as e:
|
|
# Can't test hashes the driver does not support.
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
self.assertEqual(r, ct)
|
|
|
|
def runCCMEncrypt(self, fname):
|
|
with cryptodev.KATCCMParser(fname) as parser:
|
|
self.runCCMEncryptWithParser(parser)
|
|
|
|
def runCCMEncryptWithParser(self, parser):
|
|
for data in next(parser):
|
|
Nlen = int(data['Nlen'])
|
|
if Nlen != 12:
|
|
# OCF only supports 12 byte IVs
|
|
continue
|
|
key = binascii.unhexlify(data['Key'])
|
|
nonce = binascii.unhexlify(data['Nonce'])
|
|
Alen = int(data['Alen'])
|
|
if Alen != 0:
|
|
aad = binascii.unhexlify(data['Adata'])
|
|
else:
|
|
aad = None
|
|
payload = binascii.unhexlify(data['Payload'])
|
|
ct = binascii.unhexlify(data['CT'])
|
|
|
|
try:
|
|
c = Crypto(crid=crid,
|
|
cipher=cryptodev.CRYPTO_AES_CCM_16,
|
|
key=key,
|
|
mac=cryptodev.CRYPTO_AES_CCM_CBC_MAC,
|
|
mackey=key, maclen=16)
|
|
r, tag = Crypto.encrypt(c, payload,
|
|
nonce, aad)
|
|
except EnvironmentError as e:
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
|
|
out = r + tag
|
|
self.assertEqual(out, ct,
|
|
"Count " + data['Count'] + " Actual: " + \
|
|
repr(binascii.hexlify(out)) + " Expected: " + \
|
|
repr(data) + " on " + cname)
|
|
|
|
def runCCMDecrypt(self, fname):
|
|
with cryptodev.KATCCMParser(fname) as parser:
|
|
self.runCCMDecryptWithParser(parser)
|
|
|
|
def runCCMDecryptWithParser(self, parser):
|
|
# XXX: Note that all of the current CCM
|
|
# decryption test vectors use IV and tag sizes
|
|
# that aren't supported by OCF none of the
|
|
# tests are actually ran.
|
|
for data in next(parser):
|
|
Nlen = int(data['Nlen'])
|
|
if Nlen != 12:
|
|
# OCF only supports 12 byte IVs
|
|
continue
|
|
Tlen = int(data['Tlen'])
|
|
if Tlen != 16:
|
|
# OCF only supports 16 byte tags
|
|
continue
|
|
key = binascii.unhexlify(data['Key'])
|
|
nonce = binascii.unhexlify(data['Nonce'])
|
|
Alen = int(data['Alen'])
|
|
if Alen != 0:
|
|
aad = binascii.unhexlify(data['Adata'])
|
|
else:
|
|
aad = None
|
|
ct = binascii.unhexlify(data['CT'])
|
|
tag = ct[-16:]
|
|
ct = ct[:-16]
|
|
|
|
try:
|
|
c = Crypto(crid=crid,
|
|
cipher=cryptodev.CRYPTO_AES_CCM_16,
|
|
key=key,
|
|
mac=cryptodev.CRYPTO_AES_CCM_CBC_MAC,
|
|
mackey=key, maclen=16)
|
|
except EnvironmentError as e:
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
|
|
if data['Result'] == 'Fail':
|
|
self.assertRaises(IOError,
|
|
c.decrypt, payload, nonce, aad, tag)
|
|
else:
|
|
r = Crypto.decrypt(c, payload, nonce,
|
|
aad, tag)
|
|
|
|
payload = binascii.unhexlify(data['Payload'])
|
|
plen = int(data('Plen'))
|
|
payload = payload[:plen]
|
|
self.assertEqual(r, payload,
|
|
"Count " + data['Count'] + \
|
|
" Actual: " + repr(binascii.hexlify(r)) + \
|
|
" Expected: " + repr(data) + \
|
|
" on " + cname)
|
|
|
|
###############
|
|
##### DES #####
|
|
###############
|
|
@unittest.skipIf(cname not in desmodules, 'skipping DES on %s' % (cname))
|
|
def test_tdes(self):
|
|
for i in katg('KAT_TDES', 'TCBC[a-z]*.rsp'):
|
|
self.runTDES(i)
|
|
|
|
def runTDES(self, fname):
|
|
columns = [ 'COUNT', 'KEYs', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runTDESWithParser(parser)
|
|
|
|
def runTDESWithParser(self, parser):
|
|
curfun = None
|
|
for mode, lines in next(parser):
|
|
if mode == 'ENCRYPT':
|
|
swapptct = False
|
|
curfun = Crypto.encrypt
|
|
elif mode == 'DECRYPT':
|
|
swapptct = True
|
|
curfun = Crypto.decrypt
|
|
else:
|
|
raise RuntimeError('unknown mode: %r' % repr(mode))
|
|
|
|
for data in lines:
|
|
curcnt = int(data['COUNT'])
|
|
key = data['KEYs'] * 3
|
|
cipherkey = binascii.unhexlify(key)
|
|
iv = binascii.unhexlify(data['IV'])
|
|
pt = binascii.unhexlify(data['PLAINTEXT'])
|
|
ct = binascii.unhexlify(data['CIPHERTEXT'])
|
|
|
|
if swapptct:
|
|
pt, ct = ct, pt
|
|
# run the fun
|
|
c = Crypto(cryptodev.CRYPTO_3DES_CBC, cipherkey, crid=crid)
|
|
r = curfun(c, pt, iv)
|
|
self.assertEqual(r, ct)
|
|
|
|
###############
|
|
##### SHA #####
|
|
###############
|
|
@unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
|
|
def test_sha(self):
|
|
for i in katg('shabytetestvectors', 'SHA*Msg.rsp'):
|
|
self.runSHA(i)
|
|
|
|
def runSHA(self, fname):
|
|
# Skip SHA512_(224|256) tests
|
|
if fname.find('SHA512_') != -1:
|
|
return
|
|
columns = [ 'Len', 'Msg', 'MD' ]
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runSHAWithParser(parser)
|
|
|
|
def runSHAWithParser(self, parser):
|
|
for hashlength, lines in next(parser):
|
|
# E.g., hashlength will be "L=20" (bytes)
|
|
hashlen = int(hashlength.split("=")[1])
|
|
|
|
if hashlen == 20:
|
|
alg = cryptodev.CRYPTO_SHA1
|
|
elif hashlen == 28:
|
|
alg = cryptodev.CRYPTO_SHA2_224
|
|
elif hashlen == 32:
|
|
alg = cryptodev.CRYPTO_SHA2_256
|
|
elif hashlen == 48:
|
|
alg = cryptodev.CRYPTO_SHA2_384
|
|
elif hashlen == 64:
|
|
alg = cryptodev.CRYPTO_SHA2_512
|
|
else:
|
|
# Skip unsupported hashes
|
|
# Slurp remaining input in section
|
|
for data in lines:
|
|
continue
|
|
continue
|
|
|
|
for data in lines:
|
|
msg = binascii.unhexlify(data['Msg'])
|
|
msg = msg[:int(data['Len'])]
|
|
md = binascii.unhexlify(data['MD'])
|
|
|
|
try:
|
|
c = Crypto(mac=alg, crid=crid,
|
|
maclen=hashlen)
|
|
except EnvironmentError as e:
|
|
# Can't test hashes the driver does not support.
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
|
|
_, r = c.encrypt(msg, iv="")
|
|
|
|
self.assertEqual(r, md, "Actual: " + \
|
|
repr(binascii.hexlify(r)) + " Expected: " + repr(data) + " on " + cname)
|
|
|
|
@unittest.skipIf(cname not in shamodules, 'skipping SHA-HMAC on %s' % str(cname))
|
|
def test_sha1hmac(self):
|
|
for i in katg('hmactestvectors', 'HMAC.rsp'):
|
|
self.runSHA1HMAC(i)
|
|
|
|
def runSHA1HMAC(self, fname):
|
|
columns = [ 'Count', 'Klen', 'Tlen', 'Key', 'Msg', 'Mac' ]
|
|
with cryptodev.KATParser(fname, columns) as parser:
|
|
self.runSHA1HMACWithParser(parser)
|
|
|
|
def runSHA1HMACWithParser(self, parser):
|
|
for hashlength, lines in next(parser):
|
|
# E.g., hashlength will be "L=20" (bytes)
|
|
hashlen = int(hashlength.split("=")[1])
|
|
|
|
blocksize = None
|
|
if hashlen == 20:
|
|
alg = cryptodev.CRYPTO_SHA1_HMAC
|
|
blocksize = 64
|
|
elif hashlen == 28:
|
|
alg = cryptodev.CRYPTO_SHA2_224_HMAC
|
|
blocksize = 64
|
|
elif hashlen == 32:
|
|
alg = cryptodev.CRYPTO_SHA2_256_HMAC
|
|
blocksize = 64
|
|
elif hashlen == 48:
|
|
alg = cryptodev.CRYPTO_SHA2_384_HMAC
|
|
blocksize = 128
|
|
elif hashlen == 64:
|
|
alg = cryptodev.CRYPTO_SHA2_512_HMAC
|
|
blocksize = 128
|
|
else:
|
|
# Skip unsupported hashes
|
|
# Slurp remaining input in section
|
|
for data in lines:
|
|
continue
|
|
continue
|
|
|
|
for data in lines:
|
|
key = binascii.unhexlify(data['Key'])
|
|
msg = binascii.unhexlify(data['Msg'])
|
|
mac = binascii.unhexlify(data['Mac'])
|
|
tlen = int(data['Tlen'])
|
|
|
|
if len(key) > blocksize:
|
|
continue
|
|
|
|
try:
|
|
c = Crypto(mac=alg, mackey=key,
|
|
crid=crid, maclen=hashlen)
|
|
except EnvironmentError as e:
|
|
# Can't test hashes the driver does not support.
|
|
if e.errno != errno.EOPNOTSUPP:
|
|
raise
|
|
continue
|
|
|
|
_, r = c.encrypt(msg, iv="")
|
|
|
|
self.assertEqual(r[:tlen], mac, "Actual: " + \
|
|
repr(binascii.hexlify(r)) + " Expected: " + repr(data))
|
|
|
|
return GendCryptoTestCase
|
|
|
|
cryptosoft = GenTestCase('cryptosoft0')
|
|
aesni = GenTestCase('aesni0')
|
|
ccr = GenTestCase('ccr0')
|
|
ccp = GenTestCase('ccp0')
|
|
|
|
if __name__ == '__main__':
|
|
unittest.main()
|