安全攻防视角下的NUMA架构技术选型与替代方案决策

摘要：本文从安全攻防的视角，深入探讨了非一致性内存访问（NUMA）架构在云计算与高性能计算场景下面临的安全挑战，特别是基于内存和缓存侧信道的隐蔽攻击。我们将构建一个名为"NUMA安全沙盒"的演示性项目，该项目包含一个模拟的NUMA环境、一个演示跨NUMA节点缓存侧信道攻击的模块，以及一套缓解此类攻击的软件防护策略。通过可运行的代码，直观展示攻击原理、潜在危害，并为架构师和安全工程师在面临NUMA技术选型...

摘要

本文从安全攻防的视角，深入探讨了非一致性内存访问（NUMA）架构在云计算与高性能计算场景下面临的安全挑战，特别是基于内存和缓存侧信道的隐蔽攻击。我们将构建一个名为"NUMA安全沙盒"的演示性项目，该项目包含一个模拟的NUMA环境、一个演示跨NUMA节点缓存侧信道攻击的模块，以及一套缓解此类攻击的软件防护策略。通过可运行的代码，直观展示攻击原理、潜在危害，并为架构师和安全工程师在面临NUMA技术选型或寻求替代/增强方案时，提供基于风险与性能权衡的决策依据。

1 项目概述：NUMA安全沙盒

在现代数据中心，NUMA架构被广泛应用以突破多核CPU访问集中式内存的瓶颈。它将物理服务器划分为多个"节点"，每个节点包含部分处理器核心和本地内存。核心访问本地内存速度极快，而访问其他节点的远端内存则延迟较高。然而，这种追求性能的设计引入了新的安全边界。攻击者可能利用对内存访问延迟的精确测量，推断出受害进程在不同NUMA节点上的内存访问模式，从而泄露敏感信息，例如加密算法的密钥或隐私数据。

NUMA安全沙盒 项目旨在：

环境模拟：提供一个简化的软件模拟层，在单台机器上模拟多NUMA节点内存访问的延迟差异。
攻击演示：实现一个名为 CrossNumaCacheSpy 的核心模块，演示攻击者如何利用Flush+Reload缓存侧信道技术，跨"模拟的"NUMA节点探测目标内存地址的访问情况。
防护验证：实现并验证几种软件级别的缓解措施，如内存绑定（numactl）、线程绑定以及随机化延迟注入。
决策辅助：通过量化对比攻击成功率与防护措施带来的性能开销，为技术选型提供数据支撑。

设计思路：由于在单机真实NUMA环境进行攻击演示可能干扰系统稳定性且需要特定硬件，本项目采用"延迟注入"和"内存区域标记"来模拟NUMA效应。核心攻击演示不执行真实恶意操作，而是以验证概念（Proof of Concept）为目的，展示信息泄露的可能性。

2 项目结构树

numa-security-sandbox/
├── core/
│   ├── __init__.py
│   ├── numa_simulator.py      # NUMA环境模拟器
│   ├── attack_cross_numa.py   # 跨NUMA侧信道攻击模块
│   └── defense_policies.py    # 防护策略模块
├── config/
│   └── default.yaml           # 配置文件
├── scripts/
│   ├── run_attack.py          # 攻击演示入口
│   ├── run_defense.py         # 防护验证入口
│   └── benchmark.py           # 性能基准测试
├── tests/
│   └── test_numa_sim.py       # 单元测试
├── utils/
│   ├── __init__.py
│   └── metrics.py             # 性能与安全指标计算
├── requirements.txt
└── README.md                  # 项目说明（写作要求中提示不输出，但实际项目应有）

3 核心代码实现

3.1 `core/numa_simulator.py`

该模块通过虚拟的"节点ID"和人为添加的延迟，模拟不同NUMA节点内存的访问速度差异。

import time
import random
import threading
from typing import Dict, Any, List
from dataclasses import dataclass

