金融系统中密钥管理方案与红蓝对抗演练的选型评估框架

kms_redblue_eval_framework/
├── config/
│   ├── __init__.py
│   ├── framework.yaml          # 主框架配置
│   └── attack_scenarios.yaml   # 攻击场景定义
├── core/
│   ├── __init__.py
│   ├── kms/
│   │   ├── __init__.py
│   │   ├── base.py            # KMS抽象基类
│   │   ├── local_software.py  # 本地软件KMS实现
│   │   └── simulated_hsm.py   # 模拟HSM实现
│   └── redblue/
│       ├── __init__.py
│       ├── engine.py          # 红蓝对抗引擎
│       └── attacks/           # 攻击模块目录
│           ├── __init__.py
│           ├── base.py
│           ├── key_theft.py
│           └── denial_of_service.py
├── evaluation/
│   ├── __init__.py
│   ├── model.py               # 评估模型与计算
│   └── reporter.py            # 报告生成器
├── utils/
│   ├── __init__.py
│   └── logger.py              # 日志工具
├── app.py                     # 主应用程序入口
├── requirements.txt           # Python依赖
└── README.md                  # 项目说明（根据要求，此文件不在结构树中展示）

import abc
import uuid
from enum import Enum
from typing import Optional, Tuple, Any

class KeyAlgorithm(Enum):
    """支持的密钥算法枚举"""
    RSA_2048 = "RSA_2048"
    RSA_4096 = "RSA_4096"
    AES_256 = "AES_256"
    EC_P256 = "EC_P256"

class KeyUsage(Enum):
    """密钥用途枚举"""
    ENCRYPT_DECRYPT = "ENCRYPT_DECRYPT"
    SIGN_VERIFY = "SIGN_VERIFY"

class KeyManagementService(abc.ABC):
    """密钥管理服务抽象基类"""

    @abc.abstractmethod
    def create_key(self, key_id: str, algorithm: KeyAlgorithm, usage: KeyUsage, metadata: dict = None) -> bool:
        """创建一个新的密钥。
        返回: 成功 True, 失败 False"""
        pass

    @abc.abstractmethod
    def get_key_metadata(self, key_id: str) -> Optional[dict]:
        """获取密钥元数据。"""
        pass

    @abc.abstractmethod
    def encrypt(self, key_id: str, plaintext: bytes, aad: bytes = None) -> Tuple[Optional[bytes], Optional[bytes]]:
        """使用指定密钥加密数据。
        返回: (密文, 认证标签) 或 (None, None) 失败"""
        pass

    @abc.abstractmethod
    def decrypt(self, key_id: str, ciphertext: bytes, tag: bytes = None, aad: bytes = None) -> Optional[bytes]:
        """使用指定密钥解密数据。"""
        pass

    @abc.abstractmethod
    def sign(self, key_id: str, message: bytes) -> Optional[bytes]:
        """使用指定密钥对消息签名。"""
        pass

    @abc.abstractmethod
    def verify(self, key_id: str, message: bytes, signature: bytes) -> bool:
        """验证签名。"""
        pass

    @abc.abstractmethod
    def rotate_key(self, key_id: str) -> bool:
        """轮转密钥（创建新版本）。"""
        pass

    @abc.abstractmethod
    def disable_key(self, key_id: str) -> bool:
        """禁用密钥。"""
        pass

    @abc.abstractmethod
    def get_health(self) -> dict:
        """获取服务健康状态，用于可用性评估。"""
        pass

    @abc.abstractmethod
    def get_metrics(self) -> dict:
        """获取性能指标，如平均延迟、调用次数。"""
        pass

import time
import logging
from typing import Optional, Tuple, Any
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import rsa, padding, ec
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
import os
from .base import KeyManagementService, KeyAlgorithm, KeyUsage

logger = logging.getLogger(__name__)

