全同态加密安全取证深度分析

git clone https://github.com/microsoft/SEAL.git
cd SEAL && cmake -S . -B build -DSEAL_BUILD_TESTS=ON
cmake --build build --config Release
cd build && ctest --output-on-failure

pip install fpylll lattice-estimator
python3 -c "
from estimator import *
params = LWE.Parameters(n=4096, q=2**32, Xs=ND.Uniform(0,1), Xe=ND.DiscreteGaussian(3.2))
result = LWE.estimate(params)
for k, v in result.items():
    print(f'{k}: {v}')
"

python3 -c "
import time
import statistics

def measure_ntt_timing(iterations=10000):
    timings = []
    for _ in range(iterations):
        start = time.perf_counter_ns()
        result = sum(range(4096)) % (2**32)
        end = time.perf_counter_ns()
        timings.append(end - start)
    return timings

timings = measure_ntt_timing()
print(f'Mean: {statistics.mean(timings):.1f} ns')
print(f'StdDev: {statistics.stdev(timings):.1f} ns')
print(f'Coefficient of Variation: {statistics.stdev(timings)/statistics.mean(timings)*100:.2f}%')
print(f'Min/Max ratio: {max(timings)/min(timings):.3f}')
"

#!/usr/bin/env python3
import struct
import time
import os
import hashlib

SAMPLE_RATE_MHZ = 200
SAMPLE_DURATION_US = 100
TRACES_PER_ATTACK = 10000
DATA_LENGTH_BYTES = 32

def generate_synthetic_power_trace(ciphertext_bytes, noise_seed):
    trace_length = SAMPLE_RATE_MHZ * SAMPLE_DURATION_US
    base_power = [0.0] * trace_length
    for i in range(trace_length):
        phase = (i % 256) / 256.0
        base_power[i] = 0.5 + 0.3 * (1 if phase < 0.5 else -1)
    for idx, byte_val in enumerate(ciphertext_bytes[:DATA_LENGTH_BYTES]):
        position = int((byte_val / 255.0) * trace_length) % trace_length
        for j in range(max(0, position - 5), min(trace_length, position + 6)):
            base_power[j] += 0.15 * (1.0 / (1 + abs(j - position)))
    seed_bytes = hashlib.sha256(noise_seed.to_bytes(8, 'big')).digest()
    for i in range(trace_length):
        seed_byte = seed_bytes[i % len(seed_bytes)] / 255.0
        base_power[i] += seed_byte * 0.05
    noise_amplitude = 0.02
    for i in range(trace_length):
        pseudo_random = (hashlib.md5(struct.pack('<II', i, noise_seed)).digest()[0] / 255.0 - 0.5) * 2
        base_power[i] += pseudo_random * noise_amplitude
    return base_power

def collect_power_traces(num_traces, output_file):
    traces = []
    for trace_idx in range(num_traces):
        ct_bytes = os.urandom(DATA_LENGTH_BYTES)
        noise_seed = trace_idx
        power_trace = generate_synthetic_power_trace(ct_bytes, noise_seed)
        traces.append({
            'trace_id': trace_idx,
            'ciphertext_hex': ct_bytes.hex(),
            'trace_length': len(power_trace),
            'trace_data': power_trace,
        })
    with open(output_file, 'w') as f:
        for trace in traces:
            line = f"{trace['trace_id']}|{trace['ciphertext_hex']}|"
            line += ','.join(f'{v:.6f}' for v in trace['trace_data'])
            f.write(line + '\n')
    return len(traces)

if __name__ == '__main__':
    num_traces = TRACES_PER_ATTACK
    output = 'fhe_power_traces.csv'
    collected = collect_power_traces(num_traces, output)
    print(f'Collected {collected} power traces -> {output}')

#!/bin/bash
echo "[+] FHE Cache Side-Channel Risk Assessment"
echo "============================================="

L1D_SIZE=$(sysctl -n hw.l1dcachesize 2>/dev/null || echo "unknown")
L2_SIZE=$(sysctl -n hw.l2cachesize 2>/dev/null || echo "unknown")
echo "[*] L1D Cache Size: $L1D_SIZE bytes"
echo "[*] L2 Cache Size: $L2_SIZE bytes"

if [ "$L1D_SIZE" != "unknown" ]; then
    if [ "$L1D_SIZE" -le 32768 ]; then
        echo "[HIGH] Small L1D cache increases Flush+Reload resolution"
    else
        echo "[INFO] L1D cache size: standard"
    fi
fi

echo "[*] Checking for CPU vulnerability mitigations..."
if [ -f /proc/cpuinfo ]; then
    grep -c "mitigation" /proc/cpuinfo 2>/dev/null
    echo "[*] Spectre mitigations:"
    cat /sys/devices/system/cpu/vulnerabilities/spectre_v1 2>/dev/null || echo "    Cannot read"
    cat /sys/devices/system/cpu/vulnerabilities/spectre_v2 2>/dev/null || echo "    Cannot read"
    cat /sys/devices/system/cpu/vulnerabilities/mds 2>/dev/null || echo "    Cannot read"
fi

echo ""
echo "[+] FHE Process Cache Behavior Analysis"
FHE_PROCS=$(pgrep -f "seal\|openfhe\|helib\|tfhe" 2>/dev/null)
if [ -n "$FHE_PROCS" ]; then
    for pid in $FHE_PROCS; do
        echo "[*] FHE process found: PID=$pid"
        cat /proc/$pid/status 2>/dev/null | grep -E "Threads|VmRSS|VmSize"
    done
else
    echo "[-] No FHE processes currently running"
fi

echo ""
echo "[+] Cache Partitioning Status (CAT)"
echo "    Check if Intel RDT/CAT is configured to isolate FHE workloads"
cat /sys/fs/resctrl/info/L3/cbm_mask 2>/dev/null || echo "    RDT not available"
cat /sys/fs/resctrl/info/L3/min_cbm_bits 2>/dev/null || echo "    "

#!/usr/bin/env python3
import os
import time
import json
import hashlib
from pathlib import Path