@dataclass
class NumaNode:
    """模拟一个NUMA节点"""
    node_id: int
    local_latency_ns: int  # 本地内存访问基准延迟（纳秒）
    remote_latency_ns: int # 访问此节点作为远端时的额外延迟

class NumaSimulator:
    """
    NUMA效应模拟器。
    通过为不同节点ID的内存访问引入可控延迟来模拟NUMA。
    """
    def __init__(self, config: Dict[str, Any]):
        self.nodes: Dict[int, NumaNode] = {}
        self._init_from_config(config)
        # 模拟线程到节点的绑定关系 {thread_ident: node_id}
        self._thread_binding: Dict[int, int] = {}
        self._lock = threading.Lock()

    def _init_from_config(self, config: Dict[str, Any]):
        node_configs = config.get('numa_nodes', [
            {'id': 0, 'local_latency': 100, 'remote_penalty': 200},
            {'id': 1, 'local_latency': 100, 'remote_penalty': 200}
        ])
        for nc in node_configs:
            node = NumaNode(
                node_id=nc['id'],
                local_latency_ns=nc['local_latency'],
                remote_latency_ns=nc['remote_penalty']
            )
            self.nodes[node.node_id] = node
        print(f"[Simulator] Initialized {len(self.nodes)} NUMA nodes.")

    def bind_current_thread_to_node(self, node_id: int):
        """将当前线程绑定到指定的模拟NUMA节点。"""
        with self._lock:
            self._thread_binding[threading.get_ident()] = node_id
        print(f"[Simulator] Thread {threading.current_thread().name} bound to node {node_id}")

    def get_node_for_address(self, address: int) -> int:
        """
        根据内存地址（虚拟）决定其所在的节点。
        这是一个简单的模拟：使用地址的哈希值对节点数取模。
        真实场景由操作系统和硬件决定。
        """
        return address % len(self.nodes)

    def access_memory(self, address: int, is_write: bool = False) -> float:
        """
        模拟对给定地址的一次内存访问，并返回消耗的时间（秒）。
        这是模拟延迟的核心。
        """
        target_node_id = self.get_node_for_address(address)
        current_thread_id = threading.get_ident()

        with self._lock:
            source_node_id = self._thread_binding.get(current_thread_id, 0) # 默认为节点0

        if source_node_id not in self.nodes or target_node_id not in self.nodes:
            raise ValueError("Invalid node ID")

        source_node = self.nodes[source_node_id]
        target_node = self.nodes[target_node_id]

        # 计算延迟
        if source_node_id == target_node_id:
            latency_ns = source_node.local_latency_ns
        else:
            latency_ns = source_node.local_latency_ns + target_node.remote_latency_ns

        # 添加少量随机噪声，模拟现实波动
        latency_ns += random.randint(-10, 10)
        latency_sec = latency_ns / 1_000_000_000.0

        # 真正"消耗"掉这个时间
        time.sleep(latency_sec)
        return latency_sec

# 全局模拟器实例，便于其他模块导入使用
_simulator_instance: NumaSimulator = None

def init_global_simulator(config: Dict[str, Any]):
    global _simulator_instance
    _simulator_instance = NumaSimulator(config)

def get_simulator() -> NumaSimulator:
    if _simulator_instance is None:
        raise RuntimeError("NUMA Simulator not initialized. Call init_global_simulator first.")
    return _simulator_instance

3.2 `core/attack_cross_numa.py`

该模块实现了攻击者线程，它尝试通过计时攻击来推断受害者线程对特定内存地址的访问。

import time
import threading
from core.numa_simulator import get_simulator

