AI 国防基础设施与治理：从五角大楼 AI 协议到国家级 AI 治理体系

# 国防级 AI 模型安全评估框架
# 包含对抗鲁棒性、公平性和可解释性三重评估

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

class DefenseAIModelEvaluator:
    """国防级 AI 模型安全评估器
    
    满足 DoD RAI 框架要求的三维度评估：
    1. 对抗鲁棒性 (Adversarial Robustness)
    2. 公平性 (Fairness)
    3. 可解释性 (Explainability)
    """
    
    def __init__(self, model, classification_level: str = "SECRET"):
        self.model = model
        self.classification_level = classification_level
        self.assessment_results: Dict = {}
    
    def adversarial_robustness_test(
        self, 
        test_samples: np.ndarray,
        attack_methods: List[str] = ["FGSM", "PGD", "CW"]
    ) -> Dict:
        """对抗鲁棒性测试
        
        使用多种攻击方法评估模型的对抗鲁棒性。
        DoD 要求所有部署模型必须通过对抗测试。
        """
        results = {}
        
        for attack in attack_methods:
            perturbed_samples = self._generate_adversarial_samples(
                test_samples, method=attack
            )
            
            original_preds = self.model.predict(test_samples)
            perturbed_preds = self.model.predict(perturbed_samples)
            
            # 计算攻击成功率（ASR）
            asr = np.mean(original_preds != perturbed_preds)
            
            # DoD 阈值：ASR 必须 < 5% 才能部署
            passed = asr < 0.05
            
            results[attack] = {
                "attack_success_rate": float(asr),
                "dod_threshold": 0.05,
                "passed": passed,
                "classification_required": self.classification_level
            }
        
        return results
    
    def fairness_assessment(
        self,
        test_data: np.ndarray,
        sensitive_attributes: List[str],
        decision_threshold: float = 0.5
    ) -> Dict:
        """公平性评估
        
        检查模型在不同敏感属性群体间的表现差异。
        DoD 要求：群体间准确率差异 < 5%。
        """
        predictions = self.model.predict(test_data) > decision_threshold
        
        fairness_metrics = {}
        for attr in sensitive_attributes:
            groups = self._get_attribute_groups(test_data, attr)
            group_accuracies = {}
            
            for group_name, group_mask in groups.items():
                if np.sum(group_mask) > 0:
                    group_acc = np.mean(
                        predictions[group_mask] == self._get_true_labels(test_data, group_mask)
                    )
                    group_accuracies[group_name] = float(group_acc)
            
            # 计算最大差异
            if len(group_accuracies) > 1:
                max_diff = max(group_accuracies.values()) - min(group_accuracies.values())
            else:
                max_diff = 0.0
            
            fairness_metrics[attr] = {
                "group_accuracies": group_accuracies,
                "max_disparity": float(max_diff),
                "dod_threshold": 0.05,
                "passed": max_diff < 0.05
            }
        
        return fairness_metrics
    
    def generate_assessment_report(self) -> str:
        """生成符合 DoD 标准的评估报告"""
        report_lines = [
            "=" * 60,
            "DoD AI MODEL SECURITY ASSESSMENT REPORT",
            f"Classification: {self.classification_level}",
            "=" * 60,
            "",
        ]
        
        # 对抗鲁棒性
        report_lines.append("## 1. ADVERSARIAL ROBUSTNESS")
        for attack, result in self.assessment_results.get("adversarial", {}).items():
            status = "✅ PASS" if result["passed"] else "❌ FAIL"
            report_lines.append(
                f"  {attack}: ASR={result['attack_success_rate']:.4f} "
                f"(threshold: {result['dod_threshold']}) {status}"
            )
        
        # 公平性
        report_lines.append("")
        report_lines.append("## 2. FAIRNESS ASSESSMENT")
        for attr, result in self.assessment_results.get("fairness", {}).items():
            status = "✅ PASS" if result["passed"] else "❌ FAIL"
            report_lines.append(
                f"  {attr}: max_disparity={result['max_disparity']:.4f} "
                f"(threshold: {result['dod_threshold']}) {status}"
            )
        
        # 总体结论
        all_passed = all(
            r.get("passed", False)
            for category in self.assessment_results.values()
            for r in category.values()
            if isinstance(r, dict) and "passed" in r
        )
        
        report_lines.append("")
        report_lines.append(f"## OVERALL: {'✅ APPROVED FOR DEPLOYMENT' if all_passed else '❌ NOT APPROVED'}")
        
        return "\n".join(report_lines)
    
    def _generate_adversarial_samples(self, samples, method):
        # 实现省略：FGSM/PGD/CW 攻击生成
        return samples
    
    def _get_attribute_groups(self, data, attr):
        # 实现省略：按敏感属性分组
        return {}
    
    def _get_true_labels(self, data, mask):
        # 实现省略：获取真实标签
        return []