def collect_sidechannel_evidence(output_dir):
    os.makedirs(output_dir, exist_ok=True)
    evidence = {
        'collection_time': time.time(),
        'collection_time_iso': time.strftime('%Y-%m-%dT%H:%M:%S%z'),
        'cache_evidence': [],
        'timing_evidence': [],
        'process_evidence': [],
        'io_evidence': [],
    }
    cpu_info = collect_cpu_cache_info()
    evidence['cache_evidence'].append(cpu_info)
    fhe_libs = [
        '/usr/lib/libseal.so', '/usr/local/lib/libseal.so',
        '/usr/lib/libopenfhe.so', '/usr/local/lib/libopenfhe.so',
        '/usr/lib/libtfhe.so', '/usr/local/lib/libtfhe.so',
    ]
    for lib_path in fhe_libs:
        if os.path.exists(lib_path):
            stat = os.stat(lib_path)
            with open(lib_path, 'rb') as f:
                lib_hash = hashlib.sha256(f.read(65536)).hexdigest()
            evidence['io_evidence'].append({
                'path': lib_path,
                'size': stat.st_size,
                'mtime': stat.st_mtime,
                'first_64k_hash': lib_hash,
            })
    fhe_proc_names = ['seal_server', 'fhe_worker', 'openfhe_worker', 'homomorphic']
    for proc_name in fhe_proc_names:
        try:
            result = os.popen(f'pgrep -a {proc_name}').read().strip()
            if result:
                for line in result.split('\n'):
                    parts = line.split(' ', 1)
                    pid = int(parts[0])
                    timing_start = time.perf_counter_ns()
                    try:
                        with open(f'/proc/{pid}/status') as f:
                            status = f.read()
                        timing_end = time.perf_counter_ns()
                        evidence['process_evidence'].append({
                            'pid': pid,
                            'status_read_ns': timing_end - timing_start,
                            'status_lines': len(status.split('\n')),
                        })
                        evidence['timing_evidence'].append({
                            'target': f'/proc/{pid}/status',
                            'duration_ns': timing_end - timing_start,
                        })
                    except (FileNotFoundError, PermissionError):
                        pass
        except Exception:
            continue
    timing_samples = []
    for _ in range(100):
        t_start = time.perf_counter_ns()
        _ = hashlib.sha256(b'timing_sample').digest()
        t_end = time.perf_counter_ns()
        timing_samples.append(t_end - t_start)
    avg_timing = sum(timing_samples) / len(timing_samples)
    max_deviation = max(abs(t - avg_timing) for t in timing_samples)
    evidence['timing_evidence'].append({
        'target': 'sha256_baseline',
        'avg_ns': avg_timing,
        'max_deviation_ns': max_deviation,
        'samples': len(timing_samples),
        'cv_percent': (max_deviation / avg_timing) * 100,
    })
    output_path = os.path.join(output_dir, 'sidechannel_evidence.json')
    with open(output_path, 'w') as f:
        json.dump(evidence, f, indent=2)
    print(f'[+] Evidence collected: {output_path}')
    print(f'[+] Cache evidence entries: {len(evidence["cache_evidence"])}')
    print(f'[+] Timing evidence entries: {len(evidence["timing_evidence"])}')
    print(f'[+] Process evidence entries: {len(evidence["process_evidence"])}')
    print(f'[+] I/O evidence entries: {len(evidence["io_evidence"])}')
    return evidence

def collect_cpu_cache_info():
    info = {'type': 'cpu_cache'}
    try:
        with open('/proc/cpuinfo') as f:
            content = f.read()
        import re
        cache_matches = re.findall(r'cache size\s*:\s*(\d+)\s*kB', content)
        if cache_matches:
            info['l2_cache_kb'] = int(cache_matches[0])
        if os.path.exists('/sys/devices/system/cpu/cpu0/cache/index0/size'):
            with open('/sys/devices/system/cpu/cpu0/cache/index0/size') as f:
                info['l1d_size'] = f.read().strip()
    except Exception as e:
        info['error'] = str(e)
    return info

if __name__ == '__main__':
    evidence_dir = '/tmp/fhe_sidechannel_evidence'
    collect_sidechannel_evidence(evidence_dir)

#!/usr/bin/env python3
import math
import os
import hashlib
import json
import time

def estimate_ckks_noise_budget(params, computation_depth):
    poly_modulus_degree = params.get('poly_modulus_degree', 8192)
    coeff_modulus_sizes = params.get('coeff_modulus_sizes', [60, 40, 40, 60])
    plain_modulus_bits = params.get('plain_modulus_bits', 0)
    log_q = sum(coeff_modulus_sizes)
    log_p = plain_modulus_bits if plain_modulus_bits > 0 else 0
    security_level = estimate_security_level(poly_modulus_degree, log_q)
    initial_noise_bits = 20
    noise_budget = log_q - log_p - initial_noise_bits
    per_mul_growth = 2 * initial_noise_bits
    per_add_growth = initial_noise_bits
    remaining_budget = noise_budget
    for i in range(computation_depth):
        if i % 2 == 0:
            remaining_budget -= per_mul_growth
        else:
            remaining_budget -= per_add_growth
    return {
        'poly_modulus_degree': poly_modulus_degree,
        'log_q': log_q,
        'security_level': security_level,
        'initial_noise_budget': noise_budget,
        'remaining_after_depth': remaining_budget,
        'bootstrap_required': remaining_budget < 10,
        'is_secure': security_level >= 128 and remaining_budget > 0,
    }

def estimate_security_level(n, log_q):
    logq_per_logn = log_q / math.log2(n)
    if logq_per_logn < 1.0:
        return 128
    elif logq_per_logn < 1.5:
        return 112
    elif logq_per_logn < 2.0:
        return 100
    elif logq_per_logn < 2.5:
        return 80
    else:
        return 64

def detect_noise_anomaly(noise_samples, threshold_sigma=3.0):
    if len(noise_samples) < 10:
        return {'error': 'Insufficient samples'}
    mean = sum(noise_samples) / len(noise_samples)
    variance = sum((x - mean) ** 2 for x in noise_samples) / len(noise_samples)
    std_dev = math.sqrt(variance)
    anomalies = []
    for i, sample in enumerate(noise_samples):
        z_score = (sample - mean) / std_dev if std_dev > 0 else 0
        if abs(z_score) > threshold_sigma:
            anomalies.append({
                'index': i,
                'value': sample,
                'z_score': round(z_score, 4),
                'type': 'excessive_noise' if z_score > 0 else 'suppressed_noise',
            })
    return {
        'mean': round(mean, 6),
        'std_dev': round(std_dev, 6),
        'total_samples': len(noise_samples),
        'anomaly_count': len(anomalies),
        'anomalies': anomalies,
        'anomaly_rate': round(len(anomalies) / len(noise_samples) * 100, 2),
    }

def analyze_fhe_ciphertext_integrity(ciphertext_hex, params):
    ct_bytes = bytes.fromhex(ciphertext_hex)
    poly_count = int.from_bytes(ct_bytes[:4], 'big')
    coeff_bits = params.get('coeff_modulus_bits', 60)
    bytes_per_coeff = (coeff_bits + 7) // 8
    expected_size = 4 + poly_count * params.get('poly_modulus_degree', 8192) * bytes_per_coeff
    integrity_checks = {
        'poly_count': poly_count,
        'actual_size': len(ct_bytes),
        'expected_size_approx': expected_size,
        'size_ratio': round(len(ct_bytes) / max(expected_size, 1), 4),
        'high_entropy': calculate_entropy(ct_bytes) > 7.5,
        'is_plausible': poly_count in [1, 2, 3, 4] and len(ct_bytes) > 100,
    }
    return integrity_checks

def calculate_entropy(data):
    if not data:
        return 0.0
    byte_freq = [0] * 256
    for byte in data:
        byte_freq[byte] += 1
    entropy = 0.0
    data_len = len(data)
    for freq in byte_freq:
        if freq > 0:
            p = freq / data_len
            entropy -= p * math.log2(p)
    return entropy

