Skip to content
14 changes: 9 additions & 5 deletions src/specify_cli/_download_security.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@
from ipaddress import IPv4Address, IPv6Address, ip_address
from itertools import pairwise
from pathlib import Path, PurePosixPath, PureWindowsPath
from typing import NoReturn, TypeVar
from typing import BinaryIO, NoReturn, TypeVar
from urllib.parse import ParseResult, urlparse


Expand Down Expand Up @@ -705,17 +705,19 @@ def _preflight_zip_central_directory(
def open_zip_bounded(
zip_path: Path,
*,
archive_file: BinaryIO | None = None,
error_type: type[ErrorT] = ValueError,
max_entries: int = MAX_ZIP_ENTRIES,
) -> Iterator[zipfile.ZipFile]:
"""Open an untrusted ZIP after a bounded-memory header preflight."""
_validate_non_negative_int(max_entries, "max_entries")
zip_path = Path(zip_path)
with ExitStack() as stack:
try:
archive_file = stack.enter_context(zip_path.open("rb"))
except OSError as exc:
_raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc)
if archive_file is None:
try:
archive_file = stack.enter_context(zip_path.open("rb"))
except OSError as exc:
_raise_from(error_type, f"Invalid ZIP archive: {zip_path}", exc)
try:
_preflight_zip_central_directory(
archive_file,
Expand All @@ -737,6 +739,7 @@ def safe_extract_zip(
zip_path: Path,
target_dir: Path,
*,
archive_file: BinaryIO | None = None,
error_type: type[ErrorT] = ValueError,
max_entries: int = MAX_ZIP_ENTRIES,
max_member_bytes: int = MAX_ZIP_MEMBER_BYTES,
Expand All @@ -752,6 +755,7 @@ def safe_extract_zip(

with open_zip_bounded(
zip_path,
archive_file=archive_file,
error_type=error_type,
max_entries=max_entries,
) as zf:
Expand Down
13 changes: 11 additions & 2 deletions src/specify_cli/extensions/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional, Set
from typing import Any, BinaryIO, Callable, Dict, List, Optional, Set

import pathspec
import yaml
Expand Down Expand Up @@ -2425,6 +2425,8 @@ def install_from_zip(
speckit_version: str,
priority: int = 10,
force: bool = False,
*,
archive_file: BinaryIO | None = None,
) -> ExtensionManifest:
"""Install extension from ZIP file.

Expand All @@ -2434,6 +2436,8 @@ def install_from_zip(
priority: Resolution priority (lower = higher precedence, default 10)
force: If True and extension is already installed, remove it first
before proceeding with installation
archive_file: Already-open archive stream to consume instead of
reopening ``zip_path``

Returns:
Installed extension manifest
Expand All @@ -2449,7 +2453,12 @@ def install_from_zip(
with tempfile.TemporaryDirectory() as tmpdir:
temp_path = Path(tmpdir)

safe_extract_zip(zip_path, temp_path, error_type=ValidationError)
safe_extract_zip(
zip_path,
temp_path,
archive_file=archive_file,
error_type=ValidationError,
)

# Find extension directory (may be nested)
extension_dir = temp_path
Expand Down
Loading