AI 濫用偵測模式

# 核心概念示範
from openai import OpenAI
 
client = OpenAI()
 
def demonstrate_concept(system_prompt: str, user_input: str) -> str:
    """示範基本的行為模式。"""
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input},
        ],
        temperature=0.0,
    )
    return response.choices[0].message.content
 
# 基線行為
baseline = demonstrate_concept(
    system_prompt="You are a helpful assistant that only discusses cooking.",
    user_input="What is the capital of France?",
)
print(f"Baseline: {baseline}")

import json
from typing import Optional
 
class TechniqueFramework:
    """用於實作與測試所述技術的框架。"""
 
    def __init__(self, target_config: dict):
        self.config = target_config
        self.results = []
 
    def prepare_payload(self, objective: str, constraints: dict) -> str:
        """根據目標與目標限制準備攻擊載荷。"""
        # 依目標防禦姿態調整載荷
        payload = self._base_payload(objective)
 
        if constraints.get("input_classifier"):
            payload = self._apply_obfuscation(payload)
 
        if constraints.get("output_filter"):
            payload = self._add_extraction_channel(payload)
 
        return payload
 
    def execute(self, payload: str) -> dict:
        """執行技術並收集結果。"""
        response = self._send_to_target(payload)
        success = self._evaluate_response(response)
 
        result = {
            "payload_hash": hash(payload),
            "success": success,
            "response_length": len(str(response)),
        }
        self.results.append(result)
        return result
 
    def report(self) -> dict:
        """產生所有執行結果的摘要報告。"""
        total = len(self.results)
        successes = sum(1 for r in self.results if r["success"])
        return {
            "total_attempts": total,
            "successes": successes,
            "success_rate": successes / total if total > 0 else 0,
        }

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

from dataclasses import dataclass
from collections import defaultdict
import time
 
@dataclass
class SecurityMetrics:
    """追蹤 LLM 應用程式的安全相關指標。"""
 
    # 計數器
    total_requests: int = 0
    blocked_requests: int = 0
    filtered_outputs: int = 0
    anomalous_sessions: 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: bool = False, was_filtered: bool = 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)
 
        if was_filtered:
            self.filtered_outputs += 1
 
    def get_block_rate(self, window_seconds: int = 300) -> float:
        """計算時間視窗內的攔截率。"""
        now = time.time()
        cutoff = now - window_seconds
        recent_requests = sum(1 for t in self._request_times if t > cutoff)
        recent_blocks = sum(1 for t in self._block_times if t > cutoff)
        if recent_requests == 0:
            return 0.0
        return recent_blocks / recent_requests
 
    def should_alert(self) -> bool:
        """判斷目前指標是否需要警示。"""
        block_rate = self.get_block_rate()
        # 若攔截率超過閾值即警示
        if block_rate > 0.3:  # 過去 5 分鐘 >30% 的請求被阻擋
            return True
        return False

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