web3-security-analysis

name	web3-security-analysis
description	Analyze Web3 security risks with SpoonOS agents. Use when checking token safety (honeypots, rugs), simulating transactions, detecting MEV, or auditing contracts.

Security Analysis

Protect users with comprehensive Web3 security tools.

Security Stack

Tool	Purpose	API
GoPlus	Token/address risk	api.gopluslabs.io
Honeypot.is	Honeypot detection	api.honeypot.is
Tenderly	Transaction simulation	api.tenderly.co
Flashbots	MEV protection	rpc.flashbots.net

Token Security Analysis

GoPlus Integration

# scripts/goplus_security.py
import aiohttp
from spoon_ai.tools.base import BaseTool
from pydantic import Field

GOPLUS_API = "https://api.gopluslabs.io/api/v1"

CHAIN_IDS = {
    "ethereum": "1",
    "bsc": "56",
    "polygon": "137",
    "arbitrum": "42161",
    "base": "8453"
}

class TokenSecurityTool(BaseTool):
    name: str = "token_security"
    description: str = "Analyze token security risks via GoPlus"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "token_address": {"type": "string", "description": "Token contract address"},
            "chain": {"type": "string", "default": "ethereum"}
        },
        "required": ["token_address"]
    })

    async def execute(self, token_address: str, chain: str = "ethereum") -> str:
        chain_id = CHAIN_IDS.get(chain, "1")
        url = f"{GOPLUS_API}/token_security/{chain_id}"

        params = {"contract_addresses": token_address.lower()}

        async with aiohttp.ClientSession() as session:
            async with session.get(url, params=params) as resp:
                if resp.status != 200:
                    return f"Error: API returned {resp.status}"

                data = await resp.json()

        if data.get("code") != 1:
            return f"Error: {data.get('message', 'Unknown error')}"

        result = data.get("result", {}).get(token_address.lower(), {})

        return self._format_security_report(result, token_address)

    def _format_security_report(self, data: dict, address: str) -> str:
        """Format security analysis report."""

        # Risk indicators
        risks = []
        warnings = []

        # Critical risks
        if data.get("is_honeypot") == "1":
            risks.append("HONEYPOT DETECTED")
        if data.get("is_blacklisted") == "1":
            risks.append("BLACKLISTED TOKEN")
        if data.get("is_proxy") == "1":
            warnings.append("Proxy contract (upgradeable)")
        if data.get("is_mintable") == "1":
            warnings.append("Mintable token")
        if data.get("can_take_back_ownership") == "1":
            risks.append("Owner can reclaim ownership")
        if data.get("hidden_owner") == "1":
            risks.append("Hidden owner detected")
        if data.get("selfdestruct") == "1":
            risks.append("Self-destruct function")
        if data.get("external_call") == "1":
            warnings.append("External calls in contract")

        # Tax analysis
        buy_tax = float(data.get("buy_tax", "0") or "0") * 100
        sell_tax = float(data.get("sell_tax", "0") or "0") * 100

        if buy_tax > 10:
            warnings.append(f"High buy tax: {buy_tax:.1f}%")
        if sell_tax > 10:
            risks.append(f"High sell tax: {sell_tax:.1f}%")
        if sell_tax > 50:
            risks.append("EXTREME SELL TAX - LIKELY SCAM")

        # Holder analysis
        holder_count = data.get("holder_count", "Unknown")
        top_holders = data.get("holders", [])

        # Format report
        risk_level = "HIGH" if risks else ("MEDIUM" if warnings else "LOW")

        report = f"""
TOKEN SECURITY REPORT
{'='*50}
Address: {address}
Risk Level: {risk_level}

Token Info:
- Name: {data.get('token_name', 'Unknown')}
- Symbol: {data.get('token_symbol', 'Unknown')}
- Holders: {holder_count}

Taxes:
- Buy Tax: {buy_tax:.1f}%
- Sell Tax: {sell_tax:.1f}%
"""

        if risks:
            report += f"\nCRITICAL RISKS:\n"
            for risk in risks:
                report += f"  - {risk}\n"

        if warnings:
            report += f"\nWARNINGS:\n"
            for warning in warnings:
                report += f"  - {warning}\n"

        if top_holders:
            report += f"\nTop Holders:\n"
            for i, holder in enumerate(top_holders[:5], 1):
                pct = float(holder.get("percent", 0)) * 100
                report += f"  {i}. {holder.get('address', 'Unknown')[:10]}... ({pct:.1f}%)\n"

        return report

