diff --git a/.github/CHANGELOG.md b/.github/CHANGELOG.md index 0920053..7cf220f 100644 --- a/.github/CHANGELOG.md +++ b/.github/CHANGELOG.md @@ -13,6 +13,20 @@ `NotImplementedError` directing callers to the public `download_binary()` entry point instead of returning a fabricated cache-relative path (OVH-05). +- Added a strict `mypy` gate (`strict = true`, Python 3.9 target) over the + `py7zip` runtime package with a matching CI lane, and annotated the + wrapper's public methods without changing runtime behavior. +- Landed the options contract (audit decision D5 / OVH-16): the safe + runtime accepts only `""` or a sequence of individual arguments and + raises `TypeError` for any non-empty options string; the legacy + runtime forwards a sequence as a `shell=False` argument vector while + keeping the historical `shell=True` string interpolation. +- Fixed the safe runtime's `wrapper()`/`decompress()`/`extract()` + dispatch to route decompression through `SafePy7zip.decompress()`, + whose archive-member validation raises `ArchiveTraversalError` + before extraction begins; the dispatch previously called + `SafePy7zip.run()` directly and bypassed that guard. Compression + behaviour and the options contract are unchanged. ## 0.7.3 diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d64d2ec..a9386a3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -31,6 +31,23 @@ jobs: - name: Run offline test suite run: python -m pytest tests -p no:cacheprovider + typecheck: + name: Strict type check (mypy) + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: actions/setup-python@v5 + with: + python-version: "3.12" + cache: pip + - name: Install package and type checker + # The mypy constraint and stubs live in the `typecheck` extra in + # pyproject.toml, so local and CI gates enforce the same constraint + # (mypy<2: each install resolves the newest matching release). + run: python -m pip install --upgrade pip -e ".[typecheck]" + - name: Run mypy (strict) on py7zip + run: python -m mypy + package: name: Build and inspect wheel runs-on: ubuntu-latest diff --git a/docs/USAGE.md b/docs/USAGE.md index 6eb182c..1c0433e 100644 --- a/docs/USAGE.md +++ b/docs/USAGE.md @@ -1,22 +1,67 @@ # USAGE ## Example Python script: - import py7zip + from py7zip.py7zip import Py7zip - # Sets up binary for the system if it does not already exist and assigns the 7zip class to 'z' - z = py7zip.Py7zip() + # Constructs the wrapper (offline: nothing is downloaded at + # construction). The verified binary is acquired automatically on + # first use if it is not already installed. + z = Py7zip() src = "path/to/zipfile" dst = "path/to/output/directory" # Usually contains a .zip extension - options = "" # Optional - # Compress source file or directory to destination - z.compress(src, dst, options) + # Compress source file or directory to destination. + # options is "" (no options) or a sequence of individual arguments. + z.compress(src, dst, options=["-y", "-mx=9"]) # Decompress source archive back to destination directory - z.decompress(dst, src, options) - + z.decompress(dst, src, options="") + + +## The `options` argument + +Every execution method — `wrapper`, the compress aliases +(`compress`/`archive`/`backup`) and the extract aliases +(`decompress`/`extract`) — forwards `options` unchanged to the same code +path. What `options` accepts depends on the runtime chosen at +construction. + +### Safe runtime (default) + +`Py7zip()` never invokes a shell. `options` must be one of: + +- `""` — the default. No options are forwarded to 7-Zip. +- A sequence (list, tuple, ...) of individual arguments, each forwarded + to 7-Zip as exactly one argument: `options=["-y", "-mx=9"]`. Elements + are never split, quoted, or re-interpreted; an element containing + spaces stays a single argument. + +Any non-empty string — `"-y -mx=9"`, `"-oC:\Temp"`, even `" "` — raises +`TypeError` ("options must be a sequence of individual arguments") before +any 7-Zip process is started. Shell-style option text is never parsed or +tokenized: there is deliberately no `shlex`-based bridge on the safe +runtime (audit decision D5 / OVH-16). Split the text into a sequence of +individual arguments instead. + +### Legacy runtime (`Py7zip(legacy=True)`) + +The legacy runtime exists for backward compatibility with the historical +0.6.x behaviour and is planned for deprecation and removal. + +- A string is interpolated verbatim into one shell command that runs with + `shell=True`. The shell interprets the text: quotes, `$()`, `;` and + pipes are all live. Only ever pass fully trusted, caller-controlled + text. +- A sequence is forwarded element by element as an argument vector with + `shell=False`: no element is re-interpreted by a shell, and an element + containing spaces or shell metacharacters stays a single argument. + ## 7zip Usage (Command Line Arguments; aka "Options") + + The switches listed below are the values a sequence element carries; + pass each switch as its own element, e.g. `options=["-y", "-mx=9"]`. + Usage: 7zz [...] [...] [@listfile] diff --git a/docs/planning/COMPATIBILITY.md b/docs/planning/COMPATIBILITY.md index 2ce443e..d06bcb3 100644 --- a/docs/planning/COMPATIBILITY.md +++ b/docs/planning/COMPATIBILITY.md @@ -6,7 +6,10 @@ between them. It is written before the contract changes so that each break is a recorded decision rather than an accident of refactoring. The behaviour described in "current contract" is not asserted from reading -the source; every line is pinned by a test in `tests/`. +the source; every line is pinned by a test in `tests/`. Where a piece of +the target contract has already landed, it is recorded in place as its own +section, naming the decision that authorised it and the tests that pin it +— see "Options contract" below. ## Current contract (Phase 0 baseline) @@ -26,13 +29,22 @@ the source; every line is pinned by a test in `tests/`. `decompress` and `extract` run the `x` (extract) operation; `compress`, `archive` and `backup` run the `a` (add) operation. All five forward to the -single `wrapper` method. - -**Every alias discards the caller's `options` argument.** Each one passes the -literal empty string to `wrapper`, so `obj.extract(a, b, options="-y")` runs -without `-y`. Only calling `wrapper(..., options=...)` directly forwards them. -This is characterized by -`tests/test_characterization_wrapper.py::test_aliases_silently_discard_the_callers_options`. +single `wrapper` method, and **all five forward the caller's `options` +argument unchanged, on both runtimes.** + +At the Phase 0 baseline the aliases discarded `options`: each passed the +literal empty string to `wrapper`, so `obj.extract(a, b, options="-y")` ran +without `-y` (pinned then by +`test_aliases_silently_discard_the_callers_options`). When the +safe-by-default runtime landed (`4713a75`), the aliases changed to forward +`options` and the discard pin was replaced in that same commit, per rule 3, +by forwarding pins on both runtimes: +`tests/test_characterization_wrapper.py::test_legacy_aliases_forward_the_callers_options` +and, for the safe runtime, +`tests/test_default_api.py::test_default_aliases_require_individual_option_arguments`. +This PR adds +`tests/test_default_api.py::test_default_runtime_accepts_individual_option_arguments` +as a separate pin for the safe runtime's individual-argument contract. ### Return values and errors @@ -50,15 +62,36 @@ features and are removed in Phase 4. ### Unspecified behaviour -* `wrapper` builds one shell string and passes it to - `subprocess.run(..., shell=True)`. The caller's `options` text is - interpolated verbatim, so it is interpreted by the shell. +* On the legacy runtime with string `options`, `wrapper` builds one shell + string and passes it to `subprocess.run(..., shell=True)`; the caller's + `options` text is interpolated verbatim, so it is interpreted by the + shell. Sequence `options` are the exception on both runtimes: they run + as an argument vector with `shell=False`, and the safe runtime never + builds a shell string at all (see "Options contract" below). * Source and destination are never validated. Nonexistent paths and paths containing `..` are passed through unchanged. * Downloaded bytes are written to disk and marked `0o755` with no size bound, content check, or digest. * `Py7zip.cd` changes the working directory of the entire process. +## Options contract (landed 2026-10-11 — audit decision D5 / OVH-16) + +`options` is typed `str | Sequence[str]` with default `""` and means +different things on the two runtimes. Every row below is pinned by a test. + +| Runtime | `options` value | Behaviour | Pinned by | +|--------|-----------------|-----------|-----------| +| Safe (default, `legacy=False`) — the supported runtime | `""` | No options are forwarded. | `tests/test_default_api.py::test_default_runtime_empty_string_options_are_empty_arguments` | +| Safe (default) | Sequence | Each element is forwarded as exactly one argv element to the no-shell runner; elements are never split or re-interpreted. | `tests/test_default_api.py::test_default_runtime_accepts_individual_option_arguments`; `tests/test_safe.py::test_runner_passes_an_argument_vector_without_shell` | +| Safe (default) | Any non-empty `str` (and `bytes`) | `TypeError("options must be a sequence of individual arguments")` raised before any subprocess starts. No `shlex` bridge exists — decision D5 rejects one because POSIX-mode `shlex` silently corrupts unquoted Windows backslashes (`"-oC:\Temp"`) and raises `ValueError` on an unbalanced quote. | `tests/test_default_api.py::test_default_runtime_shell_text_never_reaches_parsing_or_execution`; `tests/test_safe.py::test_runner_rejects_opaque_option_strings` | +| Legacy (`legacy=True`) — compatibility surface, slated for sunset | `str` | Interpolated verbatim into one shell string run with `shell=True` — the historical 0.6.x behaviour, shell word-splitting and command substitution included. Retained unchanged; the existing unsafe compatibility behaviour. | `tests/test_characterization_wrapper.py::test_options_are_appended_verbatim_to_the_command_string`, `::test_the_shell_word_splits_the_options_string`, `::test_options_are_evaluated_by_the_shell`, `::test_options_can_run_additional_commands` | +| Legacy (`legacy=True`) | Sequence | Forwarded element by element as an argument vector with `shell=False`; no element is re-interpreted by a shell. | `tests/test_characterization_wrapper.py::test_sequence_options_are_forwarded_as_an_argument_vector`, `::test_sequence_options_reach_the_binary_one_argument_each`, `::test_sequence_options_with_spaces_stay_one_argument` | + +All five aliases forward `options` on both runtimes (see "Aliases" above). +The legacy string path is the retained unsafe compatibility surface; its +removal is governed by the legacy-sunset gate OVH-27 (audit decision D1), +implemented by OVH-19, which requires this options story to exist first. + ## Target contract (from Phase 2 onward) | Aspect | Target | @@ -79,7 +112,7 @@ features and are removed in Phase 4. | `import py7zip` then `py7zip.Py7zip(...)` | still unsupported today; the package root exports nothing | Import from `py7zip.py7zip`. A root-level re-export may be added later as an *additive* change, not a break. | | `Py7zip()` implicitly downloading a binary | explicit `ensure_binary()` | Call `ensure_binary()` where the download used to happen implicitly. | | `Py7zip.__version__` from HTTP | package metadata; no network | No change for correct callers. Callers that relied on `"0.0.0"` as an offline sentinel must handle the metadata-backed value. | -| `wrapper(src, dst, options, method)` shell string | argv-based, typed options | Replace any free-text options string with the typed structure or explicit sequence. | +| `wrapper(src, dst, options, method)` shell string | Safe runtime (default): `""` or a sequence of individual arguments; every non-empty string raises `TypeError` before any subprocess — decision D5 landed with no `shlex` bridge. Legacy runtime: strings keep the historical `shell=True` interpolation until the legacy sunset (OVH-27/OVH-19); sequences run as `shell=False` argv (OVH-16, landed 2026-10-11). | Pass options as a sequence of individual arguments (`["-y", "-mx=9"]`), or move to the default safe runtime where that is the only accepted form. `""` still means "no options" on both runtimes. | | Ignoring the return value of `compress` / `decompress` | a structured `ArchiveResult` | Inspect the result or catch the typed error; failures will no longer be silent. | | `full`, `incremental`, `differential`, `snapshot` | removed | Stop calling them. They have never done anything. | | `Py7zip.cd` (process-wide `chdir`) | deprecated, then removed | Manage the working directory in caller code (`os.chdir` / `contextlib.chdir`) or pass explicit paths; `cd` mutates process-global state and is not part of the wrapper's archive contract. | diff --git a/py7zip/py7zip.py b/py7zip/py7zip.py index 00a6056..018211f 100644 --- a/py7zip/py7zip.py +++ b/py7zip/py7zip.py @@ -1,30 +1,41 @@ +from __future__ import annotations + import os import platform import re import subprocess import urllib.request +from collections.abc import Sequence from importlib.metadata import PackageNotFoundError from importlib.metadata import version as package_version +from pathlib import Path import requests -from .safe import SafePy7zip +from .platforms import PlatformInfo +from .safe import ArchiveResult, SafePy7zip class Py7zip: """An unofficial, cross-platform, lightweight, and easy-to-use wrapper for 7zip command-line binaries (7za) in Python.""" + # The safe runtime stores a verified Path (or None before acquisition); + # the legacy runtime stores the historical in-package path as a str. + binary_path: str | Path | None + platform_info: PlatformInfo + _safe: SafePy7zip + def __init__( self, - verbose=False, - debug=False, + verbose: bool = False, + debug: bool = False, *, - legacy=False, - cache_dir=None, - binary_path=None, - timeout=300.0, - ): + legacy: bool = False, + cache_dir: str | os.PathLike[str] | None = None, + binary_path: str | os.PathLike[str] | None = None, + timeout: float = 300.0, + ) -> None: """Initializes the class and variables.""" self.verbose = verbose self.debug = debug @@ -75,13 +86,13 @@ def __init__( self.setup() @staticmethod - def _metadata_version(): + def _metadata_version() -> str: try: return package_version("py7zip") except PackageNotFoundError: return "0.7.3" - def get_version(self, verbose=False): + def get_version(self, verbose: bool = False) -> str: """Construct the URL to fetch CHANGELOG.md and find the version number.""" if not self.legacy: return self._metadata_version() @@ -113,37 +124,46 @@ def get_version(self, verbose=False): return version - def setup(self): + def setup(self) -> None: """Checks the system for prerequisites and performs the required steps.""" if not self.legacy: return + # Only the legacy branch reaches here, and legacy __init__ always + # assigns binary_path a str. + binary_path = self.binary_path + assert binary_path is not None try: - if not os.path.exists(self.binary_path): + if not os.path.exists(binary_path): self.download_binary() except (OSError, requests.RequestException) as err: print(err) @staticmethod - def cd(directory_path): + def cd(directory_path: str | os.PathLike[str]) -> None: os.chdir(directory_path) - def download_binary(self): + def download_binary(self) -> Path | None: """Dynamically downloads the binary required for the given platform.""" if not self.legacy: self.binary_path = self._safe.ensure_binary() return self.binary_path + # Only the legacy branch reaches here, and legacy __init__ always + # assigns binary_path a str. + binary_path = self.binary_path + assert binary_path is not None binary_url = self.get_binary_url() - binary_dir = os.path.dirname(self.binary_path) + binary_dir = os.path.dirname(binary_path) os.makedirs(binary_dir, exist_ok=True) with ( urllib.request.urlopen(binary_url) as response, - open(self.binary_path, "wb") as out_file, + open(binary_path, "wb") as out_file, ): out_file.write(response.read()) - os.chmod(self.binary_path, 0o755) + os.chmod(binary_path, 0o755) + return None - def get_binary_url(self): + def get_binary_url(self) -> str: """Return the source location for the platform's 7-Zip binary. On the safe runtime (default), binary acquisition is explicit: @@ -169,30 +189,76 @@ def get_binary_url(self): raise NotImplementedError(f"Platform '{self.platform}' is not supported.") return self.url - def wrapper(self, src, dst, options="", method="decompress"): - """Method used to extract an archive.""" + def wrapper( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + method: str = "decompress", + ) -> ArchiveResult | None: + """Method used to extract an archive. + + ``options`` means different things on the two runtimes: + + * Safe runtime (default): only a sequence of individual arguments + (e.g. ``["-y", "-mx=9"]``) is accepted, each forwarded as a single + argument to an argv runner. A non-empty string raises + :class:`TypeError` (audit decision D5 / OVH-16: shell text is + rejected, never silently tokenized). The documented default + ``""`` forwards no arguments. Decompression is routed through + :meth:`SafePy7zip.decompress`, which lists the archive and + validates every member path against the destination before + extracting; an escaping member raises + :class:`ArchiveTraversalError` before extraction begins. + * Legacy runtime: a string is appended verbatim to the shell + command, preserving the historical behaviour; a sequence is + forwarded element by element as an argument vector with + ``shell=False``, so no element is re-interpreted by a shell. + """ if not self.legacy: - operation = "compress" if method == "compress" else "decompress" safe_options = () if options == "" else options - result = self._safe.run(operation, src, dst, safe_options) + if method == "compress": + result = self._safe.compress(src, dst, safe_options) + else: + result = self._safe.decompress(src, dst, safe_options) if self.debug and result.stdout: print(result.stdout) return result + command: str | tuple[str, ...] + shell = isinstance(options, str) if method == "compress": - command = f'"7za{self.extension}" a "{dst}" "{src}" {options}' + if shell: + command = f'"7za{self.extension}" a "{dst}" "{src}" {options}' + else: + command = ( + f"7za{self.extension}", + "a", + os.fspath(dst), + os.fspath(src), + *tuple(options), + ) success_msg = f"Backup created from '{src}' to '{dst}'." error_msg = f"Failed to extract archive from '{src}' to '{dst}'." else: # "decompress" method as default - command = f'"7za{self.extension}" x "{src}" -o"{dst}" {options}' + if shell: + command = f'"7za{self.extension}" x "{src}" -o"{dst}" {options}' + else: + command = ( + f"7za{self.extension}", + "x", + os.fspath(src), + f"-o{os.fspath(dst)}", + *tuple(options), + ) success_msg = f"Extracted archive from '{src}' to '{dst}'" error_msg = f"Failed to create backup from '{src}' to '{dst}'." try: - result = subprocess.run( - command, capture_output=True, text=True, shell=True, check=True + completed = subprocess.run( + command, capture_output=True, text=True, shell=shell, check=True ) if self.debug: - print(result.stdout) + print(completed.stdout) if self.verbose: print(success_msg) except subprocess.CalledProcessError as e: @@ -200,37 +266,83 @@ def wrapper(self, src, dst, options="", method="decompress"): print(f"Error: {e.stderr}.") if self.verbose: print(error_msg) + return None - def decompress(self, src, dst, options=""): + def decompress( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> ArchiveResult | None: """Method used to extract an archive.""" return self.wrapper(src, dst, options=options, method="decompress") - def extract(self, src, dst, options=""): + def extract( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> ArchiveResult | None: """Alias used for 'decompress' method.""" return self.wrapper(src, dst, options=options, method="decompress") - def compress(self, src, dst, options=""): + def compress( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> ArchiveResult | None: """Method used to backup files, folders, or directories to an archive.""" return self.wrapper(src, dst, options=options, method="compress") - def archive(self, src, dst, options=""): + def archive( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> ArchiveResult | None: """Alias used for compress method.""" return self.wrapper(src, dst, options=options, method="compress") - def backup(self, src, dst, options=""): + def backup( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> ArchiveResult | None: """Alias used for compress method.""" return self.wrapper(src, dst, options=options, method="compress") - def full(self, src, dst, options=""): + def full( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> None: return - def incremental(self, src, dst, options=""): + def incremental( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> None: return - def differential(self, src, dst, options=""): + def differential( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> None: return - def snapshot(self, src, dst, options=""): + def snapshot( + self, + src: str | os.PathLike[str], + dst: str | os.PathLike[str], + options: str | Sequence[str] = "", + ) -> None: """Streamlines the process for creating a snapshot.""" return diff --git a/py7zip/safe.py b/py7zip/safe.py index 1a08992..6f1b5d2 100644 --- a/py7zip/safe.py +++ b/py7zip/safe.py @@ -57,8 +57,7 @@ def run( """Run an archive operation without invoking a shell.""" if operation not in {"compress", "decompress"}: raise ValueError("operation must be 'compress' or 'decompress'") - if isinstance(options, (str, bytes)): - raise TypeError("options must be a sequence of individual arguments") + validate_options(options) source_text = os.fspath(source) destination_text = os.fspath(destination) @@ -121,6 +120,12 @@ def _invoke(self, command: tuple[str, ...]) -> ArchiveResult: ) +def validate_options(options: Sequence[str]) -> None: + """Reject opaque option strings before any 7-Zip process is started.""" + if isinstance(options, (str, bytes)): + raise TypeError("options must be a sequence of individual arguments") + + def validate_archive_members( members: Sequence[str], destination: str | os.PathLike[str] ) -> None: @@ -184,6 +189,7 @@ def run( options: Sequence[str] = (), ) -> ArchiveResult: """Execute an operation after explicit binary resolution.""" + validate_options(options) return ArchiveRunner(self.ensure_binary(), timeout=self.timeout).run( operation, source, destination, options ) @@ -202,6 +208,7 @@ def decompress( destination: str | os.PathLike[str], options: Sequence[str] = (), ) -> ArchiveResult: + validate_options(options) runner = ArchiveRunner(self.ensure_binary(), timeout=self.timeout) validate_archive_members(runner.list_entries(source), destination) return runner.run("decompress", source, destination, options) diff --git a/py7zip/tests/debugging.py b/py7zip/tests/debugging.py index 6330ec5..4cb2fc9 100644 --- a/py7zip/tests/debugging.py +++ b/py7zip/tests/debugging.py @@ -8,6 +8,6 @@ py7z = py7zip.Py7zip() # print(py7z.get_binary_url()) # print(py7z.setup()) - py7z.cd(os.path.dirname(py7zip.__file__)) + py7z.cd(os.path.dirname(py7zip.__file__ or ".")) py7z.decompress("7z2301-src.7z", "7zTemp1") py7z.compress("*", "7zTemp1") diff --git a/pyproject.toml b/pyproject.toml index b5ebfd6..941a5ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -38,8 +38,24 @@ Documentation = "https://github.com/aliasfoxkde/py7zip/blob/main/README.md" [tool.setuptools.packages.find] include = ["py7zip*"] +[project.optional-dependencies] +# mypy 2.x dropped the Python 3.9 target this package still supports, so the +# strict typecheck gate constrains mypy to the maintained 1.x line (`mypy<2` — +# a range, not an exact pin) plus the requests stubs. CI installs this extra +# instead of restating the constraint there. +typecheck = ["mypy<2", "types-requests"] + [tool.pytest.ini_options] minversion = "8.0" testpaths = ["tests"] addopts = "-q --strict-markers --strict-config" filterwarnings = ["error"] + +[tool.mypy] +# The runtime package is checked strictly under the oldest supported +# interpreter (requires-python >= 3.9), so the gate fails on any stdlib or +# syntax drift that would not load on 3.9. mypy 2.x dropped the 3.9 +# target, hence the <2 constraint in the typecheck extra above. +python_version = "3.9" +strict = true +files = ["py7zip"] diff --git a/tests/test_characterization_wrapper.py b/tests/test_characterization_wrapper.py index 0be1da4..55e60b8 100644 --- a/tests/test_characterization_wrapper.py +++ b/tests/test_characterization_wrapper.py @@ -1,9 +1,10 @@ """Characterisation of archive execution as it exists today. -``Py7zip.wrapper`` interpolates the binary name, source, destination and -caller-supplied options into a single string and hands it to -``subprocess.run(..., shell=True)``. These tests pin that behaviour from both -directions: +``Py7zip.wrapper`` on the legacy runtime interpolates the binary name, +source, destination and caller-supplied options string into a single string +and hands it to ``subprocess.run(..., shell=True)``. A sequence of options +is the exception: it is forwarded as an argument vector with +``shell=False``. These tests pin that behaviour from both directions: * against a recording process seam, to capture the exact shell string and the ``subprocess.run`` keyword arguments; and @@ -277,6 +278,60 @@ def test_legacy_aliases_forward_the_callers_options(make_wrapper, monkeypatch): ] +def test_sequence_options_are_forwarded_as_an_argument_vector( + make_wrapper, monkeypatch +): + """A sequence is forwarded element by element, not as its Python repr.""" + runner = FakeSubprocess().install(monkeypatch) + + wrapper = make_wrapper() + wrapper.wrapper("src.7z", "out", options=["-y", "-mx=9"], method="decompress") + wrapper.wrapper("src", "dst.7z", options=["-y", "-mx=9"], method="compress") + + assert runner.commands == [ + ("7za", "x", "src.7z", "-oout", "-y", "-mx=9"), + ("7za", "a", "dst.7z", "src", "-y", "-mx=9"), + ] + assert all( + kwargs == {"shell": False, "check": True, "capture_output": True, "text": True} + for kwargs in runner.kwargs + ) + + +@POSIX +def test_sequence_options_reach_the_binary_one_argument_each( + make_wrapper, argv_seen, tmp_path, monkeypatch +): + """Without a shell, a sequence element is never re-interpreted by anything.""" + side_effect = tmp_path / "arbitrary-command-ran" + monkeypatch.setenv("PY7ZIP_SIDE_EFFECT", str(side_effect)) + + wrapper = make_wrapper() + wrapper.wrapper( + "src.7z", + "out", + options=["-y", 'm0=on; : > "$PY7ZIP_SIDE_EFFECT"'], + method="decompress", + ) + + assert argv_seen() == [ + "x", + "src.7z", + "-oout", + "-y", + 'm0=on; : > "$PY7ZIP_SIDE_EFFECT"', + ] + assert not side_effect.exists(), "a sequence element started a shell command" + + +@POSIX +def test_sequence_options_with_spaces_stay_one_argument(make_wrapper, argv_seen): + wrapper = make_wrapper() + wrapper.wrapper("my data", "backup.7z", options=["-mx=9"], method="compress") + + assert argv_seen() == ["a", "backup.7z", "my data", "-mx=9"] + + def test_only_wrapper_honours_options(make_wrapper, monkeypatch): runner = FakeSubprocess().install(monkeypatch) diff --git a/tests/test_default_api.py b/tests/test_default_api.py index e1b1ccc..43dc44a 100644 --- a/tests/test_default_api.py +++ b/tests/test_default_api.py @@ -6,8 +6,36 @@ import pytest +import py7zip.safe as safe_module from py7zip.py7zip import Py7zip -from py7zip.safe import ArchiveRunner +from py7zip.safe import ArchiveResult, ArchiveRunner, ArchiveTraversalError + + +class RecordingSafe: + """Stand-in for ``SafePy7zip`` that records the operation it is handed. + + Only the operation-level entry points the safe wrapper dispatches to + are defined: a regression back to a direct ``SafePy7zip.run`` dispatch + fails loudly here instead of silently bypassing member validation. + """ + + def __init__(self): + self.calls = [] + + def _record(self, operation, source, destination, options=()): + self.calls.append((operation, source, destination, tuple(options))) + return ArchiveResult(command=(), returncode=0, stdout="", stderr="") + + def compress(self, source, destination, options=()): + return self._record("compress", source, destination, options) + + def decompress(self, source, destination, options=()): + return self._record("decompress", source, destination, options) + + +def traversal_listing(member): + """Build the ``l -slt -ba`` output a 7-Zip listing would print.""" + return f"Path = {member}\n" def test_default_constructor_is_offline_and_does_not_download(monkeypatch, tmp_path): @@ -30,6 +58,145 @@ def test_default_aliases_require_individual_option_arguments(tmp_path): wrapper.compress("source", "destination", options="-y") +def test_default_runtime_accepts_individual_option_arguments(tmp_path, monkeypatch): + """The sequence form reaches the safe runtime element by element.""" + wrapper = Py7zip(cache_dir=tmp_path) + recording = RecordingSafe() + monkeypatch.setattr(wrapper, "_safe", recording) + + result = wrapper.compress("src", "dst.7z", options=["-y", "-mx=9"]) + + assert recording.calls == [("compress", "src", "dst.7z", ("-y", "-mx=9"))] + assert result is not None + assert result.success + + +def test_default_runtime_shell_text_never_reaches_parsing_or_execution( + tmp_path, monkeypatch +): + """A non-empty options string raises TypeError before anything runs. + + Audit decision D5 / OVH-16 keeps ``TypeError`` for string options on the + safe path: there is no ``shlex.split`` bridge, so the documented failure + for shell text is ``TypeError`` — never ``shlex``'s undocumented failure + modes, i.e. the silent POSIX-mode corruption of unquoted Windows + backslashes (``"-oC:\\Temp"`` would lose its separator) or the + ``ValueError`` raised on an unbalanced quote. + """ + monkeypatch.setattr( + ArchiveRunner, + "_invoke", + lambda self, command: pytest.fail("a subprocess was started", pytrace=False), + ) + wrapper = Py7zip(binary_path=tmp_path / "missing") + + with pytest.raises(TypeError, match="sequence"): + wrapper.decompress("src.7z", "out", options="-oC:\\Temp") + with pytest.raises(TypeError, match="sequence"): + wrapper.decompress("src.7z", "out", options="-p'unbalanced") + + +def test_default_runtime_empty_string_options_are_empty_arguments( + tmp_path, monkeypatch +): + """The documented ``options = ""`` default forwards no arguments.""" + wrapper = Py7zip(cache_dir=tmp_path) + recording = RecordingSafe() + monkeypatch.setattr(wrapper, "_safe", recording) + + wrapper.extract("src.7z", "out") + + assert recording.calls == [("decompress", "src.7z", "out", ())] + + +@pytest.mark.parametrize( + "member", + [ + "../evil.txt", + "a/../../evil.txt", + "/abs/evil.txt", + "C:/evil.txt", + ], +) +def test_default_wrapper_rejects_traversal_members_before_extraction( + tmp_path, monkeypatch, member +): + """The public safe wrapper extracts only through member validation. + + Regression: ``Py7zip.wrapper`` dispatched the safe runtime through + ``SafePy7zip.run``, bypassing the ``SafePy7zip.decompress`` guard, so + the public ``decompress``/``extract`` entry points could write a + member outside the destination root. The wrapper must route + decompression through the validating path: the archive is listed, a + traversal member raises ``ArchiveTraversalError``, and no extraction + command is ever started. + """ + commands = [] + + def fake_run(command, **kwargs): + commands.append(tuple(command)) + return safe_module.subprocess.CompletedProcess( + command, 0, stdout=traversal_listing(member), stderr="" + ) + + monkeypatch.setattr(safe_module.subprocess, "run", fake_run) + wrapper = Py7zip(binary_path=tmp_path / "7za") + + with pytest.raises(ArchiveTraversalError, match="unsafe archive member"): + wrapper.extract("archive.7z", tmp_path / "out") + + assert commands == [(str(tmp_path / "7za"), "l", "-slt", "-ba", "archive.7z")] + + +def test_default_wrapper_extracts_after_members_validate(tmp_path, monkeypatch): + """A clean member list proceeds to the plain ``x`` extraction argv.""" + commands = [] + + def fake_run(command, **kwargs): + commands.append(tuple(command)) + return safe_module.subprocess.CompletedProcess( + command, 0, stdout=traversal_listing("data/file.txt"), stderr="" + ) + + monkeypatch.setattr(safe_module.subprocess, "run", fake_run) + wrapper = Py7zip(binary_path=tmp_path / "7za") + + result = wrapper.decompress("archive.7z", tmp_path / "out") + + assert result is not None + assert result.success + assert commands == [ + (str(tmp_path / "7za"), "l", "-slt", "-ba", "archive.7z"), + (str(tmp_path / "7za"), "x", "archive.7z", f"-o{tmp_path / 'out'}"), + ] + + +def test_default_wrapper_compress_never_lists_or_validates_members( + tmp_path, monkeypatch +): + """Compression packs local paths directly — no archive listing runs. + + Member validation is a decompression-side property; the compress + command must stay the plain ``a`` invocation with the caller's argv. + """ + commands = [] + + def fake_run(command, **kwargs): + commands.append(tuple(command)) + return safe_module.subprocess.CompletedProcess( + command, 0, stdout="", stderr="" + ) + + monkeypatch.setattr(safe_module.subprocess, "run", fake_run) + wrapper = Py7zip(binary_path=tmp_path / "7za") + + result = wrapper.compress("src dir", "dst.7z", options=["-y"]) + + assert result is not None + assert result.success + assert commands == [(str(tmp_path / "7za"), "a", "dst.7z", "src dir", "-y")] + + def test_get_binary_url_raises_on_safe_runtime(tmp_path): """The safe runtime exposes acquisition, not download URLs.