class CrossNumaCacheSpy:
    """
    演示跨NUMA节点的缓存侧信道攻击（Flush+Reload变种）。
    攻击者和受害者位于不同的模拟NUMA节点上。
    攻击者反复"探测"一个共享内存地址，通过测量访问时间来判断受害者是否访问过它。
    """
    def __init__(self, target_address: int, attacker_node: int, victim_node: int):
        """
        初始化攻击者。
        :param target_address: 要监视的共享内存地址（模拟值）。
        :param attacker_node: 攻击者线程绑定的NUMA节点。
        :param victim_node: 预期受害者所在的NUMA节点。
        """
        self.target_address = target_address
        self.attacker_node = attacker_node
        self.victim_node = victim_node
        self.simulator = get_simulator()
        # 攻击结果统计
        self.stats = {
            'total_probes': 0,
            'detected_accesses': 0,
            'access_times': []
        }
        self._stop_event = threading.Event()

    def victim_routine(self, access_interval: float, total_operations: int):
        """模拟受害者的行为：周期性地访问目标地址。"""
        print(f"[Victim on Node {self.victim_node}] Started, will access address {self.target_address} {total_operations} times.")
        self.simulator.bind_current_thread_to_node(self.victim_node)
        for i in range(total_operations):
            if self._stop_event.is_set():
                break
            # 受害者访问目标内存
            latency = self.simulator.access_memory(self.target_address)
            # print(f"[Victim] Access {i+1}, latency {latency:.3f} sec")
            time.sleep(access_interval) # 模拟工作间隔
        print(f"[Victim] Finished.")

    def _probe_access_time(self) -> float:
        """攻击者执行一次探测：访问目标地址并计时。"""
        start = time.perf_counter_ns()
        self.simulator.access_memory(self.target_address)
        end = time.perf_counter_ns()
        elapsed_ns = end - start
        return elapsed_ns

    def attacker_routine(self, probe_interval: float, threshold_ns: float):
        """
        攻击者主循环：不断探测目标地址，根据访问时间判断受害者是否访问过。
        :param probe_interval: 探测间隔（秒）。
        :param threshold_ns: 时间阈值（纳秒），低于此值则认为缓存命中（受害者可能访问过）。
        """
        print(f"[Attacker on Node {self.attacker_node}] Started probing address {self.target_address}.")
        self.simulator.bind_current_thread_to_node(self.attacker_node)
        while not self._stop_event.is_set():
            probe_time_ns = self._probe_access_time()
            self.stats['total_probes'] += 1
            self.stats['access_times'].append(probe_time_ns)

            if probe_time_ns < threshold_ns:
                self.stats['detected_accesses'] += 1
                print(f"[Attacker] Probe #{self.stats['total_probes']}: FAST access ({probe_time_ns:.0f} ns < {threshold_ns} ns). Victim might have accessed!")
            # else:
            #     print(f"[Attacker] Probe #{self.stats['total_probes']}: SLOW access ({probe_time_ns:.0f} ns).")

            time.sleep(probe_interval)

    def run_attack(self, duration: float = 10.0,
                   victim_interval: float = 0.5,
                   attacker_interval: float = 0.1,
                   victim_ops: int = 20,
                   threshold_ns: float = 300_000_000): # 300ms阈值，因模拟延迟较高
        """
        启动攻击者和受害者线程，运行一段时间。
        """
        self._stop_event.clear()
        print(f"\n=== Starting Cross-NUMA Cache Attack Simulation ===")
        print(f"Target Address: {self.target_address}")
        print(f"Attacker Node: {self.attacker_node}, Victim Node: {self.victim_node}")
        print(f"Running for ~{duration} seconds...\n")

        # 启动受害者线程
        victim_thread = threading.Thread(
            target=self.victim_routine,
            args=(victim_interval, victim_ops),
            name="VictimThread"
        )
        victim_thread.daemon = True

        # 启动攻击者线程
        attacker_thread = threading.Thread(
            target=self.attacker_routine,
            args=(attacker_interval, threshold_ns),
            name="AttackerThread"
        )
        attacker_thread.daemon = True

        victim_thread.start()
        attacker_thread.start()

        # 主线程等待一段时间后停止
        time.sleep(duration)
        self._stop_event.set()
        attacker_thread.join(timeout=2.0)
        victim_thread.join(timeout=2.0)

        # 打印攻击报告
        self._print_report()

    def _print_report(self):
        """打印攻击统计报告。"""
        print(f"\n=== Attack Report ===")
        print(f"Total probes performed: {self.stats['total_probes']}")
        print(f"Number of suspected victim accesses detected: {self.stats['detected_accesses']}")
        if self.stats['total_probes'] > 0:
            avg_time_ns = sum(self.stats['access_times']) / len(self.stats['access_times'])
            print(f"Average probe access time: {avg_time_ns:.0f} ns")
            detection_rate = self.stats['detected_accesses'] / self.stats['total_probes']
            print(f"Raw detection rate: {detection_rate:.2%}")
        print("=" * 25)

