diff --git a/tests/core/test_binary_version.py b/tests/core/test_binary_version.py index 8eae5254..ce2536f8 100644 --- a/tests/core/test_binary_version.py +++ b/tests/core/test_binary_version.py @@ -1,11 +1,12 @@ -import subprocess +import sys +import time from pathlib import Path -from typing import Any +from typing import Callable, Optional import pytest from unshackle.core import binaries -from unshackle.core.binaries import binary_version, parse_version +from unshackle.core.binaries import binary_version, parse_version, probe_output @pytest.mark.parametrize( @@ -29,33 +30,45 @@ def test_parse_version(text: str, expected: str | None) -> None: assert parse_version(text) == expected +def fake_probe(outputs: dict[str, Optional[str]], calls: list[list[str]]) -> Callable[[Path, list[str]], Optional[str]]: + def probe(path: Path, args: list[str]) -> Optional[str]: + calls.append(args) + return outputs.get(" ".join(args), "") + + return probe + + def test_bare_call_only_for_listed_tools(monkeypatch: pytest.MonkeyPatch) -> None: calls: list[list[str]] = [] + outputs = {"": "MP4 Decrypter - Version 1.4\nusage: tool [options]", "--version": "ERROR: missing output"} + monkeypatch.setattr(binaries, "probe_output", fake_probe(outputs, calls)) - def run(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess: - calls.append(cmd) - out = "MP4 Decrypter - Version 1.4\nusage: tool [options]" if len(cmd) == 1 else "ERROR: missing output" - return subprocess.CompletedProcess(cmd, 1, "", out) - - monkeypatch.setattr(binaries.subprocess, "run", run) - - listed, other = Path("/opt/bin/mp4decrypt"), Path("/opt/bin/othertool") - assert binary_version(listed) == "1.4" - assert calls == [[str(listed)]] + assert binary_version(Path("/opt/bin/mp4decrypt")) == "1.4" + assert calls == [[]] calls.clear() - assert binary_version(other) is None - assert calls == [[str(other), "--version"], [str(other), "-version"]] + assert binary_version(Path("/opt/bin/othertool")) is None + assert calls == [["--version"], ["-version"]] def test_timeout_stops_the_probe(monkeypatch: pytest.MonkeyPatch) -> None: calls: list[list[str]] = [] - - def run(cmd: list[str], **kwargs: Any) -> subprocess.CompletedProcess: - calls.append(cmd) - raise subprocess.TimeoutExpired(cmd, kwargs["timeout"]) - - monkeypatch.setattr(binaries.subprocess, "run", run) + monkeypatch.setattr(binaries, "probe_output", fake_probe({"--version": None}, calls)) assert binary_version(Path("/opt/bin/othertool")) is None - assert len(calls) == 1 + assert calls == [["--version"]] + + +def test_timeout_kills_the_whole_tree(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + monkeypatch.setattr(binaries, "PROBE_TIMEOUT", 0.5) + if sys.platform == "win32": + tool = tmp_path / "wrapper.bat" + tool.write_text("@echo off\r\nping -n 30 127.0.0.1 >nul\r\n") + else: + tool = tmp_path / "wrapper" + tool.write_text("#!/bin/sh\nsleep 30 &\nsleep 30\n") + tool.chmod(0o755) + + start = time.monotonic() + assert probe_output(tool, ["--version"]) is None + assert time.monotonic() - start < 10 diff --git a/unshackle/commands/dl.py b/unshackle/commands/dl.py index 37f49716..1eb202c4 100644 --- a/unshackle/commands/dl.py +++ b/unshackle/commands/dl.py @@ -1371,7 +1371,6 @@ class dl: }, ) - # Log binary versions for diagnostics binary_versions: dict[str, Optional[dict[str, Optional[str]]]] = {} for name, binary in [ ("shaka_packager", binaries.ShakaPackager), @@ -1380,56 +1379,9 @@ class dl: ("ffmpeg", binaries.FFMPEG), ("ffprobe", binaries.FFProbe), ]: - if binary: - version = None - try: - if name == "shaka_packager": - r = subprocess.run( - [str(binary), "--version"], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=5, - ) - version = (r.stdout or r.stderr or "").strip() - elif name in ("ffmpeg", "ffprobe"): - r = subprocess.run( - [str(binary), "-version"], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=5, - ) - version = (r.stdout or "").split("\n")[0].strip() - elif name == "mkvmerge": - r = subprocess.run( - [str(binary), "--version"], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=5, - ) - version = (r.stdout or "").strip() - elif name == "mp4decrypt": - r = subprocess.run( - [str(binary)], - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - timeout=5, - ) - output = (r.stdout or "") + (r.stderr or "") - lines = [line.strip() for line in output.split("\n") if line.strip()] - version = " | ".join(lines[:2]) if lines else None - except Exception: - version = "" - binary_versions[name] = {"path": str(binary), "version": version} - else: - binary_versions[name] = None + binary_versions[name] = ( + {"path": str(binary), "version": binaries.binary_version(binary)} if binary else None + ) self.debug_logger.log( level="DEBUG", diff --git a/unshackle/core/binaries.py b/unshackle/core/binaries.py index 09ea21ab..bb14b555 100644 --- a/unshackle/core/binaries.py +++ b/unshackle/core/binaries.py @@ -1,7 +1,10 @@ +import os import re import shutil +import signal import subprocess # nosec B404 import sys +from contextlib import suppress from pathlib import Path from typing import Optional @@ -108,6 +111,7 @@ VERSION_RE = re.compile(DOTTED_NUMBER) LABELLED_VERSION_RE = re.compile(rf"(?i)\bversion[:\s]+[a-z]?({DOTTED_NUMBER})") VERSION_WORD_RE = re.compile(r"(?i)\bversion\b") BARE_VERSION_TOOLS = {"mp4decrypt"} +PROBE_TIMEOUT = 5.0 def parse_version(text: str) -> Optional[str]: @@ -125,6 +129,39 @@ def parse_version(text: str) -> Optional[str]: return None +def probe_output(path: Path, args: list[str]) -> Optional[str]: + """Run a binary and return its stdout and stderr, or None when it does not start or times out. + + A timeout kills the whole tree: on Windows subprocess.run waits with no limit for a grandchild that holds the pipes. + """ + try: + proc = subprocess.Popen( # nosec B603 + [str(path), *args], + stdin=subprocess.DEVNULL, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + start_new_session=True, + ) + except (OSError, ValueError): + return None + with proc: + try: + out, err = proc.communicate(timeout=PROBE_TIMEOUT) + except subprocess.TimeoutExpired: + if sys.platform == "win32": + subprocess.run(["taskkill", "/F", "/T", "/PID", str(proc.pid)], capture_output=True) # nosec B603 B607 + else: + with suppress(ProcessLookupError): + os.killpg(proc.pid, signal.SIGKILL) + with suppress(subprocess.TimeoutExpired): + proc.communicate(timeout=1) + return None + return f"{out}\n{err}" + + def binary_version(path: Optional[Path]) -> Optional[str]: """Best-effort version probe of a binary. Returns None when nothing is parseable. @@ -134,20 +171,10 @@ def binary_version(path: Optional[Path]) -> Optional[str]: return None bare = path.stem.lower() in BARE_VERSION_TOOLS for args in [[]] if bare else [["--version"], ["-version"]]: - try: - proc = subprocess.run( # nosec B603 - [str(path), *args], - capture_output=True, - stdin=subprocess.DEVNULL, - text=True, - encoding="utf-8", - errors="replace", - timeout=5, - ) - except (OSError, subprocess.SubprocessError, ValueError): + output = probe_output(path, args) + if output is None: return None - version = parse_version(f"{proc.stdout}\n{proc.stderr}") - if version: + if version := parse_version(output): return version return None