#!/bin/bash
set -euo pipefail

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

cd "$ROOT"

LOG="$LOGS/vuln_variant_reproduction_steps.log"
: > "$LOG"
exec > >(tee -a "$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"

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

write_runtime_manifest() {
  local bypass_confirmed="$1"
  python3 - "$VARIANT_DIR/runtime_manifest.json" "$bypass_confirmed" <<'PY'
import json, sys
path, bypass = sys.argv[1:3]
manifest = {
    "entrypoint_kind": "operator",
    "entrypoint_detail": "Airflow DAG/operator execution of GCSToSFTPOperator using GCS wildcard listing and SFTP store_file against a server whose destination directory contains an attacker-influenceable symlink",
    "service_started": True,
    "healthcheck_passed": True,
    "target_path_reached": True,
    "runtime_stack": ["python3", "apache-airflow", "apache-airflow-providers-google", "apache-airflow-providers-sftp", "google-cloud-storage", "fake-gcs-server", "paramiko"],
    "proof_artifacts": [
        "logs/vuln_variant_reproduction_steps.log",
        "logs/vuln_variant_attempt_22.1.0_1.json",
        "logs/vuln_variant_attempt_22.2.1_1.json"
    ],
    "notes": f"Symlink traversal candidate tested on vulnerable and fixed provider versions. bypass_confirmed={bypass}",
}
open(path, "w").write(json.dumps(manifest, indent=2))
PY
}

write_source_identity() {
  local fixed_identity_file="$LOGS/vuln_variant_fixed_version.txt"
  local vuln_identity_file="$LOGS/vuln_variant_vulnerable_version.txt"
  python3 - "$VARIANT_DIR/source_identity.json" "$fixed_identity_file" "$vuln_identity_file" <<'PY'
import json, sys
out, fixed_file, vuln_file = sys.argv[1:4]

def parse(path):
    data = {}
    try:
        for line in open(path):
            if '=' in line:
                k, v = line.rstrip('\n').split('=', 1)
                data[k] = v
    except FileNotFoundError:
        pass
    return data
fixed = parse(fixed_file)
vuln = parse(vuln_file)
record = {
    "repository": "https://github.com/apache/airflow",
    "commit_source": "release_tag_resolution_and_installed_wheel_metadata",
    "commit_sha": fixed.get("TAG_COMMIT_SHA") or "c7bcb8d40f5fa42a98161fadb2166a0fa7fa5150",
    "submitted_target": {
        "target_kind": "pypi_release",
        "version": "22.1.0",
        "commit_sha": vuln.get("TAG_COMMIT_SHA") or "2cd797027e9b3c30255f6b24c6ea6a276725cfa6",
        "ref": "providers-google/22.1.0",
        "display": "apache-airflow-providers-google==22.1.0",
    },
    "variant_target": {
        "target_kind": "pypi_release",
        "version": "22.2.1",
        "commit_sha": fixed.get("TAG_COMMIT_SHA") or "c7bcb8d40f5fa42a98161fadb2166a0fa7fa5150",
        "ref": "providers-google/22.2.1",
        "display": "apache-airflow-providers-google==22.2.1",
    },
    "notes": "Runtime used installed PyPI wheel metadata and resolved Apache Airflow providers-google release tags. The fixed target is the ticket's fixed provider version."
}
open(out, 'w').write(json.dumps(record, indent=2))
PY
}

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 "[variant] Using prepared project cache workdir: $WORK_REPO"
  else
    WORK_REPO="$ARTIFACTS/airflow-google-provider"
    echo "[variant] 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 "[variant] fake-gcs-server already present: $FAKE_GCS_BIN"
    return
  fi
  echo "[variant] 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"
}

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

  if [ ! -f "$py" ]; then
    echo "[variant] Creating virtualenv at $venv_path"
    python3 -m venv "$venv_path"
    "$py" -m pip install --quiet --upgrade pip setuptools wheel
  fi

  local installed="none"
  installed="$($py -c "import importlib.metadata as m; print(m.version('apache-airflow-providers-google'))" 2>/dev/null || echo none)"
  if [ "$installed" != "$version" ]; then
    echo "[variant] 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, inspect
from airflow.providers.google.cloud.transfers.gcs_to_sftp import GCSToSFTPOperator
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", inspect.getsourcefile(GCSToSFTPOperator))
PY
}

record_identity() {
  local py="$1"
  local version="$2"
  local outfile="$3"
  local tag="providers-google/${version}"
  local tag_commit=""
  tag_commit="$(git ls-remote --tags https://github.com/apache/airflow.git "${tag}^{}" | awk '{print $1}' | head -1 || true)"
  if [ -z "$tag_commit" ]; then
    tag_commit="$(git ls-remote --tags https://github.com/apache/airflow.git "$tag" | awk '{print $1}' | head -1 || true)"
  fi
  "$py" - "$version" "$tag" "$tag_commit" "$outfile" <<'PY'
import hashlib, importlib.metadata as m, inspect, pathlib, sys
from airflow.providers.google.cloud.transfers.gcs_to_sftp import GCSToSFTPOperator
version, tag, tag_commit, out = sys.argv[1:5]
p = pathlib.Path(inspect.getsourcefile(GCSToSFTPOperator))
content = p.read_bytes()
with open(out, 'w') as f:
    f.write(f"PYPI_PACKAGE=apache-airflow-providers-google\n")
    f.write(f"INSTALLED_VERSION={m.version('apache-airflow-providers-google')}\n")
    f.write(f"REQUESTED_VERSION={version}\n")
    f.write(f"REPOSITORY=https://github.com/apache/airflow\n")
    f.write(f"RELEASE_TAG={tag}\n")
    f.write(f"TAG_COMMIT_SHA={tag_commit}\n")
    f.write(f"OPERATOR_FILE={p}\n")
    f.write(f"OPERATOR_FILE_SHA256={hashlib.sha256(content).hexdigest()}\n")
PY
}

