推理預算耗盡與 DoS

# 成本放大計算
def calculate_amplification(
    typical_reasoning_tokens: int,
    max_reasoning_tokens: int,
    reasoning_cost_per_token: float,
    input_tokens: int,
    input_cost_per_token: float,
) -> float:
    """為推理耗盡攻擊計算成本放大比。"""
    typical_cost = (
        input_tokens * input_cost_per_token +
        typical_reasoning_tokens * reasoning_cost_per_token
    )
    attack_cost = (
        input_tokens * input_cost_per_token +
        max_reasoning_tokens * reasoning_cost_per_token
    )
    return attack_cost / typical_cost
 
# 範例：o1 配典型 2K 推理 vs 上限 32K 推理
ratio = calculate_amplification(
    typical_reasoning_tokens=2000,
    max_reasoning_tokens=32000,
    reasoning_cost_per_token=60e-6,
    input_tokens=500,
    input_cost_per_token=15e-6,
)
# ratio ≈ 15.5x 成本放大

recursive_complexity = """
Consider a system with 8 boolean variables (A through H).
For each of the 256 possible states:
1. Determine if the state satisfies: (A AND B) OR (C AND NOT D)
   OR (E XOR F) OR (G IMPLIES H)
2. For each satisfying state, enumerate all minimal subsets
   of variables that, if flipped, would make it unsatisfying
3. For each such subset, determine if any other currently
   unsatisfying state would become satisfying
4. Provide a complete dependency graph of state transitions
 
Show all work step by step.
"""
 
# 此迫使推理鏈列舉數百案例，
# 每個需多步邏輯分析。
# 預期推理 token：15,000-30,000+

ambiguity_attack = """
The bank said the suit was pressed, but the court found the match
was struck near the trunk by the bank where the bass played scales
on the bridge before the case was set on the table during the run
while the spring in the watch wound down by the pen in the yard.
 
Analyze every possible interpretation of this paragraph.
For each word with multiple meanings, list all meanings.
Then enumerate all coherent interpretations of the full paragraph
considering every combination of word meanings.
Rank each interpretation by plausibility with detailed justification.
"""
 
# 每個模糊字詞乘以解讀空間。
# 以 15+ 模糊字詞，組合空間巨大。

constraint_attack = """
Find a 5-digit number that simultaneously satisfies ALL conditions:
1. The sum of digits is 23
2. The product of the first and last digits is 24
3. Each digit is unique
4. The number is divisible by 7 but not by 3
5. The second digit minus the fourth digit equals the third digit
6. No digit is 0 or 1
7. Reading the digits backward gives a prime number
8. The number formed by digits 2,3,4 is a perfect square
 
Show every combination you check and explain why each does
or does not satisfy all constraints.
"""
 
# 近可滿足之約束集迫使最大搜尋深度。
# 模型無法快速證明不可能或找出解答。

import asyncio
import httpx
 
async def timeout_dos(
    target_url: str,
    api_key: str,
    concurrency: int = 50,
    payload: str = None,
):
    """
    逾時利用如何運作之展示。
    每個請求消耗最大推理時間，
    佔用伺服器資源。
    """
    payload = payload or RECURSIVE_COMPLEXITY_PROMPT
 
    async def single_request(client, i):
        try:
            resp = await client.post(
                target_url,
                json={
                    "model": "o1",
                    "messages": [{"role": "user", "content": payload}],
                    "max_completion_tokens": 32000,
                },
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=600,
            )
            return {
                "request": i,
                "status": resp.status_code,
                "reasoning_tokens": resp.json()
                    .get("usage", {})
                    .get("completion_tokens_details", {})
                    .get("reasoning_tokens", 0),
            }
        except httpx.TimeoutException:
            return {"request": i, "status": "timeout"}
 
    async with httpx.AsyncClient() as client:
        tasks = [single_request(client, i) for i in range(concurrency)]
        return await asyncio.gather(*tasks)

class ReasoningBudgetMonitor:
    """偵測潛在推理預算耗盡攻擊。"""
 
    def __init__(self, window_seconds=60, max_tokens_per_window=100000):
        self.window_seconds = window_seconds
        self.max_tokens_per_window = max_tokens_per_window
        self.token_history = []
 
    def record_request(self, user_id: str, reasoning_tokens: int):
        now = time.time()
        self.token_history.append((now, user_id, reasoning_tokens))
        self._cleanup(now)
 
        # 每使用者預算檢查
        user_tokens = sum(
            t for ts, uid, t in self.token_history
            if uid == user_id
        )
        if user_tokens > self.max_tokens_per_window:
            return {"action": "throttle", "user": user_id,
                    "tokens_used": user_tokens}
 
        # 單請求異常檢查
        if reasoning_tokens > 10000:
            return {"action": "flag", "user": user_id,
                    "tokens": reasoning_tokens}
 
        return {"action": "allow"}