<jupyter_start><jupyter_text>SCA using hamming weight for classification.<jupyter_code>import os

import matplotlib.pyplot as plt
import numpy as np
from keras.backend import clear_session
from sklearn.utils import shuffle

from src.aes.sbox import s_box
from src.data.preprocess.hw import hamming_weights
from src.dlla.berg import make_mlp
from src.dlla.hw import prepare_traces_dl
from src.tools.plotter import init_plots
from src.trace_set.database import Database
from src.trace_set.pollution import PollutionType, Pollution
from src.trace_set.set_hw import TraceSetHW

init_plots()<jupyter_output><empty_output><jupyter_text>Guessing entropyExample from [here]() TODO<jupyter_code>def aes_labelize_ge_sr(attack_plaintext, key_hypothesis):
    key_byte = np.full(len(attack_plaintext), key_hypothesis)
    state = [int(x) ^ int(k) for x, k in zip(np.asarray(attack_plaintext), key_byte)]
    intermediate_values = s_box[state]

    return hamming_weights(intermediate_values)

# guessing entropy and success rate
def compute_ge(runs, model, correct_key, x_attack, attack_plaintext, key_rank_report_interval, key_rank_attack_traces):
    nt = len(x_attack)
    nt_interval = int(key_rank_attack_traces / key_rank_report_interval)
    key_ranking_sum = np.zeros(nt_interval)
    success_rate_sum = np.zeros(nt_interval)

    # ---------------------------------------------------------------------------------------------------------#
    # compute labels for key hypothesis
    # ---------------------------------------------------------------------------------------------------------#
    labels_key_hypothesis = np.zeros((256, nt))
    for key_byte_hypothesis in range(0, 256):
        labels_key_hypothesis[key_byte_hypothesis] = aes_labelize_ge_sr(attack_plaintext, key_byte_hypothesis)

    # ---------------------------------------------------------------------------------------------------------#
    # predict output probabilities for shuffled test or validation set
    # ---------------------------------------------------------------------------------------------------------#
    output_probabilities = model.predict(x_attack)

    probabilities_kg_all_traces = np.zeros((nt, 256))
    for index in range(nt):
        probabilities_kg_all_traces[index] = output_probabilities[index][
            np.asarray([int(leakage[index]) for leakage in labels_key_hypothesis[:]])
        ]

    # ---------------------------------------------------------------------------------------------------------#
    # run key rank "runs" times and average results.
    # ---------------------------------------------------------------------------------------------------------#
    for run in range(runs):
        probabilities_kg_all_traces_shuffled = shuffle(probabilities_kg_all_traces)

        key_probabilities = np.zeros(256)

        kr_count = 0
        for index in range(key_rank_attack_traces):

            key_probabilities += np.log(probabilities_kg_all_traces_shuffled[index] + 1e-36)
            key_probabilities_sorted = np.argsort(key_probabilities)[::-1]

            if (index + 1) % key_rank_report_interval == 0:
                key_ranking_good_key = list(key_probabilities_sorted).index(correct_key) + 1

                key_ranking_sum[kr_count] += key_ranking_good_key
                success_rate_sum[kr_count] += key_ranking_good_key == 1

                kr_count += 1

        # print("KR run: {} | final GE for correct key ({}): {})".format(run, correct_key, key_ranking_sum[nt_interval - 1] / (run + 1)))

    guessing_entropy = key_ranking_sum / runs
    success_rate = success_rate_sum / runs

    return guessing_entropy, success_rate
def store_results(database: Database, method: str, pollution: Pollution, ge: float, sr: float):
    file_name = f"./results/{database.name}.csv"

    with open(file_name, 'a') as f:
        f.write(f"{method};{pollution.type.name};{pollution.parameter};{ge};{sr}\n")