if __name__ == '__main__':
    default_params = {
        'poly_modulus_degree': 8192,
        'coeff_modulus_sizes': [60, 40, 40, 60],
        'coeff_modulus_bits': 60,
        'plain_modulus_bits': 0,
    }
    print('[+] FHE Noise Overflow Attack Detection')
    print('=' * 50)
    for depth in [1, 3, 5, 7, 10, 15]:
        result = estimate_ckks_noise_budget(default_params, depth)
        status = 'SECURE' if result['is_secure'] else 'INSECURE'
        bootstrap = ' [BOOTSTRAP REQUIRED]' if result['bootstrap_required'] else ''
        print(f'  Depth {depth:2d}: remaining_budget={result["remaining_after_depth"]:3d} bits '
              f'| {status}{bootstrap}')
    sample_sizes = [100, 200, 500, 1000]
    for size in sample_sizes:
        import random
        random.seed(42)
        normal_samples = [random.gauss(0.0, 1.0) for _ in range(size)]
        if size >= 500:
            for j in range(5):
                inject_pos = j * (size // 5)
                normal_samples[inject_pos] = random.choice([-5.0, 5.0])
        anomaly_result = detect_noise_anomaly(normal_samples)
        print(f'  Samples={size}: anomalies={anomaly_result["anomaly_count"]} '
              f'| rate={anomaly_result["anomaly_rate"]}%')

#!/usr/bin/env python3
import hashlib
import struct
import os

def compute_ciphertext_mac(ciphertext_bytes, mac_key):
    key_material = hashlib.sha256(mac_key).digest()
    length_bytes = struct.pack('<I', len(ciphertext_bytes))
    hash_input = key_material + length_bytes + ciphertext_bytes
    return hashlib.sha256(hash_input).digest()

def verify_ciphertext_integrity(ciphertext_hex, expected_mac_hex, mac_key):
    ct_bytes = bytes.fromhex(ciphertext_hex)
    expected_mac = bytes.fromhex(expected_mac_hex)
    computed_mac = compute_ciphertext_mac(ct_bytes, mac_key)
    return computed_mac == expected_mac

def detect_bit_flip(original_ct_hex, suspect_ct_hex):
    original = bytes.fromhex(original_ct_hex)
    suspect = bytes.fromhex(suspect_ct_hex)
    if len(original) != len(suspect):
        return {'detected': True, 'type': 'size_mismatch',
                'original_size': len(original), 'suspect_size': len(suspect)}
    flipped_bits = []
    for i in range(len(original)):
        diff = original[i] ^ suspect[i]
        if diff != 0:
            for bit_pos in range(8):
                if diff & (1 << bit_pos):
                    flipped_bits.append({
                        'byte_offset': i,
                        'bit_position': bit_pos,
                        'original_bit': (original[i] >> bit_pos) & 1,
                        'suspect_bit': (suspect[i] >> bit_pos) & 1,
                    })
    return {
        'detected': len(flipped_bits) > 0,
        'total_flipped_bits': len(flipped_bits),
        'affected_bytes': len(set(b['byte_offset'] for b in flipped_bits)),
        'flipped_bits': flipped_bits,
        'severity': classify_bit_flip_severity(flipped_bits),
    }

def classify_bit_flip_severity(flipped_bits):
    if not flipped_bits:
        return 'none'
    byte_offsets = set(b['byte_offset'] for b in flipped_bits)
    if len(byte_offsets) > 10:
        return 'critical'
    high_bit_flips = [b for b in flipped_bits if b['bit_position'] >= 4]
    if high_bit_flips:
        return 'high'
    low_bit_flips = [b for b in flipped_bits if b['bit_position'] < 4]
    if len(low_bit_flips) > 5:
        return 'medium'
    return 'low'

def audit_ciphertext_transmission(ciphertexts, transmission_log):
    findings = []
    for idx, ct in enumerate(ciphertexts):
        entropy = calculate_ct_entropy(bytes.fromhex(ct))
        if entropy < 7.0:
            findings.append({
                'ciphertext_index': idx,
                'issue': 'low_entropy',
                'entropy': entropy,
                'severity': 'high',
            })
        size_kb = len(bytes.fromhex(ct)) / 1024
        if size_kb > 1024:
            findings.append({
                'ciphertext_index': idx,
                'issue': 'unusual_size',
                'size_kb': size_kb,
                'severity': 'medium',
            })
    return findings

def calculate_ct_entropy(data):
    if not data:
        return 0.0
    byte_freq = [0] * 256
    for b in data:
        byte_freq[b] += 1
    entropy = 0.0
    for freq in byte_freq:
        if freq > 0:
            p = freq / len(data)
            entropy -= p * (p and __import__('math').log2(p))
    return entropy

if __name__ == '__main__':
    print('[+] FHE Ciphertext Bit-Flip Attack Detection')
    print('=' * 50)
    os.urandom(32)
    original = os.urandom(256)
    suspect = bytearray(original)
    suspect[100] ^= 0x01
    suspect[200] ^= 0x80
    result = detect_bit_flip(original.hex(), suspect.hex())
    print(f'  Bit-flip detected: {result["detected"]}')
    print(f'  Flipped bits: {result["total_flipped_bits"]}')
    print(f'  Severity: {result["severity"]}')

#!/bin/bash
echo "[+] FHE Key Management Security Audit"
echo "======================================="

FHE_KEY_DIRS=(
    "/etc/seal"
    "/etc/openfhe"
    "/var/lib/fhe/keys"
    "/opt/fhe/keys"
    "$HOME/.fhe/keys"
)

FHE_KEY_EXTENSIONS=("*.key" "*.pem" "*.bin" "*.params" "*.relin*" "*.galois*" "*.bootstrap*")

echo "[*] Searching for FHE key files..."
for dir in "${FHE_KEY_DIRS[@]}"; do
    if [ -d "$dir" ]; then
        echo "[+] Found FHE key directory: $dir"
        for ext in "${FHE_KEY_EXTENSIONS[@]}"; do
            find "$dir" -name "$ext" -type f 2>/dev/null | while read -r keyfile; do
                perms=$(stat -c '%a' "$keyfile" 2>/dev/null || stat -f '%Lp' "$keyfile" 2>/dev/null)
                owner=$(stat -c '%U:%G' "$keyfile" 2>/dev/null || stat -f '%Su:%Sg' "$keyfile" 2>/dev/null)
                size=$(stat -c '%s' "$keyfile" 2>/dev/null || stat -f '%z' "$keyfile" 2>/dev/null)
                mtime=$(stat -c '%Y' "$keyfile" 2>/dev/null || stat -f '%m' "$keyfile" 2>/dev/null)
                echo "    [KEY] $keyfile"
                echo "      Owner: $owner | Perms: $perms | Size: $size | MTime: $mtime"
                if [ "$perms" -gt 600 ] 2>/dev/null; then
                    echo "      [WARNING] Key file permissions too permissive: $perms"
                fi
            done
        done
    fi
done

echo ""
echo "[*] Checking FHE process memory for key material..."
FHE_PIDS=$(pgrep -f "seal|openfhe|helib|tfhe|fhe" 2>/dev/null)
if [ -n "$FHE_PIDS" ]; then
    for pid in $FHE_PIDS; do
        echo "[+] Analyzing PID: $pid"
        vm_peak=$(grep VmPeak /proc/$pid/status 2>/dev/null | awk '{print $2}')
        vm_rss=$(grep VmRSS /proc/$pid/status 2>/dev/null | awk '{print $2}')
        echo "    VmPeak: ${vm_peak:-N/A} kB | VmRSS: ${vm_rss:-N/A} kB"
        smaps_rollup=$(cat /proc/$pid/smaps_rollup 2>/dev/null | head -20)
        if [ -n "$smaps_rollup" ]; then
            echo "    Memory mapping summary:"
            echo "$smaps_rollup" | sed 's/^/      /'
        fi
    done
else
    echo "[-] No FHE processes running"
fi

echo ""
echo "[*] Checking key file entropy (high entropy may indicate key material)..."
find /var/lib/fhe /etc/seal /opt/fhe -type f 2>/dev/null | head -20 | while read -r f; do
    if command -v ent &>/dev/null; then
        ent_output=$(ent -b "$f" 2>/dev/null | grep "Entropy")
        echo "    $f: $ent_output"
    else
        file_size=$(wc -c < "$f" 2>/dev/null)
        sha256=$(sha256sum "$f" 2>/dev/null | cut -d' ' -f1)
        echo "    $f: size=${file_size}B sha256=${sha256:0:16}..."
    fi
done

echo ""
echo "[+] Audit complete"

#!/usr/bin/env python3
import hashlib
import json
import os
import struct
import time

def verify_key_integrity(key_path, expected_hash=None, expected_size=None):
    result = {
        'path': key_path,
        'exists': os.path.exists(key_path),
        'checks': {},
    }
    if not result['exists']:
        result['checks']['file_exists'] = False
        return result
    result['checks']['file_exists'] = True
    stat = os.stat(key_path)
    result['size'] = stat.st_size
    result['mtime'] = stat.st_mtime
    result['permissions'] = oct(stat.st_mode)[-3:]
    if int(result['permissions']) > 0o600:
        result['checks']['permissions_safe'] = False
        result['warning'] = 'Key file permissions too permissive'
    else:
        result['checks']['permissions_safe'] = True
    with open(key_path, 'rb') as f:
        data = f.read()
    computed_hash = hashlib.sha256(data).hexdigest()
    result['sha256'] = computed_hash
    if expected_hash:
        result['checks']['hash_match'] = (computed_hash == expected_hash)
    if expected_size:
        result['checks']['size_match'] = (stat.st_size == expected_size)
    entropy = calculate_byte_entropy(data)
    result['entropy'] = round(entropy, 4)
    result['checks']['entropy_adequate'] = entropy > 7.0
    result['checks']['has_fhe_header'] = detect_fhe_key_header(data)
    return result

def calculate_byte_entropy(data):
    if not data:
        return 0.0
    freq = [0] * 256
    for b in data:
        freq[b] += 1
    entropy = 0.0
    for f in freq:
        if f > 0:
            p = f / len(data)
            entropy -= p * __import__('math').log2(p)
    return entropy

def detect_fhe_key_header(data):
    fhe_headers = {
        b'SEAL': 'Microsoft SEAL',
        b'OPENFHE': 'OpenFHE',
        b'HELI': 'IBM HElib',
        b'TFHE': 'TFHE',
    }
    for header, library in fhe_headers.items():
        if data[:len(header)] == header:
            return library
    return None

def audit_key_lifecycle(key_registry_path):
    if not os.path.exists(key_registry_path):
        return {'error': 'Key registry not found'}
    with open(key_registry_path, 'r') as f:
        registry = json.load(f)
    findings = []
    current_time = time.time()
    max_key_age_days = 90
    for key_id, key_info in registry.items():
        age_days = (current_time - key_info.get('created_at', current_time)) / 86400
        if age_days > max_key_age_days:
            findings.append({
                'key_id': key_id,
                'issue': 'key_age_exceeded',
                'age_days': round(age_days, 1),
                'severity': 'high',
            })
        if key_info.get('usage_count', 0) > 1000000:
            findings.append({
                'key_id': key_id,
                'issue': 'excessive_usage',
                'usage_count': key_info['usage_count'],
                'severity': 'medium',
            })
        last_rotated = key_info.get('last_rotated', 0)
        rotation_age_days = (current_time - last_rotated) / 86400
        if rotation_age_days > 30:
            findings.append({
                'key_id': key_id,
                'issue': 'rotation_overdue',
                'days_since_rotation': round(rotation_age_days, 1),
                'severity': 'medium',
            })
    return {'findings': findings, 'total_keys': len(registry)}

if __name__ == '__main__':
    print('[+] FHE Key Management Security Audit')
    print('=' * 50)
    test_key = os.urandom(256)
    test_path = '/tmp/test_fhe_key.bin'
    with open(test_path, 'wb') as f:
        f.write(test_key)
    result = verify_key_integrity(test_path)
    print(f'  Key: {result["path"]}')
    print(f'  SHA256: {result["sha256"][:32]}...')
    print(f'  Entropy: {result["entropy"]}')
    print(f'  Permissions: {result["permissions"]}')
    os.remove(test_path)

#!/usr/bin/env python3
import hashlib
import json
import time
import uuid

class FHEAuditLogger:
    def __init__(self, session_id=None):
        self.session_id = session_id or str(uuid.uuid4())
        self.log_entries = []
        self.computation_count = 0
        self.total_noise_consumed = 0
        self.key_versions = {}
        self.parameter_hashes = {}
        self.suspicious_patterns = []

    def log_encryption(self, plaintext_hash, parameter_set, key_version, noise_budget_initial):
        entry = {
            'event_type': 'ENCRYPTION',
            'timestamp': time.time(),
            'timestamp_iso': time.strftime('%Y-%m-%dT%H:%M:%S%z'),
            'session_id': self.session_id,
            'sequence': self.computation_count,
            'plaintext_hash': plaintext_hash,
            'parameter_set': parameter_set,
            'key_version': key_version,
            'noise_budget_initial': noise_budget_initial,
        }
        self.key_versions[key_version] = time.time()
        self.log_entries.append(entry)
        self.computation_count += 1
        return entry

    def log_computation(self, operation, ciphertext_in_hash, ciphertext_out_hash,
                        noise_consumed, computation_depth, duration_ns):
        entry = {
            'event_type': 'COMPUTATION',
            'timestamp': time.time(),
            'timestamp_iso': time.strftime('%Y-%m-%dT%H:%M:%S%z'),
            'session_id': self.session_id,
            'sequence': self.computation_count,
            'operation': operation,
            'ciphertext_in_hash': ciphertext_in_hash,
            'ciphertext_out_hash': ciphertext_out_hash,
            'noise_consumed': noise_consumed,
            'computation_depth': computation_depth,
            'duration_ns': duration_ns,
        }
        self.total_noise_consumed += noise_consumed
        self.computation_count += 1
        if noise_consumed > 50:
            self.suspicious_patterns.append({
                'type': 'excessive_noise_consumption',
                'sequence': entry['sequence'],
                'noise_consumed': noise_consumed,
            })
        if duration_ns > 10000000:
            self.suspicious_patterns.append({
                'type': 'abnormal_computation_duration',
                'sequence': entry['sequence'],
                'duration_ns': duration_ns,
            })
        self.log_entries.append(entry)
        return entry

    def log_decryption(self, ciphertext_hash, key_version, noise_budget_final,
                       decryption_valid):
        entry = {
            'event_type': 'DECRYPTION',
            'timestamp': time.time(),
            'timestamp_iso': time.strftime('%Y-%m-%dT%H:%M:%S%z'),
            'session_id': self.session_id,
            'sequence': self.computation_count,
            'ciphertext_hash': ciphertext_hash,
            'key_version': key_version,
            'noise_budget_final': noise_budget_final,
            'decryption_valid': decryption_valid,
        }
        self.computation_count += 1
        if not decryption_valid:
            self.suspicious_patterns.append({
                'type': 'decryption_failure',
                'sequence': entry['sequence'],
                'key_version': key_version,
            })
        self.log_entries.append(entry)
        return entry

    def generate_audit_report(self):
        operations = {}
        for entry in self.log_entries:
            if entry['event_type'] == 'COMPUTATION':
                op = entry['operation']
                operations[op] = operations.get(op, 0) + 1
        report = {
            'session_id': self.session_id,
            'total_events': len(self.log_entries),
            'computation_count': self.computation_count,
            'total_noise_consumed': self.total_noise_consumed,
            'operation_distribution': operations,
            'key_versions_used': list(self.key_versions.keys()),
            'suspicious_patterns': self.suspicious_patterns,
            'suspicious_pattern_count': len(self.suspicious_patterns),
            'risk_level': self._assess_risk_level(),
        }
        return report

    def _assess_risk_level(self):
        if len(self.suspicious_patterns) > 5:
            return 'CRITICAL'
        elif len(self.suspicious_patterns) > 2:
            return 'HIGH'
        elif len(self.suspicious_patterns) > 0:
            return 'MEDIUM'
        return 'LOW'

    def export_log(self, filepath):
        with open(filepath, 'w') as f:
            json.dump({
                'log_entries': self.log_entries,
                'audit_report': self.generate_audit_report(),
            }, f, indent=2)

if __name__ == '__main__':
    logger = FHEAuditLogger()
    logger.log_encryption(
        plaintext_hash=hashlib.sha256(b'secret_data').hexdigest(),
        parameter_set='CKKS_8192_128bit',
        key_version='v1.0',
        noise_budget_initial=58
    )
    for i in range(5):
        ct_in = hashlib.sha256(f'ct_{i}'.encode()).hexdigest()
        ct_out = hashlib.sha256(f'ct_{i+1}'.encode()).hexdigest()
        logger.log_computation(
            operation='multiply',
            ciphertext_in_hash=ct_in,
            ciphertext_out_hash=ct_out,
            noise_consumed=12,
            computation_depth=i+1,
            duration_ns=500000 + i * 100000
        )
    logger.log_decryption(
        ciphertext_hash=hashlib.sha256(b'final_ct').hexdigest(),
        key_version='v1.0',
        noise_budget_final=2,
        decryption_valid=True
    )
    report = logger.generate_audit_report()
    print(json.dumps(report, indent=2))

#!/bin/bash
echo "[+] FHE Library Deployment Security Audit"
echo "==========================================="

FHE_LIB_PATHS=(
    "/usr/lib"
    "/usr/local/lib"
    "/opt/fhe/lib"
)

FHE_HEADERS=(
    "seal.h"
    "openfhe.h"
    "helib.h"
    "tfhe.h"
)

echo "[*] Scanning for FHE library files..."
for lib_path in "${FHE_LIB_PATHS[@]}"; do
    if [ -d "$lib_path" ]; then
        find "$lib_path" -type f \( -name "*.so" -o -name "*.a" -o -name "*.dylib" \) 2>/dev/null | while read -r lib; do
            if ldd "$lib" 2>/dev/null | grep -qi "seal\|openfhe\|helib\|tfhe\|fhe"; then
                echo "[+] FHE-related library: $lib"
                sha256sum "$lib" 2>/dev/null || shasum -a 256 "$lib" 2>/dev/null
                file "$lib"
                echo "    Size: $(wc -c < "$lib" 2>/dev/null) bytes"
                echo "    Linked libraries:"
                ldd "$lib" 2>/dev/null | head -10 | sed 's/^/      /'
                echo ""
            fi
        done
    fi
done

echo "[*] Checking FHE package versions..."
for pkg_manager in apt pip conda vcpkg; do
    case $pkg_manager in
        apt)
            dpkg -l 2>/dev/null | grep -i "seal\|openfhe\|helib\|tfhe" | head -10
            ;;
        pip)
            pip list 2>/dev/null | grep -i "seal\|openfhe\|he\|tfhe\|fhe" | head -10
            ;;
        conda)
            conda list 2>/dev/null | grep -i "seal\|openfhe\|he\|tfhe\|fhe" | head -10
            ;;
        vcpkg)
            vcpkg list 2>/dev/null | grep -i "seal\|openfhe\|he\|tfhe\|fhe" | head -10
            ;;
    esac
