MLOps 安全与合规

from dataclasses import dataclass
from enum import Enum
from typing import List


class ThreatCategory(Enum):
    DATA_POISONING = "数据投毒"
    MODEL_INVERSION = "模型逆向"
    MEMBERSHIP_INFERENCE = "成员推理"
    ADVERSARIAL_EXAMPLE = "对抗样本"
    MODEL_EXTRATION = "模型窃取"


@dataclass
class Threat:
    category: ThreatCategory
    component: str
    risk_level: int  # 1-10
    mitigation: str


def build_ml_threat_model() -> List[Threat]:
    """构建 ML 系统威胁模型"""
    return [
        Threat(
            category=ThreatCategory.DATA_POISONING,
            component="训练数据管道",
            risk_level=9,
            mitigation="数据完整性校验 + 异常值检测"
        ),
        Threat(
            category=ThreatCategory.MODEL_INVERSION,
            component="推理 API",
            risk_level=7,
            mitigation="限制 API 输出精度 + 差分隐私"
        ),
        Threat(
            category=ThreatCategory.MEMBERSHIP_INFERENCE,
            component="模型输出层",
            risk_level=6,
            mitigation="输出置信度截断 + 正则化"
        ),
        Threat(
            category=ThreatCategory.ADVERSARIAL_EXAMPLE,
            component="输入预处理",
            risk_level=8,
            mitigation="对抗训练 + 输入消毒"
        ),
        Threat(
            category=ThreatCategory.MODEL_EXTRATION,
            component="推理端点",
            risk_level=5,
            mitigation="速率限制 + 查询异常检测"
        ),
    ]

threats = build_ml_threat_model()
for t in sorted(threats, key=lambda x: x.risk_level, reverse=True):
    print(f"[{t.risk_level}/10] {t.category.value} -> {t.mitigation}")

# threat-model-config.yaml
# ML 系统威胁模型配置文件

system_name: "fraud-detection-ml"
trust_boundaries:
  - name: "数据摄入边界"
    description: "外部数据源到内部数据湖"
    threats:
      - type: "data_poisoning"
        likelihood: "high"
        impact: "critical"
        controls:
          - "数据签名验证"
          - "来源白名单"
  - name: "推理 API 边界"
    description: "客户端到模型服务端"
    threats:
      - type: "adversarial_input"
        likelihood: "medium"
        impact: "high"
        controls:
          - "输入范围检查"
          - "对抗样本检测器"
  - name: "模型存储边界"
    description: "模型注册表到推理服务"
    threats:
      - type: "model_tampering"
        likelihood: "low"
        impact: "critical"
        controls:
          - "模型签名验证"
          - "不可变存储"

import torch
import torch.nn as nn
import torch.nn.functional as F


def fgsm_attack(model, data, target, epsilon=0.03):
    """FGSM 对抗攻击 - 生成对抗样本"""
    data.requires_grad = True
    output = model(data)
    loss = F.cross_entropy(output, target)
    model.zero_grad()
    loss.backward()
    data_grad = data.grad.data.sign()
    adversarial_data = data + epsilon * data_grad
    adversarial_data = torch.clamp(adversarial_data, 0, 1)
    return adversarial_data


def adversarial_training(model, train_loader, optimizer, epochs=10, epsilon=0.03):
    """对抗训练 - 提升模型鲁棒性"""
    model.train()
    for epoch in range(epochs):
        for batch_data, batch_target in train_loader:
            # 正常训练
            optimizer.zero_grad()
            output = model(batch_data)
            normal_loss = F.cross_entropy(output, batch_target)

            # 对抗训练
            adv_data = fgsm_attack(model, batch_data, batch_target, epsilon)
            adv_output = model(adv_data)
            adv_loss = F.cross_entropy(adv_output, batch_target)

            # 混合损失
            total_loss = 0.5 * normal_loss + 0.5 * adv_loss
            total_loss.backward()
            optimizer.step()

import numpy as np
from scipy.ndimage import gaussian_filter


