AI 攻擊時間軸重建

from dataclasses import dataclass
from typing import Optional
import json
 
@dataclass
class SecurityAnalysis:
    """用於分析 LLM 系統安全特性的框架。"""
    target: str
    model: str
    defenses: list
    vulnerabilities: list
 
    def assess_risk(self, attack_type: str) -> dict:
        """評估特定攻擊類型的風險。"""
        # 檢查是否有防禦措施涵蓋此攻擊類型
        relevant_defenses = [
            d for d in self.defenses
            if attack_type in d.get("covers", [])
        ]
 
        # 風險因子
        likelihood = "high" if not relevant_defenses else "medium"
        impact = self._assess_impact(attack_type)
 
        return {
            "attack_type": attack_type,
            "likelihood": likelihood,
            "impact": impact,
            "defenses": len(relevant_defenses),
            "risk_level": self._calculate_risk(likelihood, impact),
        }
 
    def _assess_impact(self, attack_type: str) -> str:
        """評估某攻擊類型的潛在影響。"""
        high_impact = ["data_exfiltration", "unauthorized_actions", "privilege_escalation"]
        return "high" if attack_type in high_impact else "medium"
 
    def _calculate_risk(self, likelihood: str, impact: str) -> str:
        """由可能性與影響計算整體風險。"""
        risk_matrix = {
            ("high", "high"): "critical",
            ("high", "medium"): "high",
            ("medium", "high"): "high",
            ("medium", "medium"): "medium",
        }
        return risk_matrix.get((likelihood, impact), "medium")
 
    def generate_report(self) -> str:
        """產生風險評估報告。"""
        attacks = ["prompt_injection", "data_exfiltration", "unauthorized_actions"]
        assessments = [self.assess_risk(a) for a in attacks]
 
        report = f"# Risk Assessment: {self.target}\n\n"
        for assessment in assessments:
            report += (
                f"## {assessment['attack_type']}\n"
                f"- Risk: {assessment['risk_level']}\n"
                f"- Likelihood: {assessment['likelihood']}\n"
                f"- Impact: {assessment['impact']}\n"
                f"- Active defenses: {assessment['defenses']}\n\n"
            )
        return report

class PracticalFramework:
    """用於套用本文概念的實務框架。"""
 
    def __init__(self, target_config: dict):
        self.config = target_config
        self.findings = []
        self.tested_vectors = set()
 
    def test_vector(self, vector: str, payload: str) -> dict:
        """針對目標測試特定攻擊向量。"""
        self.tested_vectors.add(vector)
 
        # 送出載荷
        response = self._send(payload)
 
        # 評估結果
        finding = {
            "vector": vector,
            "payload_length": len(payload),
            "response_length": len(response),
            "success": self._evaluate(response),
            "defense_triggered": self._detect_defense(response),
        }
 
        if finding["success"]:
            self.findings.append(finding)
 
        return finding
 
    def coverage_report(self) -> dict:
        """回報測試覆蓋率。"""
        all_vectors = {
            "direct_injection", "indirect_injection", "function_abuse",
            "memory_manipulation", "context_manipulation",
        }
        return {
            "tested": list(self.tested_vectors),
            "untested": list(all_vectors - self.tested_vectors),
            "coverage": f"{len(self.tested_vectors)/len(all_vectors)*100:.0f}%",
            "findings": len(self.findings),
        }
 
    def _send(self, payload: str) -> str:
        """將載荷送至目標(實作依目標而異)。"""
        pass
 
    def _evaluate(self, response: str) -> bool:
        """評估攻擊是否成功。"""
        pass
 
    def _detect_defense(self, response: str) -> Optional[str]:
        """偵測觸發了哪個防禦機制。"""
        pass

from dataclasses import dataclass
from typing import Optional
import time
 
@dataclass
class SecurityGateway:
    """用於保護 LLM 應用程式存取的閘道模式。"""
 
    input_classifier: object  # 基於 ML 的輸入分類器
    output_filter: object     # 輸出內容過濾器
    rate_limiter: object      # 速率限制服務
    audit_logger: object      # 稽核軌跡紀錄器
 
    def process_request(self, user_id: str, message: str, session_id: str) -> dict:
        """透過所有安全層處理請求。"""
        request_id = self._generate_request_id()
 
        # 第一層:速率限制
        if not self.rate_limiter.allow(user_id):
            self.audit_logger.log(request_id, "rate_limited", user_id)
            return {"error": "Rate limit exceeded", "retry_after": 60}
 
        # 第二層:輸入分類
        classification = self.input_classifier.classify(message)
        if classification.is_adversarial:
            self.audit_logger.log(
                request_id, "input_blocked",
                user_id, classification.category
            )
            return {"error": "Request could not be processed"}
 
        # 第三層:LLM 處理
        response = self._call_llm(message, session_id)
 
        # 第四層:輸出過濾
        filtered = self.output_filter.filter(response)
        if filtered.was_modified:
            self.audit_logger.log(
                request_id, "output_filtered",
                user_id, filtered.reason
            )
 
        # 第五層:稽核日誌
        self.audit_logger.log(
            request_id, "completed",
            user_id, len(message), len(filtered.content)
        )
 
        return {"response": filtered.content}
 
    def _generate_request_id(self) -> str:
        import uuid
        return str(uuid.uuid4())
 
    def _call_llm(self, message: str, session_id: str) -> str:
        # LLM API 呼叫實作
        pass

class OrganizationalIntegration:
    """用於將 AI 安全整合進組織安全計畫的框架。"""
 
    def __init__(self, org_config: dict):
        self.config = org_config
        self.gaps = []
 
    def assess_maturity(self) -> dict:
        """評估組織的 AI 安全成熟度。"""
        domains = {
            "governance": self._check_governance(),
            "technical_controls": self._check_technical(),
            "monitoring": self._check_monitoring(),
            "incident_response": self._check_ir(),
            "training": self._check_training(),
        }
        overall = sum(d["score"] for d in domains.values()) / len(domains)
        return {"domains": domains, "overall_maturity": round(overall, 1)}
 
    def _check_governance(self) -> dict:
        has_policy = self.config.get("ai_security_policy", False)
        has_framework = self.config.get("risk_framework", False)
        score = (int(has_policy) + int(has_framework)) * 2.5
        return {"score": score, "max": 5.0}
 
    def _check_technical(self) -> dict:
        controls = ["input_classification", "output_filtering", "rate_limiting", "sandboxing"]
        active = sum(1 for c in controls if self.config.get(c, False))
        return {"score": active * 1.25, "max": 5.0}
 
    def _check_monitoring(self) -> dict:
        has_monitoring = self.config.get("ai_monitoring", False)
        has_alerting = self.config.get("ai_alerting", False)
        score = (int(has_monitoring) + int(has_alerting)) * 2.5
        return {"score": score, "max": 5.0}
 
    def _check_ir(self) -> dict:
        has_playbook = self.config.get("ai_ir_playbook", False)
        return {"score": 5.0 if has_playbook else 0.0, "max": 5.0}
 
    def _check_training(self) -> dict:
        has_training = self.config.get("ai_security_training", False)
        return {"score": 5.0 if has_training else 0.0, "max": 5.0}