3.3 `core/defense_policies.py`

该模块展示了两种软件缓解策略。

import time
import random
from core.numa_simulator import get_simulator

class DefensePolicies:
    """实现针对跨NUMA侧信道攻击的软件防护策略。"""

    @staticmethod
    def policy_memory_binding(victim_address: int, victim_node: int):
        """
        防护策略1：内存绑定。
        尝试将受害者的关键数据（对应victim_address）固定分配在它所在的本地节点。
        在真实系统中，这通常通过 `numactl --membind` 或 `libnuma` 实现。
        此处我们通过修改模拟器，使得特定地址范围总是映射到指定节点来实现模拟。
        """
        # 注意：这是一个概念性实现。在真实系统中需要操作系统API。
        print(f"[Defense: Memory Binding] Configuring simulator to pin address {victim_address} to node {victim_node}.")
        # 在我们的简单模拟器中，get_node_for_address方法已决定地址映射。
        # 更高级的模拟可以在这里覆盖该行为。此处我们仅做标记。
        # 实际上，应确保分配内存时使用正确的NUMA策略。
        return True

    @staticmethod
    def policy_thread_binding(thread_name: str, node_id: int):
        """
        防护策略2：线程绑定。
        将指定线程绑定到特定的CPU核心（从而绑定到NUMA节点）。
        这减少了线程在节点间迁移的可能性，也使得攻击者更容易预判延迟模式（但这不是本策略的主要目的）。
        """
        simulator = get_simulator()
        simulator.bind_current_thread_to_node(node_id)
        print(f"[Defense: Thread Binding] Thread '{thread_name}' bound to node {node_id}.")
        return True

    @staticmethod
    def policy_randomized_delay(mean_delay_sec: float = 0.0, stddev_sec: float = 0.05):
        """
        防护策略3：随机化延迟注入。
        在受害者或攻击者的内存访问路径上添加随机噪声，破坏攻击者计时测量的准确性。
        这模拟了某些模糊化（obfuscation）技术。
        :param mean_delay_sec: 平均延迟（秒）。
        :param stddev_sec: 延迟的标准差（秒）。
        """
        if stddev_sec <= 0:
            return
        # 生成一个服从正态分布的随机延迟
        extra_delay = random.gauss(mean_delay_sec, stddev_sec)
        # 确保延迟非负
        extra_delay = max(0.0, extra_delay)
        time.sleep(extra_delay)
        # print(f"[Defense: Randomized Delay] Injected {extra_delay:.6f} sec delay.")
        return extra_delay

    @staticmethod
    def apply_defended_victim_routine(target_address: int, victim_node: int,
                                      access_interval: float, total_operations: int,
                                      use_random_delay: bool = True):
        """
        应用了防护策略的受害者例程。
        结合了线程绑定和随机延迟。
        """
        print(f"[Defended Victim on Node {victim_node}] Started with defenses.")
        DefensePolicies.policy_thread_binding("DefendedVictimThread", victim_node)
        DefensePolicies.policy_memory_binding(target_address, victim_node)

        for i in range(total_operations):
            # 受害者访问目标内存
            latency = get_simulator().access_memory(target_address)
            # 注入随机延迟以干扰计时测量
            if use_random_delay:
                DefensePolicies.policy_randomized_delay(stddev_sec=0.03) # 30ms随机噪声
            # print(f"[Defended Victim] Access {i+1} completed.")
            time.sleep(access_interval)
        print(f"[Defended Victim] Finished.")

