Commit 4f66bd7f by lwc-tester

new test for bluepill and esp32 templates

parent d0f51b64
...@@ -21,6 +21,6 @@ source [find target/stm32f1x.cfg] ...@@ -21,6 +21,6 @@ source [find target/stm32f1x.cfg]
#tpiu config internal swodump.stm32f103-generic.log uart off 72000000 #tpiu config internal swodump.stm32f103-generic.log uart off 72000000
#reset_config srst_only srst_push_pull srst_nogate connect_assert_srst reset_config srst_only srst_push_pull srst_nogate connect_assert_srst
reset_config none srst_push_pull srst_nogate #reset_config none srst_push_pull srst_nogate
...@@ -2,197 +2,128 @@ ...@@ -2,197 +2,128 @@
import os import os
import sys import sys
import time
import struct
import serial
import subprocess import subprocess
import serial.tools.list_ports
from test_common import (
LogicMultiplexerTimeMeasurements,
parse_nist_aead_test_vectors,
DeviceUnderTestAeadUARTP,
compare_dumps,
eprint,
run_nist_aead_test_line,
)
def get_serial():
def eprint(*args, **kargs): ports = serial.tools.list_ports.comports()
print(*args, file=sys.stderr, **kargs) for port in ports:
print(port.serial_number)
devices = [
def flash(): p.device
pipe = subprocess.PIPE for p in ports
cmd = ['openocd', '-f', 'openocd.cfg', '-c' 'program ' + if p.serial_number == 'FT2XCRZ1'
'.pio/build/bluepill_f103c8/firmware.elf verify reset exit'] ]
p = subprocess.Popen(cmd, devices.sort()
stdout=sys.stderr, stdin=sys.stdout) return serial.Serial(
devices[0],
baudrate=115200,
timeout=5)
class BluePill(DeviceUnderTestAeadUARTP):
RAM_SIZE = 0x5000
def __init__(self, build_dir):
DeviceUnderTestAeadUARTP.__init__(self, get_serial())
self.firmware_path = os.path.join(
build_dir, '.pio/build/bluepill_f103c8/firmware.elf')
self.ram_pattern_path = os.path.join(
build_dir, 'empty_ram.bin')
self.ram_dump_path = os.path.join(
build_dir, 'ram_dump.bin')
self.openocd_cfg_path = os.path.join(
build_dir, 'openocd.cfg')
def flash(self):
# pipe = subprocess.PIPE
cmd = [
'openocd', '-f', 'openocd.cfg', '-c',
'program %s verify reset exit' % self.firmware_path]
p = subprocess.Popen(
cmd, stdout=sys.stderr, stdin=sys.stdout)
stdout, stderr = p.communicate("") stdout, stderr = p.communicate("")
eprint("Firmware flashed.")
cmd = [
def fill_ram(): 'openocd', '-f', self.openocd_cfg_path, '-c',
pipe = subprocess.PIPE 'program %s reset exit 0x20000000' % self.ram_pattern_path]
cmd = ['openocd', '-f', 'openocd.cfg', '-c' 'program ' + p = subprocess.Popen(
'empty_ram.bin reset exit 0x20000000'] cmd, stdout=sys.stderr, stdin=sys.stdout)
p = subprocess.Popen(cmd,
stdout=sys.stderr, stdin=sys.stdout)
stdout, stderr = p.communicate("") stdout, stderr = p.communicate("")
eprint("RAM flashed.")
def dump_ram(self):
cmd = [
'openocd', '-f', self.openocd_cfg_path,
'-c', 'init',
'-c', 'halt',
'-c', 'dump_image %s 0x20000000 0x%x' % (
self.ram_dump_path, BluePill.RAM_SIZE),
'-c', 'resume',
'-c', 'exit']
p = subprocess.Popen(
cmd, stdout=sys.stderr, stdin=sys.stdout)
stdout, stderr = p.communicate("")
eprint("RAM dumped.")
with open(self.ram_dump_path, 'rb') as ram:
ram = ram.read()
if len(ram) != BluePill.RAM_SIZE:
raise Exception(
"RAM dump was %d bytes instead of %d" %
(len(ram), BluePill.RAM_SIZE))
return ram
def get_serial():
import serial.tools.list_ports
ports = serial.tools.list_ports.comports()
devices = [ p.device for p in ports ]
devices.sort()
return devices[-1]
class UARTP:
def __init__(self, ser):
UARTP.SYN = 0xf9
UARTP.FIN = 0xf3
self.ser = ser
def uart_read(self):
r = self.ser.read(1)
if len(r) != 1:
raise Exception("Serial read error")
return r[0]
def uart_write(self, c):
b = struct.pack("B", c)
r = self.ser.write(b)
if r != len(b):
raise Exception("Serial write error")
return r
def send(self, buf):
self.uart_write(UARTP.SYN)
len_ind_0 = 0xff & len(buf)
len_ind_1 = 0xff & (len(buf) >> 7)
if len(buf) < 128:
self.uart_write(len_ind_0)
else:
self.uart_write(len_ind_0 | 0x80)
self.uart_write(len_ind_1)
fcs = 0
for i in range(len(buf)):
info = buf[i]
fcs = (fcs + info) & 0xff
self.uart_write(buf[i])
fcs = (0xff - fcs) & 0xff
self.uart_write(fcs)
self.uart_write(UARTP.FIN)
eprint("sent frame '%s'" % buf.hex())
def recv(self):
tag_old = UARTP.FIN
while 1:
tag = tag_old
while 1:
if tag_old == UARTP.FIN:
if tag == UARTP.SYN:
break
tag_old = tag
tag = self.uart_read()
tag_old = tag
l = self.uart_read()
if l & 0x80:
l &= 0x7f
l |= self.uart_read() << 7
fcs = 0
buf = []
for i in range(l):
info = self.uart_read()
buf.append(info)
fcs = (fcs + info) & 0xff
fcs = (fcs + self.uart_read()) & 0xff
tag = self.uart_read()
if fcs == 0xff:
if tag == UARTP.FIN:
buf = bytes(buf)
eprint("rcvd frame '%s'" % buf.hex())
if len(buf) >= 1 and buf[0] == 0xde:
sys.stderr.buffer.write(buf[1:])
sys.stderr.flush()
else:
return buf
def main(argv): def main(argv):
eprint(argv[0]) if len(argv) != 3:
script_dir = os.path.split(argv[0])[0] print("Usage: test LWC_AEAD_KAT.txt build_dir")
if len(script_dir) > 0:
os.chdir(script_dir)
dev = get_serial()
ser = serial.Serial(dev, baudrate=115200, timeout=5)
uartp = UARTP(ser)
flash()
fill_ram()
eprint("Flashed")
time.sleep(0.1)
ser.setDTR(False) # IO0=HIGH
ser.setRTS(True) # EN=LOW, chip in reset
time.sleep(0.1)
ser.setDTR(False) # IO0=HIGH
ser.setRTS(False) # EN=HIGH, chip out of reset
time.sleep(1)
def stdin_read(n):
b = sys.stdin.buffer.read(n)
if len(b) != n:
sys.exit(1)
return b
def stdin_readvar():
l = stdin_read(4)
(l, ) = struct.unpack("<I", l)
v = stdin_read(l)
return v
exp_hello = b"Hello, World!"
hello = ser.read(ser.in_waiting)
if hello[-13:] != exp_hello:
eprint("Improper board initialization message: %s" % hello)
return 1 return 1
eprint("Board initialized properly")
sys.stdout.write("Hello, World!\n")
sys.stdout.flush()
while 1: kat = list(parse_nist_aead_test_vectors(argv[1]))
action = stdin_read(1)[0] build_dir = argv[2]
eprint("Command %c from stdin" % action)
dut = BluePill(build_dir)
if action in b"ackmps":
v = stdin_readvar() try:
uartp.send(struct.pack("B", action) + v) tool = LogicMultiplexerTimeMeasurements(0x0003)
ack = uartp.recv() tool.begin_measurement()
if len(ack) != 1 or ack[0] != action:
raise Exception("Unacknowledged variable transfer") dut.flash()
eprint("Var %c successfully sent to board" % action) dut.prepare()
sys.stdout.write("Board prepared\n")
elif action in b"ACKMPS":
c = struct.pack("B", action)
uartp.send(c)
v = uartp.recv()
if len(v) < 1 or v[0] != action:
raise Exception("Could not obtain variable from board")
v = v[1:]
eprint("Var %c received from board: %s" % (action, v.hex()))
l = struct.pack("<I", len(v))
sys.stdout.buffer.write(l + v)
sys.stdout.flush() sys.stdout.flush()
elif action in b"ed": dump_a = dut.dump_ram()
c = struct.pack("B", action)
uartp.send(c) for i, m, ad, k, npub, c in kat:
ack = uartp.recv() tool.arm()
if len(ack) != 1 or ack[0] != action: run_nist_aead_test_line(dut, i, m, ad, k, npub, c)
raise Exception("Unacknowledged variable transfer") tool.unarm()
eprint("Operation %c completed successfully" % action)
else: if i == 1:
raise Exception("Unknown action %c" % action) dump_b = dut.dump_ram()
longest = compare_dumps(dump_a, dump_b)
print(" longest chunk of untouched memory = %d" % longest)
except Exception as ex:
print("TEST FAILED")
raise ex
return 0 finally:
tool.end_measurement()
sys.stdout.flush()
sys.stderr.flush()
if __name__ == "__main__": if __name__ == "__main__":
......
...@@ -2,187 +2,94 @@ ...@@ -2,187 +2,94 @@
import os import os
import sys import sys
import time
import struct
import serial
import subprocess import subprocess
import serial.tools.list_ports
from test_common import (
LogicMultiplexerTimeMeasurements,
parse_nist_aead_test_vectors,
DeviceUnderTestAeadUARTP,
compare_dumps,
eprint,
run_nist_aead_test_line,
)
def get_serial():
ports = serial.tools.list_ports.comports()
for port in ports:
print(port.serial_number)
devices = [
p.device
for p in ports
if (p.vid == 4292 and p.pid == 60000)
]
devices.sort()
return devices[0]
class ESP32(DeviceUnderTestAeadUARTP):
RAM_SIZE = 0x5000
def eprint(*args, **kargs): def __init__(self, build_dir):
print(*args, file=sys.stderr, **kargs) DeviceUnderTestAeadUARTP.__init__(self)
self.build_dir = build_dir
def flash(tty=None): def flash(self):
pipe = subprocess.PIPE pipe = subprocess.PIPE
previous_dir = os.path.abspath(os.curdir)
os.chdir(self.build_dir)
cmd = ['platformio', 'run', '-e', 'esp32dev', '--target', 'upload'] cmd = ['platformio', 'run', '-e', 'esp32dev', '--target', 'upload']
if tty is not None: cmd.extend(['--upload-port', get_serial()])
cmd.extend(['--upload-port', tty]) cmd.extend(['--upload-port', get_serial()])
p = subprocess.Popen(cmd, p = subprocess.Popen(
stdout=sys.stderr, stdin=pipe) cmd, stdout=sys.stderr, stdin=pipe)
stdout, stderr = p.communicate("") stdout, stderr = p.communicate("")
eprint("Firmware flashed.")
os.chdir(previous_dir)
def dump_ram(self):
return None
def get_serial():
import serial.tools.list_ports
ports = serial.tools.list_ports.comports()
devices = [ p.device for p in ports ]
devices.sort()
return devices[-1]
class UARTP:
def __init__(self, ser):
UARTP.SYN = 0xf9
UARTP.FIN = 0xf3
self.ser = ser
def uart_read(self):
r = self.ser.read(1)
if len(r) != 1:
raise Exception("Serial read error")
return r[0]
def uart_write(self, c):
b = struct.pack("B", c)
r = self.ser.write(b)
if r != len(b):
raise Exception("Serial write error")
return r
def send(self, buf):
self.uart_write(UARTP.SYN)
len_ind_0 = 0xff & len(buf)
len_ind_1 = 0xff & (len(buf) >> 7)
if len(buf) < 128:
self.uart_write(len_ind_0)
else:
self.uart_write(len_ind_0 | 0x80)
self.uart_write(len_ind_1)
fcs = 0
for i in range(len(buf)):
info = buf[i]
fcs = (fcs + info) & 0xff
self.uart_write(buf[i])
fcs = (0xff - fcs) & 0xff
self.uart_write(fcs)
self.uart_write(UARTP.FIN)
eprint("sent frame '%s'" % buf.hex())
def recv(self):
tag_old = UARTP.FIN
while 1:
tag = tag_old
while 1:
if tag_old == UARTP.FIN:
if tag == UARTP.SYN:
break
tag_old = tag
tag = self.uart_read()
tag_old = tag
l = self.uart_read()
if l & 0x80:
l &= 0x7f
l |= self.uart_read() << 7
fcs = 0
buf = []
for i in range(l):
info = self.uart_read()
buf.append(info)
fcs = (fcs + info) & 0xff
fcs = (fcs + self.uart_read()) & 0xff
tag = self.uart_read()
if fcs == 0xff:
if tag == UARTP.FIN:
buf = bytes(buf)
eprint("rcvd frame '%s'" % buf.hex())
if len(buf) >= 1 and buf[0] == 0xde:
sys.stderr.buffer.write(buf[1:])
sys.stderr.flush()
else:
return buf
def main(argv): def main(argv):
eprint(argv[0]) if len(argv) != 3:
script_dir = os.path.split(argv[0])[0] print("Usage: test LWC_AEAD_KAT.txt build_dir")
if len(script_dir) > 0:
os.chdir(script_dir)
dev = get_serial()
flash(dev)
eprint("Flashed")
time.sleep(0.1)
ser = serial.Serial(dev, baudrate=500000, timeout=5)
uartp = UARTP(ser)
ser.setDTR(False) # IO0=HIGH
ser.setRTS(True) # EN=LOW, chip in reset
time.sleep(0.1)
ser.setDTR(False) # IO0=HIGH
ser.setRTS(False) # EN=HIGH, chip out of reset
time.sleep(1)
def stdin_read(n):
b = sys.stdin.buffer.read(n)
if len(b) != n:
sys.exit(1)
return b
def stdin_readvar():
l = stdin_read(4)
(l, ) = struct.unpack("<I", l)
v = stdin_read(l)
return v
exp_hello = b"Hello, World!"
hello = ser.read(ser.in_waiting)
if hello[-13:] != exp_hello:
eprint("Improper board initialization message: %s" % hello)
return 1 return 1
eprint("Board initialized properly")
sys.stdout.write("Hello, World!\n")
sys.stdout.flush()
while 1: kat = list(parse_nist_aead_test_vectors(argv[1]))
action = stdin_read(1)[0] build_dir = argv[2]
eprint("Command %c from stdin" % action)
dut = ESP32(build_dir)
if action in b"ackmps":
v = stdin_readvar()
uartp.send(struct.pack("B", action) + v)
ack = uartp.recv()
if len(ack) != 1 or ack[0] != action:
raise Exception("Unacknowledged variable transfer")
eprint("Var %c successfully sent to board" % action)
elif action in b"ACKMPS":
c = struct.pack("B", action)
uartp.send(c)
v = uartp.recv()
if len(v) < 1 or v[0] != action:
raise Exception("Could not obtain variable from board")
v = v[1:]
eprint("Var %c received from board: %s" % (action, v.hex()))
l = struct.pack("<I", len(v))
sys.stdout.buffer.write(l + v)
sys.stdout.flush()
elif action in b"ed": try:
c = struct.pack("B", action) tool = LogicMultiplexerTimeMeasurements(0x0030)
uartp.send(c) tool.begin_measurement()
ack = uartp.recv()
if len(ack) != 1 or ack[0] != action:
raise Exception("Unacknowledged variable transfer")
eprint("Operation %c completed successfully" % action)
else: dut.flash()
raise Exception("Unknown action %c" % action)
dut.ser = serial.Serial(
get_serial(),
baudrate=500000,
timeout=5)
dut.prepare()
sys.stdout.write("Board prepared\n")
sys.stdout.flush()
return 0 for i, m, ad, k, npub, c in kat:
tool.arm()
run_nist_aead_test_line(dut, i, m, ad, k, npub, c)
tool.unarm()
except Exception as ex:
print("TEST FAILED")
raise ex
finally:
tool.end_measurement()
sys.stdout.flush()
sys.stderr.flush()
if __name__ == "__main__": if __name__ == "__main__":
......
...@@ -3,6 +3,7 @@ ...@@ -3,6 +3,7 @@
import os import os
import sys import sys
import pylink import pylink
import serial.tools.list_ports
from test_common import ( from test_common import (
LogicMultiplexerTimeMeasurements, LogicMultiplexerTimeMeasurements,
parse_nist_aead_test_vectors, parse_nist_aead_test_vectors,
...@@ -14,7 +15,6 @@ from test_common import ( ...@@ -14,7 +15,6 @@ from test_common import (
def get_serial(): def get_serial():
import serial.tools.list_ports
ports = serial.tools.list_ports.comports() ports = serial.tools.list_ports.comports()
for port in ports: for port in ports:
print(port.serial_number) print(port.serial_number)
...@@ -56,17 +56,16 @@ class F7(DeviceUnderTestAeadUARTP): ...@@ -56,17 +56,16 @@ class F7(DeviceUnderTestAeadUARTP):
def main(argv): def main(argv):
if len(argv) < 2: if len(argv) != 3:
print("Usage: test LWC_AEAD_KAT.txt") print("Usage: test LWC_AEAD_KAT.txt build_dir")
return 1
kat = list(parse_nist_aead_test_vectors(argv[1])) kat = list(parse_nist_aead_test_vectors(argv[1]))
build_dir = argv[2]
eprint(argv[0])
script_dir = os.path.split(argv[0])[0]
dut = F7( dut = F7(
os.path.join(script_dir, 'build', 'f7.bin'), os.path.join(build_dir, 'build', 'f7.bin'),
os.path.join(script_dir, 'ram_pattern.bin')) os.path.join(build_dir, 'ram_pattern.bin'))
try: try:
tool = LogicMultiplexerTimeMeasurements(0x000c) tool = LogicMultiplexerTimeMeasurements(0x000c)
......
...@@ -30,7 +30,7 @@ class DeviceUnderTest: ...@@ -30,7 +30,7 @@ class DeviceUnderTest:
class DeviceUnderTestAeadUARTP(DeviceUnderTest): class DeviceUnderTestAeadUARTP(DeviceUnderTest):
def __init__(self, ser): def __init__(self, ser=None):
self.ser = ser self.ser = ser
def prepare(self): def prepare(self):
...@@ -301,6 +301,9 @@ class LogicMultiplexerTimeMeasurements(TimeMeasurementTool): ...@@ -301,6 +301,9 @@ class LogicMultiplexerTimeMeasurements(TimeMeasurementTool):
self.sock.connect(self.server_addr) self.sock.connect(self.server_addr)
self.sock.send(struct.pack("<Q", self.mask)) self.sock.send(struct.pack("<Q", self.mask))
def arm(self):
self.capture.extend(self.recv_samples())
def unarm(self): def unarm(self):
self.capture.extend(self.recv_samples()) self.capture.extend(self.recv_samples())
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment