tests/sys/opencrypto/cryptotest.py

   1 #!/usr/local/bin/python2
   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 from __future__ import print_function
  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', 'ccr0', 'ccp0' ]
  54 desmodules = [ 'cryptosoft0', ]
  55 shamodules = [ 'cryptosoft0', 'aesni0', 'ccr0', 'ccp0' ]
  56
  57 def GenTestCase(cname):
  58     try:
  59         crid = cryptodev.Crypto.findcrid(cname)
  60     except IOError:
  61         return None
  62
  63     class GendCryptoTestCase(unittest.TestCase):
  64         ###############
  65         ##### AES #####
  66         ###############
  67         @unittest.skipIf(cname not in aesmodules, 'skipping AES-XTS on %s' % (cname))
  68         def test_xts(self):
  69             for i in katg('XTSTestVectors/format tweak value input - data unit seq no', '*.rsp'):
  70                 self.runXTS(i, cryptodev.CRYPTO_AES_XTS)
  71
  72         @unittest.skipIf(cname not in aesmodules, 'skipping AES-CBC on %s' % (cname))
  73         def test_cbc(self):
  74             for i in katg('KAT_AES', 'CBC[GKV]*.rsp'):
  75                 self.runCBC(i)
  76
  77         @unittest.skipIf(cname not in aesmodules, 'skipping AES-CCM on %s' % (cname))
  78         def test_ccm(self):
  79             for i in katg('ccmtestvectors', 'V*.rsp'):
  80                 self.runCCMEncrypt(i)
  81
  82             for i in katg('ccmtestvectors', 'D*.rsp'):
  83                 self.runCCMDecrypt(i)
  84
  85         @unittest.skipIf(cname not in aesmodules, 'skipping AES-GCM on %s' % (cname))
  86         def test_gcm(self):
  87             for i in katg('gcmtestvectors', 'gcmEncrypt*'):
  88                 self.runGCM(i, 'ENCRYPT')
  89
  90             for i in katg('gcmtestvectors', 'gcmDecrypt*'):
  91                 self.runGCM(i, 'DECRYPT')
  92
  93         _gmacsizes = { 32: cryptodev.CRYPTO_AES_256_NIST_GMAC,
  94             24: cryptodev.CRYPTO_AES_192_NIST_GMAC,
  95             16: cryptodev.CRYPTO_AES_128_NIST_GMAC,
  96         }
  97         def runGCM(self, fname, mode):
  98             curfun = None
  99             if mode == 'ENCRYPT':
 100                 swapptct = False
 101                 curfun = Crypto.encrypt
 102             elif mode == 'DECRYPT':
 103                 swapptct = True
 104                 curfun = Crypto.decrypt
 105             else:
 106                 raise RuntimeError('unknown mode: %r' % repr(mode))
 107
 108             columns = [ 'Count', 'Key', 'IV', 'CT', 'AAD', 'Tag', 'PT', ]
 109             with cryptodev.KATParser(fname, columns) as parser:
 110                 self.runGCMWithParser(parser, mode)
 111
 112         def runGCMWithParser(self, parser, mode):
 113             for _, lines in next(parser):
 114                 for data in lines:
 115                     curcnt = int(data['Count'])
 116                     cipherkey = binascii.unhexlify(data['Key'])
 117                     iv = binascii.unhexlify(data['IV'])
 118                     aad = binascii.unhexlify(data['AAD'])
 119                     tag = binascii.unhexlify(data['Tag'])
 120                     if 'FAIL' not in data:
 121                         pt = binascii.unhexlify(data['PT'])
 122                     ct = binascii.unhexlify(data['CT'])
 123
 124                     if len(iv) != 12:
 125                         # XXX - isn't supported
 126                         continue
 127
 128                     try:
 129                         c = Crypto(cryptodev.CRYPTO_AES_NIST_GCM_16,
 130                             cipherkey,
 131                             mac=self._gmacsizes[len(cipherkey)],
 132                             mackey=cipherkey, crid=crid,
 133                             maclen=16)
 134                     except EnvironmentError as e:
 135                         # Can't test algorithms the driver does not support.
 136                         if e.errno != errno.EOPNOTSUPP:
 137                             raise
 138                         continue
 139
 140                     if mode == 'ENCRYPT':
 141                         try:
 142                             rct, rtag = c.encrypt(pt, iv, aad)
 143                         except EnvironmentError as e:
 144                             # Can't test inputs the driver does not support.
 145                             if e.errno != errno.EINVAL:
 146                                 raise
 147                             continue
 148                         rtag = rtag[:len(tag)]
 149                         data['rct'] = binascii.hexlify(rct)
 150                         data['rtag'] = binascii.hexlify(rtag)
 151                         self.assertEqual(rct, ct, repr(data))
 152                         self.assertEqual(rtag, tag, repr(data))
 153                     else:
 154                         if len(tag) != 16:
 155                             continue
 156                         args = (ct, iv, aad, tag)
 157                         if 'FAIL' in data:
 158                             self.assertRaises(IOError,
 159                                 c.decrypt, *args)
 160                         else:
 161                             try:
 162                                 rpt, rtag = c.decrypt(*args)
 163                             except EnvironmentError as e:
 164                                 # Can't test inputs the driver does not support.
 165                                 if e.errno != errno.EINVAL:
 166                                     raise
 167                                 continue
 168                             data['rpt'] = binascii.hexlify(rpt)
 169                             data['rtag'] = binascii.hexlify(rtag)
 170                             self.assertEqual(rpt, pt,
 171                                 repr(data))
 172
 173         def runCBC(self, fname):
 174             columns = [ 'COUNT', 'KEY', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]
 175             with cryptodev.KATParser(fname, columns) as parser:
 176                 self.runCBCWithParser(parser)
 177
 178         def runCBCWithParser(self, parser):
 179             curfun = None
 180             for mode, lines in next(parser):
 181                 if mode == 'ENCRYPT':
 182                     swapptct = False
 183                     curfun = Crypto.encrypt
 184                 elif mode == 'DECRYPT':
 185                     swapptct = True
 186                     curfun = Crypto.decrypt
 187                 else:
 188                     raise RuntimeError('unknown mode: %r' % repr(mode))
 189
 190                 for data in lines:
 191                     curcnt = int(data['COUNT'])
 192                     cipherkey = binascii.unhexlify(data['KEY'])
 193                     iv = binascii.unhexlify(data['IV'])
 194                     pt = binascii.unhexlify(data['PLAINTEXT'])
 195                     ct = binascii.unhexlify(data['CIPHERTEXT'])
 196
 197                     if swapptct:
 198                         pt, ct = ct, pt
 199                     # run the fun
 200                     c = Crypto(cryptodev.CRYPTO_AES_CBC, cipherkey, crid=crid)
 201                     r = curfun(c, pt, iv)
 202                     self.assertEqual(r, ct)
 203
 204         def runXTS(self, fname, meth):
 205             columns = [ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT',
 206                         'CT']
 207             with cryptodev.KATParser(fname, columns) as parser:
 208                 self.runXTSWithParser(parser, meth)
 209
 210         def runXTSWithParser(self, parser, meth):
 211             curfun = None
 212             for mode, lines in next(parser):
 213                 if mode == 'ENCRYPT':
 214                     swapptct = False
 215                     curfun = Crypto.encrypt
 216                 elif mode == 'DECRYPT':
 217                     swapptct = True
 218                     curfun = Crypto.decrypt
 219                 else:
 220                     raise RuntimeError('unknown mode: %r' % repr(mode))
 221
 222                 for data in lines:
 223                     curcnt = int(data['COUNT'])
 224                     nbits = int(data['DataUnitLen'])
 225                     cipherkey = binascii.unhexlify(data['Key'])
 226                     iv = struct.pack('QQ', int(data['DataUnitSeqNumber']), 0)
 227                     pt = binascii.unhexlify(data['PT'])
 228                     ct = binascii.unhexlify(data['CT'])
 229
 230                     if nbits % 128 != 0:
 231                         # XXX - mark as skipped
 232                         continue
 233                     if swapptct:
 234                         pt, ct = ct, pt
 235                     # run the fun
 236                     try:
 237                         c = Crypto(meth, cipherkey, crid=crid)
 238                         r = curfun(c, pt, iv)
 239                     except EnvironmentError as e:
 240                         # Can't test hashes the driver does not support.
 241                         if e.errno != errno.EOPNOTSUPP:
 242                             raise
 243                         continue
 244                     self.assertEqual(r, ct)
 245
 246         def runCCMEncrypt(self, fname):
 247             with cryptodev.KATCCMParser(fname) as parser:
 248                 self.runCCMEncryptWithParser(parser)
 249
 250         def runCCMEncryptWithParser(self, parser):
 251             for data in next(parser):
 252                 Nlen = int(data['Nlen'])
 253                 if Nlen != 12:
 254                     # OCF only supports 12 byte IVs
 255                     continue
 256                 key = binascii.unhexlify(data['Key'])
 257                 nonce = binascii.unhexlify(data['Nonce'])
 258                 Alen = int(data['Alen'])
 259                 if Alen != 0:
 260                     aad = binascii.unhexlify(data['Adata'])
 261                 else:
 262                     aad = None
 263                 payload = binascii.unhexlify(data['Payload'])
 264                 ct = binascii.unhexlify(data['CT'])
 265
 266                 try:
 267                     c = Crypto(crid=crid,
 268                         cipher=cryptodev.CRYPTO_AES_CCM_16,
 269                         key=key,
 270                         mac=cryptodev.CRYPTO_AES_CCM_CBC_MAC,
 271                         mackey=key, maclen=16)
 272                     r, tag = Crypto.encrypt(c, payload,
 273                         nonce, aad)
 274                 except EnvironmentError as e:
 275                     if e.errno != errno.EOPNOTSUPP:
 276                         raise
 277                     continue
 278
 279                 out = r + tag
 280                 self.assertEqual(out, ct,
 281                     "Count " + data['Count'] + " Actual: " + \
 282                     repr(binascii.hexlify(out)) + " Expected: " + \
 283                     repr(data) + " on " + cname)
 284
 285         def runCCMDecrypt(self, fname):
 286             with cryptodev.KATCCMParser(fname) as parser:
 287                 self.runCCMDecryptWithParser(parser)
 288
 289         def runCCMDecryptWithParser(self, parser):
 290             # XXX: Note that all of the current CCM
 291             # decryption test vectors use IV and tag sizes
 292             # that aren't supported by OCF none of the
 293             # tests are actually ran.
 294             for data in next(parser):
 295                 Nlen = int(data['Nlen'])
 296                 if Nlen != 12:
 297                     # OCF only supports 12 byte IVs
 298                     continue
 299                 Tlen = int(data['Tlen'])
 300                 if Tlen != 16:
 301                     # OCF only supports 16 byte tags
 302                     continue
 303                 key = binascii.unhexlify(data['Key'])
 304                 nonce = binascii.unhexlify(data['Nonce'])
 305                 Alen = int(data['Alen'])
 306                 if Alen != 0:
 307                     aad = binascii.unhexlify(data['Adata'])
 308                 else:
 309                     aad = None
 310                 ct = binascii.unhexlify(data['CT'])
 311                 tag = ct[-16:]
 312                 ct = ct[:-16]
 313
 314                 try:
 315                     c = Crypto(crid=crid,
 316                         cipher=cryptodev.CRYPTO_AES_CCM_16,
 317                         key=key,
 318                         mac=cryptodev.CRYPTO_AES_CCM_CBC_MAC,
 319                         mackey=key, maclen=16)
 320                 except EnvironmentError as e:
 321                     if e.errno != errno.EOPNOTSUPP:
 322                         raise
 323                     continue
 324
 325                 if data['Result'] == 'Fail':
 326                     self.assertRaises(IOError,
 327                         c.decrypt, payload, nonce, aad, tag)
 328                 else:
 329                     r = Crypto.decrypt(c, payload, nonce,
 330                         aad, tag)
 331
 332                     payload = binascii.unhexlify(data['Payload'])
 333                     plen = int(data('Plen'))
 334                     payload = payload[:plen]
 335                     self.assertEqual(r, payload,
 336                         "Count " + data['Count'] + \
 337                         " Actual: " + repr(binascii.hexlify(r)) + \
 338                         " Expected: " + repr(data) + \
 339                         " on " + cname)
 340
 341         ###############
 342         ##### DES #####
 343         ###############
 344         @unittest.skipIf(cname not in desmodules, 'skipping DES on %s' % (cname))
 345         def test_tdes(self):
 346             for i in katg('KAT_TDES', 'TCBC[a-z]*.rsp'):
 347                 self.runTDES(i)
 348
 349         def runTDES(self, fname):
 350             columns = [ 'COUNT', 'KEYs', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]
 351             with cryptodev.KATParser(fname, columns) as parser:
 352                 self.runTDESWithParser(parser)
 353
 354         def runTDESWithParser(self, parser):
 355             curfun = None
 356             for mode, lines in next(parser):
 357                 if mode == 'ENCRYPT':
 358                     swapptct = False
 359                     curfun = Crypto.encrypt
 360                 elif mode == 'DECRYPT':
 361                     swapptct = True
 362                     curfun = Crypto.decrypt
 363                 else:
 364                     raise RuntimeError('unknown mode: %r' % repr(mode))
 365
 366                 for data in lines:
 367                     curcnt = int(data['COUNT'])
 368                     key = data['KEYs'] * 3
 369                     cipherkey = binascii.unhexlify(key)
 370                     iv = binascii.unhexlify(data['IV'])
 371                     pt = binascii.unhexlify(data['PLAINTEXT'])
 372                     ct = binascii.unhexlify(data['CIPHERTEXT'])
 373
 374                     if swapptct:
 375                         pt, ct = ct, pt
 376                     # run the fun
 377                     c = Crypto(cryptodev.CRYPTO_3DES_CBC, cipherkey, crid=crid)
 378                     r = curfun(c, pt, iv)
 379                     self.assertEqual(r, ct)
 380
 381         ###############
 382         ##### SHA #####
 383         ###############
 384         @unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
 385         def test_sha(self):
 386             for i in katg('shabytetestvectors', 'SHA*Msg.rsp'):
 387                 self.runSHA(i)
 388
 389         def runSHA(self, fname):
 390             # Skip SHA512_(224|256) tests
 391             if fname.find('SHA512_') != -1:
 392                 return
 393             columns = [ 'Len', 'Msg', 'MD' ]
 394             with cryptodev.KATParser(fname, columns) as parser:
 395                 self.runSHAWithParser(parser)
 396
 397         def runSHAWithParser(self, parser):
 398             for hashlength, lines in next(parser):
 399                 # E.g., hashlength will be "L=20" (bytes)
 400                 hashlen = int(hashlength.split("=")[1])
 401
 402                 if hashlen == 20:
 403                     alg = cryptodev.CRYPTO_SHA1
 404                 elif hashlen == 28:
 405                     alg = cryptodev.CRYPTO_SHA2_224
 406                 elif hashlen == 32:
 407                     alg = cryptodev.CRYPTO_SHA2_256
 408                 elif hashlen == 48:
 409                     alg = cryptodev.CRYPTO_SHA2_384
 410                 elif hashlen == 64:
 411                     alg = cryptodev.CRYPTO_SHA2_512
 412                 else:
 413                     # Skip unsupported hashes
 414                     # Slurp remaining input in section
 415                     for data in lines:
 416                         continue
 417                     continue
 418
 419                 for data in lines:
 420                     msg = binascii.unhexlify(data['Msg'])
 421                     msg = msg[:int(data['Len'])]
 422                     md = binascii.unhexlify(data['MD'])
 423
 424                     try:
 425                         c = Crypto(mac=alg, crid=crid,
 426                             maclen=hashlen)
 427                     except EnvironmentError as e:
 428                         # Can't test hashes the driver does not support.
 429                         if e.errno != errno.EOPNOTSUPP:
 430                             raise
 431                         continue
 432
 433                     _, r = c.encrypt(msg, iv="")
 434
 435                     self.assertEqual(r, md, "Actual: " + \
 436                         repr(binascii.hexlify(r)) + " Expected: " + repr(data) + " on " + cname)
 437
 438         @unittest.skipIf(cname not in shamodules, 'skipping SHA-HMAC on %s' % str(cname))
 439         def test_sha1hmac(self):
 440             for i in katg('hmactestvectors', 'HMAC.rsp'):
 441                 self.runSHA1HMAC(i)
 442
 443         def runSHA1HMAC(self, fname):
 444             columns = [ 'Count', 'Klen', 'Tlen', 'Key', 'Msg', 'Mac' ]
 445             with cryptodev.KATParser(fname, columns) as parser:
 446                 self.runSHA1HMACWithParser(parser)
 447
 448         def runSHA1HMACWithParser(self, parser):
 449             for hashlength, lines in next(parser):
 450                 # E.g., hashlength will be "L=20" (bytes)
 451                 hashlen = int(hashlength.split("=")[1])
 452
 453                 blocksize = None
 454                 if hashlen == 20:
 455                     alg = cryptodev.CRYPTO_SHA1_HMAC
 456                     blocksize = 64
 457                 elif hashlen == 28:
 458                     alg = cryptodev.CRYPTO_SHA2_224_HMAC
 459                     blocksize = 64
 460                 elif hashlen == 32:
 461                     alg = cryptodev.CRYPTO_SHA2_256_HMAC
 462                     blocksize = 64
 463                 elif hashlen == 48:
 464                     alg = cryptodev.CRYPTO_SHA2_384_HMAC
 465                     blocksize = 128
 466                 elif hashlen == 64:
 467                     alg = cryptodev.CRYPTO_SHA2_512_HMAC
 468                     blocksize = 128
 469                 else:
 470                     # Skip unsupported hashes
 471                     # Slurp remaining input in section
 472                     for data in lines:
 473                         continue
 474                     continue
 475
 476                 for data in lines:
 477                     key = binascii.unhexlify(data['Key'])
 478                     msg = binascii.unhexlify(data['Msg'])
 479                     mac = binascii.unhexlify(data['Mac'])
 480                     tlen = int(data['Tlen'])
 481
 482                     if len(key) > blocksize:
 483                         continue
 484
 485                     try:
 486                         c = Crypto(mac=alg, mackey=key,
 487                             crid=crid, maclen=hashlen)
 488                     except EnvironmentError as e:
 489                         # Can't test hashes the driver does not support.
 490                         if e.errno != errno.EOPNOTSUPP:
 491                             raise
 492                         continue
 493
 494                     _, r = c.encrypt(msg, iv="")
 495
 496                     self.assertEqual(r[:tlen], mac, "Actual: " + \
 497                         repr(binascii.hexlify(r)) + " Expected: " + repr(data))
 498
 499     return GendCryptoTestCase
 500
 501 cryptosoft = GenTestCase('cryptosoft0')
 502 aesni = GenTestCase('aesni0')
 503 ccr = GenTestCase('ccr0')
 504 ccp = GenTestCase('ccp0')
 505
 506 if __name__ == '__main__':
 507     unittest.main()