done

echo ""
echo "[*] Checking FHE compilation flags (security-relevant)..."
for lib in /usr/lib/libseal.so* /usr/local/lib/libseal.so*; do
    if [ -f "$lib" ]; then
        echo "[+] Checking $lib"
        readelf -p .comment "$lib" 2>/dev/null | head -5
        readelf -d "$lib" 2>/dev/null | grep -E "NEEDED|RPATH|RUNPATH" | head -10 | sed 's/^/    /'
        if readelf -s "$lib" 2>/dev/null | grep -q "debug\|_debug"; then
            echo "    [WARNING] Debug symbols found in production library"
        fi
    fi
done

echo ""
echo "[+] Audit complete"

环境证据 → 攻击前提 → 攻击行为 → 直接影响 → 间接影响
    ↓          ↓          ↓          ↓          ↓
系统配置   参数变更   侧信道采集   密文损坏   计算错误
部署状态   密钥获取   密文注入   密钥泄露   数据泄露
网络配置   权限提升   噪声操纵   计算篡改   业务影响

title: FHE Abnormal Computation Pattern Detection
id: a7f3c2e1-4b8d-4e9a-b5c6-d7e8f9a0b1c2
status: experimental
description: Detects abnormal computation patterns in FHE systems that may indicate ciphertext manipulation or noise overflow attacks
references:
  - https://fhe.org/security
  - https://github.com/microsoft/SEAL
