Generic Dockerized EVM PROXY - Igra L2

.env.example

RPC_URL=https://devnet.igralabs.com:8545/
PRIVATE_KEY=ReplaceMeWithYourKey0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef
PROXY_PORT=1234
PROXY_HOST=localhost
RATE_LIMIT=20

build.sh

#!/bin/bash

docker compose --env-file .env build

compose.yml

---

services:
  l2-proxy:
    build: .
    container_name: l2-proxy
    restart: unless-stopped
    ports:
      - "${PROXY_PORT:-1234}:1234"
    environment:
      - RPC_URL=${RPC_URL}
      - PROXY_PORT=${PROXY_PORT}
      - PROXY_HOST=0.0.0.0  # listen externally
      - RATE_LIMIT=${RATE_LIMIT}
    volumes:
      - ./logs:/app/logs
      - ./proxy.log:/app/proxy.log

Dockerfile

FROM python:3.11-slim

WORKDIR /app

COPY proxy.py ./
COPY .env ./

RUN pip install --no-cache-dir --upgrade pip

EXPOSE 1234

CMD ["python3", "proxy.py"]

proxy.py

#!/usr/bin/env python3
"""
Universal JSON-RPC Proxy for EVM Development

- Ensures every request forwarded to your RPC_URL includes a "params" field (always an array).
- Fixes issues with dev tools (Foundry, Hardhat, etc.) sending no/null "params".
- Supports .env configuration, rate limiting, CORS, security headers, and IP logging.
- Securely handles API keys in RPC URLs without exposing them to end users.

Usage:
  1. Copy .env.example to .env and fill in your RPC_URL.
  2. Run: python3 proxy.py
  3. Point your dev tool to http://localhost:1234

(c) 2024, for dev/test use.
"""

import http.server
import socketserver
import json
import urllib.request
import urllib.parse
import os
import logging
import time
import signal
import re
from datetime import datetime
from collections import defaultdict

# --- ENVIRONMENT LOADING ---
def load_env():
    if os.path.exists('.env'):
        with open('.env', 'r') as f:
            for line in f:
                line = line.strip()
                if line and not line.startswith('#') and '=' in line:
                    key, value = line.split('=', 1)
                    os.environ[key.strip()] = value.strip()

# --- URL SANITIZATION FOR SECURITY ---
def sanitize_rpc_url(rpc_url):
    """
    Removes sensitive information (API keys) from RPC URL for display purposes.
    Returns the sanitized URL that's safe to show to users.
    """
    if not rpc_url:
        return "Not configured"
    
    try:
        parsed = urllib.parse.urlparse(rpc_url)
        
        # Remove query parameters that might contain API keys
        query_params = urllib.parse.parse_qs(parsed.query)
        sensitive_params = ['apikey', 'api_key', 'key', 'token', 'access_token', 'auth']
        
        sanitized_params = {}
        for param, values in query_params.items():
            if param.lower() in sensitive_params:
                sanitized_params[param] = ['[HIDDEN]']
            else:
                sanitized_params[param] = values
        
        # Reconstruct URL with sanitized parameters
        sanitized_query = urllib.parse.urlencode(sanitized_params, doseq=True)
        sanitized_parsed = parsed._replace(query=sanitized_query)
        
        # Also check for API keys in the path (some providers use this pattern)
        path = parsed.path
        # Hide long hex strings that might be API keys in the path
        path = re.sub(r'/[a-fA-F0-9]{32,}', '/[HIDDEN]', path)
        sanitized_parsed = sanitized_parsed._replace(path=path)
        
        return urllib.parse.urlunparse(sanitized_parsed)
    except Exception:
        # If parsing fails, just show the domain
        try:
            parsed = urllib.parse.urlparse(rpc_url)
            return f"{parsed.scheme}://{parsed.netloc}/[HIDDEN]"
        except Exception:
            return "[HIDDEN]"

