path: root/CPA.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Correlation Power Analysis"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "This notebook demonstrates the functionality of the `CPA` class, which implements Correlation Power Analysis side-channel attack. The attack is performed and evaluated on leakage traces simulated by the `LeakageTarget` class."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Initialisation"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyecsca.ec.mult import LTRMultiplier\n",
    "from pyecsca.ec.mod import mod\n",
    "from pyecsca.ec.point import Point, InfinityPoint\n",
    "from pyecsca.ec.model import ShortWeierstrassModel\n",
    "from pyecsca.ec.curve import EllipticCurve\n",
    "from pyecsca.ec.params import DomainParameters\n",
    "from pyecsca.sca.attack.leakage_model import HammingWeight\n",
    "from pyecsca.sca.target.leakage import LeakageTarget\n",
    "from random import randint\n",
    "from pyecsca.sca.attack.CPA import CPA\n",
    "import holoviews as hv\n",
    "import multiprocessing as mp\n",
    "import matplotlib.pyplot as plt\n",
    "from pyecsca.sca.trace import Trace\n",
    "import numpy as np\n",
    "import warnings\n",
    "from scipy.stats import ConstantInputWarning"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "warnings.filterwarnings(\"ignore\", category=ConstantInputWarning)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "hv.extension(\"bokeh\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We first define the elliptic curve parameters we are going to be using for the demonstration."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = ShortWeierstrassModel()\n",
    "coords = model.coordinates[\"projective\"]\n",
    "p = 0xd7d1247f\n",
    "a = mod(0xa4a44016, p)\n",
    "b = mod(0x73f76716, p)\n",
    "n = 0xd7d2a475\n",
    "h = 1\n",
    "gx, gy, gz = mod(0x54eed6d7, p), mod(0x6f1e55ac, p), mod(1, p)\n",
    "generator = Point(coords, X=gx, Y=gy, Z=gz)\n",
    "neutral = InfinityPoint(coords)\n",
    "\n",
    "curve = EllipticCurve(model, coords, p, neutral, {\"a\": a, \"b\": b})\n",
    "params = DomainParameters(curve, generator, n, h)\n",
    "\n",
    "add = coords.formulas[\"add-2015-rcb\"]\n",
    "dbl = coords.formulas[\"dbl-2015-rcb\"]\n",
    "scl = coords.formulas[\"z\"]\n",
    "\n",
    "mult = LTRMultiplier(add, dbl, None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create and initialize an instance of the `LeakageTarget` class with the above elliptic curve parameters."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "target = LeakageTarget(model, coords, mult, HammingWeight())\n",
    "target.set_params(params)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Using the attack, we will try to recover an 8-bit scalar used in the scalar multiplication operation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "scalar_bit_length = 8\n",
    "secret_scalar = 229"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will perform scalar multiplication on 10 000 random points on the curve and simulate 10 000 corresponding leakage traces, which we will then use to evaluate the attack. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "num_of_traces = 500"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "generated_points, generated_traces = target.simulate_scalar_mult_traces(num_of_traces, secret_scalar)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Performing the attack"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We create and initialize an instance of the `CPA` class containing the attack's implementation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "mult.init(params, params.generator)\n",
    "real_pub_key = mult.multiply(secret_scalar)\n",
    "leakage_model = HammingWeight()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "cpa = CPA(generated_points, generated_traces, leakage_model, mult, params)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We perform the attack and test that the scalar we recovered is correct."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "recovered_scalar = cpa.perform(scalar_bit_length, real_pub_key)\n",
    "print(recovered_scalar)\n",
    "print(recovered_scalar == secret_scalar)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We can visualize the correlations calculated during the attack for correct and incorrect guess of the second bit of the scalar and correct guesses for fourth and sixth bit of the scalar to show where in the computation were the correlations stongest."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Correct difference of means when guessing 2. bit (guessed scalar 192 = 1100 0000)\n",
    "cpa.plot_correlations(cpa.correlations['guess_one'][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Incorrect difference of means when guessing 2. bit (guessed scalar 128 = 1000 0000)\n",
    "cpa.plot_correlations(cpa.correlations['guess_zero'][0])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Correct difference of means when guessing 4. bit (guessed scalar 224 = 1110 0000)\n",
    "cpa.plot_correlations(cpa.correlations['guess_zero'][2])"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Correct difference of means when guessing 6. bit (guessed scalar 229 = 1110 0100)\n",
    "cpa.plot_correlations(cpa.correlations['guess_one'][-2])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Evaluating the effectiveness of the attack"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We will evaluate how the success rate of the attack increases with an increasing number of traces and subsequently evaluate how the success rate for a fixed number of traces changes with increasing noise. "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "To have a convenient way of generating random subsets, we define the following function:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def select_random_sample(size, points, traces):\n",
    "    sample_points = []\n",
    "    sample_traces = []\n",
    "    selected_numbers = []\n",
    "    for _ in range(size):\n",
    "        num = randint(0, num_of_traces - 1)\n",
    "        if num not in selected_numbers:\n",
    "            selected_numbers.append(num)\n",
    "        sample_points.append(points[num])\n",
    "        sample_traces.append(traces[num])\n",
    "    return sample_points, sample_traces"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Success rate with changing number of traces"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We take subsets of different sizes and plot the success rate of the DPA corresponding to them. For each size, we perform the DPA 100 times on 100 generated subsets."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def CPA_traces(points, traces, sample_size, leakage_model, scalar_length, pub_key, multiplier, domain_params):\n",
    "    sample_points, sample_traces = select_random_sample(sample_size, points, traces)\n",
    "    sample_cpa = CPA(sample_points, sample_traces, leakage_model, multiplier, domain_params)\n",
    "    recovered = sample_cpa.perform(scalar_length, pub_key)\n",
    "    return recovered == secret_scalar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "number_of_subsets = 17\n",
    "successes_rates_traces = []\n",
    "number_of_samples = [x * 2 for x in range(1, number_of_subsets)]\n",
    "for num in number_of_samples:\n",
    "    with mp.Pool() as pool:\n",
    "        result_objects = [pool.apply_async(CPA_traces, args=(generated_points, generated_traces, num, leakage_model, scalar_bit_length, \n",
    "                                                             real_pub_key, mult, params)) for _ in range(100)]\n",
    "        results = [result.get() for result in result_objects]\n",
    "    successes_rates_traces.append(sum(results) / 100)\n",
    "    print(f\"done for {num}\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(successes_rates_traces)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We plot the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.plot(number_of_samples, successes_rates_traces, color='black')\n",
    "plt.ylim([0, 1])\n",
    "plt.xlabel('number of traces')\n",
    "plt.ylabel('CPA success rate')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Success rate with fixed number of traces and changing noise"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As we now know the the minimum amount of traces needed for 100% success rate, we generate 100 subsets with that amount of traces and perform CPA for each subset. We do this multiple times with different amount of noise."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def normalize_trace(trace: Trace):\n",
    "    min_val = np.min(trace.samples)\n",
    "    max_val = np.max(trace.samples)\n",
    "    scaled_samples = (trace.samples - min_val) / (max_val - min_val) \n",
    "    trace.samples = scaled_samples   \n",
    "\n",
    "def CPA_noise(points, traces, sample_size, standard_deviation, leakage_model, scalar_length, pub_key, multiplier, domain_params):\n",
    "    sample_points, sample_traces = select_random_sample(sample_size, points, traces)\n",
    "    for trace in sample_traces:\n",
    "        normalize_trace(trace)\n",
    "    noise = np.random.normal(0, standard_deviation, sample_traces[0].samples.shape)\n",
    "    for trace in sample_traces:\n",
    "        trace.samples = trace.samples + noise\n",
    "    sample_cpa = CPA(sample_points, sample_traces, leakage_model, multiplier, domain_params)\n",
    "    recovered = sample_cpa.perform(scalar_length, pub_key)\n",
    "    return recovered == secret_scalar"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "number_of_subsets = 17\n",
    "successes_rates_noise = []\n",
    "sample_size = 50\n",
    "noise_sds = [0.2 * x for x in range(number_of_subsets)] \n",
    "for sd in noise_sds:\n",
    "    with mp.Pool() as pool:\n",
    "        result_objects = [pool.apply_async(CPA_noise, args=(generated_points, generated_traces, sample_size, sd, leakage_model, scalar_bit_length, \n",
    "                                                             real_pub_key, mult, params)) for _ in range(100)]\n",
    "        results = [result.get() for result in result_objects]\n",
    "    successes_rates_noise.append(sum(results) / 100)\n",
    "    print(f\"done for {sd}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "We plot the results."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(successes_rates_noise)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "plt.plot(noise_sds[0:-1], successes_rates_noise[0:-1], color='black')\n",
    "plt.ylim([0, 1])\n",
    "plt.xlabel('Gaussian noise (standard deviation)')\n",
    "plt.ylabel('CPA success rate')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.11.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}