diff options
| -rw-r--r-- | Pipfile | 1 | ||||
| -rw-r--r-- | docs/Makefile | 2 | ||||
| -rw-r--r-- | docs/conf.py | 1 | ||||
| -rw-r--r-- | pyecsca/__init__.py | 2 | ||||
| -rw-r--r-- | pyecsca/combine.py | 19 | ||||
| -rw-r--r-- | pyecsca/edit.py | 22 | ||||
| -rw-r--r-- | pyecsca/filter.py | 42 | ||||
| -rw-r--r-- | pyecsca/process.py | 43 | ||||
| -rw-r--r-- | pyecsca/sampling.py | 23 | ||||
| -rw-r--r-- | pyecsca/trace.py | 6 | ||||
| -rw-r--r-- | pyecsca/trace_set/base.py | 3 | ||||
| -rw-r--r-- | pyecsca/trace_set/chipwhisperer.py | 1 | ||||
| -rw-r--r-- | pyecsca/trace_set/inspector.py | 33 | ||||
| -rw-r--r-- | pyecsca/ttest.py (renamed from pyecsca/tvla.py) | 16 | ||||
| -rw-r--r-- | test/test_process.py | 16 | ||||
| -rw-r--r-- | test/test_traceset.py | 5 | ||||
| -rw-r--r-- | test/test_ttest.py (renamed from test/test_tvla.py) | 2 |
17 files changed, 216 insertions, 21 deletions
@@ -8,6 +8,7 @@ nose2 = "*" green = "*" mypy = "*" sphinx = "*" +sphinx-autodoc-typehints = "*" [packages] numpy = "*" diff --git a/docs/Makefile b/docs/Makefile index 050823b..fff1a35 100644 --- a/docs/Makefile +++ b/docs/Makefile @@ -14,7 +14,7 @@ help: apidoc: mkdir -p api/ - sphinx-apidoc ../pyecsca/ -o api/ + sphinx-apidoc ../pyecsca/ -f -o api/ .PHONY: help apidoc Makefile diff --git a/docs/conf.py b/docs/conf.py index f4f8253..09af711 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -42,6 +42,7 @@ release = '0.1.0' # ones. extensions = [ 'sphinx.ext.autodoc', + 'sphinx_autodoc_typehints', 'sphinx.ext.todo', 'sphinx.ext.mathjax', 'sphinx.ext.ifconfig', diff --git a/pyecsca/__init__.py b/pyecsca/__init__.py index a3f890f..c2fd7ff 100644 --- a/pyecsca/__init__.py +++ b/pyecsca/__init__.py @@ -4,7 +4,7 @@ from .edit import * from .filter import * from .process import * from .sampling import * -from .tvla import * +from .ttest import * from .trace import * from .trace_set.base import * from .trace_set.inspector import * diff --git a/pyecsca/combine.py b/pyecsca/combine.py index 4b29f8c..c484748 100644 --- a/pyecsca/combine.py +++ b/pyecsca/combine.py @@ -7,6 +7,12 @@ from .trace import Trace, CombinedTrace @public def average(*traces: Trace) -> Optional[CombinedTrace]: + """ + Average `traces`, sample-wise. + + :param traces: + :return: + """ if not traces: return None if len(traces) == 1: @@ -18,11 +24,24 @@ def average(*traces: Trace) -> Optional[CombinedTrace]: @public def conditional_average(*traces: Trace, condition: Callable[[Trace], bool]) -> Optional[CombinedTrace]: + """ + Average `traces` for which the `condition` is True, sample-wise. + + :param traces: + :param condition: + :return: + """ return average(*filter(condition, traces)) @public def standard_deviation(*traces: Trace) -> Optional[CombinedTrace]: + """ + Compute the standard-deviation of the `traces`, sample-wise. + + :param traces: + :return: + """ if not traces: return None dtype = traces[0].samples.dtype diff --git a/pyecsca/edit.py b/pyecsca/edit.py index 53553a9..f01a0dc 100644 --- a/pyecsca/edit.py +++ b/pyecsca/edit.py @@ -8,6 +8,14 @@ from .trace import Trace @public def trim(trace: Trace, start: int = None, end: int = None) -> Trace: + """ + Trim the `trace` samples, output contains samples between the `start` and `end` indices. + + :param trace: + :param start: + :param end: + :return: + """ if start is None: start = 0 if end is None: @@ -19,12 +27,26 @@ def trim(trace: Trace, start: int = None, end: int = None) -> Trace: @public def reverse(trace: Trace) -> Trace: + """ + Reverse the samples of the `trace`. + + :param trace: + :return: + """ return Trace(copy(trace.title), copy(trace.data), np.flipud(trace.samples)) @public def pad(trace: Trace, lengths: Union[Tuple[int, int], int], values: Union[Tuple[Any, Any], Any] = (0, 0)) -> Trace: + """ + Pad the samples of the `trace` by `values` at the beginning and end. + + :param trace: + :param lengths: How much to pad at the beginning and end, either symmetric (if integer) or asymmetric (if tuple). + :param values: What value to pad with, either symmetric or asymmetric (if tuple). + :return: + """ if not isinstance(lengths, tuple): lengths = (lengths, lengths) if not isinstance(values, tuple): diff --git a/pyecsca/filter.py b/pyecsca/filter.py index 2635a8d..bc221ce 100644 --- a/pyecsca/filter.py +++ b/pyecsca/filter.py @@ -7,32 +7,70 @@ from .trace import Trace def filter_any(trace: Trace, sampling_frequency: int, - cutoff: Union[int, Tuple[int, int]], type: str) -> Trace: + cutoff: Union[int, Tuple[int, int]], band_type: str) -> Trace: nyq = 0.5 * sampling_frequency if not isinstance(cutoff, int): normal_cutoff = tuple(map(lambda x: x / nyq, cutoff)) else: normal_cutoff = cutoff / nyq - b, a = butter(6, normal_cutoff, btype=type, analog=False) + b, a = butter(6, normal_cutoff, btype=band_type, analog=False) result_samples = lfilter(b, a, trace.samples) return Trace(copy(trace.title), copy(trace.data), result_samples) @public def filter_lowpass(trace: Trace, sampling_frequency: int, cutoff: int) -> Trace: + """ + Apply a lowpass digital filter (Butterworth) to `trace`, given `sampling_frequency` and + `cutoff` frequency. + + :param trace: + :param sampling_frequency: + :param cutoff: + :return: + """ return filter_any(trace, sampling_frequency, cutoff, "lowpass") @public def filter_highpass(trace: Trace, sampling_frequency: int, cutoff: int) -> Trace: + """ + Apply a highpass digital filter (Butterworth) to `trace`, given `sampling_frequency` and + `cutoff` frequency. + + :param trace: + :param sampling_frequency: + :param cutoff: + :return: + """ return filter_any(trace, sampling_frequency, cutoff, "highpass") @public def filter_bandpass(trace: Trace, sampling_frequency: int, low: int, high: int) -> Trace: + """ + Apply a bandpass digital filter (Butterworth) to `trace`, given `sampling_frequency`, with the + passband from `low` to `high`. + + :param trace: + :param sampling_frequency: + :param low: + :param high: + :return: + """ return filter_any(trace, sampling_frequency, (low, high), "bandpass") @public def filter_bandstop(trace: Trace, sampling_frequency: int, low: int, high: int) -> Trace: + """ + Apply a bandstop digital filter (Butterworth) to `trace`, given `sampling_frequency`, with the + stopband from `low` to `high`. + + :param trace: + :param sampling_frequency: + :param low: + :param high: + :return: + """ return filter_any(trace, sampling_frequency, (low, high), "bandstop") diff --git a/pyecsca/process.py b/pyecsca/process.py index 811a118..8e983b2 100644 --- a/pyecsca/process.py +++ b/pyecsca/process.py @@ -7,16 +7,35 @@ from .trace import Trace @public def absolute(trace: Trace) -> Trace: + """ + Apply absolute value to samples of `trace`. + + :param trace: + :return: + """ return Trace(copy(trace.title), copy(trace.data), np.absolute(trace.samples)) @public def invert(trace: Trace) -> Trace: + """ + Invert(negate) the samples of `trace`. + + :param trace: + :return: + """ return Trace(copy(trace.title), copy(trace.data), np.negative(trace.samples)) @public def threshold(trace: Trace, value) -> Trace: + """ + Map samples of the `trace` to 1 if they are above `value` or to 0. + + :param trace: + :param value: + :return: + """ result_samples = trace.samples.copy() result_samples[result_samples <= value] = 0 result_samples[np.nonzero(result_samples)] = 1 @@ -31,13 +50,27 @@ def rolling_window(samples: np.ndarray, window: int) -> np.ndarray: @public def rolling_mean(trace: Trace, window: int) -> Trace: + """ + Compute the rolling mean of `trace` using `window`. Shortens the trace by `window` - 1. + + :param trace: + :param window: + :return: + """ return Trace(copy(trace.title), copy(trace.data), np.mean(rolling_window(trace.samples, window), -1).astype( - dtype=trace.samples.dtype)) + dtype=trace.samples.dtype)) @public def offset(trace: Trace, offset) -> Trace: + """ + Offset samples of `trace` by `offset`, sample-wise (Adds `offset` to all samples). + + :param trace: + :param offset: + :return: + """ return Trace(copy(trace.title), copy(trace.data), trace.samples + offset) @@ -47,6 +80,12 @@ def root_mean_square(trace: Trace): @public def recenter(trace: Trace) -> Trace: + """ + Subtract the root mean square of the `trace` from its samples, sample-wise. + + :param trace: + :return: + """ around = root_mean_square(trace) return offset(trace, -around) @@ -61,4 +100,4 @@ def normalize(trace: Trace) -> Trace: def normalize_wl(trace: Trace) -> Trace: return Trace(copy(trace.title), copy(trace.data), (trace.samples - np.mean(trace.samples)) / ( - np.std(trace.samples) * len(trace.samples))) + np.std(trace.samples) * len(trace.samples))) diff --git a/pyecsca/sampling.py b/pyecsca/sampling.py index b8d690d..29dc251 100644 --- a/pyecsca/sampling.py +++ b/pyecsca/sampling.py @@ -8,6 +8,14 @@ from .trace import Trace @public def downsample_average(trace: Trace, factor: int = 2) -> Trace: + """ + Downsample samples of `trace` by `factor` by averaging `factor` consecutive samples in + non-intersecting windows. + + :param trace: + :param factor: + :return: + """ resized = np.resize(trace.samples, len(trace.samples) - (len(trace.samples) % factor)) result_samples = resized.reshape(-1, factor).mean(axis=1).astype(trace.samples.dtype) return Trace(copy(trace.title), copy(trace.data), result_samples) @@ -15,11 +23,26 @@ def downsample_average(trace: Trace, factor: int = 2) -> Trace: @public def downsample_pick(trace: Trace, factor: int = 2, offset: int = 0) -> Trace: + """ + Downsample samples of `trace` by `factor` by picking each `factor`-th sample, starting at `offset`. + + :param trace: + :param factor: + :param offset: + :return: + """ result_samples = trace.samples[offset::factor].copy() return Trace(copy(trace.title), copy(trace.data), result_samples) @public def downsample_decimate(trace: Trace, factor: int = 2) -> Trace: + """ + Downsample samples of `trace` by `factor` by decimating. + + :param trace: + :param factor: + :return: + """ result_samples = decimate(trace.samples, factor) return Trace(copy(trace.title), copy(trace.data), result_samples) diff --git a/pyecsca/trace.py b/pyecsca/trace.py index ab68827..a9d6892 100644 --- a/pyecsca/trace.py +++ b/pyecsca/trace.py @@ -2,8 +2,12 @@ import weakref from numpy import ndarray from typing import Optional, Sequence +from public import public + +@public class Trace(object): + """A power trace, which has an optional title, optional data bytes and mandatory samples.""" def __init__(self, title: Optional[str], data: Optional[bytes], samples: ndarray, trace_set=None): @@ -30,7 +34,9 @@ class Trace(object): self.title, self.data, self.samples, self.trace_set) +@public class CombinedTrace(Trace): + """A power trace that was combined from other traces, `parents`.""" def __init__(self, title: Optional[str], data: Optional[bytes], samples: ndarray, trace_set=None, parents: Sequence[Trace] = None): diff --git a/pyecsca/trace_set/base.py b/pyecsca/trace_set/base.py index 1984b88..2221933 100644 --- a/pyecsca/trace_set/base.py +++ b/pyecsca/trace_set/base.py @@ -15,12 +15,15 @@ class TraceSet(object): self._keys = list(kwargs.keys()) def __len__(self): + """Return the number of traces.""" return len(self._traces) def __getitem__(self, index) -> Trace: + """Get the trace at `index`.""" return self._traces[index] def __iter__(self): + """Iterate over the traces.""" yield from self._traces def __repr__(self): diff --git a/pyecsca/trace_set/chipwhisperer.py b/pyecsca/trace_set/chipwhisperer.py index 52300f6..31e2479 100644 --- a/pyecsca/trace_set/chipwhisperer.py +++ b/pyecsca/trace_set/chipwhisperer.py @@ -9,6 +9,7 @@ from ..trace import Trace @public class ChipWhispererTraceSet(TraceSet): + """ChipWhisperer trace set (native) format.""" def __init__(self, path: str = None, name: str = None): if path is None and name is None: diff --git a/pyecsca/trace_set/inspector.py b/pyecsca/trace_set/inspector.py index fdf9552..dcb7918 100644 --- a/pyecsca/trace_set/inspector.py +++ b/pyecsca/trace_set/inspector.py @@ -4,6 +4,7 @@ from enum import IntEnum from io import BytesIO, RawIOBase, BufferedIOBase, UnsupportedOperation from pathlib import Path from public import public +from typing import Union, Optional, BinaryIO, List from .base import TraceSet from ..trace import Trace @@ -60,6 +61,8 @@ class Parsers(object): @public class InspectorTraceSet(TraceSet): + """Riscure Inspector trace set format (.trs).""" + num_traces: int num_samples: int sample_coding: SampleCoding @@ -97,7 +100,7 @@ class InspectorTraceSet(TraceSet): external_clock_frequencty: float = 0 external_clock_time_base: int = 0 - _raw_traces = None + _raw_traces: List[Trace] = None _tag_parsers: dict = { 0x41: ("num_traces", 4, Parsers.read_int, Parsers.write_int), 0x42: ("num_samples", 4, Parsers.read_int, Parsers.write_int), @@ -135,7 +138,14 @@ class InspectorTraceSet(TraceSet): } _set_tags: set = set() - def __init__(self, input=None, keep_raw_traces=True): + def __init__(self, input: Optional[Union[str, Path, bytes, BinaryIO]] = None, + keep_raw_traces: bool = True): + """ + Read Inspector trace set from file path, bytes or file-like object. + + :param input: Input file path, bytes or file-like object. + :param keep_raw_traces: Whether to store the raw (unscaled) traces as well. + """ traces = None if isinstance(input, bytes): with BytesIO(input) as f: @@ -220,10 +230,16 @@ class InspectorTraceSet(TraceSet): def _scale(self, traces): return list(map(lambda trace: Trace(trace.title, trace.data, - trace.samples.astype("f4") * self.y_scale, trace_set=self), + trace.samples.astype("f4") * self.y_scale, + trace_set=self), traces)) - def save(self, output): + def save(self, output: Union[Path, str, BinaryIO]): + """ + Save this trace set into a file. + + :param output: An output path or file-like object. + """ if isinstance(output, (Path, str)): with open(output, "wb") as f: self._write(f) @@ -233,19 +249,22 @@ class InspectorTraceSet(TraceSet): raise ValueError("Cannot save data, unknown output: {}".format(output)) def __bytes__(self): + """Return the byte-representation of this trace set file.""" with BytesIO() as b: self.save(b) return b.getvalue() @property - def raw(self): + def raw(self) -> Optional[List[Trace]]: + """The raw (unscaled) traces, as read from the trace set file.""" if self._raw_traces is None: return None return list(self._raw_traces) @property - def sampling_frequency(self): - return int(1/self.x_scale) + def sampling_frequency(self) -> int: + """The sampling frequency of the trace set.""" + return int(1 / self.x_scale) def __repr__(self): args = ", ".join(["{}={!r}".format(self._tag_parsers[set_tag][0], diff --git a/pyecsca/tvla.py b/pyecsca/ttest.py index 0a80ff9..b9ecd36 100644 --- a/pyecsca/tvla.py +++ b/pyecsca/ttest.py @@ -17,8 +17,24 @@ def ttest(first_set: Sequence[Trace], second_set: Sequence[Trace], @public def welch_ttest(first_set: Sequence[Trace], second_set: Sequence[Trace]) -> CombinedTrace: + """ + Perform the Welch's t-test sample wise on two sets of traces `first_set` and `second_set`. + Useful for Test Vector Leakage Analysis (TVLA). + + :param first_set: + :param second_set: + :return: Welch's t-values (samplewise) + """ return ttest(first_set, second_set, False) @public def student_ttest(first_set: Sequence[Trace], second_set: Sequence[Trace]) -> CombinedTrace: + """ + Perform the Students's t-test sample wise on two sets of traces `first_set` and `second_set`. + Useful for Test Vector Leakage Analysis (TVLA). + + :param first_set: + :param second_set: + :return: Student's t-values (samplewise) + """ return ttest(first_set, second_set, True) diff --git a/test/test_process.py b/test/test_process.py index 130d51f..a3ba85f 100644 --- a/test/test_process.py +++ b/test/test_process.py @@ -1,7 +1,8 @@ from unittest import TestCase import numpy as np -from pyecsca import Trace, absolute, invert, threshold, rolling_mean, offset, recenter +from pyecsca import Trace, absolute, invert, threshold, rolling_mean, offset, recenter, normalize, normalize_wl +from .utils import plot class ProcessTests(TestCase): @@ -12,26 +13,22 @@ class ProcessTests(TestCase): def test_absolute(self): result = absolute(self._trace) self.assertIsNotNone(result) - self.assertIsInstance(result, Trace) self.assertEqual(result.samples[1], 60) def test_invert(self): result = invert(self._trace) self.assertIsNotNone(result) - self.assertIsInstance(result, Trace) np.testing.assert_equal(result.samples, [-30, 60, -145, -247]) def test_threshold(self): result = threshold(self._trace, 128) self.assertIsNotNone(result) - self.assertIsInstance(result, Trace) self.assertEqual(result.samples[0], 0) self.assertEqual(result.samples[2], 1) def test_rolling_mean(self): result = rolling_mean(self._trace, 2) self.assertIsNotNone(result) - self.assertIsInstance(result, Trace) self.assertEqual(len(result.samples), 3) self.assertEqual(result.samples[0], -15) self.assertEqual(result.samples[1], 42) @@ -40,8 +37,15 @@ class ProcessTests(TestCase): def test_offset(self): result = offset(self._trace, 5) self.assertIsNotNone(result) - self.assertIsInstance(result, Trace) np.testing.assert_equal(result.samples, np.array([35, -55, 150, 252], dtype=np.dtype("i2"))) def test_recenter(self): self.assertIsNotNone(recenter(self._trace)) + + def test_normalize(self): + result = normalize(self._trace) + self.assertIsNotNone(result) + + def test_normalize_wl(self): + result = normalize_wl(self._trace) + self.assertIsNotNone(result)
\ No newline at end of file diff --git a/test/test_traceset.py b/test/test_traceset.py index c9b3fd4..fe847c7 100644 --- a/test/test_traceset.py +++ b/test/test_traceset.py @@ -12,6 +12,9 @@ class TraceSetTests(TestCase): self.assertIsNotNone(InspectorTraceSet()) self.assertIsNotNone(ChipWhispererTraceSet()) + +class InspectorTraceSetTests(TestCase): + def test_load_fname(self): result = InspectorTraceSet("test/data/example.trs") self.assertIsNotNone(result) @@ -52,4 +55,4 @@ class ChipWhispererTraceSetTest(TestCase): def test_load_fname(self): result = ChipWhispererTraceSet("test/data/", "chipwhisperer") self.assertIsNotNone(result) - print(result[:])
\ No newline at end of file + self.assertEqual(len(result), 2) diff --git a/test/test_tvla.py b/test/test_ttest.py index 4ff1359..49cc4ea 100644 --- a/test/test_tvla.py +++ b/test/test_ttest.py @@ -4,7 +4,7 @@ import numpy as np from pyecsca import Trace, welch_ttest, student_ttest -class TvlaTests(TestCase): +class TTestTests(TestCase): def setUp(self): self.a = Trace(None, b"\xff", np.array([20, 80], dtype=np.dtype("i1"))) |