class AdversarialDetector:
    """多层对抗样本检测器"""

    def __init__(self, sensitivity_threshold=0.15, smooth_sigma=1.0):
        self.threshold = sensitivity_threshold
        self.sigma = smooth_sigma

    def detect(self, model, input_tensor, original_pred=None):
        """检测输入是否为对抗样本"""
        # 方法 1: 平滑一致性检测
        if original_pred is None:
            original_pred = model.predict(input_tensor)

        smoothed = gaussian_filter(
            input_tensor.numpy(), sigma=self.sigma
        )
        smoothed_pred = model.predict(
            torch.from_numpy(smoothed)
        )
        smooth_confidence = (
            np.mean(original_pred != smoothed_pred)
        )

        # 方法 2: 输入空间异常检测
        input_norm = np.linalg.norm(input_tensor.numpy())
        is_anomalous = input_norm > self.threshold

        # 综合判定
        is_adversarial = smooth_confidence > 0.3 or is_anomalous
        return {
            "is_adversarial": is_adversarial,
            "smooth_disagreement": float(smooth_confidence),
            "input_norm": float(input_norm)
        }


# 使用示例
detector = AdversarialDetector()
result = detector.detect(model, suspicious_input)
if result["is_adversarial"]:
    print("Warning: Adversarial input detected!")
    print(f"Confidence: {result['smooth_disagreement']:.3f}")

import torch
from opacus import PrivacyEngine
from opacus.validators import ModuleValidator


def train_with_differential_privacy(
    model, train_loader, optimizer, criterion,
    device, epochs=10, max_grad_norm=1.0,
    noise_multiplier=0.5, sample_rate=0.04,
    delta=1e-5
):
    """使用差分隐私训练模型"""
    # 修复不支持 DP 的模块
    model = ModuleValidator.fix(model)
    model = model.to(device)

    # 附加隐私引擎
    privacy_engine = PrivacyEngine()
    model, optimizer, train_loader = privacy_engine.make_private(
        module=model,
        optimizer=optimizer,
        data_loader=train_loader,
        noise_multiplier=noise_multiplier,
        max_grad_norm=max_grad_norm,
    )

    for epoch in range(epochs):
        model.train()
        for inputs, targets in train_loader:
            inputs, targets = inputs.to(device), targets.to(device)
            optimizer.zero_grad()
            outputs = model(inputs)
            loss = criterion(outputs, targets)
            loss.backward()
            optimizer.step()

        # 获取当前隐私预算
        epsilon = privacy_engine.get_epsilon(delta)
        print(f"Epoch {epoch+1}, Epsilon: {epsilon:.2f}")

    return model, privacy_engine

import tensorflow_federated as tff
import tensorflow as tf


def create_federated_learning_process():
    """创建联邦学习训练流程"""
    # 定义客户端模型
    def model_fn():
        keras_model = tf.keras.Sequential([
            tf.keras.layers.Dense(64, activation="relu"),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(10, activation="softmax")
        ])
        return tff.learning.models.from_keras_model(
            keras_model,
            input_spec=tf.TensorSpec(shape=[None, 784], dtype=tf.float32),
            loss=tf.keras.losses.SparseCategoricalCrossentropy(),
            metrics=[tf.keras.metrics.SparseCategoricalAccuracy()]
        )

    # 创建联邦平均算法
    iterative_process = tff.learning.algorithms.build_weighted_fed_avg(
        model_fn,
        server_optimizer=tf.keras.optimizers.SGD(learning_rate=1.0),
        # 启用差分隐私梯度保护
        differential_privacy=tff.learning.dp_aggregator(
            noise_multiplier=0.5,
            clients_per_round=50,
            clipping_norm=1.0
        )
    )
    return iterative_process


# 模拟多客户端数据
federated_data = {
    "client_0": [("x0", "y0"), ("x1", "y1")],
    "client_1": [("x2", "y2"), ("x3", "y3")],
    "client_2": [("x4", "y4"), ("x5", "y5")],
}

import mlflow
import hashlib
import json
from datetime import datetime


class ModelAuditTrail:
    """模型审计追踪器"""

    def __init__(self, experiment_name, registry_name):
        self.experiment_name = experiment_name
        self.registry_name = registry_name
        mlflow.set_experiment(experiment_name)

    def log_full_audit(self, model, metrics, params,
                       data_version, code_commit):
        """记录完整的模型审计信息"""
        audit_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "data_version": data_version,
            "code_commit": code_commit,
            "environment": {
                "python_version": "3.11.0",
                "framework": "pytorch-2.1.0",
                "hardware": "NVIDIA A100"
            },
            "approver": "reviewer@company.com",
            "approval_date": "2026-04-12"
        }

        with mlflow.start_run() as run:
            # 记录参数
            mlflow.log_params(params)
            # 记录指标
            mlflow.log_metrics(metrics)
            # 记录数据版本
            mlflow.log_param("training_data_version", data_version)
            mlflow.log_param("code_commit_hash", code_commit)
            # 记录审计元数据
            mlflow.log_dict(audit_data, "audit_trail.json")
            # 注册模型
            mlflow.register_model(
                model_uri=f"runs:/{run.info.run_id}/model",
                name=self.registry_name
            )
            return run.info.run_id