class LocalSoftwareKMS(KeyManagementService):
    """本地软件KMS实现。警告：密钥存在于进程内存，仅用于演示评估。"""

    def __init__(self, kms_id: str = "local-software-kms"):
        self.kms_id = kms_id
        self._keys = {}  # key_id -> {‘version‘, ‘algorithm‘, ‘material‘, ‘enabled‘, ...}
        self._metrics = {"encrypt_calls": 0, "decrypt_calls": 0, "total_time": 0.0}
        self._start_time = time.time()

    def create_key(self, key_id: str, algorithm: KeyAlgorithm, usage: KeyUsage, metadata: dict = None) -> bool:
        logger.info(f"[{self.kms_id}] Creating key {key_id} with alg {algorithm}")
        if key_id in self._keys:
            logger.warning(f"Key {key_id} already exists.")
            return False

        key_material = None
        try:
            if algorithm == KeyAlgorithm.RSA_2048:
                private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048)
                key_material = private_key.private_bytes(
                    encoding=serialization.Encoding.PEM,
                    format=serialization.PrivateFormat.PKCS8,
                    encryption_algorithm=serialization.NoEncryption()
                )
            elif algorithm == KeyAlgorithm.AES_256:
                key_material = os.urandom(32)  # 256 bits
            else:
                # 简化处理，其他算法类似
                logger.error(f"Algorithm {algorithm} not fully implemented in demo.")
                return False

            self._keys[key_id] = {
                'version': 'v1',
                'algorithm': algorithm,
                'usage': usage,
                'material': key_material,
                'enabled': True,
                'created_at': time.time(),
                'metadata': metadata or {}
            }
            return True
        except Exception as e:
            logger.exception(f"Failed to create key {key_id}: {e}")
            return False

    def encrypt(self, key_id: str, plaintext: bytes, aad: bytes = None) -> Tuple[Optional[bytes], Optional[bytes]]:
        start = time.time()
        self._metrics["encrypt_calls"] += 1
        if key_id not in self._keys or not self._keys[key_id]['enabled']:
            logger.error(f"Key {key_id} not found or disabled.")
            return None, None

        key_info = self._keys[key_id]
        try:
            if key_info['algorithm'] == KeyAlgorithm.AES_256:
                iv = os.urandom(12)
                encryptor = Cipher(
                    algorithms.AES(key_info['material']),
                    modes.GCM(iv)
                ).encryptor()
                if aad:
                    encryptor.authenticate_additional_data(aad)
                ciphertext = encryptor.update(plaintext) + encryptor.finalize()
                return ciphertext, encryptor.tag
            else:
                # RSA加密等逻辑（略）
                logger.warning("RSA encryption logic omitted for brevity.")
                return b"simulated_ciphertext", b"simulated_tag"
        except Exception as e:
            logger.exception(f"Encrypt failed for key {key_id}: {e}")
            return None, None
        finally:
            self._metrics["total_time"] += (time.time() - start)

    def decrypt(self, key_id: str, ciphertext: bytes, tag: bytes = None, aad: bytes = None) -> Optional[bytes]:
        # 实现逻辑与encrypt对称，略
        self._metrics["decrypt_calls"] += 1
        # 模拟解密成功
        return b"decrypted_dummy_data"

    def sign(self, key_id: str, message: bytes) -> Optional[bytes]:
        # 签名实现（略）
        return b"simulated_signature"

    def verify(self, key_id: str, message: bytes, signature: bytes) -> bool:
        # 验证实现（略），演示中始终返回True
        return True

    def rotate_key(self, key_id: str) -> bool:
        if key_id not in self._keys:
            return False
        old_key = self._keys[key_id]
        # 简化：仅更新版本号，实际应生成新密钥材料
        old_key['version'] = f"v{int(old_key['version'][1:]) + 1}"
        logger.info(f"Key {key_id} rotated to {old_key['version']}")
        return True

    def disable_key(self, key_id: str) -> bool:
        if key_id in self._keys:
            self._keys[key_id]['enabled'] = False
            return True
        return False

    def get_key_metadata(self, key_id: str) -> Optional[dict]:
        return self._keys.get(key_id)

    def get_health(self) -> dict:
        uptime = time.time() - self._start_time
        return {"status": "HEALTHY", "uptime_seconds": uptime, "keys_managed": len(self._keys)}

    def get_metrics(self) -> dict:
        total_calls = self._metrics["encrypt_calls"] + self._metrics["decrypt_calls"]
        avg_latency = (self._metrics["total_time"] / total_calls) if total_calls > 0 else 0
        return {
            **self._metrics,
            "average_latency_seconds": avg_latency,
            "kms_id": self.kms_id
        }