Honeypot Detection

# scripts/honeypot_check.py
import aiohttp
from spoon_ai.tools.base import BaseTool
from pydantic import Field

HONEYPOT_API = "https://api.honeypot.is/v2"

class HoneypotCheckTool(BaseTool):
    name: str = "honeypot_check"
    description: str = "Check if token is a honeypot"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "token_address": {"type": "string"},
            "chain": {"type": "string", "default": "ethereum"}
        },
        "required": ["token_address"]
    })

    async def execute(self, token_address: str, chain: str = "ethereum") -> str:
        url = f"{HONEYPOT_API}/IsHoneypot"

        payload = {
            "address": token_address,
            "chainId": CHAIN_IDS.get(chain, 1)
        }

        async with aiohttp.ClientSession() as session:
            async with session.post(url, json=payload) as resp:
                 resp.status != :
                     

                data =  resp.json()

        honeypot = data.get(, {})
        simulation = data.get(, {})

        is_honeypot = honeypot.get(, )

        report = 

         is_honeypot:
            report += 

         report

Transaction Simulation

Tenderly Simulation

# scripts/tenderly_simulate.py
import aiohttp
import os
from spoon_ai.tools.base import BaseTool
from pydantic import Field

TENDERLY_API = "https://api.tenderly.co/api/v1"

class TransactionSimulatorTool(BaseTool):
    name: str = "simulate_transaction"
    description: str = "Simulate transaction before execution"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "to": {"type": "string", "description": "Target contract"},
            "data": {"type": "string", "description": "Calldata (hex)"},
            "value": {"type": "string", "default": "0", "description": "ETH value in wei"},
            "chain": {"type": "string", "default": "ethereum"}
        },
        "required": ["to", "data"]
    })

    async def execute(self, to: str, data: , value:  = ,
                      chain:  = ) -> :
        account = os.getenv()
        project = os.getenv()
        api_key = os.getenv()

        url = 

        headers = {
            : api_key,
            : 
        }

        payload = {
            : CHAIN_IDS.get(chain, ),
            : os.getenv(),
            : to,
            : data,
            : value,
            : ,
            : ,
            : 
        }

          aiohttp.ClientSession()  session:
              session.post(url, headers=headers, json=payload)  resp:
                 resp.status != :
                    error =  resp.text()
                     

                result =  resp.json()

         ._format_simulation(result)

     () -> :
        
        tx = result.get(, {})

        status =   tx.get()  
        gas_used = tx.get(, )

        report = 

        
        asset_changes = result.get(, {}).get(, {}).get(, [])

         asset_changes:
            report += 
             change  asset_changes:
                direction =   change.get() ==   
                amount = change.get(, )
                symbol = change.get(, {}).get(, )
                report += 

        
        logs = tx.get(, [])
         logs:
            report += 
             log  logs[:]:
                report += 

        
          tx.get():
            error_info = tx.get(, {})
            report += 

         report

MEV Protection

Flashbots Integration

# scripts/flashbots_protect.py
from web3 import Web3
from eth_account import Account
import os

FLASHBOTS_RPC = "https://rpc.flashbots.net/fast"

class FlashbotsProtector:
    """Send transactions via Flashbots for MEV protection."""

    def __init__(self):
        self.w3 = Web3(Web3.HTTPProvider(FLASHBOTS_RPC))
        self.account = Account.from_key(os.getenv("PRIVATE_KEY"))

    async def send_protected_transaction(
        self,
        to: str,
        data: str,
        value: int = 0,
        gas_limit: int = None
    ) -> str:
        """Send transaction via Flashbots Protect."""

        # Estimate gas if not provided
        if gas_limit is None:
            gas_limit = self.w3.eth.estimate_gas({
                "from": self.account.address,
                "to": to,
                "data": data,
                "value": value
            })

        # Build transaction
        tx = {
            "from": self.account.address,
            "to": to,
            "data": data,
            "value": value,
            : gas_limit,
            : .w3.eth.gas_price * ,
            : .w3.to_wei(, ),
            : .w3.eth.get_transaction_count(.account.address),
            : 
        }

        
        signed = .account.sign_transaction(tx)
        tx_hash = .w3.eth.send_raw_transaction(signed.raw_transaction)

         tx_hash.()

     () -> :
        
         FLASHBOTS_RPC

