| 1 | from __future__ import annotations |
| 2 | |
| 3 | from pathlib import Path |
| 4 | import shutil |
| 5 | import stat |
| 6 | import subprocess |
| 7 | import tarfile |
| 8 | import zipfile |
| 9 | |
| 10 | from helpers import extension, files, runtime |
| 11 | from helpers.api import ApiHandler, Input, Output, Request |
| 12 | from api import get_work_dir_files |
| 13 | |
| 14 | |
| 15 | ARCHIVE_SUFFIXES = ( |
| 16 | ".tar.gz", ".tar.bz2", ".tar.xz", ".tar.zst", ".tar", ".tgz", ".tbz", ".tbz2", ".txz", |
| 17 | ".zip", ".rar", ".7z", ".gz", ".bz2", ".xz", ".zst", |
| 18 | ) |
| 19 | TAR_SUFFIXES = (".tar.gz", ".tar.bz2", ".tar.xz", ".tar", ".tgz", ".tbz", ".tbz2", ".txz") |
| 20 | |
| 21 | |
| 22 | class ExtractWorkDirArchive(ApiHandler): |
| 23 | async def process(self, input: Input, request: Request) -> Output: |
| 24 | path = str(input.get("path") or "").strip() |
| 25 | if not path: |
| 26 | return {"error": "Archive path is required"} |
| 27 | if not path.startswith("/"): |
| 28 | path = f"/{path}" |
| 29 | |
| 30 | try: |
| 31 | extracted_path = await runtime.call_development_function(extract_archive, path) |
| 32 | except (OSError, ValueError) as exc: |
| 33 | return {"error": str(exc)} |
| 34 | |
| 35 | current_path = str(input.get("currentPath") or "") |
| 36 | await extension.call_extensions_async( |
| 37 | "workdir_file_mutation_after", |
| 38 | agent=None, |
| 39 | data={ |
| 40 | "action": "extract", |
| 41 | "path": extracted_path, |
| 42 | "paths": [path, extracted_path], |
| 43 | "current_path": current_path, |
| 44 | }, |
| 45 | ) |
| 46 | listing = await runtime.call_development_function(get_work_dir_files.get_files, current_path) |
| 47 | return {"data": listing, "extracted_path": extracted_path} |
| 48 | |
| 49 | |
| 50 | def extract_archive(path: str) -> str: |
| 51 | source = resolve_archive_path(path) |
| 52 | target = create_target_directory(source) |
| 53 | try: |
| 54 | kind = archive_kind(source) |
| 55 | if kind == "zip": |
| 56 | extract_zip(source, target) |
| 57 | elif kind == "tar": |
| 58 | extract_tar(source, target) |
| 59 | else: |
| 60 | extract_with_7zip(source, target) |
| 61 | except Exception: |
| 62 | shutil.rmtree(target, ignore_errors=True) |
| 63 | raise |
| 64 | return str(target) |
| 65 | |
| 66 | |
| 67 | def resolve_archive_path(path: str) -> Path: |
| 68 | base = Path(files.get_base_dir()).resolve() |
| 69 | candidate = Path(path) |
| 70 | resolved = candidate.resolve() if candidate.is_absolute() else (base / candidate).resolve() |
| 71 | try: |
| 72 | resolved.relative_to(base) |
| 73 | except ValueError as exc: |
| 74 | raise ValueError("Invalid archive path") from exc |
| 75 | if not resolved.is_file(): |
| 76 | raise ValueError("Archive file was not found") |
| 77 | return resolved |
| 78 | |
| 79 | |
| 80 | def archive_kind(path: Path) -> str: |
| 81 | name = path.name.lower() |
| 82 | if name.endswith(".zip"): |
| 83 | return "zip" |
| 84 | if name.endswith(TAR_SUFFIXES): |
| 85 | return "tar" |
| 86 | if name.endswith(ARCHIVE_SUFFIXES): |
| 87 | return "7zip" |
| 88 | raise ValueError("Unsupported archive format") |
| 89 | |
| 90 | |
| 91 | def create_target_directory(source: Path) -> Path: |
| 92 | name = source.name |
| 93 | for suffix in ARCHIVE_SUFFIXES: |
| 94 | if name.lower().endswith(suffix): |
| 95 | name = name[:-len(suffix)] |
| 96 | break |
| 97 | name = name or "extracted" |
| 98 | target = source.parent / name |
| 99 | index = 2 |
| 100 | while target.exists(): |
| 101 | target = source.parent / f"{name}-{index}" |
| 102 | index += 1 |
| 103 | target.mkdir() |
| 104 | return target |
| 105 | |
| 106 | |
| 107 | def safe_member_path(target: Path, name: str) -> Path: |
| 108 | if not name or name.startswith(("/", "\\")) or "\\" in name or ".." in Path(name).parts: |
| 109 | raise ValueError("Archive contains an unsafe path") |
| 110 | destination = (target / name).resolve(strict=False) |
| 111 | try: |
| 112 | destination.relative_to(target.resolve()) |
| 113 | except ValueError as exc: |
| 114 | raise ValueError("Archive contains an unsafe path") from exc |
| 115 | return destination |
| 116 | |
| 117 | |
| 118 | def extract_zip(source: Path, target: Path) -> None: |
| 119 | with zipfile.ZipFile(source) as archive: |
| 120 | for member in archive.infolist(): |
| 121 | safe_member_path(target, member.filename) |
| 122 | if stat.S_ISLNK(member.external_attr >> 16): |
| 123 | raise ValueError("Archive contains a symbolic link") |
| 124 | archive.extractall(target) |
| 125 | |
| 126 | |
| 127 | def extract_tar(source: Path, target: Path) -> None: |
| 128 | with tarfile.open(source, "r:*") as archive: |
| 129 | for member in archive.getmembers(): |
| 130 | safe_member_path(target, member.name) |
| 131 | if member.issym() or member.islnk() or member.isdev(): |
| 132 | raise ValueError("Archive contains a symbolic link or device") |
| 133 | archive.extractall(target, filter="data") |
| 134 | |
| 135 | |
| 136 | def extract_with_7zip(source: Path, target: Path) -> None: |
| 137 | binary = shutil.which("7z") or shutil.which("7zz") |
| 138 | if not binary: |
| 139 | raise ValueError("This archive format requires 7-Zip in the runtime image") |
| 140 | listing = subprocess.run( |
| 141 | [binary, "l", "-slt", str(source)], |
| 142 | check=True, |
| 143 | capture_output=True, |
| 144 | text=True, |
| 145 | ).stdout |
| 146 | marker = "----------" |
| 147 | if marker not in listing: |
| 148 | raise ValueError("Could not inspect archive safely") |
| 149 | for line in listing.split(marker, 1)[1].splitlines(): |
| 150 | if line.startswith("Path = "): |
| 151 | safe_member_path(target, line.removeprefix("Path = ")) |
| 152 | subprocess.run([binary, "x", "-y", f"-o{target}", str(source)], check=True, capture_output=True) |
| 153 | if any(path.is_symlink() for path in target.rglob("*")): |
| 154 | raise ValueError("Archive contains a symbolic link") |