import time
import logging
from typing import Optional, Tuple
from .base import KeyManagementService, KeyAlgorithm, KeyUsage

logger = logging.getLogger(__name__)

class SimulatedHSM(KeyManagementService):
    """模拟的硬件安全模块(HSM)实现。"""

    def __init__(self, kms_id: str = "simulated-hsm", slot_count: int = 5):
        self.kms_id = kms_id
        self.slot_count = slot_count
        # 模拟HSM内部存储：slot -> {key_id: key_info}
        self._slots = [{} for _ in range(slot_count)]
        self._metrics = {"operations": 0, "total_delay": 0.0}
        # HSM特性：固定基础延迟（毫秒）
        self._base_delay_sec = 0.005  # 5ms

    def _simulate_hsm_delay(self):
        """模拟HSM操作的硬件延迟。"""
        delay = self._base_delay_sec
        time.sleep(delay)
        self._metrics["total_delay"] += delay
        self._metrics["operations"] += 1

    def _find_key_slot(self, key_id: str) -> Tuple[Optional[int], Optional[dict]]:
        for slot_idx, slot in enumerate(self._slots):
            if key_id in slot:
                return slot_idx, slot[key_id]
        return None, None

    def create_key(self, key_id: str, algorithm: KeyAlgorithm, usage: KeyUsage, metadata: dict = None) -> bool:
        logger.info(f"[{self.kms_id}] HSM Creating key {key_id}")
        self._simulate_hsm_delay()
        # 找到第一个未满的槽位
        for slot_idx, slot in enumerate(self._slots):
            if len(slot) < 10:  # 假设每个槽位最多10个密钥
                # HSM中不存储完整的密钥材料，只存储引用句柄
                self._slots[slot_idx][key_id] = {
                    'version': 'v1',
                    'algorithm': algorithm,
                    'usage': usage,
                    'handle': f"hsm_handle_{slot_idx}_{key_id}",
                    'enabled': True,
                    'created_at': time.time(),
                    'metadata': metadata or {},
                    '__material_never_exportable__': True  # 强调不可导出性
                }
                return True
        logger.error("All HSM slots are full!")
        return False

    def encrypt(self, key_id: str, plaintext: bytes, aad: bytes = None) -> Tuple[Optional[bytes], Optional[bytes]]:
        slot_idx, key_info = self._find_key_slot(key_id)
        if not key_info or not key_info['enabled']:
            return None, None
        self._simulate_hsm_delay()  # HSM操作延迟
        # 模拟HSM内部加密，返回密文和标签
        # 注意：此模拟中实际未加密，但强调密钥材料未离开"硬件"
        simulated_cipher = f"<HSM-ENCRYPTED({key_info['handle']}):{plaintext[:5]}...>".encode()
        return simulated_cipher, b"hsm_auth_tag"

    def decrypt(self, key_id: str, ciphertext: bytes, tag: bytes = None, aad: bytes = None) -> Optional[bytes]:
        # 类似encrypt，略
        self._simulate_hsm_delay()
        return b"<HSM-DECRYPTED-DATA>"

    def sign(self, key_id: str, message: bytes) -> Optional[bytes]:
        self._simulate_hsm_delay()
        return f"<HSM-SIGNATURE({key_id})>".encode()

    # 其他方法如rotate_key, disable_key等也需模拟延迟，实现略...

    def get_health(self) -> dict:
        used_slots = sum(1 for slot in self._slots if len(slot) > 0)
        return {
            "status": "HEALTHY",
            "slots_used": used_slots,
            "slots_total": self.slot_count,
            "type": "SIMULATED_HSM"
        }

    def get_metrics(self) -> dict:
        avg_op_delay = (self._metrics["total_delay"] / self._metrics["operations"]) if self._metrics["operations"] > 0 else 0
        return {
            **self._metrics,
            "average_operation_delay_seconds": avg_op_delay,
            "kms_id": self.kms_id
        }

