Enhanced unzip and install with manifest validation and cleanup.
(
cached_zip_path: Path,
cache_manager: PlatformIOCache,
is_framework: bool,
env_section: str,
)
| 577 | |
| 578 | |
| 579 | def unzip_and_install( |
| 580 | cached_zip_path: Path, |
| 581 | cache_manager: PlatformIOCache, |
| 582 | is_framework: bool, |
| 583 | env_section: str, |
| 584 | ) -> bool: |
| 585 | """ |
| 586 | Enhanced unzip and install with manifest validation and cleanup. |
| 587 | """ |
| 588 | cached_zip_path_obj = cached_zip_path |
| 589 | # Extract to a directory alongside the zip file |
| 590 | artifact_dir = cached_zip_path_obj.parent |
| 591 | extracted_dir = artifact_dir / "extracted" |
| 592 | temp_unzip_dir = artifact_dir / "temp_extract" |
| 593 | |
| 594 | try: |
| 595 | # Clean extraction directory |
| 596 | if temp_unzip_dir.exists(): |
| 597 | shutil.rmtree(temp_unzip_dir) |
| 598 | temp_unzip_dir.mkdir(parents=True) |
| 599 | |
| 600 | print(f"Extracting {cached_zip_path_obj} to {temp_unzip_dir}") |
| 601 | |
| 602 | # Extract with better error handling |
| 603 | try: |
| 604 | with zipfile.ZipFile(cached_zip_path_obj, "r") as zip_ref: |
| 605 | # Check for zip bombs (basic protection) |
| 606 | total_size = sum(info.file_size for info in zip_ref.infolist()) |
| 607 | if total_size > 500 * 1024 * 1024: # 500MB limit |
| 608 | raise ValueError("Archive too large, possible zip bomb") |
| 609 | |
| 610 | zip_ref.extractall(temp_unzip_dir) |
| 611 | print("Extraction completed successfully") |
| 612 | |
| 613 | except zipfile.BadZipFile: |
| 614 | logger.error(f"Invalid zip file: {cached_zip_path_obj}") |
| 615 | return False |
| 616 | except KeyboardInterrupt as ki: |
| 617 | handle_keyboard_interrupt(ki) |
| 618 | raise |
| 619 | except Exception as e: |
| 620 | logger.error(f"Extraction failed: {e}") |
| 621 | return False |
| 622 | |
| 623 | # Find content directory (handle nested structures like GitHub archives) |
| 624 | content_items = list(temp_unzip_dir.iterdir()) |
| 625 | if len(content_items) == 1 and content_items[0].is_dir(): |
| 626 | # Single root directory (common with GitHub/GitLab archives) - use it |
| 627 | unzipped_content_path = content_items[0] |
| 628 | print(f"Found nested directory structure: {content_items[0].name}") |
| 629 | else: |
| 630 | # Multiple items at root - use temp dir |
| 631 | unzipped_content_path = temp_unzip_dir |
| 632 | print("Using flat directory structure") |
| 633 | |
| 634 | # Validate manifest files and auto-detect type |
| 635 | manifest_result = validate_and_detect_manifest(unzipped_content_path) |
| 636 | if not manifest_result.is_valid: |
nothing calls this directly
no test coverage detected