3.4 `config/default.yaml`

项目的配置文件，定义了模拟参数。

# NUMA Security Sandbox Configuration
numa_simulation:
  numa_nodes:

    - id: 0
      local_latency: 100    # 本地访问基准延迟 (ns)
      remote_penalty: 200   # 远端访问额外延迟 (ns)

    - id: 1
      local_latency: 100
      remote_penalty: 200

attack:
  target_address: 0x1000    # 模拟的共享内存地址
  attacker_node: 0
  victim_node: 1
  run_duration_sec: 15      # 攻击演示运行时长
  victim_access_interval_sec: 0.7
  attacker_probe_interval_sec: 0.15
  victim_total_operations: 15
  detection_threshold_ns: 250_000_000 # 攻击判定阈值(纳秒)

defense:
  use_random_delay: true
  random_delay_stddev_sec: 0.03

benchmark:
  iterations: 1000

3.5 `scripts/run_attack.py`

攻击演示的入口脚本。

#!/usr/bin/env python3
"""
启动无防护状态的跨NUMA缓存侧信道攻击演示。
"""
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

import yaml
from core.numa_simulator import init_global_simulator
from core.attack_cross_numa import CrossNumaCacheSpy

def main():
    # 加载配置
    config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'default.yaml')
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)

    # 初始化NUMA模拟器
    init_global_simulator(config['numa_simulation'])

    # 提取攻击配置
    attack_cfg = config['attack']
    spy = CrossNumaCacheSpy(
        target_address=attack_cfg['target_address'],
        attacker_node=attack_cfg['attacker_node'],
        victim_node=attack_cfg['victim_node']
    )

    # 运行攻击
    spy.run_attack(
        duration=attack_cfg['run_duration_sec'],
        victim_interval=attack_cfg['victim_access_interval_sec'],
        attacker_interval=attack_cfg['attacker_probe_interval_sec'],
        victim_ops=attack_cfg['victim_total_operations'],
        threshold_ns=attack_cfg['detection_threshold_ns']
    )

if __name__ == '__main__':
    main()

3.6 `scripts/run_defense.py`

防护验证的入口脚本。它与攻击脚本类似，但受害者使用防护策略。

#!/usr/bin/env python3
"""
启动带防护的受害者，并运行攻击以验证防护效果。
"""
import sys
import os
import threading
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

import yaml
from core.numa_simulator import init_global_simulator
from core.attack_cross_numa import CrossNumaCacheSpy
from core.defense_policies import DefensePolicies

def main():
    config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'default.yaml')
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)

    init_global_simulator(config['numa_simulation'])
    attack_cfg = config['attack']
    defense_cfg = config['defense']

    spy = CrossNumaCacheSpy(
        target_address=attack_cfg['target_address'],
        attacker_node=attack_cfg['attacker_node'],
        victim_node=attack_cfg['victim_node']
    )

    # 使用防护策略启动受害者线程
    victim_thread = threading.Thread(
        target=DefensePolicies.apply_defended_victim_routine,
        args=(
            attack_cfg['target_address'],
            attack_cfg['victim_node'],
            attack_cfg['victim_access_interval_sec'],
            attack_cfg['victim_total_operations'],
            defense_cfg['use_random_delay']
        ),
        name="DefendedVictimThread"
    )
    victim_thread.daemon = True
    victim_thread.start()

    # 给受害者一点时间启动并绑定
    time.sleep(1)

    # 攻击者开始探测（攻击者例程与之前相同）
    # 我们需要修改CrossNumaCacheSpy以允许外部控制停止事件，这里简化处理：直接运行攻击者例程
    print(f"\n=== Starting Attack against DEFENDED Victim ===")
    spy._stop_event.clear()
    attacker_thread = threading.Thread(
        target=spy.attacker_routine,
        args=(attack_cfg['attacker_probe_interval_sec'], attack_cfg['detection_threshold_ns']),
        name="AttackerThread"
    )
    attacker_thread.daemon = True
    attacker_thread.start()

    # 主线程等待一段时间
    time.sleep(attack_cfg['run_duration_sec'])
    spy._stop_event.set()
    attacker_thread.join(timeout=2.0)
    victim_thread.join(timeout=2.0)

    spy._print_report()