# 国防级 AI 模型部署管道配置
# 符合 DoD RAI 框架和零信任架构要求

apiVersion: defense.ai/v1
kind: ModelDeploymentPipeline
metadata:
  name: target-recognition-v3
  classification: SECRET
  rai_champion: "COL_Johnson_4281"

spec:
  # 安全环境要求
  securityContext:
    cloudProvider: aws-govcloud
    ilLevel: IL6
    networkIsolation: true
    encryptionInTransit: TLS1.3
    encryptionAtRest: AES256-GCM
    keyManagement: government-controlled

  # 模型注册要求
  modelRegistry:
    source: "classified-training-cluster-7"
    trainingDataClassification: TOP_SECRET//SCI
    modelVersion: "3.2.1"
    requiredAssessments:
      - adversarialRobustness
      - fairness
      - explainability
      - supplyChainVerification

  # 部署审批流程
  approvalWorkflow:
    - step: automatedSecurityScan
      required: true
      timeout: 2h
    - step: raiChampionReview
      required: true
      approver: "rai_champion"
      timeout: 24h
    - step: securityOfficerApproval
      required: true
      approver: "security_officer"
      timeout: 48h
    - step: deploymentCommanderSignoff
      required: true
      approver: "deployment_commander"
      timeout: 72h

  # 持续监控要求
  continuousMonitoring:
    performanceDriftThreshold: 0.03
    adversarialDetection: true
    auditLogRetention: 7years
    alertEscalationPath:
      - level: automated
        response: "pause_and_notify"
      - level: rai_champion
        response: "manual_review"
      - level: security_officer
        response: "immediate_rollback"

# 国防级 AI 数据分级与访问控制系统
# 实现强制访问控制（MAC）和需要-to-know 原则

from enum import Enum
from dataclasses import dataclass
from typing import Set, Optional
import hashlib

class ClassificationLevel(Enum):
    """数据分类级别"""
    UNCLASSIFIED = 0      # 非密
    CUI = 1               # 受控非密
    SECRET = 2            # 机密
    TOP_SECRET = 3        # 最高机密
    TOP_SECRET_SCI = 4    # 最高机密//敏感隔离信息

@dataclass
class SecurityClearance:
    """人员安全许可"""
    level: ClassificationLevel
    compartments: Set[str]  # 隔离 compartments（如 SCI 的特定子集）
    need_to_know: Set[str]  # 需要-to-know 的数据标签

@dataclass
class DataLabel:
    """数据标签"""
    classification: ClassificationLevel
    compartments: Set[str]
    handling_codes: Set[str]  # 处理代码（如 NOFORN、ORCON）