import abc
import logging
from typing import Dict, Any

logger = logging.getLogger(__name__)

class AttackModule(abc.ABC):
    """红队攻击模块基类。"""

    def __init__(self, attack_id: str, description: str):
        self.attack_id = attack_id
        self.description = description
        self.success = False
        self.evidence = []

    @abc.abstractmethod
    def execute(self, target_kms, config: Dict[str, Any]) -> Dict[str, Any]:
        """执行攻击。
        参数:
            target_kms: 目标KMS实例 (KeyManagementService)
            config: 该攻击模块的配置
        返回:
            包含攻击结果的字典
        """
        pass

    def _log_evidence(self, msg: str):
        logger.warning(f"[Attack {self.attack_id}] {msg}")
        self.evidence.append(msg)

    def get_result(self) -> Dict[str, Any]:
        return {
            "attack_id": self.attack_id,
            "description": self.description,
            "success": self.success,
            "evidence": self.evidence
        }

import logging
from .base import AttackModule
from core.kms.base import KeyManagementService

logger = logging.getLogger(__name__)

class KeyTheftAttack(AttackModule):
    """模拟密钥材料窃取攻击。"""

    def __init__(self):
        super().__init__(
            attack_id="KEY-THEFT-01",
            description="Attempt to extract or access raw key material from the KMS."
        )

    def execute(self, target_kms: KeyManagementService, config: dict) -> dict:
        logger.info(f"Executing {self.attack_id} against {target_kms.kms_id}")
        test_key_id = config.get("test_key_id", "test_key_rsa_2048")

        # 攻击向量1: 尝试直接访问内部存储（仅对LocalSoftwareKMS可能有效）
        if hasattr(target_kms, '_keys'):
            try:
                # 模拟利用漏洞或错误配置访问内部字典
                if test_key_id in target_kms._keys:
                    key_material = target_kms._keys[test_key_id].get('material')
                    if key_material:
                        self._log_evidence(f"SUCCESS: Potentially accessed raw key material for {test_key_id} via internal attribute '_keys'. Material preview: {key_material[:20] if key_material else 'None'}")
                        self.success = True
                    else:
                        self._log_evidence(f"Key found in internal dict, but no 'material' field (maybe HSM-like).")
                else:
                    self._log_evidence(f"Key {test_key_id} not found in internal dict.")
            except Exception as e:
                self._log_evidence(f"Failed to access internal '_keys': {e}")

        # 攻击向量2: 尝试通过元数据接口获取敏感信息
        metadata = target_kms.get_key_metadata(test_key_id)
        if metadata:
            # 检查元数据是否意外泄露敏感信息
            sensitive_keys = ['material', 'private_key', 'clear_key']
            for sk in sensitive_keys:
                if sk in metadata:
                    self._log_evidence(f"WARNING: Sensitive key '{sk}' found in public metadata!")
                    self.success = True

        if not self.success:
            self._log_evidence("Attack did not succeed in extracting key material.")

        return self.get_result()

import yaml
import importlib
import logging
import time
from typing import List, Dict, Any
from .attacks.base import AttackModule

logger = logging.getLogger(__name__)