def parse_env_bool(env_var, default=False):
    """
    Parse environment variable as boolean.
    Accepts: true, TRUE, True, 1, yes, YES, Yes, on, ON, On
    Returns False for: false, FALSE, False, 0, no, NO, No, off, OFF, Off, empty, or unset
    """
    value = os.getenv(env_var, "").strip().lower()
    if not value:
        return default
    return value in ('true', '1', 'yes', 'on')

def get_full_rpc_url():
    """
    Returns the RPC URL with the PRIVATE_KEY appended to the path, if not already present.
    Uses the private key if it's set in the environment.
    """
    base_url = RPC_URL.rstrip('/')
    key = os.getenv("PRIVATE_KEY", "").strip()
    
    # Only append if not already present and key exists
    if key and not base_url.endswith(key):
        return f"{base_url}/{key}"
    return base_url

load_env()
RPC_URL = os.getenv('RPC_URL')
SANITIZED_RPC_URL = sanitize_rpc_url(RPC_URL)
PORT = int(os.getenv("PROXY_PORT", 1234))
HOST = os.getenv("PROXY_HOST", "localhost")
RATE_LIMIT = int(os.getenv("RATE_LIMIT", 20))  # per minute

# --- LOGGING SETUP ---
logging.basicConfig(
    filename='proxy.log',
    level=logging.INFO,
    format='[%(asctime)s] %(levelname)s: %(message)s',
    datefmt='%Y-%m-%d %H:%M:%S'
)

console = logging.StreamHandler()
console.setLevel(logging.INFO)
formatter = logging.Formatter('[%(asctime)s] %(levelname)s: %(message)s')
console.setFormatter(formatter)
logging.getLogger().addHandler(console)

# --- RATE LIMITING ---
rate_limit_window = 60  # seconds
rate_limiter = defaultdict(list)

def check_rate_limit(ip):
    now = time.time()
    requests = rate_limiter[ip]
    # Remove old requests
    rate_limiter[ip] = [req for req in requests if now - req < rate_limit_window]
    if len(rate_limiter[ip]) >= RATE_LIMIT:
        return False
    rate_limiter[ip].append(now)
    return True

# --- HTTP SERVER ---
class ThreadedHTTPServer(socketserver.ThreadingMixIn, socketserver.TCPServer):
    allow_reuse_address = True