class DefenseDataAccessController:
    """国防级数据访问控制器
    
    实现强制访问控制（MAC）：
    - 简单安全规则（SSR）：主体只能读取 ≤ 其许可级别的数据
    - *-property：主体只能写入 ≥ 其许可级别的数据
    - Need-to-Know：即使许可级别足够，也需要业务需要
    """
    
    def __init__(self):
        self.access_log: list = []
    
    def can_read(
        self, 
        clearance: SecurityClearance, 
        data_label: DataLabel
    ) -> tuple[bool, str]:
        """检查读访问权限
        
        必须同时满足：
        1. 安全许可级别 ≥ 数据分类级别
        2. 主体的 compartments 覆盖数据的所有 compartments
        3. 数据标签在主体的 need-to-know 范围内
        4. 处理代码不限制读取
        """
        # 检查安全许可级别
        if clearance.level.value < data_label.classification.value:
            reason = (
                f"安全许可不足: "
                f"{clearance.level.name} < {data_label.classification.name}"
            )
            self._log_access(False, reason)
            return False, reason
        
        # 检查 compartments
        if not data_label.compartments.issubset(clearance.compartments):
            missing = data_label.compartments - clearance.compartments
            reason = f"缺少 compartment 访问权限: {missing}"
            self._log_access(False, reason)
            return False, reason
        
        # 检查 need-to-know
        if not data_label.compartments.issubset(clearance.need_to_know):
            missing = data_label.compartments - clearance.need_to_know
            reason = f"不在 need-to-know 范围内: {missing}"
            self._log_access(False, reason)
            return False, reason
        
        # 检查处理代码
        if "NOFORN" in data_label.handling_codes:
            reason = "数据标记为 NOFORN（不可对外国国民公开）"
            self._log_access(False, reason)
            return False, reason
        
        self._log_access(True, "访问已批准")
        return True, "访问已批准"
    
    def can_write(
        self, 
        clearance: SecurityClearance, 
        target_label: DataLabel
    ) -> tuple[bool, str]:
        """检查写访问权限（*-property）
        
        主体只能向 ≥ 其许可级别的系统写入数据，
        防止高密级数据意外流向低密级系统。
        """
        if clearance.level.value > target_label.classification.value:
            reason = (
                f"写入级别过低: "
                f"主体级别 {clearance.level.name} > "
                f"目标级别 {target_label.classification.name} "
                f"(违反 *-property)"
            )
            self._log_access(False, reason)
            return False, reason
        
        self._log_access(True, "写入已批准")
        return True, "写入已批准"
    
    def classify_ai_training_data(
        self, 
        raw_data: dict,
        source_classification: ClassificationLevel
    ) -> DataLabel:
        """为 AI 训练数据生成安全标签
        
        AI 训练数据的分类级别由其最高密级的
        训练样本决定。混合训练集必须标记为
        其中最高的安全级别。
        """
        compartments = {f"AI_TRAINING_{source_classification.name}"}
        handling_codes = {"ORCON"}  # 控制分发
        
        return DataLabel(
            classification=source_classification,
            compartments=compartments,
            handling_codes=handling_codes
        )
    
    def _log_access(self, granted: bool, reason: str):
        log_entry = {
            "granted": granted,
            "reason": reason,
            "timestamp": "2026-05-02T05:00:00Z"
        }
        self.access_log.append(log_entry)

# 使用示例
if __name__ == "__main__":
    controller = DefenseDataAccessController()
    
    # 创建安全许可（上校级别，拥有特定 compartment 访问权）
    colonel_clearance = SecurityClearance(
        level=ClassificationLevel.TOP_SECRET,
        compartments={"GAMMA", "AI_SYSTEMS", "TARGET_RECOGNITION"},
        need_to_know={"GAMMA", "AI_SYSTEMS"}
    )
    
    # 创建数据标签（机密级 AI 训练数据）
    secret_training_data = DataLabel(
        classification=ClassificationLevel.SECRET,
        compartments={"AI_SYSTEMS"},
        handling_codes={"ORCON"}
    )
    
    # 检查读访问
    can_read, reason = controller.can_read(
        colonel_clearance, secret_training_data
    )
    print(f"读访问: {can_read} - {reason}")
    
    # 尝试向低密级系统写入（应被拒绝）
    unclassified_label = DataLabel(
        classification=ClassificationLevel.UNCLASSIFIED,
        compartments=set(),
        handling_codes=set()
    )
    can_write, reason = controller.can_write(
        colonel_clearance, unclassified_label
    )
    print(f"写入非密系统: {can_write} - {reason}")