mirror of
https://git.FreeBSD.org/src.git
synced 2024-12-17 10:26:15 +00:00
Revert r346443
My wide sweeping stylistic change (while well intended) is impeding others from working on `tests/sys/opencrypto`. The plan is to revert the change in ^/head, then reintroduce the changes after the other changes get merged into ^/head . Approved by: emaste (mentor; implicit) Requested by: jhb MFC after: 2 months
This commit is contained in:
Enji Cooper
parent
60ade167fd
commit
03accca747
Notes:
svn2git
2020-12-20 02:59:44 +00:00
svn path=/head/; revision=346452
@ -43,65 +43,65 @@
|
||||
__all__ = [ 'Crypto', 'MismatchError', ]
|
||||
|
||||
class FindOp(dpkt.Packet):
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('crid', 'i', 0),
|
||||
('name', '32s', 0),
|
||||
)
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('crid', 'i', 0),
|
||||
('name', '32s', 0),
|
||||
)
|
||||
|
||||
class SessionOp(dpkt.Packet):
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('cipher', 'I', 0),
|
||||
('mac', 'I', 0),
|
||||
('keylen', 'I', 0),
|
||||
('key', 'P', 0),
|
||||
('mackeylen', 'i', 0),
|
||||
('mackey', 'P', 0),
|
||||
('ses', 'I', 0),
|
||||
)
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('cipher', 'I', 0),
|
||||
('mac', 'I', 0),
|
||||
('keylen', 'I', 0),
|
||||
('key', 'P', 0),
|
||||
('mackeylen', 'i', 0),
|
||||
('mackey', 'P', 0),
|
||||
('ses', 'I', 0),
|
||||
)
|
||||
|
||||
class SessionOp2(dpkt.Packet):
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('cipher', 'I', 0),
|
||||
('mac', 'I', 0),
|
||||
('keylen', 'I', 0),
|
||||
('key', 'P', 0),
|
||||
('mackeylen', 'i', 0),
|
||||
('mackey', 'P', 0),
|
||||
('ses', 'I', 0),
|
||||
('crid', 'i', 0),
|
||||
('pad0', 'i', 0),
|
||||
('pad1', 'i', 0),
|
||||
('pad2', 'i', 0),
|
||||
('pad3', 'i', 0),
|
||||
)
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('cipher', 'I', 0),
|
||||
('mac', 'I', 0),
|
||||
('keylen', 'I', 0),
|
||||
('key', 'P', 0),
|
||||
('mackeylen', 'i', 0),
|
||||
('mackey', 'P', 0),
|
||||
('ses', 'I', 0),
|
||||
('crid', 'i', 0),
|
||||
('pad0', 'i', 0),
|
||||
('pad1', 'i', 0),
|
||||
('pad2', 'i', 0),
|
||||
('pad3', 'i', 0),
|
||||
)
|
||||
|
||||
class CryptOp(dpkt.Packet):
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('ses', 'I', 0),
|
||||
('op', 'H', 0),
|
||||
('flags', 'H', 0),
|
||||
('len', 'I', 0),
|
||||
('src', 'P', 0),
|
||||
('dst', 'P', 0),
|
||||
('mac', 'P', 0),
|
||||
('iv', 'P', 0),
|
||||
)
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = ( ('ses', 'I', 0),
|
||||
('op', 'H', 0),
|
||||
('flags', 'H', 0),
|
||||
('len', 'I', 0),
|
||||
('src', 'P', 0),
|
||||
('dst', 'P', 0),
|
||||
('mac', 'P', 0),
|
||||
('iv', 'P', 0),
|
||||
)
|
||||
|
||||
class CryptAEAD(dpkt.Packet):
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = (
|
||||
('ses', 'I', 0),
|
||||
('op', 'H', 0),
|
||||
('flags', 'H', 0),
|
||||
('len', 'I', 0),
|
||||
('aadlen', 'I', 0),
|
||||
('ivlen', 'I', 0),
|
||||
('src', 'P', 0),
|
||||
('dst', 'P', 0),
|
||||
('aad', 'P', 0),
|
||||
('tag', 'P', 0),
|
||||
('iv', 'P', 0),
|
||||
)
|
||||
__byte_order__ = '@'
|
||||
__hdr__ = (
|
||||
('ses', 'I', 0),
|
||||
('op', 'H', 0),
|
||||
('flags', 'H', 0),
|
||||
('len', 'I', 0),
|
||||
('aadlen', 'I', 0),
|
||||
('ivlen', 'I', 0),
|
||||
('src', 'P', 0),
|
||||
('dst', 'P', 0),
|
||||
('aad', 'P', 0),
|
||||
('tag', 'P', 0),
|
||||
('iv', 'P', 0),
|
||||
)
|
||||
|
||||
# h2py.py can't handle multiarg macros
|
||||
CRIOGET = 3221513060
|
||||
@ -116,445 +116,445 @@ class CryptAEAD(dpkt.Packet):
|
||||
CIOCCRYPTAEAD = 3225445229
|
||||
|
||||
def _getdev():
|
||||
fd = os.open('/dev/crypto', os.O_RDWR)
|
||||
buf = array.array('I', [0])
|
||||
ioctl(fd, CRIOGET, buf, 1)
|
||||
os.close(fd)
|
||||
fd = os.open('/dev/crypto', os.O_RDWR)
|
||||
buf = array.array('I', [0])
|
||||
ioctl(fd, CRIOGET, buf, 1)
|
||||
os.close(fd)
|
||||
|
||||
return buf[0]
|
||||
return buf[0]
|
||||
|
||||
_cryptodev = _getdev()
|
||||
|
||||
def _findop(crid, name):
|
||||
fop = FindOp()
|
||||
fop.crid = crid
|
||||
fop.name = name
|
||||
s = array.array('B', fop.pack_hdr())
|
||||
ioctl(_cryptodev, CIOCFINDDEV, s, 1)
|
||||
fop.unpack(s)
|
||||
fop = FindOp()
|
||||
fop.crid = crid
|
||||
fop.name = name
|
||||
s = array.array('B', fop.pack_hdr())
|
||||
ioctl(_cryptodev, CIOCFINDDEV, s, 1)
|
||||
fop.unpack(s)
|
||||
|
||||
try:
|
||||
idx = fop.name.index('\x00')
|
||||
name = fop.name[:idx]
|
||||
except ValueError:
|
||||
name = fop.name
|
||||
try:
|
||||
idx = fop.name.index('\x00')
|
||||
name = fop.name[:idx]
|
||||
except ValueError:
|
||||
name = fop.name
|
||||
|
||||
return fop.crid, name
|
||||
return fop.crid, name
|
||||
|
||||
class Crypto:
|
||||
@staticmethod
|
||||
def findcrid(name):
|
||||
return _findop(-1, name)[0]
|
||||
@staticmethod
|
||||
def findcrid(name):
|
||||
return _findop(-1, name)[0]
|
||||
|
||||
@staticmethod
|
||||
def getcridname(crid):
|
||||
return _findop(crid, '')[1]
|
||||
@staticmethod
|
||||
def getcridname(crid):
|
||||
return _findop(crid, '')[1]
|
||||
|
||||
def __init__(self, cipher=0, key=None, mac=0, mackey=None,
|
||||
crid=CRYPTOCAP_F_SOFTWARE | CRYPTOCAP_F_HARDWARE):
|
||||
self._ses = None
|
||||
ses = SessionOp2()
|
||||
ses.cipher = cipher
|
||||
ses.mac = mac
|
||||
def __init__(self, cipher=0, key=None, mac=0, mackey=None,
|
||||
crid=CRYPTOCAP_F_SOFTWARE | CRYPTOCAP_F_HARDWARE):
|
||||
self._ses = None
|
||||
ses = SessionOp2()
|
||||
ses.cipher = cipher
|
||||
ses.mac = mac
|
||||
|
||||
if key is not None:
|
||||
ses.keylen = len(key)
|
||||
k = array.array('B', key)
|
||||
ses.key = k.buffer_info()[0]
|
||||
else:
|
||||
self.key = None
|
||||
if key is not None:
|
||||
ses.keylen = len(key)
|
||||
k = array.array('B', key)
|
||||
ses.key = k.buffer_info()[0]
|
||||
else:
|
||||
self.key = None
|
||||
|
||||
if mackey is not None:
|
||||
ses.mackeylen = len(mackey)
|
||||
mk = array.array('B', mackey)
|
||||
ses.mackey = mk.buffer_info()[0]
|
||||
self._maclen = 16 # parameterize?
|
||||
else:
|
||||
self._maclen = None
|
||||
if mackey is not None:
|
||||
ses.mackeylen = len(mackey)
|
||||
mk = array.array('B', mackey)
|
||||
ses.mackey = mk.buffer_info()[0]
|
||||
self._maclen = 16 # parameterize?
|
||||
else:
|
||||
self._maclen = None
|
||||
|
||||
if not cipher and not mac:
|
||||
raise ValueError('one of cipher or mac MUST be specified.')
|
||||
ses.crid = crid
|
||||
#print(ses)
|
||||
s = array.array('B', ses.pack_hdr())
|
||||
#print(s)
|
||||
ioctl(_cryptodev, CIOCGSESSION2, s, 1)
|
||||
ses.unpack(s)
|
||||
if not cipher and not mac:
|
||||
raise ValueError('one of cipher or mac MUST be specified.')
|
||||
ses.crid = crid
|
||||
#print(ses)
|
||||
s = array.array('B', ses.pack_hdr())
|
||||
#print(s)
|
||||
ioctl(_cryptodev, CIOCGSESSION2, s, 1)
|
||||
ses.unpack(s)
|
||||
|
||||
self._ses = ses.ses
|
||||
self._ses = ses.ses
|
||||
|
||||
def __del__(self):
|
||||
if self._ses is None:
|
||||
return
|
||||
def __del__(self):
|
||||
if self._ses is None:
|
||||
return
|
||||
|
||||
try:
|
||||
ioctl(_cryptodev, CIOCFSESSION, _pack('I', self._ses))
|
||||
except TypeError:
|
||||
pass
|
||||
self._ses = None
|
||||
try:
|
||||
ioctl(_cryptodev, CIOCFSESSION, _pack('I', self._ses))
|
||||
except TypeError:
|
||||
pass
|
||||
self._ses = None
|
||||
|
||||
def _doop(self, op, src, iv):
|
||||
cop = CryptOp()
|
||||
cop.ses = self._ses
|
||||
cop.op = op
|
||||
cop.flags = 0
|
||||
cop.len = len(src)
|
||||
s = array.array('B', src)
|
||||
cop.src = cop.dst = s.buffer_info()[0]
|
||||
if self._maclen is not None:
|
||||
m = array.array('B', [0] * self._maclen)
|
||||
cop.mac = m.buffer_info()[0]
|
||||
ivbuf = array.array('B', iv)
|
||||
cop.iv = ivbuf.buffer_info()[0]
|
||||
def _doop(self, op, src, iv):
|
||||
cop = CryptOp()
|
||||
cop.ses = self._ses
|
||||
cop.op = op
|
||||
cop.flags = 0
|
||||
cop.len = len(src)
|
||||
s = array.array('B', src)
|
||||
cop.src = cop.dst = s.buffer_info()[0]
|
||||
if self._maclen is not None:
|
||||
m = array.array('B', [0] * self._maclen)
|
||||
cop.mac = m.buffer_info()[0]
|
||||
ivbuf = array.array('B', iv)
|
||||
cop.iv = ivbuf.buffer_info()[0]
|
||||
|
||||
#print('cop:', cop)
|
||||
ioctl(_cryptodev, CIOCCRYPT, str(cop))
|
||||
#print('cop:', cop)
|
||||
ioctl(_cryptodev, CIOCCRYPT, str(cop))
|
||||
|
||||
s = s.tostring()
|
||||
if self._maclen is not None:
|
||||
return s, m.tostring()
|
||||
s = s.tostring()
|
||||
if self._maclen is not None:
|
||||
return s, m.tostring()
|
||||
|
||||
return s
|
||||
return s
|
||||
|
||||
def _doaead(self, op, src, aad, iv, tag=None):
|
||||
caead = CryptAEAD()
|
||||
caead.ses = self._ses
|
||||
caead.op = op
|
||||
caead.flags = CRD_F_IV_EXPLICIT
|
||||
caead.flags = 0
|
||||
caead.len = len(src)
|
||||
s = array.array('B', src)
|
||||
caead.src = caead.dst = s.buffer_info()[0]
|
||||
caead.aadlen = len(aad)
|
||||
saad = array.array('B', aad)
|
||||
caead.aad = saad.buffer_info()[0]
|
||||
def _doaead(self, op, src, aad, iv, tag=None):
|
||||
caead = CryptAEAD()
|
||||
caead.ses = self._ses
|
||||
caead.op = op
|
||||
caead.flags = CRD_F_IV_EXPLICIT
|
||||
caead.flags = 0
|
||||
caead.len = len(src)
|
||||
s = array.array('B', src)
|
||||
caead.src = caead.dst = s.buffer_info()[0]
|
||||
caead.aadlen = len(aad)
|
||||
saad = array.array('B', aad)
|
||||
caead.aad = saad.buffer_info()[0]
|
||||
|
||||
if self._maclen is None:
|
||||
raise ValueError('must have a tag length')
|
||||
if self._maclen is None:
|
||||
raise ValueError('must have a tag length')
|
||||
|
||||
if tag is None:
|
||||
tag = array.array('B', [0] * self._maclen)
|
||||
else:
|
||||
assert len(tag) == self._maclen, \
|
||||
if tag is None:
|
||||
tag = array.array('B', [0] * self._maclen)
|
||||
else:
|
||||
assert len(tag) == self._maclen, \
|
||||
'%d != %d' % (len(tag), self._maclen)
|
||||
tag = array.array('B', tag)
|
||||
tag = array.array('B', tag)
|
||||
|
||||
caead.tag = tag.buffer_info()[0]
|
||||
caead.tag = tag.buffer_info()[0]
|
||||
|
||||
ivbuf = array.array('B', iv)
|
||||
caead.ivlen = len(iv)
|
||||
caead.iv = ivbuf.buffer_info()[0]
|
||||
ivbuf = array.array('B', iv)
|
||||
caead.ivlen = len(iv)
|
||||
caead.iv = ivbuf.buffer_info()[0]
|
||||
|
||||
ioctl(_cryptodev, CIOCCRYPTAEAD, str(caead))
|
||||
ioctl(_cryptodev, CIOCCRYPTAEAD, str(caead))
|
||||
|
||||
s = s.tostring()
|
||||
s = s.tostring()
|
||||
|
||||
return s, tag.tostring()
|
||||
return s, tag.tostring()
|
||||
|
||||
def perftest(self, op, size, timeo=3):
|
||||
import random
|
||||
import time
|
||||
def perftest(self, op, size, timeo=3):
|
||||
import random
|
||||
import time
|
||||
|
||||
inp = array.array('B', (random.randint(0, 255) for x in xrange(size)))
|
||||
out = array.array('B', inp)
|
||||
inp = array.array('B', (random.randint(0, 255) for x in xrange(size)))
|
||||
out = array.array('B', inp)
|
||||
|
||||
# prep ioctl
|
||||
cop = CryptOp()
|
||||
cop.ses = self._ses
|
||||
cop.op = op
|
||||
cop.flags = 0
|
||||
cop.len = len(inp)
|
||||
s = array.array('B', inp)
|
||||
cop.src = s.buffer_info()[0]
|
||||
cop.dst = out.buffer_info()[0]
|
||||
if self._maclen is not None:
|
||||
m = array.array('B', [0] * self._maclen)
|
||||
cop.mac = m.buffer_info()[0]
|
||||
ivbuf = array.array('B', (random.randint(0, 255) for x in xrange(16)))
|
||||
cop.iv = ivbuf.buffer_info()[0]
|
||||
# prep ioctl
|
||||
cop = CryptOp()
|
||||
cop.ses = self._ses
|
||||
cop.op = op
|
||||
cop.flags = 0
|
||||
cop.len = len(inp)
|
||||
s = array.array('B', inp)
|
||||
cop.src = s.buffer_info()[0]
|
||||
cop.dst = out.buffer_info()[0]
|
||||
if self._maclen is not None:
|
||||
m = array.array('B', [0] * self._maclen)
|
||||
cop.mac = m.buffer_info()[0]
|
||||
ivbuf = array.array('B', (random.randint(0, 255) for x in xrange(16)))
|
||||
cop.iv = ivbuf.buffer_info()[0]
|
||||
|
||||
exit = [ False ]
|
||||
def alarmhandle(a, b, exit=exit):
|
||||
exit[0] = True
|
||||
exit = [ False ]
|
||||
def alarmhandle(a, b, exit=exit):
|
||||
exit[0] = True
|
||||
|
||||
oldalarm = signal.signal(signal.SIGALRM, alarmhandle)
|
||||
signal.alarm(timeo)
|
||||
oldalarm = signal.signal(signal.SIGALRM, alarmhandle)
|
||||
signal.alarm(timeo)
|
||||
|
||||
start = time.time()
|
||||
reps = 0
|
||||
while not exit[0]:
|
||||
ioctl(_cryptodev, CIOCCRYPT, str(cop))
|
||||
reps += 1
|
||||
start = time.time()
|
||||
reps = 0
|
||||
while not exit[0]:
|
||||
ioctl(_cryptodev, CIOCCRYPT, str(cop))
|
||||
reps += 1
|
||||
|
||||
end = time.time()
|
||||
end = time.time()
|
||||
|
||||
signal.signal(signal.SIGALRM, oldalarm)
|
||||
signal.signal(signal.SIGALRM, oldalarm)
|
||||
|
||||
print('time:', end - start)
|
||||
print('perf MB/sec:', (reps * size) / (end - start) / 1024 / 1024)
|
||||
print('time:', end - start)
|
||||
print('perf MB/sec:', (reps * size) / (end - start) / 1024 / 1024)
|
||||
|
||||
def encrypt(self, data, iv, aad=None):
|
||||
if aad is None:
|
||||
return self._doop(COP_ENCRYPT, data, iv)
|
||||
else:
|
||||
return self._doaead(COP_ENCRYPT, data, aad,
|
||||
iv)
|
||||
def encrypt(self, data, iv, aad=None):
|
||||
if aad is None:
|
||||
return self._doop(COP_ENCRYPT, data, iv)
|
||||
else:
|
||||
return self._doaead(COP_ENCRYPT, data, aad,
|
||||
iv)
|
||||
|
||||
def decrypt(self, data, iv, aad=None, tag=None):
|
||||
if aad is None:
|
||||
return self._doop(COP_DECRYPT, data, iv)
|
||||
else:
|
||||
return self._doaead(COP_DECRYPT, data, aad,
|
||||
iv, tag=tag)
|
||||
def decrypt(self, data, iv, aad=None, tag=None):
|
||||
if aad is None:
|
||||
return self._doop(COP_DECRYPT, data, iv)
|
||||
else:
|
||||
return self._doaead(COP_DECRYPT, data, aad,
|
||||
iv, tag=tag)
|
||||
|
||||
class MismatchError(Exception):
|
||||
pass
|
||||
pass
|
||||
|
||||
class KATParser:
|
||||
def __init__(self, fname, fields):
|
||||
self.fp = open(fname)
|
||||
self.fields = set(fields)
|
||||
self._pending = None
|
||||
def __init__(self, fname, fields):
|
||||
self.fp = open(fname)
|
||||
self.fields = set(fields)
|
||||
self._pending = None
|
||||
|
||||
def __iter__(self):
|
||||
while True:
|
||||
didread = False
|
||||
if self._pending is not None:
|
||||
i = self._pending
|
||||
self._pending = None
|
||||
else:
|
||||
i = self.fp.readline()
|
||||
didread = True
|
||||
def __iter__(self):
|
||||
while True:
|
||||
didread = False
|
||||
if self._pending is not None:
|
||||
i = self._pending
|
||||
self._pending = None
|
||||
else:
|
||||
i = self.fp.readline()
|
||||
didread = True
|
||||
|
||||
if didread and not i:
|
||||
return
|
||||
if didread and not i:
|
||||
return
|
||||
|
||||
if (i and i[0] == '#') or not i.strip():
|
||||
continue
|
||||
if i[0] == '[':
|
||||
yield i[1:].split(']', 1)[0], self.fielditer()
|
||||
else:
|
||||
raise ValueError('unknown line: %r' % repr(i))
|
||||
if (i and i[0] == '#') or not i.strip():
|
||||
continue
|
||||
if i[0] == '[':
|
||||
yield i[1:].split(']', 1)[0], self.fielditer()
|
||||
else:
|
||||
raise ValueError('unknown line: %r' % repr(i))
|
||||
|
||||
def eatblanks(self):
|
||||
while True:
|
||||
line = self.fp.readline()
|
||||
if line == '':
|
||||
break
|
||||
def eatblanks(self):
|
||||
while True:
|
||||
line = self.fp.readline()
|
||||
if line == '':
|
||||
break
|
||||
|
||||
line = line.strip()
|
||||
if line:
|
||||
break
|
||||
line = line.strip()
|
||||
if line:
|
||||
break
|
||||
|
||||
return line
|
||||
return line
|
||||
|
||||
def fielditer(self):
|
||||
while True:
|
||||
values = {}
|
||||
def fielditer(self):
|
||||
while True:
|
||||
values = {}
|
||||
|
||||
line = self.eatblanks()
|
||||
if not line or line[0] == '[':
|
||||
self._pending = line
|
||||
return
|
||||
line = self.eatblanks()
|
||||
if not line or line[0] == '[':
|
||||
self._pending = line
|
||||
return
|
||||
|
||||
while True:
|
||||
try:
|
||||
f, v = line.split(' =')
|
||||
except:
|
||||
if line == 'FAIL':
|
||||
f, v = 'FAIL', ''
|
||||
else:
|
||||
print('line:', repr(line))
|
||||
raise
|
||||
v = v.strip()
|
||||
while True:
|
||||
try:
|
||||
f, v = line.split(' =')
|
||||
except:
|
||||
if line == 'FAIL':
|
||||
f, v = 'FAIL', ''
|
||||
else:
|
||||
print('line:', repr(line))
|
||||
raise
|
||||
v = v.strip()
|
||||
|
||||
if f in values:
|
||||
raise ValueError('already present: %r' % repr(f))
|
||||
values[f] = v
|
||||
line = self.fp.readline().strip()
|
||||
if not line:
|
||||
break
|
||||
if f in values:
|
||||
raise ValueError('already present: %r' % repr(f))
|
||||
values[f] = v
|
||||
line = self.fp.readline().strip()
|
||||
if not line:
|
||||
break
|
||||
|
||||
# we should have everything
|
||||
remain = self.fields.copy() - set(values.keys())
|
||||
# XXX - special case GCM decrypt
|
||||
if remain and not ('FAIL' in values and 'PT' in remain):
|
||||
raise ValueError('not all fields found: %r' % repr(remain))
|
||||
# we should have everything
|
||||
remain = self.fields.copy() - set(values.keys())
|
||||
# XXX - special case GCM decrypt
|
||||
if remain and not ('FAIL' in values and 'PT' in remain):
|
||||
raise ValueError('not all fields found: %r' % repr(remain))
|
||||
|
||||
yield values
|
||||
yield values
|
||||
|
||||
def _spdechex(s):
|
||||
return ''.join(s.split()).decode('hex')
|
||||
return ''.join(s.split()).decode('hex')
|
||||
|
||||
if __name__ == '__main__':
|
||||
if True:
|
||||
try:
|
||||
crid = Crypto.findcrid('aesni0')
|
||||
print('aesni:', crid)
|
||||
except IOError:
|
||||
print('aesni0 not found')
|
||||
if True:
|
||||
try:
|
||||
crid = Crypto.findcrid('aesni0')
|
||||
print('aesni:', crid)
|
||||
except IOError:
|
||||
print('aesni0 not found')
|
||||
|
||||
for i in xrange(10):
|
||||
try:
|
||||
name = Crypto.getcridname(i)
|
||||
print('%2d: %r' % (i, repr(name)))
|
||||
except IOError:
|
||||
pass
|
||||
elif False:
|
||||
kp = KATParser('/usr/home/jmg/aesni.testing/format tweak value input - data unit seq no/XTSGenAES128.rsp', [ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT', 'CT' ])
|
||||
for mode, ni in kp:
|
||||
print(i, ni)
|
||||
for j in ni:
|
||||
print(j)
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('00000000000000000000000000000001')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e')
|
||||
#pt = _spdechex('00000000000000000000000000000000')
|
||||
ct = _spdechex('f42c33853ecc5ce2949865fdb83de3bff1089e9360c94f830baebfaff72836ab5236f77212f1e7396c8c54ac73d81986375a6e9e299cfeca5ba051ed25e8d1affa5beaf6c1d2b45e90802408f2ced21663497e906de5f29341e5e52ddfea5363d628b3eb7806835e17bae051b3a6da3f8e2941fe44384eac17a9d298d2c331ca8320c775b5d53263a5e905059d891b21dede2d8110fd427c7bd5a9a274ddb47b1945ee79522203b6e297d0e399ef')
|
||||
for i in xrange(10):
|
||||
try:
|
||||
name = Crypto.getcridname(i)
|
||||
print('%2d: %r' % (i, repr(name)))
|
||||
except IOError:
|
||||
pass
|
||||
elif False:
|
||||
kp = KATParser('/usr/home/jmg/aesni.testing/format tweak value input - data unit seq no/XTSGenAES128.rsp', [ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT', 'CT' ])
|
||||
for mode, ni in kp:
|
||||
print(i, ni)
|
||||
for j in ni:
|
||||
print(j)
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('00000000000000000000000000000001')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e')
|
||||
#pt = _spdechex('00000000000000000000000000000000')
|
||||
ct = _spdechex('f42c33853ecc5ce2949865fdb83de3bff1089e9360c94f830baebfaff72836ab5236f77212f1e7396c8c54ac73d81986375a6e9e299cfeca5ba051ed25e8d1affa5beaf6c1d2b45e90802408f2ced21663497e906de5f29341e5e52ddfea5363d628b3eb7806835e17bae051b3a6da3f8e2941fe44384eac17a9d298d2c331ca8320c775b5d53263a5e905059d891b21dede2d8110fd427c7bd5a9a274ddb47b1945ee79522203b6e297d0e399ef')
|
||||
|
||||
c = Crypto(CRYPTO_AES_ICM, key)
|
||||
enc = c.encrypt(pt, iv)
|
||||
c = Crypto(CRYPTO_AES_ICM, key)
|
||||
enc = c.encrypt(pt, iv)
|
||||
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
|
||||
assert ct == enc
|
||||
assert ct == enc
|
||||
|
||||
dec = c.decrypt(ct, iv)
|
||||
dec = c.decrypt(ct, iv)
|
||||
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('00000000000000000000000000000001')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e0a3f')
|
||||
#pt = _spdechex('00000000000000000000000000000000')
|
||||
ct = _spdechex('f42c33853ecc5ce2949865fdb83de3bff1089e9360c94f830baebfaff72836ab5236f77212f1e7396c8c54ac73d81986375a6e9e299cfeca5ba051ed25e8d1affa5beaf6c1d2b45e90802408f2ced21663497e906de5f29341e5e52ddfea5363d628b3eb7806835e17bae051b3a6da3f8e2941fe44384eac17a9d298d2c331ca8320c775b5d53263a5e905059d891b21dede2d8110fd427c7bd5a9a274ddb47b1945ee79522203b6e297d0e399ef3768')
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('00000000000000000000000000000001')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e0a3f')
|
||||
#pt = _spdechex('00000000000000000000000000000000')
|
||||
ct = _spdechex('f42c33853ecc5ce2949865fdb83de3bff1089e9360c94f830baebfaff72836ab5236f77212f1e7396c8c54ac73d81986375a6e9e299cfeca5ba051ed25e8d1affa5beaf6c1d2b45e90802408f2ced21663497e906de5f29341e5e52ddfea5363d628b3eb7806835e17bae051b3a6da3f8e2941fe44384eac17a9d298d2c331ca8320c775b5d53263a5e905059d891b21dede2d8110fd427c7bd5a9a274ddb47b1945ee79522203b6e297d0e399ef3768')
|
||||
|
||||
c = Crypto(CRYPTO_AES_ICM, key)
|
||||
enc = c.encrypt(pt, iv)
|
||||
c = Crypto(CRYPTO_AES_ICM, key)
|
||||
enc = c.encrypt(pt, iv)
|
||||
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
|
||||
assert ct == enc
|
||||
assert ct == enc
|
||||
|
||||
dec = c.decrypt(ct, iv)
|
||||
dec = c.decrypt(ct, iv)
|
||||
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('6eba2716ec0bd6fa5cdef5e6d3a795bc')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e0a3f')
|
||||
ct = _spdechex('f1f81f12e72e992dbdc304032705dc75dc3e4180eff8ee4819906af6aee876d5b00b7c36d282a445ce3620327be481e8e53a8e5a8e5ca9abfeb2281be88d12ffa8f46d958d8224738c1f7eea48bda03edbf9adeb900985f4fa25648b406d13a886c25e70cfdecdde0ad0f2991420eb48a61c64fd797237cf2798c2675b9bb744360b0a3f329ac53bbceb4e3e7456e6514f1a9d2f06c236c31d0f080b79c15dce1096357416602520daa098b17d1af427')
|
||||
c = Crypto(CRYPTO_AES_CBC, key)
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('6eba2716ec0bd6fa5cdef5e6d3a795bc')
|
||||
pt = _spdechex('ab3cabed693a32946055524052afe3c9cb49664f09fc8b7da824d924006b7496353b8c1657c5dec564d8f38d7432e1de35aae9d95590e66278d4acce883e51abaf94977fcd3679660109a92bf7b2973ccd547f065ec6cee4cb4a72a5e9f45e615d920d76cb34cba482467b3e21422a7242e7d931330c0fbf465c3a3a46fae943029fd899626dda542750a1eee253df323c6ef1573f1c8c156613e2ea0a6cdbf2ae9701020be2d6a83ecb7f3f9d8e0a3f')
|
||||
ct = _spdechex('f1f81f12e72e992dbdc304032705dc75dc3e4180eff8ee4819906af6aee876d5b00b7c36d282a445ce3620327be481e8e53a8e5a8e5ca9abfeb2281be88d12ffa8f46d958d8224738c1f7eea48bda03edbf9adeb900985f4fa25648b406d13a886c25e70cfdecdde0ad0f2991420eb48a61c64fd797237cf2798c2675b9bb744360b0a3f329ac53bbceb4e3e7456e6514f1a9d2f06c236c31d0f080b79c15dce1096357416602520daa098b17d1af427')
|
||||
c = Crypto(CRYPTO_AES_CBC, key)
|
||||
|
||||
enc = c.encrypt(pt, iv)
|
||||
enc = c.encrypt(pt, iv)
|
||||
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
|
||||
assert ct == enc
|
||||
assert ct == enc
|
||||
|
||||
dec = c.decrypt(ct, iv)
|
||||
dec = c.decrypt(ct, iv)
|
||||
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('b3d8cc017cbb89b39e0f67e2')
|
||||
pt = _spdechex('c3b3c41f113a31b73d9a5cd4321030')
|
||||
aad = _spdechex('24825602bd12a984e0092d3e448eda5f')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa7354')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa73')
|
||||
tag = _spdechex('0032a1dc85f1c9786925a2e71d8272dd')
|
||||
tag = _spdechex('8d11a0929cb3fbe1fef01a4a38d5f8ea')
|
||||
assert pt == dec
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('b3d8cc017cbb89b39e0f67e2')
|
||||
pt = _spdechex('c3b3c41f113a31b73d9a5cd4321030')
|
||||
aad = _spdechex('24825602bd12a984e0092d3e448eda5f')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa7354')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa73')
|
||||
tag = _spdechex('0032a1dc85f1c9786925a2e71d8272dd')
|
||||
tag = _spdechex('8d11a0929cb3fbe1fef01a4a38d5f8ea')
|
||||
|
||||
c = Crypto(CRYPTO_AES_NIST_GCM_16, key,
|
||||
mac=CRYPTO_AES_128_NIST_GMAC, mackey=key)
|
||||
c = Crypto(CRYPTO_AES_NIST_GCM_16, key,
|
||||
mac=CRYPTO_AES_128_NIST_GMAC, mackey=key)
|
||||
|
||||
enc, enctag = c.encrypt(pt, iv, aad=aad)
|
||||
enc, enctag = c.encrypt(pt, iv, aad=aad)
|
||||
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
|
||||
assert enc == ct
|
||||
assert enc == ct
|
||||
|
||||
print('etg:', enctag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
assert enctag == tag
|
||||
print('etg:', enctag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
assert enctag == tag
|
||||
|
||||
# Make sure we get EBADMSG
|
||||
#enctag = enctag[:-1] + 'a'
|
||||
dec, dectag = c.decrypt(ct, iv, aad=aad, tag=enctag)
|
||||
# Make sure we get EBADMSG
|
||||
#enctag = enctag[:-1] + 'a'
|
||||
dec, dectag = c.decrypt(ct, iv, aad=aad, tag=enctag)
|
||||
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
print('dec:', dec.encode('hex'))
|
||||
print(' pt:', pt.encode('hex'))
|
||||
|
||||
assert dec == pt
|
||||
assert dec == pt
|
||||
|
||||
print('dtg:', dectag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
print('dtg:', dectag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
|
||||
assert dectag == tag
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('b3d8cc017cbb89b39e0f67e2')
|
||||
key = key + iv[:4]
|
||||
iv = iv[4:]
|
||||
pt = _spdechex('c3b3c41f113a31b73d9a5cd432103069')
|
||||
aad = _spdechex('24825602bd12a984e0092d3e448eda5f')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa7354')
|
||||
tag = _spdechex('0032a1dc85f1c9786925a2e71d8272dd')
|
||||
assert dectag == tag
|
||||
elif False:
|
||||
key = _spdechex('c939cc13397c1d37de6ae0e1cb7c423c')
|
||||
iv = _spdechex('b3d8cc017cbb89b39e0f67e2')
|
||||
key = key + iv[:4]
|
||||
iv = iv[4:]
|
||||
pt = _spdechex('c3b3c41f113a31b73d9a5cd432103069')
|
||||
aad = _spdechex('24825602bd12a984e0092d3e448eda5f')
|
||||
ct = _spdechex('93fe7d9e9bfd10348a5606e5cafa7354')
|
||||
tag = _spdechex('0032a1dc85f1c9786925a2e71d8272dd')
|
||||
|
||||
c = Crypto(CRYPTO_AES_GCM_16, key, mac=CRYPTO_AES_128_GMAC, mackey=key)
|
||||
c = Crypto(CRYPTO_AES_GCM_16, key, mac=CRYPTO_AES_128_GMAC, mackey=key)
|
||||
|
||||
enc, enctag = c.encrypt(pt, iv, aad=aad)
|
||||
enc, enctag = c.encrypt(pt, iv, aad=aad)
|
||||
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
print('enc:', enc.encode('hex'))
|
||||
print(' ct:', ct.encode('hex'))
|
||||
|
||||
assert enc == ct
|
||||
assert enc == ct
|
||||
|
||||
print('etg:', enctag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
assert enctag == tag
|
||||
elif False:
|
||||
for i in xrange(100000):
|
||||
c = Crypto(CRYPTO_AES_XTS, '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex'))
|
||||
data = '52a42bca4e9425a25bbc8c8bf6129dec'.decode('hex')
|
||||
ct = '517e602becd066b65fa4f4f56ddfe240'.decode('hex')
|
||||
iv = _pack('QQ', 71, 0)
|
||||
print('etg:', enctag.encode('hex'))
|
||||
print('tag:', tag.encode('hex'))
|
||||
assert enctag == tag
|
||||
elif False:
|
||||
for i in xrange(100000):
|
||||
c = Crypto(CRYPTO_AES_XTS, '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex'))
|
||||
data = '52a42bca4e9425a25bbc8c8bf6129dec'.decode('hex')
|
||||
ct = '517e602becd066b65fa4f4f56ddfe240'.decode('hex')
|
||||
iv = _pack('QQ', 71, 0)
|
||||
|
||||
enc = c.encrypt(data, iv)
|
||||
assert enc == ct
|
||||
elif True:
|
||||
c = Crypto(CRYPTO_AES_XTS, '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex'))
|
||||
data = '52a42bca4e9425a25bbc8c8bf6129dec'.decode('hex')
|
||||
ct = '517e602becd066b65fa4f4f56ddfe240'.decode('hex')
|
||||
iv = _pack('QQ', 71, 0)
|
||||
enc = c.encrypt(data, iv)
|
||||
assert enc == ct
|
||||
elif True:
|
||||
c = Crypto(CRYPTO_AES_XTS, '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex'))
|
||||
data = '52a42bca4e9425a25bbc8c8bf6129dec'.decode('hex')
|
||||
ct = '517e602becd066b65fa4f4f56ddfe240'.decode('hex')
|
||||
iv = _pack('QQ', 71, 0)
|
||||
|
||||
enc = c.encrypt(data, iv)
|
||||
assert enc == ct
|
||||
enc = c.encrypt(data, iv)
|
||||
assert enc == ct
|
||||
|
||||
dec = c.decrypt(enc, iv)
|
||||
assert dec == data
|
||||
dec = c.decrypt(enc, iv)
|
||||
assert dec == data
|
||||
|
||||
#c.perftest(COP_ENCRYPT, 192*1024, reps=30000)
|
||||
#c.perftest(COP_ENCRYPT, 192*1024, reps=30000)
|
||||
|
||||
else:
|
||||
key = '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex')
|
||||
print('XTS %d testing:' % (len(key) * 8))
|
||||
c = Crypto(CRYPTO_AES_XTS, key)
|
||||
for i in [ 8192, 192*1024]:
|
||||
print('block size: %d' % i)
|
||||
c.perftest(COP_ENCRYPT, i)
|
||||
c.perftest(COP_DECRYPT, i)
|
||||
else:
|
||||
key = '1bbfeadf539daedcae33ced497343f3ca1f2474ad932b903997d44707db41382'.decode('hex')
|
||||
print('XTS %d testing:' % (len(key) * 8))
|
||||
c = Crypto(CRYPTO_AES_XTS, key)
|
||||
for i in [ 8192, 192*1024]:
|
||||
print('block size: %d' % i)
|
||||
c.perftest(COP_ENCRYPT, i)
|
||||
c.perftest(COP_DECRYPT, i)
|
||||
|
||||
@ -42,290 +42,290 @@
|
||||
katdir = '/usr/local/share/nist-kat'
|
||||
|
||||
def katg(base, glob):
|
||||
assert os.path.exists(os.path.join(katdir, base)), "Please 'pkg install nist-kat'"
|
||||
return iglob(os.path.join(katdir, base, glob))
|
||||
assert os.path.exists(os.path.join(katdir, base)), "Please 'pkg install nist-kat'"
|
||||
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
|
||||
try:
|
||||
crid = cryptodev.Crypto.findcrid(cname)
|
||||
except IOError:
|
||||
return None
|
||||
|
||||
class GendCryptoTestCase(unittest.TestCase):
|
||||
###############
|
||||
##### AES #####
|
||||
###############
|
||||
@unittest.skipIf(cname not in aesmodules, 'skipping AES 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)
|
||||
class GendCryptoTestCase(unittest.TestCase):
|
||||
###############
|
||||
##### AES #####
|
||||
###############
|
||||
@unittest.skipIf(cname not in aesmodules, 'skipping AES 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 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 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 on %s' % (cname))
|
||||
def test_gcm(self):
|
||||
for i in katg('gcmtestvectors', 'gcmEncrypt*'):
|
||||
self.runGCM(i, 'ENCRYPT')
|
||||
@unittest.skipIf(cname not in aesmodules, 'skipping AES 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')
|
||||
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))
|
||||
_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))
|
||||
|
||||
for bogusmode, lines in cryptodev.KATParser(fname,
|
||||
[ 'Count', 'Key', 'IV', 'CT', 'AAD', 'Tag', 'PT', ]):
|
||||
for data in lines:
|
||||
curcnt = int(data['Count'])
|
||||
cipherkey = data['Key'].decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
aad = data['AAD'].decode('hex')
|
||||
tag = data['Tag'].decode('hex')
|
||||
if 'FAIL' not in data:
|
||||
pt = data['PT'].decode('hex')
|
||||
ct = data['CT'].decode('hex')
|
||||
for bogusmode, lines in cryptodev.KATParser(fname,
|
||||
[ 'Count', 'Key', 'IV', 'CT', 'AAD', 'Tag', 'PT', ]):
|
||||
for data in lines:
|
||||
curcnt = int(data['Count'])
|
||||
cipherkey = data['Key'].decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
aad = data['AAD'].decode('hex')
|
||||
tag = data['Tag'].decode('hex')
|
||||
if 'FAIL' not in data:
|
||||
pt = data['PT'].decode('hex')
|
||||
ct = data['CT'].decode('hex')
|
||||
|
||||
if len(iv) != 12:
|
||||
# XXX - isn't supported
|
||||
continue
|
||||
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)
|
||||
except EnvironmentError, e:
|
||||
# Can't test algorithms the driver does not support.
|
||||
if e.errno != errno.EOPNOTSUPP:
|
||||
raise
|
||||
continue
|
||||
try:
|
||||
c = Crypto(cryptodev.CRYPTO_AES_NIST_GCM_16,
|
||||
cipherkey,
|
||||
mac=self._gmacsizes[len(cipherkey)],
|
||||
mackey=cipherkey, crid=crid)
|
||||
except EnvironmentError, 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, e:
|
||||
# Can't test inputs the driver does not support.
|
||||
if e.errno != errno.EINVAL:
|
||||
raise
|
||||
continue
|
||||
rtag = rtag[:len(tag)]
|
||||
data['rct'] = rct.encode('hex')
|
||||
data['rtag'] = rtag.encode('hex')
|
||||
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, e:
|
||||
# Can't test inputs the driver does not support.
|
||||
if e.errno != errno.EINVAL:
|
||||
raise
|
||||
continue
|
||||
data['rpt'] = rpt.encode('hex')
|
||||
data['rtag'] = rtag.encode('hex')
|
||||
self.assertEqual(rpt, pt,
|
||||
repr(data))
|
||||
if mode == 'ENCRYPT':
|
||||
try:
|
||||
rct, rtag = c.encrypt(pt, iv, aad)
|
||||
except EnvironmentError, e:
|
||||
# Can't test inputs the driver does not support.
|
||||
if e.errno != errno.EINVAL:
|
||||
raise
|
||||
continue
|
||||
rtag = rtag[:len(tag)]
|
||||
data['rct'] = rct.encode('hex')
|
||||
data['rtag'] = rtag.encode('hex')
|
||||
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, e:
|
||||
# Can't test inputs the driver does not support.
|
||||
if e.errno != errno.EINVAL:
|
||||
raise
|
||||
continue
|
||||
data['rpt'] = rpt.encode('hex')
|
||||
data['rtag'] = rtag.encode('hex')
|
||||
self.assertEqual(rpt, pt,
|
||||
repr(data))
|
||||
|
||||
def runCBC(self, fname):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'KEY', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]):
|
||||
if mode == 'ENCRYPT':
|
||||
swapptct = False
|
||||
curfun = Crypto.encrypt
|
||||
elif mode == 'DECRYPT':
|
||||
swapptct = True
|
||||
curfun = Crypto.decrypt
|
||||
else:
|
||||
raise RuntimeError('unknown mode: %r' % repr(mode))
|
||||
def runCBC(self, fname):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'KEY', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]):
|
||||
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 = data['KEY'].decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
pt = data['PLAINTEXT'].decode('hex')
|
||||
ct = data['CIPHERTEXT'].decode('hex')
|
||||
for data in lines:
|
||||
curcnt = int(data['COUNT'])
|
||||
cipherkey = data['KEY'].decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
pt = data['PLAINTEXT'].decode('hex')
|
||||
ct = data['CIPHERTEXT'].decode('hex')
|
||||
|
||||
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)
|
||||
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):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT',
|
||||
'CT' ]):
|
||||
if mode == 'ENCRYPT':
|
||||
swapptct = False
|
||||
curfun = Crypto.encrypt
|
||||
elif mode == 'DECRYPT':
|
||||
swapptct = True
|
||||
curfun = Crypto.decrypt
|
||||
else:
|
||||
raise RuntimeError('unknown mode: %r' % repr(mode))
|
||||
def runXTS(self, fname, meth):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'DataUnitLen', 'Key', 'DataUnitSeqNumber', 'PT',
|
||||
'CT' ]):
|
||||
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 = data['Key'].decode('hex')
|
||||
iv = struct.pack('QQ', int(data['DataUnitSeqNumber']), 0)
|
||||
pt = data['PT'].decode('hex')
|
||||
ct = data['CT'].decode('hex')
|
||||
for data in lines:
|
||||
curcnt = int(data['COUNT'])
|
||||
nbits = int(data['DataUnitLen'])
|
||||
cipherkey = data['Key'].decode('hex')
|
||||
iv = struct.pack('QQ', int(data['DataUnitSeqNumber']), 0)
|
||||
pt = data['PT'].decode('hex')
|
||||
ct = data['CT'].decode('hex')
|
||||
|
||||
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, e:
|
||||
# Can't test hashes the driver does not support.
|
||||
if e.errno != errno.EOPNOTSUPP:
|
||||
raise
|
||||
continue
|
||||
self.assertEqual(r, 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, e:
|
||||
# Can't test hashes the driver does not support.
|
||||
if e.errno != errno.EOPNOTSUPP:
|
||||
raise
|
||||
continue
|
||||
self.assertEqual(r, ct)
|
||||
|
||||
###############
|
||||
##### 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)
|
||||
###############
|
||||
##### 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):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'KEYs', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]):
|
||||
if mode == 'ENCRYPT':
|
||||
swapptct = False
|
||||
curfun = Crypto.encrypt
|
||||
elif mode == 'DECRYPT':
|
||||
swapptct = True
|
||||
curfun = Crypto.decrypt
|
||||
else:
|
||||
raise RuntimeError('unknown mode: %r' % repr(mode))
|
||||
def runTDES(self, fname):
|
||||
curfun = None
|
||||
for mode, lines in cryptodev.KATParser(fname,
|
||||
[ 'COUNT', 'KEYs', 'IV', 'PLAINTEXT', 'CIPHERTEXT', ]):
|
||||
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 = key.decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
pt = data['PLAINTEXT'].decode('hex')
|
||||
ct = data['CIPHERTEXT'].decode('hex')
|
||||
for data in lines:
|
||||
curcnt = int(data['COUNT'])
|
||||
key = data['KEYs'] * 3
|
||||
cipherkey = key.decode('hex')
|
||||
iv = data['IV'].decode('hex')
|
||||
pt = data['PLAINTEXT'].decode('hex')
|
||||
ct = data['CIPHERTEXT'].decode('hex')
|
||||
|
||||
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)
|
||||
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):
|
||||
# SHA not available in software
|
||||
pass
|
||||
#for i in iglob('SHA1*'):
|
||||
# self.runSHA(i)
|
||||
###############
|
||||
##### SHA #####
|
||||
###############
|
||||
@unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
|
||||
def test_sha(self):
|
||||
# SHA not available in software
|
||||
pass
|
||||
#for i in iglob('SHA1*'):
|
||||
# self.runSHA(i)
|
||||
|
||||
@unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
|
||||
def test_sha1hmac(self):
|
||||
for i in katg('hmactestvectors', 'HMAC.rsp'):
|
||||
self.runSHA1HMAC(i)
|
||||
@unittest.skipIf(cname not in shamodules, 'skipping SHA on %s' % str(cname))
|
||||
def test_sha1hmac(self):
|
||||
for i in katg('hmactestvectors', 'HMAC.rsp'):
|
||||
self.runSHA1HMAC(i)
|
||||
|
||||
def runSHA1HMAC(self, fname):
|
||||
for hashlength, lines in cryptodev.KATParser(fname,
|
||||
[ 'Count', 'Klen', 'Tlen', 'Key', 'Msg', 'Mac' ]):
|
||||
# E.g., hashlength will be "L=20" (bytes)
|
||||
hashlen = int(hashlength.split("=")[1])
|
||||
def runSHA1HMAC(self, fname):
|
||||
for hashlength, lines in cryptodev.KATParser(fname,
|
||||
[ 'Count', 'Klen', 'Tlen', 'Key', 'Msg', 'Mac' ]):
|
||||
# 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
|
||||
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 = data['Key'].decode('hex')
|
||||
msg = data['Msg'].decode('hex')
|
||||
mac = data['Mac'].decode('hex')
|
||||
tlen = int(data['Tlen'])
|
||||
for data in lines:
|
||||
key = data['Key'].decode('hex')
|
||||
msg = data['Msg'].decode('hex')
|
||||
mac = data['Mac'].decode('hex')
|
||||
tlen = int(data['Tlen'])
|
||||
|
||||
if len(key) > blocksize:
|
||||
continue
|
||||
if len(key) > blocksize:
|
||||
continue
|
||||
|
||||
try:
|
||||
c = Crypto(mac=alg, mackey=key,
|
||||
crid=crid)
|
||||
except EnvironmentError, e:
|
||||
# Can't test hashes the driver does not support.
|
||||
if e.errno != errno.EOPNOTSUPP:
|
||||
raise
|
||||
continue
|
||||
try:
|
||||
c = Crypto(mac=alg, mackey=key,
|
||||
crid=crid)
|
||||
except EnvironmentError, e:
|
||||
# Can't test hashes the driver does not support.
|
||||
if e.errno != errno.EOPNOTSUPP:
|
||||
raise
|
||||
continue
|
||||
|
||||
_, r = c.encrypt(msg, iv="")
|
||||
_, r = c.encrypt(msg, iv="")
|
||||
|
||||
# A limitation in cryptodev.py means we
|
||||
# can only store MACs up to 16 bytes.
|
||||
# That's good enough to validate the
|
||||
# correct behavior, more or less.
|
||||
maclen = min(tlen, 16)
|
||||
self.assertEqual(r[:maclen], mac[:maclen], "Actual: " + \
|
||||
repr(r[:maclen].encode("hex")) + " Expected: " + repr(data))
|
||||
# A limitation in cryptodev.py means we
|
||||
# can only store MACs up to 16 bytes.
|
||||
# That's good enough to validate the
|
||||
# correct behavior, more or less.
|
||||
maclen = min(tlen, 16)
|
||||
self.assertEqual(r[:maclen], mac[:maclen], "Actual: " + \
|
||||
repr(r[:maclen].encode("hex")) + " Expected: " + repr(data))
|
||||
|
||||
return GendCryptoTestCase
|
||||
return GendCryptoTestCase
|
||||
|
||||
cryptosoft = GenTestCase('cryptosoft0')
|
||||
aesni = GenTestCase('aesni0')
|
||||
@ -333,4 +333,4 @@ def runSHA1HMAC(self, fname):
|
||||
ccp = GenTestCase('ccp0')
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
unittest.main()
|
||||
|
||||
Loading…
Reference in New Issue
Block a user