author: Blue Team FHE Analysis
date: 2026/07/30
tags:
  - attack.defense_evasion
  - attack.t1027
  - fhe.security
  - cryptography.forensics
logsource:
  category: application
  product: fhe
detection:
  selection_noise_overflow:
    EventID: 4001
    NoiseConsumed|gt: 50
  selection_abnormal_depth:
    EventID: 4002
    ComputationDepth|gt: 20
  selection_bootstrap_anomaly:
    EventID: 4003
    BootstrapFrequency|gt: 100
    TimeWindow: 60
  selection_ct_integrity_fail:
    EventID: 4004
    CiphertextIntegrityCheck: "FAILED"
  selection_key_access_anomaly:
    EventID: 4005
    AccessType: "UNAUTHORIZED"
    KeyType: relinearization_key
  condition: selection_noise_overflow or selection_abnormal_depth or selection_bootstrap_anomaly or selection_ct_integrity_fail or selection_key_access_anomaly
level: high
falsepositives:
  - Legitimate deep FHE circuits with high computation depth
  - Noise budget calculation approximations
fields:
  - EventID
  - NoiseConsumed
  - ComputationDepth
  - BootstrapFrequency
  - CiphertextIntegrityCheck
  - AccessType
  - KeyType

title: FHE Key Usage Anomaly Detection
id: b8e4d3f2-5c9e-4f0a-c6d7-e8f9a0b1c2d3
status: experimental
description: Detects anomalous FHE key usage patterns including unauthorized access, rotation failures, and potential key extraction
references:
  - https://fhe.org/security
  - https://github.com/openfheorg/openfhe-development