class RedBlueEngine:
    """红蓝对抗演练引擎。"""

    def __init__(self, scenario_config_path: str):
        self.scenario_config_path = scenario_config_path
        self.scenarios = self._load_scenarios()
        self.attack_results = []

    def _load_scenarios(self) -> List[Dict[str, Any]]:
        with open(self.scenario_config_path, 'r') as f:
            config = yaml.safe_load(f)
        return config.get('attack_scenarios', [])

    def _load_attack_module(self, module_path: str) -> AttackModule:
        """动态加载攻击模块。"""
        try:
            module_name, class_name = module_path.rsplit('.', 1)
            module = importlib.import_module(module_name)
            attack_class = getattr(module, class_name)
            return attack_class()
        except (ImportError, AttributeError) as e:
            logger.error(f"Failed to load attack module {module_path}: {e}")
            raise

    def run_scenarios(self, target_kms) -> List[Dict[str, Any]]:
        """对所有加载的场景，针对目标KMS运行攻击。"""
        logger.info(f"Starting Red Team scenarios against {target_kms.kms_id}")
        for scenario in self.scenarios:
            if not scenario.get('enabled', True):
                continue
            logger.info(f"  Running scenario: {scenario['name']}")
            attack_module = self._load_attack_module(scenario['attack_module'])
            start_time = time.time()
            result = attack_module.execute(target_kms, scenario.get('config', {}))
            result['execution_time_seconds'] = time.time() - start_time
            result['target_kms_id'] = target_kms.kms_id
            self.attack_results.append(result)
            logger.info(f"    Result: Success={result['success']}")
        return self.attack_results

from typing import Dict, Any, List
import numpy as np

class EvaluationModel:
    """评估模型，计算安全、性能、可用性等维度分数。"""

    # 权重配置 (可根据实际调整)
    WEIGHTS = {
        'security': 0.40,
        'performance': 0.25,
        'availability': 0.20,
        'cost_compliance': 0.15
    }

    @staticmethod
    def calculate_security_score(attack_results: List[Dict[str, Any]]) -> float:
        """基于攻击成功率计算安全分数 (0-100)。"""
        if not attack_results:
            return 100.0  # 无攻击场景，默认满分
        successful_attacks = sum(1 for r in attack_results if r['success'])
        total_attacks = len(attack_results)
        success_ratio = successful_attacks / total_attacks
        # 成功越少，分数越高
        security_score = (1 - success_ratio) * 100
        return max(0.0, min(100.0, security_score))  # 钳制在0-100

    @staticmethod
    def calculate_performance_score(kms_metrics: Dict[str, Any]) -> float:
        """基于延迟和吞吐量计算性能分数 (0-100)。"""
        avg_latency = kms_metrics.get('average_latency_seconds') or kms_metrics.get('average_operation_delay_seconds', 1.0)
        # 假设理想延迟 < 0.01秒 (10ms) 得100分， > 0.5秒得0分
        ideal_latency = 0.01
        max_tolerable_latency = 0.5
        if avg_latency <= ideal_latency:
            return 100.0
        elif avg_latency >= max_tolerable_latency:
            return 0.0
        else:
            # 线性插值
            score = 100 * (1 - (avg_latency - ideal_latency) / (max_tolerable_latency - ideal_latency))
            return round(score, 2)

    @staticmethod
    def calculate_availability_score(health_info: Dict[str, Any]) -> float:
        """基于健康状态计算可用性分数。"""
        status = health_info.get('status', 'UNKNOWN').upper()
        if status == 'HEALTHY':
            return 100.0
        elif status == 'DEGRADED':
            return 60.0
        else:
            return 10.0

    @staticmethod
    def calculate_cost_compliance_score(kms_type: str) -> float:
        """简化版的成本与合规分数 (示例逻辑)。"""
        # 示例逻辑：HSM通常更合规但成本高，软件KMS成本低但合规性要求需额外验证
        if 'hsm' in kms_type.lower():
            return 85.0  # 合规性高，成本高
        else:
            return 70.0  # 成本低，合规性中等

    def evaluate(self, kms_id: str, kms_type: str, metrics: dict, health: dict, attack_results: list) -> Dict[str, Any]:
        """执行综合评估。"""
        scores = {
            'security': self.calculate_security_score(attack_results),
            'performance': self.calculate_performance_score(metrics),
            'availability': self.calculate_availability_score(health),
            'cost_compliance': self.calculate_cost_compliance_score(kms_type)
        }
        # 计算加权总分
        weighted_total = sum(scores[dim] * self.WEIGHTS[dim] for dim in self.WEIGHTS)
        scores['total_weighted_score'] = round(weighted_total, 2)

        return {
            'kms_id': kms_id,
            'kms_type': kms_type,
            'scores': scores,
            'raw_metrics': metrics,
            'raw_health': health,
            'attack_summary': {
                'total_attacks': len(attack_results),
                'successful_attacks': sum(1 for r in attack_results if r['success'])
            }
        }