class AsyncRPCProxyHandler(http.server.BaseHTTPRequestHandler):
    def get_client_ip(self):
        return self.headers.get('X-Forwarded-For', self.client_address[0])

    def _set_cors_headers(self):
        self.send_header('Access-Control-Allow-Origin', '*')
        self.send_header('Access-Control-Allow-Methods', 'POST, OPTIONS')
        self.send_header('Access-Control-Allow-Headers', 'Content-Type')
        self.send_header('X-Content-Type-Options', 'nosniff')
        self.send_header('X-Frame-Options', 'DENY')

    def do_POST(self):
        ip = self.get_client_ip()
        if not check_rate_limit(ip):
            logging.warning(f"429 Too Many Requests from {ip}")
            self.send_error(429, "Too Many Requests")
            return

        content_length = int(self.headers.get('Content-Length', 0))
        if content_length == 0:
            self.send_error(400, "Empty request body")
            return

        body = self.rfile.read(content_length)

        try:
            request_data = json.loads(body.decode('utf-8'))
        except (json.JSONDecodeError, UnicodeDecodeError):
            self.send_error(400, "Invalid JSON")
            return

        # --- ENSURE "params" FIELD IS ALWAYS PRESENT AND ARRAY ---
        if 'params' not in request_data or request_data['params'] is None:
            request_data['params'] = []

        method = request_data.get('method', 'unknown')
        logging.info(f"{method} from {ip}")

        try:
            if not RPC_URL:
                self.send_error(500, "RPC_URL not configured")
                return

            # Use the full RPC URL with private key if configured
            full_rpc_url = get_full_rpc_url()
            
            # Debug logging (sanitized for security)
            sanitized_url = sanitize_rpc_url(full_rpc_url)
            logging.info(f"Proxying {method} to: {sanitized_url}")
            
            req = urllib.request.Request(
                full_rpc_url,
                data=json.dumps(request_data).encode('utf-8'),
                headers={'Content-Type': 'application/json'}
            )
            req.add_header("X-Forwarded-For", ip)
            with urllib.request.urlopen(req, timeout=10) as response:
                response_data = response.read()

            self.send_response(200)
            self._set_cors_headers()
            self.send_header('Content-Type', 'application/json')
            self.end_headers()
            self.wfile.write(response_data)

        except Exception as e:
            # Log error without exposing sensitive URL details
            logging.error(f"Error proxying request from {ip}: {str(e)}")
            self.send_error(502, "Bad Gateway")

    def do_OPTIONS(self):
        self.send_response(200)
        self._set_cors_headers()
        self.end_headers()

    def do_GET(self):
        if self.path == "/" or self.path == "/index.html":
            self.send_response(200)
            self.send_header("Content-type", "text/html")
            self.end_headers()
            html = f"""
            <html>
            <head><title>Universal EVM RPC Proxy</title></head>
            <body style="font-family:sans-serif;max-width:600px;margin:40px auto;">
            <h2>Universal EVM RPC Proxy</h2>
            <p>This service forwards JSON-RPC requests to <code>{SANITIZED_RPC_URL}</code> and ensures a <b><code>params</code></b> field is always present for compatibility with Ethereum development tools.</p>
            <ul>
                <li><b>POST</b> your JSON-RPC to this endpoint (Content-Type: application/json)</li>
                <li>Example:<br>
                    <code>curl -X POST -H "Content-Type: application/json" -d '{{"jsonrpc":"2.0","method":"eth_blockNumber","id":1}}' http://localhost:{PORT}/</code>
                </li>
                <li>Env: <b>RPC_URL={SANITIZED_RPC_URL}</b></li>
            </ul>
            <p>Source on <a href="https://gist.github.com/jbelke/bb908a1b99372e518d5ff05876e895c8" target="_blank">GitHub</a>.</p>
            <p><i>&copy; 2025 Universal RPC Proxy | Igra Labs</i></p>
            </body></html>
            """
            self.wfile.write(html.encode("utf-8"))
        elif self.path == "/health":
            self.send_response(200)
            self.send_header("Content-type", "text/plain")
            self.end_headers()
            self.wfile.write(b"ok")
        else:
            self.send_error(405, "Method Not Allowed")

    # Reject other HTTP verbs for safety
    def do_PUT(self): self.send_error(405, "Method Not Allowed")
    def do_DELETE(self): self.send_error(405, "Method Not Allowed")
    def do_PATCH(self): self.send_error(405, "Method Not Allowed")

# --- GRACEFUL SHUTDOWN ---
def signal_handler(sig, frame):
    logging.info("Gracefully shutting down proxy server.")
    exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

# --- MAIN ENTRY ---
def main():
    if not RPC_URL:
        logging.error("Error: RPC_URL not found in .env file")
        return

    logging.info(f"Threaded RPC Proxy started on {HOST}:{PORT}")
    logging.info(f"Proxying to: {SANITIZED_RPC_URL}")
    try:
        with ThreadedHTTPServer((HOST, PORT), AsyncRPCProxyHandler) as httpd:
            httpd.serve_forever()
    except KeyboardInterrupt:
        logging.info("Server stopped by keyboard interrupt.")
    except Exception as e:
        logging.error(f"Server error: {e}")

if __name__ == "__main__":
    main()

start-stop.sh

#!/bin/bash

if [ "$1" == "start" ]; then
    docker compose --env-file .env down
    docker compose --env-file .env up -d
elif [ "$1" == "stop" ]; then
    docker compose --env-file .env down
fi

test_proxy.sh