author: Blue Team FHE Analysis
date: 2026/07/30
tags:
  - attack.credential_access
  - attack.t1557
  - fhe.key_management
  - cryptography.forensics
logsource:
  category: application
  product: fhe
detection:
  selection_unauthorized_key_access:
    EventID: 5001
    KeyType: "SECRET_KEY"
    AccessResult: "UNAUTHORIZED"
  selection_key_export_anomaly:
    EventID: 5002
    KeyType: "RELINEARIZATION_KEY"
    ExportDestination: "REMOTE"
  selection_key_rotation_failure:
    EventID: 5003
    EventType: "KEY_ROTATION"
    Result: "FAILED"
  selection_key_permission_change:
    EventID: 5004
    ObjectType: "KEY_FILE"
    ChangeType: "PERMISSIONS"
    NewPermissions|gt: "0600"
  selection_galois_key_mass_export:
    EventID: 5005
    KeyType: "GALOIS_KEY"
    ExportCount|gt: 10
    TimeWindow: 300
  condition: selection_unauthorized_key_access or selection_key_export_anomaly or selection_key_rotation_failure or selection_key_permission_change or selection_galois_key_mass_export
level: critical
falsepositives:
  - Planned key rotation operations
  - Bulk key distribution during initial deployment
fields:
  - EventID
  - KeyType
  - AccessResult
  - ExportDestination
  - Result
  - ChangeType
  - NewPermissions
  - ExportCount

#!/bin/bash
echo "[+] FHE Library Integrity & Version Audit"
echo "==========================================="

REPORT_FILE="/tmp/fhe_audit_report_$(date +%Y%m%d_%H%M%S).txt"
mkdir -p "$(dirname "$REPORT_FILE")"

{
echo "FHE Library Security Audit Report"
echo "Generated: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
echo "Hostname: $(hostname)"
echo "========================================"
echo ""

FHE_LIBRARIES=(
    "libseal:libseal.so:libseal.a:libseal.dylib"
    "libopenfhe:libopenfhe.so:libopenfhe.a:libopenfhe.dylib"
    "libhelib:libHElib.so:libHElib.a:libHElib.dylib"
    "libtfhe:libtfhe.so:libtfhe.a:libtfhe.dylib"
)

echo "[1] FHE Library Search"
echo "----------------------"
for lib_entry in "${FHE_LIBRARIES[@]}"; do
    lib_name=$(echo "$lib_entry" | cut -d: -f1)
    IFS=: read -ra lib_files <<< "$lib_entry"
    found=0
    for lib_file in "${lib_files[@]:1}"; do
        while IFS= read -r -d '' lib_path; do
            echo "[FOUND] $lib_name: $lib_path"
            sha256=$(sha256sum "$lib_path" 2>/dev/null || shasum -a 256 "$lib_path" 2>/dev/null)
            echo "  SHA256: $sha256"
            lib_size=$(wc -c < "$lib_path" 2>/dev/null)
            echo "  Size: $lib_size bytes"
            lib_mtime=$(stat -c '%Y' "$lib_path" 2>/dev/null || stat -f '%m' "$lib_path" 2>/dev/null)
            echo "  Modified: $(date -d "@$lib_mtime" '+%Y-%m-%d %H:%M:%S' 2>/dev/null || date -r "$lib_mtime" '+%Y-%m-%d %H:%M:%S' 2>/dev/null)"
            file_output=$(file "$lib_path" 2>/dev/null)
            echo "  Type: $file_output"
            if ldd "$lib_path" 2>/dev/null | grep -qi "not found"; then
                echo "  [WARNING] Missing shared library dependencies:"
                ldd "$lib_path" 2>/dev/null | grep "not found" | sed 's/^/    /'
            fi
            readelf -d "$lib_path" 2>/dev/null | grep -E "RPATH|RUNPATH" | sed 's/^/  /'
            echo ""
            found=1
            break
        done < <(find /usr/lib /usr/local/lib /opt -maxdepth 3 -name "$lib_file" -type f -print0 2>/dev/null)
        [ $found -eq 1 ] && break
    done
    [ $found -eq 0 ] && echo "[NOT FOUND] $lib_name"
done

echo ""
echo "[2] FHE Package Version Check"
echo "-----------------------------"
echo "--- pip packages ---"
pip3 list 2>/dev/null | grep -iE "seal|openfhe|he_|tfhe|fhe|concrete" || echo "  No pip FHE packages found"
echo "--- apt packages ---"
dpkg -l 2>/dev/null | grep -iE "seal|openfhe|he|tfhe|fhe" || echo "  No apt FHE packages found"
echo "--- conda packages ---"
conda list 2>/dev/null | grep -iE "seal|openfhe|he|tfhe|fhe" || echo "  No conda FHE packages found"
echo "--- brew packages ---"
brew list 2>/dev/null | grep -iE "seal|openfhe|he|tfhe|fhe" || echo "  No brew FHE packages found"

echo ""
echo "[3] Known Vulnerable Versions Check"
echo "------------------------------------"
VULN_DB=(
    "SEAL:4.0.0:CVE-2024-XXXX:NTT buffer overflow"
    "SEAL:3.x:EOL-SEAL3:End-of-life, no security patches"
    "HElib:2.0.0:EOL-HELIB:End-of-life, maintenance only"
    "TFHE-rs:0.1.0:EARLY-VERSION:Pre-release security limitations"
)

for entry in "${VULN_DB[@]}"; do
    IFS=: read -r lib min_ver cve desc <<< "$entry"
    echo "  [$lib] Version $min_ver+: $cve - $desc"
done

echo ""
echo "[4] FHE Process Security Check"
echo "-------------------------------"
FHE_PROCS=$(ps aux 2>/dev/null | grep -iE "seal|openfhe|helib|tfhe|fhe" | grep -v grep)
if [ -n "$FHE_PROCS" ]; then
    echo "$FHE_PROCS" | while read -r line; do
        echo "  $line"
        pid=$(echo "$line" | awk '{print $2}')
        proc_user=$(echo "$line" | awk '{print $1}')
        echo "    User: $proc_user"
        if [ "$proc_user" = "root" ]; then
            echo "    [WARNING] FHE process running as root"
        fi
        cat /proc/$pid/status 2>/dev/null | grep -E "Threads|VmRSS|VmSize" | sed 's/^/    /'
    done
else
    echo "  No FHE processes currently running"
fi

echo ""
echo "[5] FHE Configuration Files Audit"
echo "----------------------------------"
CONF_DIRS=("/etc/seal" "/etc/openfhe" "/opt/fhe/conf" "/etc/fhe")
for dir in "${CONF_DIRS[@]}"; do
    if [ -d "$dir" ]; then
        echo "  Config directory: $dir"
        find "$dir" -type f -name "*.json" -o -name "*.yaml" -o -name "*.conf" -o -name "*.params" 2>/dev/null | while read -r conf; do
            echo "    [$conf]"
            file_perm=$(stat -c '%a' "$conf" 2>/dev/null || stat -f '%Lp' "$conf" 2>/dev/null)
            echo "      Permissions: $file_perm"
            if [ "$file_perm" -gt 644 ] 2>/dev/null; then
                echo "      [WARNING] Configuration file has overly permissive access"
            fi
        done
    fi
done

echo ""
echo "========================================"
echo "Audit complete: $(date -u '+%Y-%m-%dT%H:%M:%SZ')"
} | tee "$REPORT_FILE"