if __name__ == '__main__':
    main()

3.7 `scripts/benchmark.py`

性能基准测试脚本，用于量化防护措施的开销。

#!/usr/bin/env python3
"""
性能基准测试：比较有无防护策略时，内存访问的吞吐量或延迟。
"""
import sys
import os
import time
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

import yaml
from core.numa_simulator import init_global_simulator, get_simulator
from core.defense_policies import DefensePolicies

def benchmark_access(use_defense: bool, iterations: int, node_id: int, address: int) -> dict:
    """基准测试函数"""
    simulator = get_simulator()
    simulator.bind_current_thread_to_node(node_id)

    latencies = []
    start_total = time.perf_counter_ns()

    for i in range(iterations):
        start = time.perf_counter_ns()
        simulator.access_memory(address)
        if use_defense:
            # 在每次访问后应用随机延迟防护
            DefensePolicies.policy_randomized_delay(stddev_sec=0.03)
        end = time.perf_counter_ns()
        latencies.append(end - start)

    end_total = time.perf_counter_ns()
    total_time_ns = end_total - start_total

    avg_latency_ns = sum(latencies) / len(latencies)
    throughput = iterations / (total_time_ns / 1_000_000_000) # operations per second

    return {
        'avg_latency_ns': avg_latency_ns,
        'total_time_sec': total_time_ns / 1_000_000_000,
        'throughput_ops_per_sec': throughput,
        'min_latency_ns': min(latencies),
        'max_latency_ns': max(latencies)
    }

def main():
    config_path = os.path.join(os.path.dirname(__file__), '..', 'config', 'default.yaml')
    with open(config_path, 'r') as f:
        config = yaml.safe_load(f)

    init_global_simulator(config['numa_simulation'])
    benchmark_cfg = config['benchmark']
    attack_cfg = config['attack']
    iterations = benchmark_cfg['iterations']

    print("=== Performance Benchmark ===")
    print(f"Iterations per test: {iterations}")
    print(f"Test Address: {attack_cfg['target_address']}")
    print(f"Local Node (for benchmark): {attack_cfg['attacker_node']}\n")

    # 基准测试：无防护
    print("1. Benchmark WITHOUT defense:")
    result_no_def = benchmark_access(False, iterations, attack_cfg['attacker_node'], attack_cfg['target_address'])
    print(f"   Average Latency: {result_no_def['avg_latency_ns'] / 1_000_000:.3f} ms")
    print(f"   Throughput: {result_no_def['throughput_ops_per_sec']:.2f} ops/sec")
    print(f"   Total Time: {result_no_def['total_time_sec']:.3f} sec\n")

    # 基准测试：有防护（随机延迟）
    print("2. Benchmark WITH defense (Randomized Delay):")
    result_def = benchmark_access(True, iterations, attack_cfg['attacker_node'], attack_cfg['target_address'])
    print(f"   Average Latency: {result_def['avg_latency_ns'] / 1_000_000:.3f} ms")
    print(f"   Throughput: {result_def['throughput_ops_per_sec']:.2f} ops/sec")
    print(f"   Total Time: {result_def['total_time_sec']:.3f} sec\n")

    # 计算开销
    latency_overhead = ((result_def['avg_latency_ns'] - result_no_def['avg_latency_ns']) / result_no_def['avg_latency_ns']) * 100
    throughput_degradation = ((result_no_def['throughput_ops_per_sec'] - result_def['throughput_ops_per_sec']) / result_no_def['throughput_ops_per_sec']) * 100

    print("=== Overhead Summary ===")
    print(f"Latency Increased by: {latency_overhead:.1f}%")
    print(f"Throughput Degraded by: {throughput_degradation:.1f}%")

