diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 5ba8bbc..4c1e29d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,7 +14,7 @@ jobs: strategy: matrix: os: [ubuntu-latest, windows-latest, macos-latest] - python-version: ["3.8", "3.9", "3.10", "3.11", "3.12"] + python-version: ["3.12", "3.13"] steps: - uses: actions/checkout@v4 @@ -42,7 +42,7 @@ jobs: pytest -v --cov=tzst --cov-report=xml - name: Upload coverage to Codecov - if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.11' + if: matrix.os == 'ubuntu-latest' && matrix.python-version == '3.12' uses: codecov/codecov-action@v3 with: file: ./coverage.xml @@ -58,7 +58,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v4 with: - python-version: "3.11" + python-version: "3.12" - name: Install build dependencies run: | @@ -116,7 +116,7 @@ jobs: - name: Set up Python uses: actions/setup-python@v4 with: - python-version: "3.11" + python-version: "3.12" - name: Fetch PyPI version id: pypi_version diff --git a/pyproject.toml b/pyproject.toml index 6beace8..e72e1f3 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -8,7 +8,7 @@ dynamic = ["version"] description = "A Python library for creating and manipulating .tzst/.tar.zst archives" readme = "README.md" license = "BSD-3-Clause" -requires-python = ">=3.8" +requires-python = ">=3.12" authors = [{ name = "tzst contributors" }] keywords = ["tar", "zstandard", "compression", "archive"] classifiers = [ @@ -17,11 +17,8 @@ classifiers = [ "License :: OSI Approved :: BSD License", "Operating System :: OS Independent", "Programming Language :: Python :: 3", - "Programming Language :: Python :: 3.8", - "Programming Language :: Python :: 3.9", - "Programming Language :: Python :: 3.10", - "Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.12", + "Programming Language :: Python :: 3.13", "Topic :: System :: Archiving :: Compression", "Topic :: Software Development :: Libraries :: Python Modules", ] @@ -56,7 +53,7 @@ python_functions = ["test_*"] addopts = "--cov=tzst --cov-report=term-missing --cov-report=html" [tool.ruff] -target-version = "py38" +target-version = "py312" line-length = 88 [tool.ruff.lint] @@ -77,5 +74,5 @@ ignore = [ "tests/**/*" = ["E501"] [tool.black] -target-version = ["py38", "py39", "py310", "py311", "py312"] +target-version = ["py312", "py313"] line-length = 88 diff --git a/src/tzst/cli.py b/src/tzst/cli.py index def93e8..47ab676 100644 --- a/src/tzst/cli.py +++ b/src/tzst/cli.py @@ -3,7 +3,6 @@ import argparse import sys from pathlib import Path -from typing import List, Optional from . import __version__ from .core import create_archive, extract_archive, list_archive, test_archive @@ -30,7 +29,7 @@ def cmd_add(args) -> int: """Add/create archive command with atomic file operations.""" try: archive_path = Path(args.archive) - files: List[Path] = [Path(f) for f in args.files] + files: list[Path] = [Path(f) for f in args.files] # Check if files exist missing_files = [f for f in files if not f.exists()] @@ -407,7 +406,7 @@ Documentation: return parser -def main(argv: Optional[List[str]] = None) -> int: +def main(argv: list[str] | None = None) -> int: """Main entry point for the CLI.""" print_banner() print() diff --git a/src/tzst/core.py b/src/tzst/core.py index aa90fad..524ced6 100644 --- a/src/tzst/core.py +++ b/src/tzst/core.py @@ -5,23 +5,21 @@ import os import tarfile import tempfile import time +from collections.abc import Callable, Sequence from pathlib import Path -from typing import BinaryIO, Callable, List, Optional, Sequence, Union +from typing import BinaryIO import zstandard as zstd from .exceptions import TzstArchiveError, TzstDecompressionError -# Check if extraction filters are supported (Python 3.12+) -EXTRACTION_FILTERS_SUPPORTED = hasattr(tarfile, "data_filter") - class TzstArchive: """A class for handling .tzst/.tar.zst archives.""" def __init__( self, - filename: Union[str, Path], + filename: str | Path, mode: str = "r", compression_level: int = 3, streaming: bool = False, @@ -41,11 +39,14 @@ class TzstArchive: self.mode = mode self.compression_level = compression_level self.streaming = streaming - self._tarfile: Optional[tarfile.TarFile] = None - self._fileobj: Optional[BinaryIO] = None - self._compressed_stream: Optional[ - Union[zstd.ZstdCompressionWriter, zstd.ZstdDecompressionReader, io.BytesIO] - ] = None + self._tarfile: tarfile.TarFile | None = None + self._fileobj: BinaryIO | None = None + self._compressed_stream: ( + zstd.ZstdCompressionWriter + | zstd.ZstdDecompressionReader + | io.BytesIO + | None + ) = None # Validate mode valid_modes = ["r", "w", "a"] @@ -162,8 +163,8 @@ class TzstArchive: def add( self, - name: Union[str, Path], - arcname: Optional[str] = None, + name: str | Path, + arcname: str | None = None, recursive: bool = True, ): """ @@ -187,11 +188,11 @@ class TzstArchive: def extract( self, - member: Optional[str] = None, - path: Union[str, Path] = ".", + member: str | None = None, + path: str | Path = ".", set_attrs: bool = True, numeric_owner: bool = False, - filter: Optional[Union[str, Callable]] = "data", + filter: str | Callable | None = "data", ): """ Extract files from the archive. @@ -233,18 +234,8 @@ class TzstArchive: "Please use non-streaming mode for selective extraction, or extract all files." ) - # Prepare extraction arguments - extract_kwargs = {} - - # Add filter argument if supported (Python 3.12+) - if EXTRACTION_FILTERS_SUPPORTED: - extract_kwargs["filter"] = filter - elif filter is not None and filter != "data": - # Warn if user specified a filter but it's not supported - print( - "Warning: Extraction filters are not supported in this Python version. " - "Consider upgrading to Python 3.12+ for enhanced security features.", - ) + # Prepare extraction arguments - filters are always supported in Python 3.12+ + extract_kwargs = {"filter": filter} try: if member: @@ -263,7 +254,7 @@ class TzstArchive: else: raise - def extractfile(self, member: Union[str, tarfile.TarInfo]): + def extractfile(self, member: str | tarfile.TarInfo): """ Extract a file-like object from the archive. @@ -280,7 +271,7 @@ class TzstArchive: return self._tarfile.extractfile(member) - def getmembers(self) -> List[tarfile.TarInfo]: + def getmembers(self) -> list[tarfile.TarInfo]: """Get list of all members in the archive.""" if not self._tarfile: raise RuntimeError("Archive not open") @@ -289,7 +280,7 @@ class TzstArchive: return self._tarfile.getmembers() - def getnames(self) -> List[str]: + def getnames(self) -> list[str]: """Get list of all member names in the archive.""" if not self._tarfile: raise RuntimeError("Archive not open") @@ -298,7 +289,7 @@ class TzstArchive: return self._tarfile.getnames() - def list(self, verbose: bool = False) -> List[dict]: + def list(self, verbose: bool = False) -> list[dict]: """ List contents of the archive. @@ -379,8 +370,8 @@ class TzstArchive: def create_archive( - archive_path: Union[str, Path], - files: Sequence[Union[str, Path]], + archive_path: str | Path, + files: Sequence[str | Path], compression_level: int = 3, use_temp_file: bool = True, ) -> None: @@ -442,7 +433,7 @@ def create_archive( def _create_archive_impl( archive_path: Path, - files: Sequence[Union[str, Path]], + files: Sequence[str | Path], compression_level: int, ) -> None: """Internal implementation for creating archives.""" @@ -477,12 +468,12 @@ def _create_archive_impl( def extract_archive( - archive_path: Union[str, Path], - extract_path: Union[str, Path] = ".", - members: Optional[List[str]] = None, + archive_path: str | Path, + extract_path: str | Path = ".", + members: list[str] | None = None, flatten: bool = False, streaming: bool = False, - filter: Optional[Union[str, Callable]] = "data", + filter: str | Callable | None = "data", ) -> None: """ Extract files from a .tzst archive. @@ -534,8 +525,8 @@ def extract_archive( def list_archive( - archive_path: Union[str, Path], verbose: bool = False, streaming: bool = False -) -> List[dict]: + archive_path: str | Path, verbose: bool = False, streaming: bool = False +) -> list[dict]: """ List contents of a .tzst archive. @@ -551,7 +542,7 @@ def list_archive( return archive.list(verbose=verbose) -def test_archive(archive_path: Union[str, Path], streaming: bool = False) -> bool: +def test_archive(archive_path: str | Path, streaming: bool = False) -> bool: """ Test the integrity of a .tzst archive. diff --git a/tests/test_security.py b/tests/test_security.py index a3613d7..fbc7390 100644 --- a/tests/test_security.py +++ b/tests/test_security.py @@ -3,7 +3,6 @@ from unittest.mock import patch from tzst import TzstArchive, extract_archive -from tzst.core import EXTRACTION_FILTERS_SUPPORTED class TestExtractionFilters: @@ -23,16 +22,15 @@ class TestExtractionFilters: # Test that default filter is 'data' extract_dir = temp_dir / "extracted_default" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir) + with patch("tarfile.TarFile.extractall") as mock_extractall: + with TzstArchive(archive_path, "r") as archive: + archive.extract(path=extract_dir) - # Verify that 'data' filter was used - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert "filter" in call_args[1] - assert call_args[1]["filter"] == "data" + # Verify that 'data' filter was used + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert "filter" in call_args[1] + assert call_args[1]["filter"] == "data" def test_data_filter_explicit(self, sample_files, temp_dir): """Test explicitly setting 'data' filter.""" @@ -48,14 +46,13 @@ class TestExtractionFilters: # Test extraction with explicit 'data' filter extract_dir = temp_dir / "extracted_data" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter="data") + with patch("tarfile.TarFile.extractall") as mock_extractall: + with TzstArchive(archive_path, "r") as archive: + archive.extract(path=extract_dir, filter="data") - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert call_args[1]["filter"] == "data" + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert call_args[1]["filter"] == "data" def test_tar_filter(self, sample_files, temp_dir): """Test 'tar' filter for Unix-like features.""" @@ -71,14 +68,13 @@ class TestExtractionFilters: # Test extraction with 'tar' filter extract_dir = temp_dir / "extracted_tar" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter="tar") + with patch("tarfile.TarFile.extractall") as mock_extractall: + with TzstArchive(archive_path, "r") as archive: + archive.extract(path=extract_dir, filter="tar") - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert call_args[1]["filter"] == "tar" + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert call_args[1]["filter"] == "tar" def test_fully_trusted_filter(self, sample_files, temp_dir): """Test 'fully_trusted' filter (dangerous but complete).""" @@ -94,14 +90,13 @@ class TestExtractionFilters: # Test extraction with 'fully_trusted' filter extract_dir = temp_dir / "extracted_trusted" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter="fully_trusted") + with patch("tarfile.TarFile.extractall") as mock_extractall: + with TzstArchive(archive_path, "r") as archive: + archive.extract(path=extract_dir, filter="fully_trusted") - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert call_args[1]["filter"] == "fully_trusted" + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert call_args[1]["filter"] == "fully_trusted" def test_none_filter_with_warning(self, sample_files, temp_dir, capsys): """Test None filter shows deprecation warning.""" @@ -117,14 +112,13 @@ class TestExtractionFilters: # Test extraction with None filter extract_dir = temp_dir / "extracted_none" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter=None) + with patch("tarfile.TarFile.extractall") as mock_extractall: + with TzstArchive(archive_path, "r") as archive: + archive.extract(path=extract_dir, filter=None) - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert call_args[1]["filter"] is None + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert call_args[1]["filter"] is None def test_custom_filter_function(self, sample_files, temp_dir): """Test custom filter function.""" @@ -147,36 +141,13 @@ class TestExtractionFilters: # Test extraction with custom filter extract_dir = temp_dir / "extracted_custom" - if EXTRACTION_FILTERS_SUPPORTED: - with patch("tarfile.TarFile.extractall") as mock_extractall: - with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter=custom_filter) - - mock_extractall.assert_called_once() - call_args = mock_extractall.call_args - assert call_args[1]["filter"] == custom_filter - - def test_filter_not_supported_warning(self, sample_files, temp_dir, capsys): - """Test warning when filters not supported in older Python versions.""" - archive_path = temp_dir / "test_warning.tzst" - file_paths = [f for f in sample_files if f.is_file()] - - # Create archive - with TzstArchive(archive_path, "w") as archive: - for file_path in file_paths: - relative_path = file_path.relative_to(sample_files[0].parent) - archive.add(file_path, arcname=str(relative_path)) - - # Mock EXTRACTION_FILTERS_SUPPORTED to False - with patch("tzst.core.EXTRACTION_FILTERS_SUPPORTED", False): - extract_dir = temp_dir / "extracted_warning" - + with patch("tarfile.TarFile.extractall") as mock_extractall: with TzstArchive(archive_path, "r") as archive: - archive.extract(path=extract_dir, filter="tar") + archive.extract(path=extract_dir, filter=custom_filter) - # Check warning was printed - captured = capsys.readouterr() - assert "Warning: Extraction filters are not supported" in captured.out + mock_extractall.assert_called_once() + call_args = mock_extractall.call_args + assert call_args[1]["filter"] == custom_filter def test_convenience_function_filter(self, sample_files, temp_dir): """Test filter parameter in extract_archive convenience function.""" @@ -190,17 +161,16 @@ class TestExtractionFilters: archive.add(file_path, arcname=str(relative_path)) # Test extract_archive with different filters - if EXTRACTION_FILTERS_SUPPORTED: - for filter_type in ["data", "tar", "fully_trusted"]: - extract_dir = temp_dir / f"extracted_conv_{filter_type}" + for filter_type in ["data", "tar", "fully_trusted"]: + extract_dir = temp_dir / f"extracted_conv_{filter_type}" - # This should not raise an exception - extract_archive(archive_path, extract_dir, filter=filter_type) + # This should not raise an exception + extract_archive(archive_path, extract_dir, filter=filter_type) - # Verify files were extracted - assert extract_dir.exists() - extracted_files = list(extract_dir.rglob("*")) - assert len([f for f in extracted_files if f.is_file()]) > 0 + # Verify files were extracted + assert extract_dir.exists() + extracted_files = list(extract_dir.rglob("*")) + assert len([f for f in extracted_files if f.is_file()]) > 0 class TestSecurityDocumentation: @@ -208,20 +178,19 @@ class TestSecurityDocumentation: def test_extract_method_has_security_warning(self): """Test that extract method has proper security warning in docstring.""" - assert ( - "Never extract archives from untrusted sources" - in TzstArchive.extract.__doc__ - ) - assert "filter" in TzstArchive.extract.__doc__ - assert "data" in TzstArchive.extract.__doc__ + docstring = TzstArchive.extract.__doc__ + assert docstring is not None + assert "Never extract archives from untrusted sources" in docstring + assert "filter" in docstring + assert "data" in docstring def test_extract_archive_has_security_warning(self): """Test that extract_archive function has proper security warning.""" - assert ( - "Never extract archives from untrusted sources" in extract_archive.__doc__ - ) - assert "path traversal attacks" in extract_archive.__doc__ - assert "data" in extract_archive.__doc__ + docstring = extract_archive.__doc__ + assert docstring is not None + assert "Never extract archives from untrusted sources" in docstring + assert "path traversal attacks" in docstring + assert "data" in docstring class TestSecurityEdgeCases: @@ -241,15 +210,14 @@ class TestSecurityEdgeCases: # Test extraction in streaming mode with filter extract_dir = temp_dir / "extracted_streaming" - if EXTRACTION_FILTERS_SUPPORTED: - # This should work without errors - with TzstArchive(archive_path, "r", streaming=True) as archive: - archive.extract(path=extract_dir, filter="data") + # This should work without errors + with TzstArchive(archive_path, "r", streaming=True) as archive: + archive.extract(path=extract_dir, filter="data") - # Verify files were extracted - assert extract_dir.exists() - extracted_files = list(extract_dir.rglob("*")) - assert len([f for f in extracted_files if f.is_file()]) > 0 + # Verify files were extracted + assert extract_dir.exists() + extracted_files = list(extract_dir.rglob("*")) + assert len([f for f in extracted_files if f.is_file()]) > 0 def test_filter_with_specific_member_extraction(self, sample_files, temp_dir): """Test filter when extracting specific members.""" @@ -265,14 +233,13 @@ class TestSecurityEdgeCases: # Test extracting specific member with filter extract_dir = temp_dir / "extracted_member" - if EXTRACTION_FILTERS_SUPPORTED and file_paths: + if file_paths: with TzstArchive(archive_path, "r") as archive: members = archive.getnames() if members: # Extract first member with data filter - archive.extract( - member=members[0], path=extract_dir, filter="data" - ) # Verify file was extracted + archive.extract(member=members[0], path=extract_dir, filter="data") + # Verify file was extracted assert extract_dir.exists() extracted_file = extract_dir / members[0] assert extracted_file.exists()