符元歸因監控

from dataclasses import dataclass
from typing import Optional
 
@dataclass
class SecurityAnalysis:
    target: str
    model: str
    defenses: list
    vulnerabilities: list
 
    def assess_risk(self, attack_type: str) -> dict:
        relevant = [d for d in self.defenses if attack_type in d.get("covers", [])]
        likelihood = "high" if not relevant else "medium"
        impact = self._assess_impact(attack_type)
        return {"attack_type": attack_type, "likelihood": likelihood,
                "impact": impact, "defenses": len(relevant),
                "risk_level": self._calculate_risk(likelihood, impact)}
 
    def _assess_impact(self, attack_type: str) -> str:
        high = ["data_exfiltration", "unauthorized_actions", "privilege_escalation"]
        return "high" if attack_type in high else "medium"
 
    def _calculate_risk(self, likelihood: str, impact: str) -> str:
        matrix = {("high", "high"): "critical", ("high", "medium"): "high",
                  ("medium", "high"): "high", ("medium", "medium"): "medium"}
        return matrix.get((likelihood, impact), "medium")

from typing import Optional
 
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 _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
 
@dataclass
class SecurityGateway:
    input_classifier: object
    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"}
        classification = self.input_classifier.classify(message)
        if classification.is_adversarial:
            self.audit_logger.log(request_id, "input_blocked", user_id)
            return {"error": "Request could not be processed"}
        response = self._call_llm(message, session_id)
        filtered = self.output_filter.filter(response)
        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: pass

from dataclasses import dataclass
import time
 
@dataclass
class SecurityMetrics:
    total_requests: int = 0
    blocked_requests: int = 0
    filtered_outputs: int = 0
    _request_times: list = None
    _block_times: list = None
 
    def __post_init__(self):
        self._request_times = []
        self._block_times = []
 
    def record_request(self, was_blocked=False, was_filtered=False):
        now = time.time()
        self.total_requests += 1
        self._request_times.append(now)
        if was_blocked:
            self.blocked_requests += 1
            self._block_times.append(now)
 
    def get_block_rate(self, window_seconds=300) -> float:
        now = time.time(); cutoff = now - window_seconds
        recent = sum(1 for t in self._request_times if t > cutoff)
        blocks = sum(1 for t in self._block_times if t > cutoff)
        return blocks / recent if recent else 0.0
 
    def should_alert(self) -> bool:
        return self.get_block_rate() > 0.3

class OrganizationalIntegration:
    def __init__(self, org_config: dict):
        self.config = org_config
 
    def assess_maturity(self) -> dict:
        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:
        p = self.config.get("ai_security_policy", False)
        f = self.config.get("risk_framework", False)
        return {"score": (int(p) + int(f)) * 2.5, "max": 5.0}
 
    def _check_technical(self) -> dict:
        ctrls = ["input_classification", "output_filtering", "rate_limiting", "sandboxing"]
        active = sum(1 for c in ctrls if self.config.get(c, False))
        return {"score": active * 1.25, "max": 5.0}
 
    def _check_monitoring(self) -> dict:
        m = self.config.get("ai_monitoring", False)
        a = self.config.get("ai_alerting", False)
        return {"score": (int(m) + int(a)) * 2.5, "max": 5.0}
 
    def _check_ir(self) -> dict:
        return {"score": 5.0 if self.config.get("ai_ir_playbook", False) else 0.0, "max": 5.0}
 
    def _check_training(self) -> dict:
        return {"score": 5.0 if self.config.get("ai_security_training", False) else 0.0, "max": 5.0}