import json
from datetime import datetime
from typing import Dict, Any, List

class HTMLReporter:
    """生成HTML评估报告。"""

    @staticmethod
    def generate(evaluation_results: List[Dict[str, Any]], output_path: str = "kms_evaluation_report.html"):
        """生成并保存HTML报告。"""
        timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
        html_content = f"""
        <!DOCTYPE html>
        <html lang="en">
        <head>
            <meta charset="UTF-8">
            <meta name="viewport" content="width=device-width, initial-scale=1.0">
            <title>KMS 选型评估报告</title>
            <style>
                body {{ font-family: sans-serif; margin: 40px; background-color: #f5f5f5; }}
                .container {{ max-width: 1200px; margin: auto; background: white; padding: 30px; border-radius: 10px; box-shadow: 0 0 20px rgba(0,0,0,0.1); }}
                h1 {{ color: #2c3e50; border-bottom: 3px solid #3498db; padding-bottom: 10px; }}
                .summary {{ background: #f8f9fa; padding: 20px; border-radius: 8px; margin-bottom: 30px; }}
                .kms-result {{ border: 1px solid #ddd; border-radius: 8px; padding: 20px; margin-bottom: 25px; background: #fff; }}
                .score-card {{ display: inline-block; background: #3498db; color: white; padding: 10px 20px; border-radius: 5px; font-size: 1.5em; margin: 10px 0; }}
                .dim-scores {{ display: flex; flex-wrap: wrap; gap: 15px; margin-top: 15px; }}
                .dim-box {{ background: #ecf0f1; padding: 15px; border-radius: 5px; flex: 1; min-width: 150px; }}
                .dim-name {{ font-weight: bold; color: #2c3e50; }}
                .dim-value {{ font-size: 1.2em; color: #27ae60; }}
                table {{ width: 100%; border-collapse: collapse; margin-top: 20px; }}
                th, td {{ padding: 12px 15px; border: 1px solid #ddd; text-align: left; }}
                th {{ background-color: #3498db; color: white; }}
                tr:nth-child(even) {{ background-color: #f9f9f9; }}
                .footer {{ margin-top: 40px; text-align: center; color: #7f8c8d; font-size: 0.9em; }}
            </style>
        </head>
        <body>
            <div class="container">
                <h1>📊 金融系统密钥管理方案 (KMS/HSM) 选型评估报告</h1>
                <p><strong>生成时间:</strong> {timestamp}</p>
                <p><strong>评估概述:</strong> 本报告通过模拟红蓝对抗演练与性能监控，对不同的密钥管理后端进行了多维度量化评估，旨在为选型决策提供数据支持。</p>

                <div class="summary">
                    <h2>🏆 综合排名</h2>
        """
        # 按总分排序
        sorted_results = sorted(evaluation_results, key=lambda x: x['scores']['total_weighted_score'], reverse=True)
        for i, result in enumerate(sorted_results):
            rank = i + 1
            html_content += f"""
                    <p>{rank}. <strong>{result['kms_id']}</strong> - 加权总分: <span class="score-card">{result['scores']['total_weighted_score']}</span></p>
            """

        html_content += """
                </div>
        """
        # 每个KMS的详细结果
        for result in sorted_results:
            scores = result['scores']
            html_content += f"""
                <div class="kms-result">
                    <h2>🔧 KMS: {result['kms_id']} (类型: {result['kms_type']})</h2>
                    <p><strong>加权总分:</strong> <span class="score-card">{scores['total_weighted_score']}/100</span></p>
                    <div class="dim-scores">
                        <div class="dim-box"><span class="dim-name">🔐 安全</span><br><span class="dim-value">{scores['security']}</span></div>
                        <div class="dim-box"><span class="dim-name">⚡ 性能</span><br><span class="dim-value">{scores['performance']}</span></div>
                        <div class="dim-box"><span class="dim-name">🛡️ 可用性</span><br><span class="dim-value">{scores['availability']}</span></div>
                        <div class="dim-box"><span class="dim-name">💰 成本与合规</span><br><span class="dim-value">{scores['cost_compliance']}</span></div>
                    </div>
                    <h3>攻击演练摘要</h3>
                    <p>总攻击次数: {result['attack_summary']['total_attacks']}, 成功次数: <strong style="color: {'red' if result['attack_summary']['successful_attacks'] > 0 else 'green'}">{result['attack_summary']['successful_attacks']}</strong></p>
                    <h3>原始指标 (片段)</h3>
                    <pre>{json.dumps(result['raw_metrics'], indent=2)[:500]}...</pre>
                </div>
            """
        html_content += """
                <div class="footer">
                    <p>--- 报告结束 ---</p>
                    <p><i>本报告由 KMS 红蓝对抗选型评估框架自动生成。结果基于模拟环境，生产环境选型请结合具体需求。</i></p>
                </div>
            </div>
        </body>
        </html>
        """
        with open(output_path, 'w') as f:
            f.write(html_content)
        print(f"[+] 评估报告已生成: {output_path}")

