attacker.py

import argparse
import threading
import time

import requests


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Simple HTTP traffic generator.")
    parser.add_argument("target", help="Target URL (e.g. http://localhost:5000)")
    parser.add_argument(
        "--workers",
        type=int,
        default=50,
        help="Number of concurrent attack threads to run (default: 50)",
    )
    parser.add_argument(
        "--timeout",
        type=float,
        default=2.0,
        help="Request timeout in seconds (default: 2.0)",
    )
    return parser.parse_args()


def attack(target: str, timeout: float, session: requests.Session) -> None:
    while True:
        try:
            session.get(target, timeout=timeout)
        except requests.RequestException:
            # This script is intentionally fire-and-forget for load simulation.
            continue


def main() -> None:
    args = parse_args()

    if args.workers < 1:
        raise ValueError("--workers must be greater than 0")

    print(f"Attack started against {args.target} with {args.workers} workers.")

    session = requests.Session()
    for worker_id in range(1, args.workers + 1):
        thread = threading.Thread(
            target=attack,
            kwargs={"target": args.target, "timeout": args.timeout, "session": session},
            daemon=True,
            name=f"attacker-{worker_id}",
        )
        thread.start()
        print(f"Worker {worker_id} online")

    while True:
        time.sleep(1)


if __name__ == "__main__":
    main()