def get_ge(db: Database, pollution: Pollution, do_plot=False):
    # Load the profiling traces
    trace_set = TraceSetHW(db, pollution)
    attack_plaintext, attack_key = TraceSetHW(db).attack_meta()

    correct_key = attack_key[0]
    # In the ascad dataset, all attack keys should be the same.
    assert np.all(attack_key == correct_key)

    x_profiling, y_profiling, x_attack, y_attack = prepare_traces_dl(*trace_set.profile(), *trace_set.attack())

    num_profiling = 200000
    x_profiling = x_profiling[:num_profiling]
    y_profiling = y_profiling[:num_profiling]

    num_attack_traces = 1000

    model = make_mlp(x_profiling, y_profiling, progress=False)

    key_rank_num_traces = num_attack_traces
    key_rank_runs = 100

    ge, sr = compute_ge(key_rank_runs, model, correct_key, x_attack, attack_plaintext, 1, num_attack_traces)

    if do_plot:
        plt.plot(np.arange(1, key_rank_num_traces + 1), ge, label="GE")
        plt.xlabel("Traces")
        plt.ylabel("Guessing Entropy")
        plt.xlim([0, key_rank_num_traces])
        plt.legend()
        plt.show()

    print(f"{pollution.get_name()}: ge=({ge[-1]}), sr=({sr[-1]})")
    store_results(db, "sca_hw", pollution, ge[-1], sr[-1])

    return ge


# POLLUTION_PARAMETER = 0
# POLLUTION_TYPE = PollutionType.gauss
#
# GE = get_ge(Database.ascad, Pollution(POLLUTION_TYPE, POLLUTION_PARAMETER))
# FILE_SUFFIX = file_suffix(POLLUTION_TYPE, POLLUTION_PARAMETER)
# pd.DataFrame({"Profiled SCA": GE}).to_csv(f"results/sca-ge{FILE_SUFFIX}.csv")
# sns.lineplot(data=GE)
PARAMS = {
    Database.ascad: {
        PollutionType.desync: np.arange(0, 2.05, .05),
        PollutionType.gauss: np.arange(5, 8.5, .5) # np.arange(0, 5.1, .1)
    },
    Database.ascad_none: {
        PollutionType.desync: np.arange(0, 205, 5),
        PollutionType.gauss: np.arange(0, 205, 5)
    },
    Database.aisy: {
        PollutionType.desync: np.arange(0, 460, 10),
        PollutionType.gauss: np.union1d(np.arange(0, 4100, 100), np.arange(0, 410, 10))
    }
}

while True:
    for DB in [Database.ascad]: # [Database.ascad_none, Database.ascad]:
        for POLL_TYPE in [PollutionType.gauss]: # [PollutionType.desync, PollutionType.gauss]:
            for PARAM in PARAMS[DB][POLL_TYPE]:
                print(f"Calculating GE for {DB}, {POLL_TYPE}, {PARAM}")
                POLL = Pollution(POLL_TYPE, PARAM)
                TRS = TraceSetHW(DB, POLL)

                if os.path.exists(TRS.path):
                    get_ge(DB, POLL)
                    clear_session()<jupyter_output>Calculating GE for Database.ascad, PollutionType.gauss, 5.0
gauss_5-0: ge=(58.95), sr=(0.01)
Calculating GE for Database.ascad, PollutionType.gauss, 5.5
gauss_5-5: ge=(52.72), sr=(0.04)
Calculating GE for Database.ascad, PollutionType.gauss, 6.0
gauss_6-0: ge=(84.77), sr=(0.01)
Calculating GE for Database.ascad, PollutionType.gauss, 6.5
gauss_6-5: ge=(101.44), sr=(0.03)
Calculating GE for Database.ascad, PollutionType.gauss, 7.0
gauss_7-0: ge=(114.35), sr=(0.0)
Calculating GE for Database.ascad, PollutionType.gauss, 7.5
gauss_7-5: ge=(101.56), sr=(0.01)
Calculating GE for Database.ascad, PollutionType.gauss, 8.0
gauss_8-0: ge=(129.55), sr=(0.01)
Calculating GE for Database.ascad, PollutionType.gauss, 5.0
gauss_5-0: ge=(62.04), sr=(0.04)
Calculating GE for Database.ascad, PollutionType.gauss, 5.5
gauss_5-5: ge=(62.26), sr=(0.03)
Calculating GE for Database.ascad, PollutionType.gauss, 6.0
gauss_6-0: ge=(82.33), sr=(0.04)
Calculating GE for Database.ascad, PollutionType.gauss, 6.5
gauss_6[...]
{'fraction_non_alphanumeric': 0.19322124366159593, 'fraction_numerical': 0.027888977848945822, 'mean_word_length': 2.65609756097561, 'pattern_counts': {'":': 1, '<': 8, '<?xml version=': 0, '>': 8, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '623317', 'license': 'MIT', 'n_tokens_mistral': 2719, 'n_tokens_neox': 2468, 'n_words': 492, 'path': 'report/chapter-4/1-compute.ipynb', 'repo_name': 'ccberg/LA', 'size': '7494'}