if __name__ == '__main__':
    main()

3.8 `requirements.txt`

项目依赖。

pyyaml>=5.4

4 安装依赖与运行步骤

环境要求：Python 3.7+
克隆/创建项目目录：

mkdir numa-security-sandbox && cd numa-security-sandbox
    # 将上述所有代码文件按结构树放入对应目录。

安装依赖：

pip install -r requirements.txt

运行攻击演示（无防护）：

python scripts/run_attack.py

观察控制台输出。攻击者会报告探测次数和通过快速访问推断出的受害者访问次数。

运行防护验证：

python scripts/run_defense.py

观察输出。由于随机延迟的引入，攻击者的计时测量将受到干扰，`detected_accesses`（检测到的访问）数量应该会显著下降或变得不可靠，表明防护生效。

运行性能基准测试：

python scripts/benchmark.py

查看防护措施（此处主要是随机延迟）引入的性能开销（延迟增加、吞吐量下降百分比）。

5 测试与验证

我们提供了一个简单的单元测试来验证NUMA模拟器的基本逻辑。

5.1 `tests/test_numa_sim.py`

import unittest
import sys
import os
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))

from core.numa_simulator import NumaSimulator, NumaNode

class TestNumaSimulator(unittest.TestCase):

    def setUp(self):
        config = {
            'numa_nodes': [
                {'id': 0, 'local_latency': 50, 'remote_penalty': 150},
                {'id': 1, 'local_latency': 50, 'remote_penalty': 150}
            ]
        }
        self.sim = NumaSimulator(config)

    def test_node_initialization(self):
        self.assertEqual(len(self.sim.nodes), 2)
        self.assertIn(0, self.sim.nodes)
        self.assertIn(1, self.sim.nodes)
        self.assertEqual(self.sim.nodes[0].local_latency_ns, 50)

    def test_address_to_node_mapping(self):
        # 地址0应映射到节点0 (0 % 2 = 0)
        self.assertEqual(self.sim.get_node_for_address(0), 0)
        # 地址1应映射到节点1 (1 % 2 = 1)
        self.assertEqual(self.sim.get_node_for_address(1), 1)
        # 地址5应映射到节点1 (5 % 2 = 1)
        self.assertEqual(self.sim.get_node_for_address(5), 1)

    def test_thread_binding(self):
        import threading
        self.sim.bind_current_thread_to_node(1)
        # 在测试中验证绑定字典是否正确更新
        # 注意：由于`_thread_binding`是私有变量，实际测试中可能需要通过公共接口间接测试。
        # 这里我们假设绑定成功，主要测试不抛出异常。
        self.assertTrue(True)

if __name__ == '__main__':
    unittest.main()

运行测试：

python -m pytest tests/test_numa_sim.py -v

6 技术选型与替代方案决策分析

通过运行上述项目，我们可以获得关于NUMA架构下安全风险的直观认识，并对防护措施的有效性和成本进行量化。这为技术决策提供了依据。

graph TD A[面临多核高并发场景] --> B{内存架构选型} B --> C[UMA / 集中式内存] B --> D[NUMA架构] C --> E[优点: 编程模型简单, 无远端访问开销] C --> F[缺点: 可扩展性差, 核心数多时总线争用严重] F --> G[可能导致整体性能瓶颈] D --> H[优点: 优秀的可扩展性, 高核心数下性能更佳] D --> I[缺点: 引入新的安全攻击面 （侧信道, 如本项目演示） 编程复杂性增加] I --> J{安全风险评估} J -->|高敏感应用| K[考虑替代/增强方案] J -->|风险可接受| L[接受NUMA, 并实施软件防护 （如本项目策略）] K --> M[替代方案1: 硬件隔离 （如CCIX, CXL 内存池化）] K --> N[替代方案2: 软件定义内存 （如PMEM, 用户态内存管理）] K --> O[增强方案: 安全NUMA感知调度器 （将敏感任务与攻击者隔离到不同节点）] L --> P[决策输出: 使用NUMA + 防护策略] M --> Q[决策输出: 采用新互联架构] N --> Q O --> P P --> R[需承受性能开销] Q --> S[可能引入新复杂度与成本] style D fill:#f9f,stroke:#333,stroke-width:2px style I fill:#ff9,stroke:#333,stroke-width:2px style P fill:#9f9,stroke:#333,stroke-width:2px