framework:
  name: "KMS RedBlue Evaluation Framework v1.0"
  log_level: "INFO"

evaluation:
  kms_backends_to_evaluate:

    - type: "local_software"
      id: "app-software-kms-01"
      config:
        some_param: "value"

    - type: "simulated_hsm"
      id: "on-prem-hsm-sim-01"
      config:
        slot_count: 8

  # 评估前准备：创建测试密钥
  test_keys:

    - key_id: "test_key_rsa_2048"
      algorithm: "RSA_2048"
      usage: "ENCRYPT_DECRYPT"

    - key_id: "test_key_aes_256"
      algorithm: "AES_256"
      usage: "ENCRYPT_DECRYPT"

attack_scenarios:

  - name: "密钥材料窃取攻击"
    enabled: true
    description: "尝试通过内部属性或API漏洞获取明文密钥材料。"
    attack_module: "core.redblue.attacks.key_theft.KeyTheftAttack"
    config:
      test_key_id: "test_key_rsa_2048"

  - name: "拒绝服务攻击 (模拟)"
    enabled: true
    description: "模拟高频调用，观察KMS性能降级或不可用。"
    attack_module: "core.redblue.attacks.denial_of_service.DenialOfServiceAttack"
    config:
      request_count: 1000
      target_operation: "encrypt"

  # 可在此处添加更多场景，例如：
  # - name: "权限绕过攻击"
  #   attack_module: "core.redblue.attacks.priviledge_escalation.PrivEscalationAttack"

import yaml
import logging
import time
from typing import Dict, Any
from core.kms.local_software import LocalSoftwareKMS
from core.kms.simulated_hsm import SimulatedHSM
from core.redblue.engine import RedBlueEngine
from evaluation.model import EvaluationModel
from evaluation.reporter import HTMLReporter
from utils.logger import setup_logging