echo ""
echo "[+] Report saved to: $REPORT_FILE"

#!/usr/bin/env python3
import math
import os
import json
import hashlib
import time
import struct
import re

SECURITY_LEVELS = {
    128: {"name": "128-bit", "min_poly_mod_degree": 4096, "min_security_margin": 1.0},
    112: {"name": "112-bit", "min_poly_mod_degree": 2048, "min_security_margin": 0.8},
    100: {"name": "100-bit", "min_poly_mod_degree": 1024, "min_security_margin": 0.6},
    80: {"name": "80-bit", "min_poly_mod_degree": 512, "min_security_margin": 0.4},
}

KNOWN_VULNERABLE_CONFIGS = [
    {"poly_mod_degree": 1024, "coeff_bits": 60, "vuln": "CFSGG20", "severity": "critical"},
    {"poly_mod_degree": 2048, "coeff_bits": 40, "vuln": "INSUFFICIENT_SECURITY", "severity": "high"},
    {"poly_mod_degree": 8192, "coeff_bits": 120, "vuln": "OVERSIZED_PARAMETERS", "severity": "low"},
]


def estimate_lwe_security(n, log_q):
    ratio = log_q / math.log2(n) if n > 1 else float('inf')
    if ratio < 0.8:
        return 150
    elif ratio < 1.0:
        return 128
    elif ratio < 1.3:
        return 112
    elif ratio < 1.8:
        return 100
    elif ratio < 2.5:
        return 80
    elif ratio < 3.5:
        return 64
    else:
        return 40


def assess_fhe_parameters(params):
    poly_mod_degree = params.get("poly_modulus_degree", 8192)
    coeff_modulus_sizes = params.get("coeff_modulus_sizes", [60, 40, 40, 60])
    log_q = sum(coeff_modulus_sizes)
    security_level = estimate_lwe_security(poly_mod_degree, log_q)
    assessment = {
        "poly_modulus_degree": poly_mod_degree,
        "coeff_modulus_sizes": coeff_modulus_sizes,
        "log_q": log_q,
        "estimated_security_level": security_level,
        "security_level_name": f"{security_level}-bit",
        "findings": [],
        "risk_score": 0,
    }
    min_required = 128
    if security_level < min_required:
        assessment["findings"].append({
            "type": "INSUFFICIENT_SECURITY",
            "severity": "critical",
            "detail": f"Security level {security_level} below minimum {min_required}",
        })
        assessment["risk_score"] += 50
    if poly_mod_degree < 4096:
        assessment["findings"].append({
            "type": "SMALL_POLYNOMIAL_MODULUS",
            "severity": "high",
            "detail": f"PolyModulusDegree {poly_mod_degree} may be vulnerable to lattice attacks",
        })
        assessment["risk_score"] += 30
    if log_q / math.log2(poly_mod_degree) > 3.0:
        assessment["findings"].append({
            "type": "OVERSIZED_PARAMETERS",
            "severity": "low",
            "detail": "Parameters oversized, potential performance issue",
        })
        assessment["risk_score"] += 5
    if any(cb > 61 for cb in coeff_modulus_sizes):
        assessment["findings"].append({
            "type": "LARGE_COEFFICIENT_BITS",
            "severity": "medium",
            "detail": "Coefficient bits exceed 61, may impact NTT performance",
        })
        assessment["risk_score"] += 10
    if len(coeff_modulus_sizes) > 6:
        assessment["findings"].append({
            "type": "EXCESSIVE_MODULI_CHAIN",
            "severity": "medium",
            "detail": f"Chain length {len(coeff_modulus_sizes)} may indicate overly deep circuits",
        })
        assessment["risk_score"] += 10
    for vuln_config in KNOWN_VULNERABLE_CONFIGS:
        if (poly_mod_degree == vuln_config["poly_mod_degree"] and
                log_q == vuln_config["coeff_bits"]):
            assessment["findings"].append({
                "type": "KNOWN_VULNERABLE_CONFIG",
                "severity": vuln_config["severity"],
                "detail": f"Matches known vulnerable configuration: {vuln_config['vuln']}",
            })
            assessment["risk_score"] += 40
    assessment["risk_level"] = (
        "CRITICAL" if assessment["risk_score"] >= 50 else
        "HIGH" if assessment["risk_score"] >= 30 else
        "MEDIUM" if assessment["risk_score"] >= 15 else
        "LOW"
    )
    return assessment


def scan_sidechannel_risk(fhe_lib_path):
    scan_result = {
        "lib_path": fhe_lib_path,
        "exists": os.path.exists(fhe_lib_path),
        "findings": [],
        "risk_score": 0,
    }
    if not scan_result["exists"]:
        scan_result["findings"].append({
            "type": "LIBRARY_NOT_FOUND",
            "severity": "info",
            "detail": f"Library not found at {fhe_lib_path}",
        })
        return scan_result
    try:
        with open(fhe_lib_path, "rb") as f:
            lib_data = f.read()
    except PermissionError:
        scan_result["findings"].append({
            "type": "PERMISSION_DENIED",
            "severity": "medium",
            "detail": "Cannot read library for analysis",
        })
        return scan_result
    stat = os.stat(fhe_lib_path)
    scan_result["file_size"] = stat.st_size
    scan_result["sha256"] = hashlib.sha256(lib_data).hexdigest()
    if stat.st_mode & 0o002:
        scan_result["findings"].append({
            "type": "WORLD_WRITABLE",
            "severity": "critical",
            "detail": "Library is world-writable, can be replaced with malicious version",
        })
        scan_result["risk_score"] += 40
    if stat.st_mode & 0o004:
        scan_result["findings"].append({
            "type": "WORLD_READABLE",
            "severity": "low",
            "detail": "Library is world-readable, may facilitate reverse engineering",
        })
        scan_result["risk_score"] += 5
    timing_patterns = [b"rdtsc", b"clock_gettime", b"QueryPerformanceCounter", b"__rdtsc"]
    for pattern in timing_patterns:
        if pattern in lib_data:
            scan_result["findings"].append({
                "type": "TIMING_INSTRUCTION",
                "severity": "medium",
                "detail": f"High-resolution timing instruction found: {pattern.decode()}",
            })
            scan_result["risk_score"] += 10
    debug_patterns = [b"__assert", b"assert(", b"fprintf(stderr", b"abort()"]
    for pattern in debug_patterns:
        if pattern in lib_data:
            scan_result["findings"].append({
                "type": "DEBUG_CODE_PRESENT",
                "severity": "medium",
                "detail": f"Debug code found: {pattern.decode()[:20]}",
            })
            scan_result["risk_score"] += 10
    if b"strcpy" in lib_data or b"strcat" in lib_data or b"gets" in lib_data:
        scan_result["findings"].append({
            "type": "UNSAFE_STRING_FUNCTION",
            "severity": "high",
            "detail": "Unsafe C string functions detected, potential buffer overflow risk",
        })
        scan_result["risk_score"] += 25
    if b"ntt" in lib_data.lower():
        scan_result["findings"].append({
            "type": "NTT_IMPLEMENTATION_DETECTED",
            "severity": "info",
            "detail": "NTT implementation detected, check for timing side channels",
        })
    scan_result["risk_level"] = (
        "CRITICAL" if scan_result["risk_score"] >= 50 else
        "HIGH" if scan_result["risk_score"] >= 30 else
        "MEDIUM" if scan_result["risk_score"] >= 15 else
        "LOW"
    )
    return scan_result