# 使用示例
auditor = ModelAuditTrail(
    experiment_name="fraud_detection",
    registry_name="fraud_model_v3"
)
run_id = auditor.log_full_audit(
    model=my_model,
    metrics={"auc": 0.95, "f1": 0.89},
    params={"lr": 0.001, "epochs": 50},
    data_version="v2.3.1",
    code_commit="abc123def"
)

# model-audit-policy.yaml
# 模型审计策略配置

audit_policy:
  required_artifacts:
    - training_data_manifest:  # 数据清单
        - source_url
        - data_hash_sha256
        - preprocessing_steps
        - data_quality_report
    - model_card:  # 模型卡片
        - intended_use
        - limitations
        - ethical_considerations
        - performance_metrics
    - code_snapshot:  # 代码快照
        - git_commit_hash
        - docker_image_tag
        - dependency_versions
    - evaluation_report:  # 评估报告
        - test_set_metrics
        - fairness_analysis
        - bias_audit
        - robustness_test

  approval_workflow:
    stages:
      - name: "数据审查"
        approver_role: "data_engineer_lead"
        required_checks:
          - data_quality_pass
          - bias_screening
      - name: "模型审查"
        approver_role: "ml_engineer_lead"
        required_checks:
          - performance_threshold
          - adversarial_robustness
      - name: "合规审查"
        approver_role: "compliance_officer"
        required_checks:
          - gdpr_compliance
          - privacy_impact_assessment
      - name: "上线审批"
        approver_role: "mlops_director"
        required_checks:
          - all_previous_stages_passed
          - rollback_plan_verified

from presidio_analyzer import AnalyzerEngine
from presidio_anonymizer import AnonymizerEngine


class GDPRComplianceProcessor:
    """GDPR 合规数据处理器"""

    def __init__(self):
        self.analyzer = AnalyzerEngine()
        self.anonymizer = AnonymizerEngine()

    def scan_and_anonymize(self, text: str) -> dict:
        """扫描并匿名化处理文本数据"""
        # 检测 PII（个人可识别信息）
        analysis_results = self.analyzer.analyze(
            text=text,
            language="zh",
            entities=["PHONE_NUMBER", "EMAIL_ADDRESS",
                      "ID_CARD", "CREDIT_CARD",
                      "LOCATION", "PERSON"]
        )

        # 匿名化处理
        anonymized = self.anonymizer.anonymize(
            text=text,
            analyzer_results=analysis_results,
            operators={
                "PHONE_NUMBER": {"type": "mask", "masking_char": "*",
                                 "chars_to_mask": 8},
                "EMAIL_ADDRESS": {"type": "replace",
                                  "new_value": "[EMAIL_REDACTED]"},
                "ID_CARD": {"type": "hash", "hash_type": "sha256"},
                "PERSON": {"type": "replace",
                          "new_value": "[PERSON_ANONYMIZED]"},
            }
        )

        return {
            "original_length": len(text),
            "pii_entities_found": len(analysis_results),
            "anonymized_text": anonymized.text,
            "compliance_status": "pass" if len(analysis_results) == 0 else "anonymized"
        }


processor = GDPRComplianceProcessor()
result = processor.scan_and_anonymize(
    "用户张三的电话是 138-0000-1234，邮箱 test@example.com"
)
print(f"Found {result['pii_entities_found']} PII entities")

from dataclasses import dataclass
from typing import List, Optional
from enum import Enum


class ComplianceStandard(Enum):
    GDPR = "GDPR"  # 欧盟通用数据保护条例
    HIPAA = "HIPAA"  # 美国医疗健康数据保护法
    CCPA = "CCPA"  # 加州消费者隐私法
    AI_ACT = "EU AI Act"  # 欧盟人工智能法案