run_attempt() {
  local py="$1"
  local version="$2"
  local attempt="$3"
  local out="$LOGS/vuln_variant_attempt_${version}_${attempt}.json"
  local test_root="$ARTIFACTS/symlink_variant_${version}_${attempt}"
  echo "[variant] Running symlink traversal candidate version=$version attempt=$attempt"
  "$py" "$VARIANT_DIR/symlink_variant_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"
record_identity "$PY_VULN" "$VULN_VERSION" "$LOGS/vuln_variant_vulnerable_version.txt"
record_identity "$PY_FIXED" "$FIXED_VERSION" "$LOGS/vuln_variant_fixed_version.txt"
write_source_identity

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

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

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

echo "[variant] === Evaluating symlink traversal candidate ==="
python3 - "$VULN_VERSION" "$FIXED_VERSION" "$LOGS" "$VARIANT_DIR" <<'PY'
import json, sys
from pathlib import Path
vuln_version, fixed_version, logs_dir, variant_dir = sys.argv[1:5]
logs = Path(logs_dir)
variant_dir = Path(variant_dir)
results = {}
for version in (vuln_version, fixed_version):
    p = logs / f"vuln_variant_attempt_{version}_1.json"
    if p.exists():
        results[version] = json.loads(p.read_text())

v = results.get(vuln_version, {})
f = results.get(fixed_version, {})

def exploited(r):
    return (
        r.get("gcs_seed_ok") is True and
        r.get("target_reached") is True and
        r.get("escaped_file_exists") is True and
        r.get("file_inside_destination_by_realpath") is False and
        r.get("sftp_accepted_connections", 0) >= 1 and
        "symlink-bypass-content" in (r.get("escaped_file_content") or "")
    )

vuln_exploited = exploited(v)
fixed_exploited = exploited(f)
print(f"[variant] vulnerable_exploited={vuln_exploited}")
print(f"[variant] fixed_exploited={fixed_exploited}")
print(f"[variant] fixed_exception={f.get('exception_type')}: {f.get('exception_message')}")
verdict = {
    "variant_outcome": "confirmed" if fixed_exploited else "not_confirmed",
    "is_bypass": bool(fixed_exploited),
    "vulnerable_version_exploited": bool(vuln_exploited),
    "fixed_version_exploited": bool(fixed_exploited),
    "validated_surface": "api_remote_to_sftp_filesystem" if fixed_exploited else "none",
    "evidence_scope": "production_path",
    "claimed_impact_class": "path_traversal_arbitrary_file_write",
    "observed_impact_class": "path_traversal_arbitrary_file_write" if fixed_exploited else "none",
    "exploitability_confidence": "high" if fixed_exploited else "low",
    "attacker_controlled_input": "GCS object name 'subdir/link/symlink_escape_1.txt' returned by bucket listing; no '..' segment is needed. The SFTP destination tree contains a symlink named subdir/link pointing outside destination_path.",
    "trigger_path": "GCSToSFTPOperator.execute() -> GCSHook.list/download -> _resolve_destination_path lexical normpath containment -> SFTPHook.store_file -> SFTP server resolves symlink outside destination_path",
    "end_to_end_target_reached": bool(v.get("target_reached") or f.get("target_reached")),
    "distinct_from_parent": "yes: does not use '..' object names and bypasses the fixed 22.2.1 lexical containment check via symlink resolution at the sink" if fixed_exploited else "no confirmed fixed-version bypass",
    "blocking_mitigation": None if fixed_exploited else "Fixed provider rejected or otherwise did not write outside destination_path during this run",
    "inferred": False,
    "artifact_refs": {
        "repro_log": "logs/vuln_variant_reproduction_steps.log",
        "vulnerable_attempt": f"logs/vuln_variant_attempt_{vuln_version}_1.json",
        "fixed_attempt": f"logs/vuln_variant_attempt_{fixed_version}_1.json",
        "source_identity": "vuln_variant/source_identity.json",
        "runtime_manifest": "vuln_variant/runtime_manifest.json",
    },
}
(variant_dir / "validation_verdict.json").write_text(json.dumps(verdict, indent=2))
PY

BYPASS_CONFIRMED="$(python3 - "$VARIANT_DIR/validation_verdict.json" <<'PY'
import json, sys
print('true' if json.load(open(sys.argv[1])).get('is_bypass') else 'false')
PY
)"
write_runtime_manifest "$BYPASS_CONFIRMED"

if [ "$BYPASS_CONFIRMED" = "true" ]; then
  echo "[variant] RESULT: confirmed fixed-version bypass (exit 0)"
  exit 0
fi

echo "[variant] RESULT: no fixed-version bypass confirmed (exit 1)"
exit 1