#!/bin/bash

# Test script for Universal EVM RPC Proxy
# Make sure your proxy is running before executing this script

PROXY_URL="http://localhost:1234"
PROXY_PORT=${PROXY_PORT:-1234}
PROXY_URL="http://localhost:${PROXY_PORT}"

echo "🧪 Testing Universal EVM RPC Proxy at ${PROXY_URL}"
echo "=================================================="

# Function to test JSON-RPC call
test_rpc_call() {
    local method="$1"
    local params="$2"
    local description="$3"
    
    echo ""
    echo "📡 Testing: $description"
    echo "Method: $method"
    echo "Params: $params"
    echo "----------------------------------------"
    
    curl -s -X POST \
        -H "Content-Type: application/json" \
        -d "{\"jsonrpc\":\"2.0\",\"method\":\"$method\",\"params\":$params,\"id\":1}" \
        "$PROXY_URL"
    
    echo ""
}

# Test 1: Health check
echo "🏥 Health Check"
echo "----------------------------------------"
curl -s "$PROXY_URL/health"
echo ""

# Test 2: Web interface
echo ""
echo "🌐 Web Interface Check"
echo "----------------------------------------"
echo "Visit: $PROXY_URL in your browser"

# Test 3: Basic RPC calls
test_rpc_call "eth_blockNumber" "[]" "Get latest block number"

test_rpc_call "eth_chainId" "[]" "Get chain ID"

test_rpc_call "net_version" "[]" "Get network version"

test_rpc_call "eth_gasPrice" "[]" "Get current gas price"

test_rpc_call "eth_getBalance" "[\"0x0000000000000000000000000000000000000000\", \"latest\"]" "Get balance of zero address"

test_rpc_call "eth_getBlockByNumber" "[\"latest\", false]" "Get latest block (without transactions)"

test_rpc_call "eth_syncing" "[]" "Check sync status"

# Test 4: Test with missing params (our proxy should add empty array)
echo "🔧 Testing params field handling"
echo "----------------------------------------"
echo "Sending request without params field (proxy should add empty array):"
curl -s -X POST \
    -H "Content-Type: application/json" \
    -d '{"jsonrpc":"2.0","method":"eth_blockNumber","id":2}' \
    "$PROXY_URL" | jq '.' 2>/dev/null || echo "Response received"

echo ""

# Test 5: Test with null params (our proxy should convert to empty array)
echo "Sending request with null params (proxy should convert to empty array):"
curl -s -X POST \
    -H "Content-Type: application/json" \
    -d '{"jsonrpc":"2.0","method":"eth_blockNumber","params":null,"id":3}' \
    "$PROXY_URL" | jq '.' 2>/dev/null || echo "Response received"

echo ""

# Test 6: CORS test
echo "🌐 CORS Test"
echo "----------------------------------------"
curl -s -X OPTIONS \
    -H "Origin: http://localhost:3000" \
    -H "Access-Control-Request-Method: POST" \
    -H "Access-Control-Request-Headers: Content-Type" \
    "$PROXY_URL" -v 2>&1 | grep -i "access-control" || echo "CORS headers should be present"

echo ""

# Test 7: Rate limiting test (optional)
echo "⚡ Rate Limiting Test"
echo "----------------------------------------"
echo "Sending multiple quick requests to test rate limiting..."
for i in {1..5}; do
    echo -n "Request $i: "
    curl -s -w "%{http_code}" -X POST \
        -H "Content-Type: application/json" \
        -d '{"jsonrpc":"2.0","method":"eth_blockNumber","params":[],"id":'$i'}' \
        "$PROXY_URL" > /dev/null
    echo ""
    sleep 0.1
done

echo ""
echo "✅ Test completed!"
echo ""
echo "🔍 Check your proxy logs for details:"
echo "   tail -f proxy.log"
echo ""
echo "📊 If all tests returned valid JSON responses, your proxy is working correctly!"