tests/sys/opencrypto/cryptotest.py

   1 #!/usr/local/bin/python3
   2 #
   3 # Copyright (c) 2014 The FreeBSD Foundation
   4 # All rights reserved.
   5 # Copyright 2019 Enji Cooper
   6 #
   7 # This software was developed by John-Mark Gurney under
   8 # the sponsorship from the FreeBSD Foundation.
   9 # Redistribution and use in source and binary forms, with or without
  10 # modification, are permitted provided that the following conditions
  11 # are met:
  12 # 1.  Redistributions of source code must retain the above copyright
  13 #     notice, this list of conditions and the following disclaimer.
  14 # 2.  Redistributions in binary form must reproduce the above copyright
  15 #     notice, this list of conditions and the following disclaimer in the
  16 #     documentation and/or other materials provided with the distribution.
  17 #
  18 # THIS SOFTWARE IS PROVIDED BY THE AUTHOR AND CONTRIBUTORS ``AS IS'' AND
  19 # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
  20 # IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
  21 # ARE DISCLAIMED. IN NO EVENT SHALL THE AUTHOR OR CONTRIBUTORS BE LIABLE
  22 # FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
  23 # DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS
  24 # OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION)
  25 # HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT
  26 # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY
  27 # OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF
  28 # SUCH DAMAGE.
  29 #
  30 # $FreeBSD$
  31 #
  32
  33
  34
  35 import binascii
  36 import errno
  37 import cryptodev
  38 import itertools
  39 import os
  40 import struct
  41 import unittest
  42 from cryptodev import *
  43 from glob import iglob
  44
  45 katdir = '/usr/local/share/nist-kat'
  46
  47 def katg(base, glob):
  48     assert os.path.exists(katdir), "Please 'pkg install nist-kat'"
  49     if not os.path.exists(os.path.join(katdir, base)):
  50         raise unittest.SkipTest("Missing %s test vectors" % (base))
  51     return iglob(os.path.join(katdir, base, glob))
  52
  53 aesmodules = [ 'cryptosoft0', 'aesni0', 'armv8crypto0', 'ccr0', 'ccp0', 'safexcel0', 'qat0' ]
  54 shamodules = [ 'cryptosoft0', 'aesni0', 'armv8crypto0', 'ccr0', 'ccp0', 'ossl0', 'safexcel0', 'qat0' ]
  55
  56 def GenTestCase(cname):
  57     try:
  58         crid = cryptodev.Crypto.findcrid(cname)
  59     except IOError:
  60         return None
  61
  62     class GendCryptoTestCase(unittest.TestCase):
  63         ###############
  64         ##### AES #####
  65         ###############
  66         @unittest.skipIf(cname not in aesmodules, 'skipping AES-XTS on %s' % (cname))
  67         def test_xts(self):
  68             for i in katg('XTSTestVectors/format tweak value input - data unit seq no', '*.rsp'):
  69                 self.runXTS(i, cryptodev.CRYPTO_AES_XTS)
  70
  71         @unittest.skipIf(cname not in aesmodules, 'skipping AES-CBC on %s' % (cname))
  72         def test_cbc(self):
  73             for i in katg('KAT_AES', 'CBC[GKV]*.rsp'):
  74                 self.runCBC(i)
  75
  76         @unittest.skipIf(cname not in aesmodules, 'skipping AES-CCM on %s' % (cname))
  77         def test_ccm(self):
  78             for i in katg('ccmtestvectors', 'V*.rsp'):
  79                 self.runCCMEncrypt(i)
  80
  81             for i in katg('ccmtestvectors', 'D*.rsp'):
  82                 self.runCCMDecrypt(i)
  83
  84         @unittest.skipIf(cname not in aesmodules, 'skipping AES-GCM on %s' % (cname))
  85         def test_gcm(self):
  86             for i in katg('gcmtestvectors', 'gcmEncrypt*'):
  87                 self.runGCM(i, 'ENCRYPT')
  88
  89             for i in katg('gcmtestvectors', 'gcmDecrypt*'):
  90                 self.runGCM(i, 'DECRYPT')
  91
  92         def runGCM(self, fname, mode):
  93             curfun = None
  94             if mode == 'ENCRYPT':
  95                 swapptct = False
  96                 curfun = Crypto.encrypt
  97             elif mode == 'DECRYPT':
  98                 swapptct = True
  99                 curfun = Crypto.decrypt
 100             else:
 101                 raise RuntimeError('unknown mode: %r' % repr(mode))
 102
 103             columns = [ 'Count', 'Key', 'IV', 'CT', 'AAD', 'Tag', 'PT', ]
 104             with cryptodev.KATParser(fname, columns) as parser:
 105                 self.runGCMWithParser(parser, mode)
 106
 107         def runGCMWithParser(self, parser, mode):
 108             for _, lines in next(parser):
 109                 for data in lines:
 110                     curcnt = int(data['Count'])
 111                     cipherkey = binascii.unhexlify(data['Key'])
 112                     iv = binascii.unhexlify(data['IV'])
 113                     aad = binascii.unhexlify(data['AAD'])
 114                     tag = binascii.unhexlify(data['Tag'])
 115                     if 'FAIL' not in data:
 116                         pt = binascii.unhexlify(data['PT'])
 117                     ct = binascii.unhexlify(data['CT'])
 118
 119                     if len(iv) != 12:
 120                         # XXX - isn't supported
 121                         continue
 122
 123                     try:
 124                         c = Crypto(cryptodev.CRYPTO_AES_NIST_GCM_16,
 125                             cipherkey, crid=crid,
 126                             maclen=16)
 127                     except EnvironmentError as e:
 128                         # Can't test algorithms the driver does not support.
 129                         if e.errno != errno.EOPNOTSUPP:
 130                             raise
 131                         continue
 132
 133                     if mode == 'ENCRYPT':
 134                         try:
 135                             rct, rtag = c.encrypt(pt, iv, aad)
 136                         except EnvironmentError as e:
 137                             # Can't test inputs the driver does not support.
 138                             if e.errno != errno.EINVAL:
 139                                 raise
 140                             continue
 141                         rtag = rtag[:len(tag)]
 142                         data['rct'] = binascii.hexlify(rct)
 143                         data['rtag'] = binascii.hexlify(rtag)
 144                         self.assertEqual(rct, ct, repr(data))
 145                         self.assertEqual(rtag, tag, repr(data))
 146                     else:
 147                         if len(tag) != 16:
 148                             continue
 149                         args = (ct, iv, aad, tag)
 150                         if 'FAIL' in data:
 151                             self.assertRaises(IOError,
 152                                 c.decrypt, *args)
 153                         else:
 154                             try:
 155                                 rpt, rtag = c.decrypt(*args)
 156                             except EnvironmentError as e:
 157                                 # Can't test inputs the driver does not support.
 158                                 if e.errno != errno.EINVAL:
 159                                     raise
 160                                 continue
 161                             data['rpt'] = binascii.hexlify(rpt)
 162                             data['rtag'] = binascii.hexlify(rtag)
 163                             self.assertEqual(rpt, pt,
 164                                 repr(data))
 165
 166         def runCBC(self, fname):
 167             columns = [ 'COUNT', 'KEY', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]
 168             with cryptodev.KATParser(fname, columns) as parser:
 169                 self.runCBCWithParser(parser)
 170
 171         def runCBCWithParser(self, parser):
 172             curfun = None
 173             for mode, lines in next(parser):
 174                 if mode == 'ENCRYPT':
 175                     swapptct = False
 176                     curfun = Crypto.encrypt
 177                 elif mode == 'DECRYPT':
 178                     swapptct = True
 179                     curfun = Crypto.decrypt
 180                 else:
 181                     raise RuntimeError('unknown mode: %r' % repr(mode))
 182
 183                 for data in lines:
 184                     curcnt = int(data['COUNT'])
 185                     cipherkey = binascii.unhexlify(data['KEY'])
 186                     iv = binascii.unhexlify(data['IV'])
 187                     pt = binascii.unhexlify(data['PLAINTEXT'])
 188                     ct = binascii.unhexlify(data['CIPHERTEXT'])
 189
 190                     if swapptct:
 191                         pt, ct = ct, pt
 192                     # run the fun
 193                     c = Crypto(cryptodev.CRYPTO_AES_CBC, cipherkey, crid=crid)
 194                     r = curfun(c, pt, iv)
 195                     self.assertEqual(r, ct)
 196
 197         def runXTS(self, fname, meth):
 198             columns = [ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT',
 199                         'CT']
 200             with cryptodev.KATParser(fname, columns) as parser:
 201                 self.runXTSWithParser(parser, meth)
 202
 203         def runXTSWithParser(self, parser, meth):
 204             curfun = None
 205             for mode, lines in next(parser):
 206                 if mode == 'ENCRYPT':
 207                     swapptct = False
 208                     curfun = Crypto.encrypt
 209                 elif mode == 'DECRYPT':
 210                     swapptct = True
 211                     curfun = Crypto.decrypt
 212                 else:
 213                     raise RuntimeError('unknown mode: %r' % repr(mode))
 214
 215                 for data in lines:
 216                     curcnt = int(data['COUNT'])
 217                     nbits = int(data['DataUnitLen'])
 218                     cipherkey = binascii.unhexlify(data['Key'])
 219                     iv = struct.pack('QQ', int(data['DataUnitSeqNumber']), 0)
 220                     pt = binascii.unhexlify(data['PT'])
 221                     ct = binascii.unhexlify(data['CT'])
 222
 223                     if nbits % 128 != 0:
 224                         # XXX - mark as skipped
 225                         continue
 226                     if swapptct:
 227                         pt, ct = ct, pt
 228                     # run the fun
 229                     try:
 230                         c = Crypto(meth, cipherkey, crid=crid)
 231                         r = curfun(c, pt, iv)
 232                     except EnvironmentError as e:
 233                         # Can't test hashes the driver does not support.
 234                         if e.errno != errno.EOPNOTSUPP:
 235                             raise
 236                         continue
 237                     self.assertEqual(r, ct)
 238
 239         def runCCMEncrypt(self, fname):
 240             with cryptodev.KATCCMParser(fname) as parser:
 241                 self.runCCMEncryptWithParser(parser)
 242
 243         def runCCMEncryptWithParser(self, parser):
 244             for data in next(parser):
 245                 Nlen = int(data['Nlen'])
 246                 Tlen = int(data['Tlen'])
 247                 key = binascii.unhexlify(data['Key'])
 248                 nonce = binascii.unhexlify(data['Nonce'])
 249                 Alen = int(data['Alen'])
 250                 Plen = int(data['Plen'])
 251                 if Alen != 0:
 252                     aad = binascii.unhexlify(data['Adata'])
 253                 else:
 254                     aad = None
 255                 if Plen != 0:
 256                     payload = binascii.unhexlify(data['Payload'])
 257                 else:
 258                     payload = None
 259                 ct = binascii.unhexlify(data['CT'])
 260
 261                 try:
 262                     c = Crypto(crid=crid,
 263                         cipher=cryptodev.CRYPTO_AES_CCM_16,
 264                         key=key,
 265                         mackey=key, maclen=Tlen, ivlen=Nlen)
 266                     r, tag = Crypto.encrypt(c, payload,
 267                         nonce, aad)
 268                 except EnvironmentError as e:
 269                     if e.errno != errno.EOPNOTSUPP:
 270                         raise
 271                     continue
 272
 273                 out = r + tag
 274                 self.assertEqual(out, ct,
 275                     "Count " + data['Count'] + " Actual: " + \
 276                     repr(binascii.hexlify(out)) + " Expected: " + \
 277                     repr(data) + " on " + cname)
 278
 279         def runCCMDecrypt(self, fname):
 280             with cryptodev.KATCCMParser(fname) as parser:
 281                 self.runCCMDecryptWithParser(parser)
 282
 283         def runCCMDecryptWithParser(self, parser):
 284             for data in next(parser):
 285                 Nlen = int(data['Nlen'])
 286                 Tlen = int(data['Tlen'])
 287                 key = binascii.unhexlify(data['Key'])
 288                 nonce = binascii.unhexlify(data['Nonce'])
 289                 Alen = int(data['Alen'])
 290                 Plen = int(data['Plen'])
 291                 if Alen != 0:
 292                     aad = binascii.unhexlify(data['Adata'])
 293                 else:
 294                     aad = None
 295                 ct = binascii.unhexlify(data['CT'])
 296                 tag = ct[-Tlen:]
 297                 if Plen != 0:
 298                     payload = ct[:-Tlen]
 299                 else:
 300                     payload = None
 301
 302                 try:
 303                     c = Crypto(crid=crid,
 304                         cipher=cryptodev.CRYPTO_AES_CCM_16,
 305                         key=key,
 306                         mackey=key, maclen=Tlen, ivlen=Nlen)
 307                 except EnvironmentError as e:
 308                     if e.errno != errno.EOPNOTSUPP:
 309                         raise
 310                     continue
 311
 312                 if data['Result'] == 'Fail':
 313                     self.assertRaises(IOError,
 314                         c.decrypt, payload, nonce, aad, tag)
 315                 else:
 316                     r, tag = Crypto.decrypt(c, payload, nonce,
 317                                             aad, tag)
 318
 319                     payload = binascii.unhexlify(data['Payload'])
 320                     payload = payload[:Plen]
 321                     self.assertEqual(r, payload,
 322                         "Count " + data['Count'] + \
 323                         " Actual: " + repr(binascii.hexlify(r)) + \
 324                         " Expected: " + repr(data) + \
 325                         " on " + cname)
 326
 327         ###############
 328         ##### SHA #####
 329         ###############
 330         @unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
 331         def test_sha(self):
 332             for i in katg('shabytetestvectors', 'SHA*Msg.rsp'):
 333                 self.runSHA(i)
 334
 335         def runSHA(self, fname):
 336             # Skip SHA512_(224|256) tests
 337             if fname.find('SHA512_') != -1:
 338                 return
 339             columns = [ 'Len', 'Msg', 'MD' ]
 340             with cryptodev.KATParser(fname, columns) as parser:
 341                 self.runSHAWithParser(parser)
 342
 343         def runSHAWithParser(self, parser):
 344             for hashlength, lines in next(parser):
 345                 # E.g., hashlength will be "L=20" (bytes)
 346                 hashlen = int(hashlength.split("=")[1])
 347
 348                 if hashlen == 20:
 349                     alg = cryptodev.CRYPTO_SHA1
 350                 elif hashlen == 28:
 351                     alg = cryptodev.CRYPTO_SHA2_224
 352                 elif hashlen == 32:
 353                     alg = cryptodev.CRYPTO_SHA2_256
 354                 elif hashlen == 48:
 355                     alg = cryptodev.CRYPTO_SHA2_384
 356                 elif hashlen == 64:
 357                     alg = cryptodev.CRYPTO_SHA2_512
 358                 else:
 359                     # Skip unsupported hashes
 360                     # Slurp remaining input in section
 361                     for data in lines:
 362                         continue
 363                     continue
 364
 365                 for data in lines:
 366                     msg = binascii.unhexlify(data['Msg'])
 367                     msg = msg[:int(data['Len'])]
 368                     md = binascii.unhexlify(data['MD'])
 369
 370                     try:
 371                         c = Crypto(mac=alg, crid=crid,
 372                             maclen=hashlen)
 373                     except EnvironmentError as e:
 374                         # Can't test hashes the driver does not support.
 375                         if e.errno != errno.EOPNOTSUPP:
 376                             raise
 377                         continue
 378
 379                     _, r = c.encrypt(msg, iv="")
 380
 381                     self.assertEqual(r, md, "Actual: " + \
 382                         repr(binascii.hexlify(r)) + " Expected: " + repr(data) + " on " + cname)
 383
 384         @unittest.skipIf(cname not in shamodules, 'skipping SHA-HMAC on %s' % str(cname))
 385         def test_sha1hmac(self):
 386             for i in katg('hmactestvectors', 'HMAC.rsp'):
 387                 self.runSHA1HMAC(i)
 388
 389         def runSHA1HMAC(self, fname):
 390             columns = [ 'Count', 'Klen', 'Tlen', 'Key', 'Msg', 'Mac' ]
 391             with cryptodev.KATParser(fname, columns) as parser:
 392                 self.runSHA1HMACWithParser(parser)
 393
 394         def runSHA1HMACWithParser(self, parser):
 395             for hashlength, lines in next(parser):
 396                 # E.g., hashlength will be "L=20" (bytes)
 397                 hashlen = int(hashlength.split("=")[1])
 398
 399                 blocksize = None
 400                 if hashlen == 20:
 401                     alg = cryptodev.CRYPTO_SHA1_HMAC
 402                     blocksize = 64
 403                 elif hashlen == 28:
 404                     alg = cryptodev.CRYPTO_SHA2_224_HMAC
 405                     blocksize = 64
 406                 elif hashlen == 32:
 407                     alg = cryptodev.CRYPTO_SHA2_256_HMAC
 408                     blocksize = 64
 409                 elif hashlen == 48:
 410                     alg = cryptodev.CRYPTO_SHA2_384_HMAC
 411                     blocksize = 128
 412                 elif hashlen == 64:
 413                     alg = cryptodev.CRYPTO_SHA2_512_HMAC
 414                     blocksize = 128
 415                 else:
 416                     # Skip unsupported hashes
 417                     # Slurp remaining input in section
 418                     for data in lines:
 419                         continue
 420                     continue
 421
 422                 for data in lines:
 423                     key = binascii.unhexlify(data['Key'])
 424                     msg = binascii.unhexlify(data['Msg'])
 425                     mac = binascii.unhexlify(data['Mac'])
 426                     tlen = int(data['Tlen'])
 427
 428                     if len(key) > blocksize:
 429                         continue
 430
 431                     try:
 432                         c = Crypto(mac=alg, mackey=key,
 433                             crid=crid, maclen=hashlen)
 434                     except EnvironmentError as e:
 435                         # Can't test hashes the driver does not support.
 436                         if e.errno != errno.EOPNOTSUPP:
 437                             raise
 438                         continue
 439
 440                     _, r = c.encrypt(msg, iv="")
 441
 442                     self.assertEqual(r[:tlen], mac, "Actual: " + \
 443                         repr(binascii.hexlify(r)) + " Expected: " + repr(data))
 444
 445     return GendCryptoTestCase
 446
 447 cryptosoft = GenTestCase('cryptosoft0')
 448 aesni = GenTestCase('aesni0')
 449 armv8crypto = GenTestCase('armv8crypto0')
 450 ccr = GenTestCase('ccr0')
 451 ccp = GenTestCase('ccp0')
 452 ossl = GenTestCase('ossl0')
 453 safexcel = GenTestCase('safexcel0')
 454 qat = GenTestCase('qat0')
 455
 456 if __name__ == '__main__':
 457     unittest.main()