紅隊 Automation Strategy

┌──────────────────────────────────────────────────────────────┐
│ Layer 3: Human Analysis                                       │
│ Novel attacks, contextual 評估, reporting               │
│ (Triggered by findings from Layers 1-2)                       │
├──────────────────────────────────────────────────────────────┤
│ Layer 2: Continuous Automated 紅隊演練 (CART)              │
│ Scheduled and triggered scans against production models       │
│ (Runs continuously or on model updates)                       │
├──────────────────────────────────────────────────────────────┤
│ Layer 1: CI/CD Integration                                    │
│ Automated checks in the deployment pipeline                   │
│ (Runs on every deployment)                                    │
└──────────────────────────────────────────────────────────────┘

# 範例: GitHub Actions workflow for AI 安全 checks
name: AI 安全 Checks
on:
  pull_request:
    paths:
      - 'models/**'
      - 'prompts/**'
      - 'configs/安全/**'
  push:
    branches: [main]
 
jobs:
  安全-regression:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
 
      - name: Run 安全 regression suite
        run: |
          python -m pytest tests/安全/ \
            --model-endpoint ${{ secrets.STAGING_MODEL_ENDPOINT }} \
            --threshold 0.95 \
            --report-輸出 安全-report.json
 
      - name: Run 越獄 regression
        run: |
          garak --model-type api \
            --model-endpoint ${{ secrets.STAGING_MODEL_ENDPOINT }} \
            --probes known_jailbreaks \
            --report-prefix ci-越獄
 
      - name: Check results
        run: |
          python scripts/check_safety_results.py \
            --安全-report 安全-report.json \
            --越獄-report ci-越獄.json \
            --fail-on-regression

class CARTPipeline:
    """Continuous Automated 紅隊演練 pipeline."""
 
    def __init__(self, model_endpoints: dict, alert_webhook: str):
        self.endpoints = model_endpoints
        self.alert_webhook = alert_webhook
        self.attack_registry = AttackRegistry()
        self.results_store = ResultsStore()
 
    def run_cycle(self):
        """Run one CART cycle across all monitored endpoints."""
        for endpoint_name, endpoint_config in self.endpoints.items():
            results = self._test_endpoint(endpoint_name, endpoint_config)
            self.results_store.save(endpoint_name, results)
 
            # Alert on new findings
            new_findings = self._identify_new_findings(endpoint_name, results)
            if new_findings:
                self._send_alert(endpoint_name, new_findings)
 
    def _test_endpoint(self, name: str, config: dict) -> list:
        """測試 a single endpoint with all applicable attack categories."""
        results = []
 
        for attack in self.attack_registry.get_attacks(config["model_type"]):
            try:
                outcome = attack.execute(config["url"], config["auth"])
                results.append({
                    "attack": attack.name,
                    "category": attack.category,
                    "success": outcome.succeeded,
                    "response": outcome.response[:500],
                    "timestamp": datetime.now().isoformat()
                })
            except Exception as e:
                results.append({
                    "attack": attack.name,
                    "error": str(e),
                    "timestamp": datetime.now().isoformat()
                })
 
        return results
 
    def _identify_new_findings(self, endpoint: str, results: list) -> list:
        """Compare results with previous cycle to find new 漏洞."""
        previous = self.results_store.get_latest(endpoint)
        if previous is None:
            return [r for r in results if r.get("success")]
 
        prev_successes = set(r["attack"] for r in previous if r.get("success"))
        new_successes = [
            r for r in results
            if r.get("success") and r["attack"] not in prev_successes
        ]
 
        return new_successes
 
    def _send_alert(self, endpoint: str, findings: list):
        """Send alert for new findings."""
        import requests
        requests.post(self.alert_webhook, json={
            "endpoint": endpoint,
            "new_findings": len(findings),
            "details": findings,
            "severity": "high" if any(f["category"] == "safety_bypass" for f in findings) else "medium"
        })

Automation → Alert → Triage (human) → Investigation (human) → Report (human)
     ↓                                       ↓
  New attack patterns                   Feed back into
  from human research                   automation registry

class TriageWorkflow:
    """Human-in-the-loop triage for automated findings."""
 
    SEVERITY_THRESHOLDS = {
        "auto_escalate": 0.95,    # Very high confidence: auto-escalate with human notification
        "human_review": 0.70,     # Moderate confidence: require human review before escalation
        "auto_dismiss": 0.30,     # Low confidence: auto-dismiss with periodic audit
    }
 
    def triage(self, finding: dict) -> dict:
        """Route a finding based on confidence level."""
        confidence = finding.get("confidence", 0.5)
 
        if confidence >= self.SEVERITY_THRESHOLDS["auto_escalate"]:
            return {
                "action": "escalate",
                "human_required": False,
                "notification": True,
                "reason": "High-confidence automated finding"
            }
        elif confidence >= self.SEVERITY_THRESHOLDS["human_review"]:
            return {
                "action": "queue_for_review",
                "human_required": True,
                "priority": "normal",
                "reason": "Moderate confidence, requires human verification"
            }
        else:
            return {
                "action": "log_and_dismiss",
                "human_required": False,
                "audit_flag": True,
                "reason": "Low confidence, logged for periodic audit review"
            }