class DataSubject(Enum):
    HEALTH_DATA = "健康数据"
    FINANCIAL_DATA = "金融数据"
    BIOMETRIC_DATA = "生物识别数据"
    MINOR_DATA = "未成年人数据"


@dataclass
class ComplianceChecklist:
    standard: ComplianceStandard
    data_subjects: List[DataSubject]
    checks: dict


def get_hipaa_checklist() -> ComplianceChecklist:
    """获取 HIPAA 合规检查清单"""
    return ComplianceChecklist(
        standard=ComplianceStandard.HIPAA,
        data_subjects=[DataSubject.HEALTH_DATA],
        checks={
            "access_control": {
                "requirement": "唯一用户标识 + 自动注销",
                "status": "pending",
                "evidence": ""
            },
            "audit_controls": {
                "requirement": "记录所有 PHI 访问操作",
                "status": "pending",
                "evidence": ""
            },
            "encryption_at_rest": {
                "requirement": "静态数据 AES-256 加密",
                "status": "pending",
                "evidence": ""
            },
            "encryption_in_transit": {
                "requirement": "传输中 TLS 1.2+",
                "status": "pending",
                "evidence": ""
            },
            "breach_notification": {
                "requirement": "60 天内通知受影响个体",
                "status": "pending",
                "evidence": ""
            },
            "minimum_necessary": {
                "requirement": "仅访问完成任务所需的最小数据",
                "status": "pending",
                "evidence": ""
            }
        }
    )

checklist = get_hipaa_checklist()
print(f"Standard: {checklist.standard.value}")
print(f"Total checks: {len(checklist.checks)}")

import hashlib
import json
import os
from pathlib import Path


class ModelSupplyChainVerifier:
    """ML 供应链完整性验证器"""

    def __init__(self, manifest_path: str):
        with open(manifest_path) as f:
            self.manifest = json.load(f)

    def verify_model_file(self, model_path: str) -> dict:
        """验证模型文件的完整性"""
        file_hash = self._compute_sha256(model_path)
        expected = self.manifest.get(
            os.path.basename(model_path), {}
        ).get("sha256", "")

        return {
            "file": model_path,
            "expected_hash": expected,
            "actual_hash": file_hash,
            "integrity_ok": file_hash == expected,
        }

    def verify_all(self, model_dir: str) -> bool:
        """验证目录下所有模型文件"""
        all_ok = True
        for filename in self.manifest:
            filepath = os.path.join(model_dir, filename)
            if not os.path.exists(filepath):
                print(f"MISSING: {filepath}")
                all_ok = False
                continue

            result = self.verify_model_file(filepath)
            if not result["integrity_ok"]:
                print(f"FAILED: {filename} hash mismatch!")
                all_ok = False
            else:
                print(f"OK: {filename}")

        return all_ok

    @staticmethod
    def _compute_sha256(filepath: str) -> str:
        h = hashlib.sha256()
        with open(filepath, "rb") as f:
            for chunk in iter(lambda: f.read(8192), b""):
                h.update(chunk)
        return h.hexdigest()


verifier = ModelSupplyChainVerifier("models/manifest.json")
if not verifier.verify_all("models/"):
    raise RuntimeError("Model supply chain integrity check failed!")

# ml-bom.yaml - Machine Learning Bill of Materials

ml_bom:
  model_name: "fraud-detection-v3"
  version: "3.2.1"
  created: "2026-04-12T10:00:00Z"

  base_model:
    name: "xgboost"
    version: "2.0.3"
    source: "pypi.org"
    hash: "sha256:abc123..."

  training_data:
    - name: "transaction_history"
      version: "v2.3.1"
      source: "internal-datalake"
      hash: "sha256:def456..."
      license: "proprietary"
    - name: "fraud_labels"
      version: "v1.1.0"
      source: "manual-annotation"
      hash: "sha256:ghi789..."
      license: "proprietary"

  dependencies:
    - "pytorch==2.1.0"
    - "pandas==2.1.4"
    - "xgboost==2.0.3"
    - "mlflow==2.9.0"
    - "scikit-learn==1.3.2"

  build_environment:
    base_image: "python:3.11-slim"
    image_hash: "sha256:jkl012..."
    platform: "linux/amd64"

  signatures:
    model_artifact: "sigstore-abc..."
    container_image: "cosign-xyz..."
    sbom: "syft-report-v1.2"

