xref: /freebsd/tests/sys/opencrypto/cryptotest.py (revision 6966ac055c3b7a39266fb982493330df7a097997)
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
33from __future__ import print_function
34
35import binascii
36import errno
37import cryptodev
38import itertools
39import os
40import struct
41import unittest
42from cryptodev import *
43from glob import iglob
44
45katdir = '/usr/local/share/nist-kat'
46
47def 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
53aesmodules = [ 'cryptosoft0', 'aesni0', 'armv8crypto0', 'ccr0', 'ccp0' ]
54desmodules = [ 'cryptosoft0', ]
55shamodules = [ 'cryptosoft0', 'aesni0', 'armv8crypto0', 'ccr0', 'ccp0' ]
56
57def 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
501cryptosoft = GenTestCase('cryptosoft0')
502aesni = GenTestCase('aesni0')
503armv8crypto = GenTestCase('armv8crypto0')
504ccr = GenTestCase('ccr0')
505ccp = GenTestCase('ccp0')
506
507if __name__ == '__main__':
508    unittest.main()
509