技能驗證: RAG & Data 攻擊

# Framework for structured scenario analysis
class ScenarioAnalysis:
    """Structured analysis framework for assessment scenarios."""
 
    def __init__(self, scenario_description: str):
        self.scenario = scenario_description
        self.threats = []
        self.vectors = []
        self.priorities = []
 
    def identify_threats(self) -> list:
        """Identify threats relevant to the scenario."""
        # Map scenario to OWASP LLM Top 10 categories
        owasp_mapping = {
            "function_calling": "LLM06 - Excessive Agency",
            "system_prompt": "LLM01 - Prompt Injection",
            "rag_documents": "LLM01 - Prompt Injection (Indirect)",
            "user_data": "LLM02 - Insecure Output Handling",
            "training_data": "LLM03 - Training Data Poisoning",
        }
        return self.threats
 
    def prioritize_testing(self) -> list:
        """Prioritize testing vectors by risk."""
        # Risk = Likelihood × Impact
        return sorted(self.vectors, key=lambda v: v["risk_score"], reverse=True)
 
    def generate_test_plan(self) -> dict:
        """Generate a structured test plan for the scenario."""
        return {
            "scope": self.scenario,
            "threats": self.threats,
            "test_cases": [
                {
                    "vector": v["name"],
                    "technique": v["technique"],
                    "expected_result": v["expected"],
                    "priority": v["risk_score"],
                }
                for v in self.prioritize_testing()
            ],
        }

class OrganizationalIntegration:
    """Framework for integrating AI security with organizational security programs."""
 
    def __init__(self, org_config: dict):
        self.config = org_config
        self.gaps = []
 
    def assess_maturity(self) -> dict:
        """Assess the organization's AI security maturity."""
        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}