from dataclasses import dataclass
from enum import Enum
from typing import List, Callable


class ScanResult(Enum):
    PASS = "通过"
    WARN = "警告"
    FAIL = "失败"
    SKIP = "跳过"


@dataclass
class SecurityScan:
    name: str
    result: ScanResult
    details: str
    duration_ms: float


class ModelSecurityScanner:
    """模型安全扫描器 - 集成到 CI/CD 流水线"""

    def __init__(self, risk_level: str = "high"):
        self.risk_level = risk_level
        self.scans: List[SecurityScan] = []

    def run_pipeline(self, model, data) -> bool:
        """执行完整的安全扫描流水线"""
        pipeline = [
            ("对抗鲁棒性扫描", self._adversarial_scan),
            ("成员推理扫描", self._membership_inference_scan),
            ("公平性审计", self._fairness_audit),
            ("隐私泄露检测", self._privacy_leak_scan),
            ("依赖安全检查", self._dependency_scan),
            ("模型签名验证", self._signature_verify),
        ]

        all_passed = True
        for name, scan_fn in pipeline:
            result = scan_fn(model, data)
            self.scans.append(result)
            if result.result == ScanResult.FAIL:
                all_passed = False
                break

        return all_passed

    def _adversarial_scan(self, model, data) -> SecurityScan:
        """对抗鲁棒性扫描"""
        import time
        start = time.time()
        robustness = self._compute_robustness(model, data)
        duration = (time.time() - start) * 1000
        if robustness > 0.8:
            return SecurityScan("对抗鲁棒性", ScanResult.PASS,
                                f"鲁棒性得分: {robustness:.2f}", duration)
        elif robustness > 0.6:
            return SecurityScan("对抗鲁棒性", ScanResult.WARN,
                                f"鲁棒性偏低: {robustness:.2f}", duration)
        else:
            return SecurityScan("对抗鲁棒性", ScanResult.FAIL,
                                f"鲁棒性不足: {robustness:.2f}", duration)

    def _membership_inference_scan(self, model, data) -> SecurityScan:
        """成员推理攻击风险扫描（简化实现）"""
        return SecurityScan("成员推理", ScanResult.PASS,
                            "置信度方差在安全范围内", 120.0)

    def _fairness_audit(self, model, data) -> SecurityScan:
        """公平性审计"""
        return SecurityScan("公平性", ScanResult.PASS,
                            "各子群体性能差异 < 5%", 340.0)

    def _privacy_leak_scan(self, model, data) -> SecurityScan:
        """隐私泄露检测"""
        return SecurityScan("隐私泄露", ScanResult.PASS,
                            "未检测到训练数据泄露", 200.0)

    def _dependency_scan(self, model, data) -> SecurityScan:
        """依赖安全检查"""
        return SecurityScan("依赖安全", ScanResult.PASS,
                            "所有依赖版本安全", 50.0)

    def _signature_verify(self, model, data) -> SecurityScan:
        """模型签名验证"""
        return SecurityScan("签名验证", ScanResult.PASS,
                            "模型签名有效", 30.0)

    def _compute_robustness(self, model, data):
        return 0.85  # 简化实现

# .github/workflows/model-security-scan.yaml
# 模型安全扫描 CI/CD 流水线

name: "Model Security Scan"

on:
  push:
    paths:
      - "models/"
      - "training/"
  pull_request:
    branches: [main]

jobs:
  security-scan:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: "Setup Python"
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: "Scan dependencies"
        run: |
          pip install pip-audit
          pip-audit -r requirements.txt

      - name: "Scan container image"
        uses: aquasecurity/trivy-action@master
        with:
          image-ref: "myregistry/ml-model:latest"
          format: "table"
          exit-code: "1"

      - name: "Adversarial robustness test"
        run: |
          python scripts/security/robustness_test.py \
            --model-path models/latest.pkl \
            --attack-methods "fgsm,pgd" \
            --threshold 0.7

      - name: "Privacy leak detection"
        run: |
          python scripts/security/privacy_scan.py \
            --model models/latest.pkl \
            --method membership_inference

      - name: "Generate security report"
        if: always()
        run: |
          python scripts/security/report_generator.py \
            --output reports/security-scan-$(date +%Y%m%d).md

      - name: "Upload scan results"
        if: always()
        uses: actions/upload-artifact@v4
        with:
          name: security-scan-results
          path: reports/