Add TRNG attack files

TRNG_attack/attack_student.py

@@ -0,0 +1,27 @@
+import numpy as np
+
+TRNG_PAIR_CNT = 64
+
+
+if __name__ == '__main__':
+    # reading info file - length of trace, sampling frequency (not necessary to know in our case), random value generated by the TRNG
+    with open("data_info.txt", "r") as fin:
+        tracelen = int(fin.readline())
+        fs = int(fin.readline())
+        trng_val = fin.readline()
+
+    traces = np.fromfile("data.bin", dtype='uint8') # reading traces for individual ROs
+    traces = np.reshape(traces, (traces.size//tracelen, tracelen)) # reshape of matrix, each row contains the trace for one RO
+
+    traces_bin = ??? # conversion of waveforms to rectangles - everything below threshold is 0, otherwise 1 (they are boolean values actually)
+    rising_edges = np.logical_not(???) & ??? # finding rising edges, each rising edge is represented by True
+
+    cnt = np.count_nonzero(???, axis=1) # count the number of rising edges in rows
+    # cnt is now a 1D vector
+    cnt = cnt.reshape(TRNG_PAIR_CNT,2).min(axis=1) # Reshape of the count array into matrix, where each row contains 2 values - the number of rising edges for two ROs in a pair. Then we select the smaller value.
+
+    cnt_sel = cnt & ?x???? # select only the two least significant bits
+
+    estimate = ''.join([np.binary_repr(x, width=2) for x in cnt_sel]) # binary representation of the values (the last 2 bits) and joining them into one string
+    print('{0:0>32x}'.format(int(estimate, 2)))
+    print(trng_val) # from data_info, output of the RNG in FPGA

TRNG_attack/measurement.py

@@ -0,0 +1,87 @@
+import oscilloscope
+import serial
+import serial.tools.list_ports
+from time import sleep
+
+SAMPLE_FREQ = 625*10**6
+
+RO_CNT = 64
+TRNG_PAIR_CNT = 64
+
+def list_resources(resources: list, resource_name: str):
+    if not resources:
+        print('no', resource_name, 'available')
+    else:
+        print('available', resource_name + ':')
+        for resource_id in range(len(resources)):
+            print('[', resource_id, '] ', resources[resource_id])
+
+def list_scopes():
+    found_oscilloscopes = oscilloscope.get_oscilloscopes()
+    list_resources(found_oscilloscopes, 'oscilloscopes')
+
+def channel_meas(scope, n):
+    scope.command_check(":WAVeform:SOURce", 'CHANnel{}'.format(n))
+    trace = scope.query_binary(':WAVeform:DATA?')
+    return trace
+
+
+# Infinite test run
+# The cycle iterates over all ROs. Can be interrupted by pressing CTRL-C
+def run(fpga_comm):
+    print ("Infinite run, press CTRL-C to break.")
+    try:
+        i = 0
+        while True:
+            fpga_comm.write(bytes([i,i]))
+            i = (i + 1) % RO_CNT
+            sleep(1)
+    except KeyboardInterrupt:
+        pass
+    
+
+def trng_read(scope, fpga_comm):
+
+    with open('data_info.txt', "w") as finfo, open ('data.bin', "wb") as fdata:
+    
+        for i in range(TRNG_PAIR_CNT):
+            print('--------------------------MEAS {}-------------------------------'.format(i))
+            scope.write(':SINGle')
+            sleep(0.1)
+            fpga_comm.write(bytes([i,i]))
+            trace1 = channel_meas(scope, 1)
+            trace2 = channel_meas(scope, 2)
+            if i == 0:
+                tracelength = scope.query(':WAVeform:POINts?')
+                fs = scope.query(':ACQuire:SRATe?')
+                print(tracelength, file = finfo)
+                print(int(float(fs)), file = finfo)
+            fdata.write(trace1)
+            fdata.write(trace2)
+
+        val = fpga_comm.read(16)
+        print(val.hex())
+        print(val.hex(), file = finfo)
+
+
+
+
+if __name__ == '__main__':
+    list_scopes()
+    ports = serial.tools.list_ports.comports()
+    list_resources(ports, "COM")
+
+    # modify the device numbers in the following two lines:
+    s = serial.Serial(ports[0].device, 923076)
+    scope = oscilloscope.Oscilloscope(0)
+
+    # scope.setup_measurement()
+    # scope.save_conf('scope_setup.conf')
+    scope.load_conf('scope_setup.conf')
+    sleep(2) # wait for the oscilloscope to process the setup
+
+    # test run -- only TRNG, no recording
+    # run(s) 
+
+    # measurement -- RESET the FPGA first! (USB disconnect+connect)
+    # trng_read(scope, s)

TRNG_attack/oscilloscope.py

@@ -0,0 +1,101 @@
+import logging
+import pyvisa
+from pyvisa import constants
+
+
+logger = logging.getLogger(__name__)
+logger.addHandler(logging.NullHandler())
+
+
+def get_oscilloscopes():
+    rm = pyvisa.ResourceManager('@py')
+    return rm.list_resources(query = '?*::INSTR')
+
+
+class Oscilloscope:
+    def __init__(self, oscilloscope_id):
+        rm = pyvisa.ResourceManager('@py')
+        resources = rm.list_resources(query = '?*::INSTR')
+        self.resource = rm.open_resource(
+                resources[oscilloscope_id],
+                read_termination = '\n',
+                write_termination = '\n')
+        self.resource.timeout = 20000
+        logger.debug('timeout set to %dms', self.resource.timeout)
+        self.resource.query_delay = 0.1
+        logger.debug('query_delay set to %fs', self.resource.query_delay)
+        # self.query = self.resource.query
+        self.write('*CLS')
+        print('connected to the oscilloscope with *IDN:',
+            self.query('*IDN?'))
+
+    def __del__(self):
+        self.close()
+
+    def close(self):
+        logger.debug('closing oscilloscope...')
+        self.resource.close()
+
+    def write(self, *args):
+        logger.debug('%s', ', '.join(map(str, args)))
+        return self.resource.write(*args)
+
+    def query(self, *args):
+        logger.debug('%s...', ', '.join(map(str, args)))
+        data = self.resource.query(*args)
+        logger.debug('%s %s', args[0], data)
+        return data
+
+    def command_binary(self, query, data: bytes):
+        logger.debug('%s, len: %d', query, len(data))
+        return self.resource.write_binary_values(
+                query,
+                data,
+                datatype='B')
+
+    def query_binary(self, query):
+        logger.debug('%s...', query)
+        data = self.resource.query_binary_values(
+                query,
+                datatype='B',
+                container = bytes)
+        logger.debug('%s, len: %d', query, len(data))
+        return data
+
+    def query_check(self, command):
+        print(command, self.resource.query(command+'?'))
+
+    def command_check(self, command, value):
+        data = self.resource.write(command + ' ' + value)
+        self.query_check(command)
+        return data
+
+    def save_conf(self, filename):
+        logger.debug('to filename %s', filename)
+        data = self.query_binary(':SYSTem:SETup?')
+        out_file = open(filename, 'wb')
+        len_written = out_file.write(data)
+        out_file.close()
+        logger.debug('read %d, written %d', len(data), len_written)
+        return len(data) - len_written
+
+    def load_conf(self, filename):
+        logger.debug('from filename %s', filename)
+        in_file = open(filename, 'rb')
+        data = in_file.read()
+        len_written = self.command_binary(':SYSTem:SETup ', data)
+        in_file.close()
+        logger.debug('read %d, written %d', len(data), len_written)
+        return len(data) - len_written
+
+    def setup_measurement(self):
+        logger.debug('')
+        self.command_check(":ACQuire:TYPE", "Normal")
+        # self.command_check(":ACQuire:COUNt", "2")
+        self.command_check(":TIMebase:MODE", "MAIN")
+        self.command_check(":WAVeform:UNSigned", "1")
+        self.command_check(":WAVeform:BYTeorder", "LSBFirst")
+        self.command_check(":WAVeform:FORMat", "BYTE")
+        self.command_check(":WAVeform:SOURce", "CHANnel1")
+        self.command_check(":WAVeform:POINts:MODE", "RAW")
+        self.command_check(":ACQuire:COMPlete", "100")