决策关键点：

性能需求 vs 安全需求：如果应用处理绝密数据，即使NUMA性能再好，也可能因侧信道风险而被否决。此时应优先考虑UMA或具有更强硬件隔离特性的新兴架构（如CXL）。
成本考量：软件防护（如内存绑定、噪声注入）成本低，但会损失一部分NUMA带来的性能收益。硬件解决方案可能性能更好，但采购和开发成本高。
可维护性：复杂的NUMA优化代码和防护代码会提高软件复杂度。团队需要具备相应的专业知识。
攻击面评估：并非所有NUMA应用都面临同等风险。如果攻击者无法在目标节点上运行代码（例如，通过严格的容器或VM隔离），风险则大大降低。

7 攻击流程与防护作用时序图

下图清晰地展示了攻击者与受害者在跨NUMA场景下的交互，以及防护策略如何介入并破坏攻击链。

sequenceDiagram participant A as 攻击者 (Node 0) participant C as CPU缓存/内存子系统 participant V as 受害者 (Node 1) participant D as 防护模块 Note over A,V: 初始状态: 目标地址数据不在任何缓存中 A->>C: 1. 发起探测访问 (计时开始) C->>C: 检查缓存 (Miss) C->>C: 从远端Node 1内存加载数据 C->>A: 数据返回 (计时结束) - 时间**长** Note right of A: 记录: "慢访问" V->>D: 2. 受害者准备访问 D->>V: 执行线程绑定 (绑定到Node 1) V->>C: 访问目标地址 C->>C: 加载到Node 1本地缓存 C->>V: 数据返回 D->>V: 注入随机延迟噪声 A->>C: 3. 再次探测访问 (计时开始) C->>C: 检查缓存 (Hit! 数据在Node 1缓存) C->>A: 数据快速返回 (计时结束) - 时间**短** Note right of A: 记录: "快访问" -> 推测受害者访问过！ Note over A,V: 防护生效场景 V->>D: 4. 受害者(防护模式)准备访问 D->>V: 执行线程绑定 + 内存绑定 V->>C: 访问目标地址 C->>V: 数据返回 D->>V: 注入显著的随机延迟 A->>C: 5. 攻击者探测 (计时开始) C->>C: 检查缓存 (Hit) C->>A: 数据快速返回 (计时结束) - 本应时间短 Note right of A: 但！受害者的随机延迟 使两次访问间隔变长， 缓存可能已被换出。 Note right of A: 或者，即使缓存命中， 攻击者的计时被噪声干扰， 无法稳定区分快/慢。

结论：NUMA架构是一把双刃剑。NUMA安全沙盒项目通过可运行的代码证明，在追求极致性能的同时，必须清醒评估其引入的侧信道安全风险。架构师应根据实际应用的安全等级、性能预算和运维能力，在传统NUMA、软件加固方案及新兴的替代硬件架构之间做出审慎抉择。软件防护可以作为一种有效的补偿控制措施，但其性能开销需纳入整体评估。

安全攻防视角下的NUMA架构技术选型与替代方案决策

摘要

1 项目概述：NUMA安全沙盒

2 项目结构树

3 核心代码实现

3.1 core/numa_simulator.py

3.2 core/attack_cross_numa.py

3.3 core/defense_policies.py

3.4 config/default.yaml

3.5 scripts/run_attack.py

3.6 scripts/run_defense.py

3.7 scripts/benchmark.py

3.8 requirements.txt