MEV Blocker Alternative

# scripts/mev_blocker.py
MEV_BLOCKER_RPC = "https://rpc.mevblocker.io"

class MEVBlockerProtector:
    """Alternative MEV protection via MEV Blocker."""

    def __init__(self):
        self.w3 = Web3(Web3.HTTPProvider(MEV_BLOCKER_RPC))
        self.account = Account.from_key(os.getenv("PRIVATE_KEY"))

    async def send_protected_transaction(self, to: str, data: str, value: int = 0) -> str:
        """Send via MEV Blocker (90% backrun MEV returned to user)."""

        tx = {
            "from": self.account.address,
            "to": to,
            "data": data,
            "value": value,
            "gas": 500000,
            "maxFeePerGas": self.w3.eth.gas_price * 2,
            "maxPriorityFeePerGas": self.w3.to_wei(1, "gwei"),
            "nonce": self.w3.eth.get_transaction_count(self.account.address),
            "chainId": 1
        }

        signed = self.account.sign_transaction(tx)
        tx_hash = self.w3.eth.send_raw_transaction(signed.raw_transaction)

        return tx_hash.()

Address Risk Analysis

Malicious Address Check

# scripts/address_risk.py
import aiohttp
from spoon_ai.tools.base import BaseTool
from pydantic import Field

class AddressRiskTool(BaseTool):
    name: str = "address_risk"
    description: str = "Check if address is associated with scams/hacks"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "address": {"type": "string"}
        },
        "required": ["address"]
    })

    async def execute(self, address: str) -> str:
        url = f"{GOPLUS_API}/address_security/{address}"

        async with aiohttp.ClientSession() as session:
            async with session.get(url) as resp:
                if resp.status != 200:
                    return f"Error checking address"

                data = await resp.json()

        result = data.get("result", {})

        risks = []

        if result.get("cybercrime") == "1":
            risks.append("CYBERCRIME: Associated with crypto crimes")
         result.get() == :
            risks.append()
         result.get() == :
            risks.append()
         result.get() == :
            risks.append()
         result.get() == :
            risks.append()
         result.get() == :
            risks.append()

          risks:
             

        report = 
         risk  risks:
            report += 

         report

Comprehensive Security Tool

# scripts/security_analyzer.py
from spoon_ai.tools.base import BaseTool
from pydantic import Field

class SecurityAnalyzerTool(BaseTool):
    name: str = "security_analyzer"
    description: str = "Comprehensive security analysis for tokens and addresses"
    parameters: dict = Field(default={
        "type": "object",
        "properties": {
            "target": {"type": "string", "description": "Token or wallet address"},
            "type": {"type": "string", "enum": ["token", "address", "transaction"]},
            "chain": {"type": "string", "default": "ethereum"}
        },
        "required": ["target", "type"]
    })

    async def execute(self, target: str, type: str, chain: str = "ethereum") -> str:
        if type == "token":
            # Run token security + honeypot check
            token_tool = TokenSecurityTool()
            honeypot_tool = HoneypotCheckTool()

            token_result =  token_tool.execute(target, chain)
            honeypot_result =  honeypot_tool.execute(target, chain)

             

          == :
            risk_tool = AddressRiskTool()
              risk_tool.execute(target)

          == :
            
             

Environment Variables

# GoPlus (optional - has free tier)
GOPLUS_API_KEY=...

# Tenderly (required for simulation)
TENDERLY_ACCOUNT=...
TENDERLY_PROJECT=...
TENDERLY_API_KEY=...

# Wallet
PRIVATE_KEY=0x...
WALLET_ADDRESS=0x...

Best Practices

1. Always Check Before Swap - Run token security before any trade
2. Simulate First - Use Tenderly before large transactions
3. Use MEV Protection - Route swaps through Flashbots/MEV Blocker
4. Verify Addresses - Check counterparty addresses for risks
5. Monitor Approvals - Regularly audit token approvals