def scan_fhe_source_directory(source_dir):
    scan_result = {
        "directory": source_dir,
        "files_scanned": 0,
        "findings": [],
    }
    fhe_extensions = [".cpp", ".h", ".hpp", ".c", ".rs", ".py"]
    security_patterns = {
        r"reinterpret_cast<.*\*>": "REINTERPRET_CAST",
        r"memcpy\s*\(": "UNSAFE_MEMCPY",
        r"alloca\s*\(": "STACK_ALLOCATION",
        r"#pragma\s+ omp\s+ parallel": "OPENMP_PARALLEL",
        r"__m256": "AVX_INSTRUCTION",
        r"__m512": "AVX512_INSTRUCTION",
        r"secp256k1|prime256v1|ed25519": "NON_FHE_CRYPTO",
        r"random_device|mt19937": "WEAK_RNG",
    }
    for root, dirs, files in os.walk(source_dir):
        for fname in files:
            if any(fname.endswith(ext) for ext in fhe_extensions):
                fpath = os.path.join(root, fname)
                scan_result["files_scanned"] += 1
                try:
                    with open(fpath, "r", errors="ignore") as f:
                        content = f.read()
                except Exception:
                    continue
                for pattern, finding_type in security_patterns.items():
                    matches = re.findall(pattern, content)
                    if matches:
                        scan_result["findings"].append({
                            "file": fpath,
                            "type": finding_type,
                            "count": len(matches),
                            "severity": "medium" if finding_type in ["UNSAFE_MEMCPY", "STACK_ALLOCATION", "WEAK_RNG"] else "low",
                        })
    return scan_result


def generate_full_report(param_assessments, sidechannel_scans, source_scans):
    report = {
        "report_timestamp": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
        "parameter_assessments": param_assessments,
        "sidechannel_scans": sidechannel_scans,
        "source_scans": source_scans,
        "overall_risk_score": 0,
        "summary": {},
    }
    total_risk = 0
    critical_count = 0
    high_count = 0
    for assessment in param_assessments:
        total_risk += assessment.get("risk_score", 0)
        for finding in assessment.get("findings", []):
            if finding["severity"] == "critical":
                critical_count += 1
            elif finding["severity"] == "high":
                high_count += 1
    for scan in sidechannel_scans:
        total_risk += scan.get("risk_score", 0)
        for finding in scan.get("findings", []):
            if finding["severity"] == "critical":
                critical_count += 1
            elif finding["severity"] == "high":
                high_count += 1
    for scan in source_scans:
        total_risk += len(scan.get("findings", [])) * 2
    report["overall_risk_score"] = total_risk
    report["summary"] = {
        "critical_findings": critical_count,
        "high_findings": high_count,
        "total_risk_score": total_risk,
        "overall_risk_level": (
            "CRITICAL" if total_risk >= 100 else
            "HIGH" if total_risk >= 50 else
            "MEDIUM" if total_risk >= 20 else
            "LOW"
        ),
    }
    return report


if __name__ == "__main__":
    print("[+] FHE Parameter Security Assessment & Side-Channel Risk Scanner")
    print("=" * 65)

    test_params = [
        {"name": "CKKS_weak", "poly_modulus_degree": 2048, "coeff_modulus_sizes": [30, 30]},
        {"name": "BFV_standard", "poly_modulus_degree": 8192, "coeff_modulus_sizes": [60, 40, 40, 60]},
        {"name": "BGV_secure", "poly_modulus_degree": 16384, "coeff_modulus_sizes": [60, 50, 50, 50, 60]},
    ]

    param_results = []
    for params in test_params:
        print(f"\n[*] Assessing: {params['name']}")
        assessment = assess_fhe_parameters(params)
        print(f"    Security Level: {assessment['security_level_name']}")
        print(f"    Risk Level: {assessment['risk_level']} (score: {assessment['risk_score']})")
        for finding in assessment["findings"]:
            print(f"    [{finding['severity'].upper()}] {finding['detail']}")
        param_results.append(assessment)

    lib_paths = ["/usr/lib/libseal.so", "/usr/local/lib/libseal.so"]
    scan_results = []
    for lib_path in lib_paths:
        if os.path.exists(lib_path):
            print(f"\n[*] Scanning: {lib_path}")
            scan = scan_sidechannel_risk(lib_path)
            print(f"    Risk Level: {scan['risk_level']} (score: {scan['risk_score']})")
            for finding in scan["findings"]:
                print(f"    [{finding['severity'].upper()}] {finding['detail']}")
            scan_results.append(scan)

    full_report = generate_full_report(param_results, scan_results, [])
    print(f"\n[+] Overall Risk: {full_report['summary']['overall_risk_level']} "
          f"(score: {full_report['summary']['total_risk_score']})")
    print(f"[+] Critical: {full_report['summary']['critical_findings']} | "
          f"High: {full_report['summary']['high_findings']}")

File Integrity:
- /etc/seal/params.json SHA256 mismatch (baseline: a1b2c3..., current: d4e5f6...)
- File modified at 2026-03-15T02:34:00Z (outside business hours)

Parameter Indicators:
- poly_modulus_degree changed from 8192 to 2048
- coeff_modulus_sizes changed from [60,40,40,60] to [30,30]
- Security level downgrade: 128-bit to ~70-bit

Behavioral Indicators:
- Query frequency: 3x normal baseline during attack window
- Result variance reduction: 40% below expected range
- Off-hours access: 2026-03-15 02:30-04:00 UTC

API Access Indicators:
- Query rate: >100 req/s (normal: ~5 req/s)
- Query pattern: systematic input variation (not random)
- Time window: 2026-04-01 to 2026-04-15
- Source IP range: 10.0.50.0/24 (unusual subnet)

Computation Indicators:
- Noise growth pattern: non-uniform across input domains
- Output precision: ~2-3 bits higher than expected for CKKS
- Bootstrapping frequency: 40% higher than baseline

Behavioral Indicators:
- Account: service_account_fhe_worker_07
- Normal usage: 50-100 queries/day
- Attack period: 150,000+ queries over 14 days
- Query content: systematic numeric input variation