Add parsing of ChipWhisperer trace sets.

13 files changed, 248 insertions, 24 deletions

diff --git a/pyecsca/__init__.py b/pyecsca/__init__.py
index 89149e4..a3f890f 100644
--- a/pyecsca/__init__.py
+++ b/pyecsca/__init__.py
@@ -8,3 +8,4 @@ from .tvla import *
 from .trace import *
 from .trace_set.base import *
 from .trace_set.inspector import *
+from .trace_set.chipwhisperer import *
\ No newline at end of file
diff --git a/pyecsca/align.py b/pyecsca/align.py
index 1c4b891..ff586ab 100644
--- a/pyecsca/align.py
+++ b/pyecsca/align.py
@@ -63,6 +63,7 @@ def align_correlation(reference: Trace, *traces: Trace,
         correlation = np.correlate(trace_part, reference_part, "same")
         max_correlation_offset = correlation.argmax(axis=0)
         max_correlation = correlation[max_correlation_offset]
+        del trace_part
         if max_correlation < min_correlation:
             return False, 0
         left_space = min(max_offset, reference_offset)
@@ -103,25 +104,26 @@ def align_peaks(reference: Trace, *traces: Trace,
 
 
 @public
-def align_sad(reference: Trace, *traces: Trace,
-              reference_offset: int, reference_length: int, max_offset: int) -> List[Trace]:
+def align_offset(reference: Trace, *traces: Trace,
+                 reference_offset: int, reference_length: int, max_offset: int,
+                 dist_func: Callable[[np.ndarray, np.ndarray], float], max_dist: float = float("inf")) -> List[Trace]:
     """
-    Align `traces` to the reference `trace` so that the Sum Of Absolute Differences between the
-    reference trace window from `reference_offset` of `reference_length` and the trace being aligned
-    within `max_offset` of the reference window is maximized.
+    Align `traces` to the reference `trace` so that the value of the `dist_func` is minimized
+    between the reference trace window from `reference_offset` of `reference_length` and the trace
+    being aligned within `max_offset` of the reference window.
 
     :param reference:
     :param traces:
     :param reference_offset:
     :param reference_length:
     :param max_offset:
+    :param dist_func:
     :return:
     """
     reference_part = reference.samples[reference_offset: reference_offset + reference_length]
-
     def align_func(trace):
         length = len(trace.samples)
-        best_sad = 0
+        best_distance = 0
         best_offset = 0
         for offset in range(-max_offset, max_offset):
             start = reference_offset + offset
@@ -129,17 +131,41 @@ def align_sad(reference: Trace, *traces: Trace,
             if start < 0 or stop >= length:
                 continue
             trace_part = trace.samples[start:stop]
-            # todo: add other distance functions here
-            sad = np.sum(np.abs(reference_part - trace_part))
-            if sad > best_sad:
-                best_sad = sad
+            distance = dist_func(reference_part, trace_part)
+            if distance < best_distance:
+                best_distance = distance
                 best_offset = offset
-        return True, best_offset
-
+        if best_distance < max_dist:
+            return True, best_offset
+        else:
+            return False, 0
     return align_reference(reference, *traces, align_func=align_func)
 
 
 @public
+def align_sad(reference: Trace, *traces: Trace,
+              reference_offset: int, reference_length: int, max_offset: int) -> List[Trace]:
+    """
+    Align `traces` to the reference `trace` so that the Sum Of Absolute Differences between the
+    reference trace window from `reference_offset` of `reference_length` and the trace being aligned
+    within `max_offset` of the reference window is maximized.
+
+    :param reference:
+    :param traces:
+    :param reference_offset:
+    :param reference_length:
+    :param max_offset:
+    :return:
+    """
+    def sad(reference_part, trace_part):
+        return float(np.sum(np.abs(reference_part - trace_part)))
+
+    return align_offset(reference, *traces,
+                        reference_offset=reference_offset, reference_length=reference_length,
+                        max_offset=max_offset, dist_func=sad)
+
+
+@public
 def align_dtw_scale(reference: Trace, *traces: Trace, radius: int = 1,
                     fast: bool = True) -> List[Trace]:
     """
diff --git a/pyecsca/trace_set/chipwhisperer.py b/pyecsca/trace_set/chipwhisperer.py
new file mode 100644
index 0000000..52300f6
--- /dev/null
+++ b/pyecsca/trace_set/chipwhisperer.py
@@ -0,0 +1,42 @@
+import numpy as np
+from .base import TraceSet
+from os.path import exists, isfile, join
+from configparser import ConfigParser
+from public import public
+
+from ..trace import Trace
+
+
+@public
+class ChipWhispererTraceSet(TraceSet):
+
+    def __init__(self, path: str = None, name: str = None):
+        if path is None and name is None:
+            super().__init__()
+        else:
+            data = self._read_data(path, name)
+            trace_data = data["traces"]
+            traces = [Trace(None, None, trace_samples, trace_set=self) for trace_samples in trace_data]
+            del data["traces"]
+            config = self._read_config(path, name)
+            super().__init__(*traces, **data, **config)
+
+    def _read_data(self, path, name):
+        types = {"keylist": None, "knownkey": None, "textin": None, "textout": None, "traces": None}
+        for type in types.keys():
+            type_path = join(path, name + "_" + type + ".npy")
+            if exists(type_path) and isfile(type_path):
+                types[type] = np.load(type_path)
+        return types
+
+    def _read_config(self, path, name):
+        config_path = join(path, "config_" + name + "_.cfg")
+        if exists(config_path) and isfile(config_path):
+            config = ConfigParser()
+            config.read(config_path)
+            return config["Trace Config"]
+        else:
+            return {}
+
+    def __repr__(self):
+        return "ChipWhispererTraceSet()"
diff --git a/test/data/chipwhisperer_keylist.npy b/test/data/chipwhisperer_keylist.npy
new file mode 100644
index 0000000..c709dc7
--- /dev/null
+++ b/test/data/chipwhisperer_keylist.npy
diff --git a/test/data/chipwhisperer_knownkey.npy b/test/data/chipwhisperer_knownkey.npy
new file mode 100644
index 0000000..82c97cd
--- /dev/null
+++ b/test/data/chipwhisperer_knownkey.npy
diff --git a/test/data/chipwhisperer_settings.cwset b/test/data/chipwhisperer_settings.cwset
new file mode 100644
index 0000000..839fca1
--- /dev/null
+++ b/test/data/chipwhisperer_settings.cwset
@@ -0,0 +1,132 @@
+[Aux Settings]
+[Trace Settings]
+[[ChipWhisperer/Native]]
+Config File = 
+Format = 
+[Trace Settings]
+[Target Settings]
+[[Simple Serial]]
+Connection = NewAE USB (CWLite/CW1200)
+Key Length (Bytes) = 16
+Input Length (Bytes) = 16
+Output Length (Bytes) = 16
+[[[Protocol Version]]]
+Version = auto
+Timeout (ms) = 20
+[[Simple Serial]]
+Preset Mode = Custom
+Init Command = 
+Load Key Command = 
+Load Input Command = 
+Go Command = p$TEXT$\n
+Output Format = 
+[[[NewAE USB (CWLite/CW1200)]]]
+baud = 38400
+[Target Settings]
+[Scope Settings]
+[[ChipWhisperer/OpenADC]]
+Connection = NewAE USB (CWLite/CW1200)
+Auto-Refresh DCM Status = True
+[[[NewAE USB (CWLite/CW1200)]]]
+Serial Number = Auto
+[[[[ChipWisperer-Lite USB]]]]
+[[ChipWhisperer/OpenADC]]
+[[[OpenADC]]]
+[[[[HW Information]]]]
+Version = (1, 8, 'ChipWhisperer-Lite', 0)
+Synth Date = unknown
+System Freq = 96000000
+Max Samples = 24400
+[[[OpenADC]]]
+[[[[Gain Setting]]]]
+Mode = low
+Setting = 45
+Result = 22.50390625
+[[[OpenADC]]]
+[[[[Trigger Setup]]]]
+Trigger Pin State = False
+Mode = rising edge
+Timeout (secs) = 2
+Offset = 0
+Pre-Trigger Samples = 0
+Total Samples = 24000
+Downsample Factor = 1
+Trigger Active Count = 22292
+[[[OpenADC]]]
+[[[[Clock Setup]]]]
+[[[[[ADC Clock]]]]]
+Source = CLKGEN x1 via DCM
+Phase Adjust = 0
+ADC Freq = 7384609
+ADC Sample Rate = 7384609
+DCM Locked = True
+[[[[Clock Setup]]]]
+Freq Counter = 0
+Freq Counter Src = EXTCLK Input
+[[[[[CLKGEN Settings]]]]]
+Input Source = system
+Input Frequency = 10000000.0
+Multiply = 2
+Divide = 26
+Desired Frequency = 7384615
+Current Frequency = 7384615
+DCM Locked = True
+[[ChipWhisperer/OpenADC]]
+[[[CW Extra]]]
+[[[[CW Extra Settings]]]]
+[[[[[Trigger Pins]]]]]
+Target IO1 (Serial TXD) = False
+Target IO2 (Serial RXD) = False
+Target IO3 (SmartCard Serial) = False
+Target IO4 (Trigger Line) = True
+Collection Mode = OR
+[[[[CW Extra Settings]]]]
+Trigger Module = Basic (Edge/Level)
+Clock Source = Target IO-IN
+Target HS IO-Out = CLKGEN
+HS-Glitch Out Enable (High Power) = 0
+HS-Glitch Out Enable (Low Power) = 0
+[[[[[Target IOn Pins]]]]]
+Target IO1 = Serial RXD
+Target IO2 = Serial TXD
+Target IO3 = High-Z
+Target IO4 = High-Z
+[[[[CW Extra Settings]]]]
+[[[[[Target IOn GPIO Mode]]]]]
+Target IO1: GPIO = Disabled
+Target IO2: GPIO = Disabled
+Target IO3: GPIO = Disabled
+Target IO4: GPIO = Disabled
+nRST: GPIO = Default
+PDID: GPIO = Default
+PDIC: GPIO = Default
+[[[[CW Extra Settings]]]]
+Target Power State = True
+[[[CW Extra]]]
+[[[[Glitch Module]]]]
+Clock Source = Target IO-IN
+Glitch Width (as % of period) = 10
+Glitch Width (fine adjust) = 0
+Glitch Offset (as % of period) = 10
+Glitch Offset (fine adjust) = 0
+Glitch Trigger = Manual
+Single-Shot Arm = After Scope Arm
+Ext Trigger Offset = 0
+Repeat = 1
+Output Mode = Clock XORd
+[Scope Settings]
+[Generic Settings]
+Scope Module = ChipWhisperer/OpenADC
+Target Module = Simple Serial
+Trace Format = ChipWhisperer/Native
+Auxiliary Module = None
+[[Acquisition Settings]]
+Number of Traces = 2
+Number of Sets = 1
+Traces per Set = 2
+Key/Text Pattern = Basic
+[Generic Settings]
+[[Basic]]
+Key = Random
+Plaintext = Fixed
+Fixed Plaintext = 00 00 00 00 00 00 00 00 00 00 00 00 00 00 80 00
diff --git a/test/data/chipwhisperer_textin.npy b/test/data/chipwhisperer_textin.npy
new file mode 100644
index 0000000..22b93e1
--- /dev/null
+++ b/test/data/chipwhisperer_textin.npy
diff --git a/test/data/chipwhisperer_textout.npy b/test/data/chipwhisperer_textout.npy
new file mode 100644
index 0000000..0ac8831
--- /dev/null
+++ b/test/data/chipwhisperer_textout.npy
diff --git a/test/data/chipwhisperer_traces.npy b/test/data/chipwhisperer_traces.npy
new file mode 100644
index 0000000..df72f86
--- /dev/null
+++ b/test/data/chipwhisperer_traces.npy
diff --git a/test/data/config_chipwhisperer_.cfg b/test/data/config_chipwhisperer_.cfg
new file mode 100644
index 0000000..dcc3129
--- /dev/null
+++ b/test/data/config_chipwhisperer_.cfg
@@ -0,0 +1,13 @@
+[Trace Config]
+numTraces = 2
+format = native
+numPoints = 24000
+prefix = chipwhisperer_
+date = 2017-07-15 21:56:26
+scopeSampleRate = 7384609
+notes = "AckPattern: Basic (Key=Random, Plaintext=Fixed:00 00 00 00 00 00 00 00 00 00 00 00 00 00 80 00); Aux: "
+scopeName = ChipWhisperer/OpenADC
+scopeXUnits = 0
+targetSW = unknown
+targetHW = Simple Serial
+scopeYUnits = 0
diff --git a/test/example.trs b/test/data/example.trs
index a432e69..a432e69 100644
--- a/test/example.trs
+++ b/test/data/example.trs
diff --git a/test/test_align.py b/test/test_align.py
index ae7baba..d86d177 100644
--- a/test/test_align.py
+++ b/test/test_align.py
@@ -23,13 +23,13 @@ class AlignTests(TestCase):
 
     @slow
     def test_large_align(self):
-        example = InspectorTraceSet("test/example.trs")
+        example = InspectorTraceSet("test/data/example.trs")
         result = align_correlation(*example, reference_offset=100000, reference_length=20000, max_offset=15000)
         self.assertIsNotNone(result)
 
     @slow
     def test_large_dtw_align(self):
-        example = InspectorTraceSet("test/example.trs")
+        example = InspectorTraceSet("test/data/example.trs")
         result = align_dtw(*example[:5])
         self.assertIsNotNone(result)
 
@@ -46,7 +46,8 @@ class AlignTests(TestCase):
         second_arr = np.array([10, 10, 10, 10, 90, 40, 50, 20, 10, 17, 16, 10, 10], dtype=np.dtype("i1"))
         a = Trace(None, None, first_arr)
         b = Trace(None, None, second_arr)
-        align_sad(a, b, reference_offset=2, reference_length=5, max_offset=3)
+        result = align_sad(a, b, reference_offset=2, reference_length=5, max_offset=3)
+        self.assertEqual(len(result), 2)
 
     def test_dtw_align_scale(self):
         first_arr = np.array([10, 64, 14, 120, 15, 30, 10, 15, 20, 15, 15, 10, 10, 8, 10, 12, 10, 13, 9], dtype=np.dtype("i1"))
diff --git a/test/test_traceset.py b/test/test_traceset.py
index 547c4c2..c9b3fd4 100644
--- a/test/test_traceset.py
+++ b/test/test_traceset.py
@@ -2,7 +2,7 @@ from unittest import TestCase
 import os.path
 import tempfile
 
-from pyecsca import TraceSet, InspectorTraceSet
+from pyecsca import TraceSet, InspectorTraceSet, ChipWhispererTraceSet
 
 
 class TraceSetTests(TestCase):
@@ -10,9 +10,10 @@ class TraceSetTests(TestCase):
     def test_create(self):
         self.assertIsNotNone(TraceSet())
         self.assertIsNotNone(InspectorTraceSet())
+        self.assertIsNotNone(ChipWhispererTraceSet())
 
     def test_load_fname(self):
-        result = InspectorTraceSet("test/example.trs")
+        result = InspectorTraceSet("test/data/example.trs")
         self.assertIsNotNone(result)
         self.assertEqual(result.global_title, "Example trace set")
         self.assertEqual(len(result), 10)
@@ -21,26 +22,34 @@ class TraceSetTests(TestCase):
         self.assertEqual(result.sampling_frequency, 12500000)
 
     def test_load_file(self):
-        with open("test/example.trs", "rb") as f:
+        with open("test/data/example.trs", "rb") as f:
             self.assertIsNotNone(InspectorTraceSet(f))
 
     def test_load_bytes(self):
-        with open("test/example.trs", "rb") as f:
+        with open("test/data/example.trs", "rb") as f:
             self.assertIsNotNone(InspectorTraceSet(f.read()))
 
     def test_get_bytes(self):
-        self.assertIsNotNone(bytes(InspectorTraceSet("test/example.trs")))
+        self.assertIsNotNone(bytes(InspectorTraceSet("test/data/example.trs")))
 
     def test_keep_traces(self):
-        trace_set = InspectorTraceSet("test/example.trs")
+        trace_set = InspectorTraceSet("test/data/example.trs")
         self.assertIsNotNone(trace_set.raw)
-        trace_set = InspectorTraceSet("test/example.trs", keep_raw_traces=False)
+        trace_set = InspectorTraceSet("test/data/example.trs", keep_raw_traces=False)
         self.assertIsNone(trace_set.raw)
 
     def test_save(self):
-        trace_set = InspectorTraceSet("test/example.trs")
+        trace_set = InspectorTraceSet("test/data/example.trs")
         with tempfile.TemporaryDirectory() as dirname:
             path = os.path.join(dirname, "out.trs")
             trace_set.save(path)
             self.assertTrue(os.path.exists(path))
             self.assertIsNotNone(InspectorTraceSet(path))
+
+
+class ChipWhispererTraceSetTest(TestCase):
+
+    def test_load_fname(self):
+        result = ChipWhispererTraceSet("test/data/", "chipwhisperer")
+        self.assertIsNotNone(result)
+        print(result[:])
\ No newline at end of file