def main():
    """主函数：加载配置，初始化KMS，运行演练，评估并生成报告。"""
    setup_logging()
    logger = logging.getLogger(__name__)
    logger.info("Starting KMS RedBlue Evaluation Framework...")

    # 1. 加载框架配置
    with open('config/framework.yaml', 'r') as f:
        framework_config = yaml.safe_load(f)

    # 2. 初始化待评估的KMS后端
    kms_instances = []
    for backend_cfg in framework_config['evaluation']['kms_backends_to_evaluate']:
        kms_type = backend_cfg['type']
        kms_id = backend_cfg['id']
        if kms_type == 'local_software':
            kms = LocalSoftwareKMS(kms_id=kms_id)
        elif kms_type == 'simulated_hsm':
            slot_count = backend_cfg['config'].get('slot_count', 5)
            kms = SimulatedHSM(kms_id=kms_id, slot_count=slot_count)
        else:
            logger.error(f"Unsupported KMS type: {kms_type}")
            continue
        kms_instances.append((kms_type, kms))
        logger.info(f"Initialized KMS: {kms_id} ({kms_type})")

    # 3. 为每个KMS创建测试密钥
    test_keys = framework_config['evaluation'].get('test_keys', [])
    for key_spec in test_keys:
        for kms_type, kms in kms_instances:
            # 简化：使用相同的key_id，实际生产应区分
            kms.create_key(**key_spec)

    # 4. 初始化红蓝对抗引擎并运行攻击
    engine = RedBlueEngine('config/attack_scenarios.yaml')
    all_evaluation_results = []
    for kms_type, kms_instance in kms_instances:
        logger.info(f"\n{'='*50}")
        logger.info(f"Evaluating KMS: {kms_instance.kms_id}")
        logger.info(f"{'='*50}")

        # 运行攻击场景
        attack_results = engine.run_scenarios(kms_instance)

        # 收集KMS性能与健康数据
        metrics = kms_instance.get_metrics()
        health = kms_instance.get_health()

        # 5. 使用评估模型进行计算
        evaluator = EvaluationModel()
        result = evaluator.evaluate(
            kms_id=kms_instance.kms_id,
            kms_type=kms_type,
            metrics=metrics,
            health=health,
            attack_results=attack_results
        )
        all_evaluation_results.append(result)

        # 打印简要结果
        logger.info(f"Evaluation for {kms_instance.kms_id}:")
        logger.info(f"  Security Score: {result['scores']['security']}")
        logger.info(f"  Performance Score: {result['scores']['performance']}")
        logger.info(f"  Total Weighted Score: {result['scores']['total_weighted_score']}")

    # 6. 生成最终HTML报告
    if all_evaluation_results:
        reporter = HTMLReporter()
        report_path = f"reports/kms_evaluation_{int(time.time())}.html"
        reporter.generate(all_evaluation_results, output_path=report_path)
        logger.info(f"\n✅ 评估完成！详细报告见: {report_path}")
    else:
        logger.error("No evaluation results were generated.")

if __name__ == "__main__":
    main()

# 创建并进入项目目录
mkdir kms_redblue_eval && cd kms_redblue_eval

# 将上述所有代码文件按项目结构树放置到对应目录

# 安装依赖
pip install -r requirements.txt

cryptography>=41.0.0
PyYAML>=6.0
numpy>=1.24.0

python app.py

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

from core.kms.local_software import LocalSoftwareKMS, KeyAlgorithm, KeyUsage

class TestLocalSoftwareKMS(unittest.TestCase):

    def setUp(self):
        self.kms = LocalSoftwareKMS(kms_id="test-kms")
        self.test_key_id = "unit_test_key"
        # 确保测试前没有残留密钥
        # (简化，实际可能需要清理方法)

    def test_create_and_encrypt_decrypt(self):
        success = self.kms.create_key(
            key_id=self.test_key_id,
            algorithm=KeyAlgorithm.AES_256,
            usage=KeyUsage.ENCRYPT_DECRYPT
        )
        self.assertTrue(success)

        plaintext = b"Hello, Secret World!"
        ciphertext, tag = self.kms.encrypt(self.test_key_id, plaintext)
        self.assertIsNotNone(ciphertext)
        self.assertIsNotNone(tag)

        decrypted = self.kms.decrypt(self.test_key_id, ciphertext, tag)
        self.assertEqual(decrypted, plaintext)

    def test_key_metadata(self):
        self.kms.create_key(self.test_key_id, KeyAlgorithm.RSA_2048, KeyUsage.SIGN_VERIFY, metadata={"owner": "test"})
        meta = self.kms.get_key_metadata(self.test_key_id)
        self.assertIsNotNone(meta)
        self.assertEqual(meta['metadata']['owner'], 'test')
        self.assertTrue(meta['enabled'])

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

python -m pytest tests/test_kms.py -v