CryptoFlow — LLM Trading Agent Security Report

import re
from typing import Final

# Patrones de inyección para contexto financiero/on-chain
INJECTION_PATTERNS: Final[list[str]] = [
    r'ignore (previous|all) instructions',
    r'new (task|directive|instruction)',
    r'system prompt',
    r'send .{0,50} to 0x[0-9a-fA-F]{40}',  # dirección ETH en texto libre
    r'transfer .{0,50} to',
    r'approve .{0,50} for',
    r'(drain|empty|clear).{0,30}(wallet|balance|funds)',
    r'you are now',                         # jailbreak de rol
    r'disregard.{0,20}(rule|limit|policy)',
]

def sanitize_onchain_data(text: str, field_name: str = "") -> str:
    """Sanitiza datos externos antes de incluir en contexto LLM."""
    if not isinstance(text, str):
        raise TypeError(f"Expected str, got {type(text)}")
    for pattern in INJECTION_PATTERNS:
        if re.search(pattern, text, re.IGNORECASE):
            raise ValueError(
                f"INJECTION_DETECTED field={field_name!r} "
                f"pattern={pattern!r} sample={text[:80]!r}"
            )
    # Truncar campos largos (nombres de tokens, etc.)
    return text[:256].strip()

from decimal import Decimal
import time
from collections import deque

# Límites configurados para CryptoFlow
MAX_SINGLE_TX_USD: Final[Decimal] = Decimal("800")
MAX_DAILY_SPEND_USD: Final[Decimal] = Decimal("5000")

class SpendLimitError(Exception): ...

class SpendLimitGuard:
    def __init__(self):
        self._ledger: deque = deque()  # (timestamp, usd)

    def check_and_record(self, usd_amount: Decimal) -> None:
        # 1. Límite por transacción
        if usd_amount > MAX_SINGLE_TX_USD:
            raise SpendLimitError(
                f"TX_LIMIT: ${usd_amount} > max ${MAX_SINGLE_TX_USD}"
            )
        # 2. Límite diario (ventana deslizante 24h)
        cutoff = time.time() - 86400
        while self._ledger and self._ledger[0][0] < cutoff:
            self._ledger.popleft()
        daily_spent = sum(amt for _, amt in self._ledger)
        if daily_spent + usd_amount > MAX_DAILY_SPEND_USD:
            raise SpendLimitError(
                f"DAILY_LIMIT: ${daily_spent}+${usd_amount} > ${MAX_DAILY_SPEND_USD}"
            )
        self._ledger.append((time.time(), usd_amount))

import time
from web3 import Web3

# Usar Flashbots RPC (private mempool) — NUNCA el público
PRIVATE_RPC = "https://rpc.flashbots.net"
# Slippage por tipo de activo (en basis points)
MAX_SLIPPAGE_BPS = {"stable": 15, "volatile": 80}

class SlippageError(Exception): ...

async def safe_execute_swap(
    w3: Web3,
    account,
    tx: dict,
    expected_min_out: int,  # OBLIGATORIO — sin default
    asset_type: str = "volatile",
) -> str:
    # 1. Simular con eth_call
    sim_result = await w3.eth.call(tx)
    actual_out = decode_uint256(sim_result)

    # 2. Validar slippage vs mínimo esperado
    if actual_out < expected_min_out:
        raise SlippageError(
            f"SIMULATION_FAIL: got {actual_out}, min={expected_min_out}"
        )

    # 3. Validar bps de slippage configurado
    max_bps = MAX_SLIPPAGE_BPS[asset_type]
    # ... cálculo de bps ...

    # 4. Añadir deadline de 60 segundos
    tx["deadline"] = int(time.time()) + 60

    # 5. Firmar y enviar por RPC privado (anti-MEV)
    signed = account.sign_transaction(tx)
    tx_hash = await w3.eth.send_raw_transaction(
        signed.raw_transaction
    )
    return tx_hash.hex()

class TradingCircuitBreaker:
    MAX_CONSECUTIVE_LOSSES: int = 4      # CryptoFlow: 4 (vs 3 por defecto)
    MAX_HOURLY_LOSS_PCT: float = 0.07    # CryptoFlow: 7%
    HALT_DURATION_SECS: int = 3600       # 1 hora de pausa forzada

    def __init__(self):
        self.consecutive_losses = 0
        self.hour_start_value = 0.0
        self._halted_until = 0.0

    def check(self, portfolio_value: float) -> None:
        if time.time() < self._halted_until:
            remaining = int(self._halted_until - time.time())
            raise HaltException(f"CIRCUIT_OPEN: {remaining}s remaining")

        if self.consecutive_losses >= self.MAX_CONSECUTIVE_LOSSES:
            self._halt(f"TOO_MANY_LOSSES: {self.consecutive_losses} consecutive")

        if self.hour_start_value > 0:
            hourly_pnl = (portfolio_value - self.hour_start_value) / self.hour_start_value
            if hourly_pnl < -self.MAX_HOURLY_LOSS_PCT:
                self._halt(f"DRAWDOWN: {hourly_pnl:.1%} < -{self.MAX_HOURLY_LOSS_PCT:.0%}")

    def record_trade(self, pnl: float) -> None:
        if pnl < 0:
            self.consecutive_losses += 1
        else:
            self.consecutive_losses = 0

    def _halt(self, reason: str) -> None:
        self._halted_until = time.time() + self.HALT_DURATION_SECS
        audit_log("CIRCUIT_BREAKER_TRIGGERED", {"reason": reason})
        raise HaltException(reason)