diff --git a/docs/dev/service-architecture.md b/docs/dev/service-architecture.md index e38b3379..e83ffdd7 100644 --- a/docs/dev/service-architecture.md +++ b/docs/dev/service-architecture.md @@ -97,7 +97,7 @@ These are abstract. A service will not load unless it defines all three: line, and a REST download job shows it as the job's `notice`. - **Event callbacks**: `on_segment_downloaded`, `on_track_downloaded`, `on_track_decrypted`, `on_track_repacked`, `on_track_multiplex` let a service react to - pipeline events (for example, MonaLisa decrypts each segment in `on_segment_downloaded`). + pipeline events. - **`resolve_server_keys(title)`**: an escape hatch for services whose decryption keys are resolved server-side *ahead of* the normal per-track license flow. Unlike everything else here it is **not declared on the `Service` base class at all**: `dl.py` probes for it with diff --git a/tests/core/test_monalisa_decrypt.py b/tests/core/test_monalisa_decrypt.py new file mode 100644 index 00000000..6d7c005f --- /dev/null +++ b/tests/core/test_monalisa_decrypt.py @@ -0,0 +1,95 @@ +"""MonaLisa's external decrypt tool: a missing tool raises, and a failed run leaves the ciphertext untouched.""" + +import subprocess +from pathlib import Path +from uuid import UUID + +import pytest + +from unshackle.core.drm.monalisa import MonaLisa +from unshackle.core.utils import subprocess as tool_runner + +pytestmark = pytest.mark.unit + + +@pytest.fixture +def track(tmp_path: Path) -> Path: + path = tmp_path / "track.mp4" + path.write_bytes(b"cipher") + return path + + +@pytest.fixture +def make_drm(monkeypatch: pytest.MonkeyPatch): + """Build a MonaLisa with a known key; extract_keys needs a real device file, so it is skipped.""" + monkeypatch.setattr(MonaLisa, "extract_keys", lambda self: None) + + def build(tool: Path) -> MonaLisa: + drm = MonaLisa("ticket", Path("unused.mld"), tool) + drm._kid = UUID(int=1) + drm._key = "00" * 16 + return drm + + return build + + +@pytest.fixture +def tool(tmp_path: Path) -> Path: + path = tmp_path / "decrypt-tool" + path.write_bytes(b"") + return path + + +def fake_tool(monkeypatch: pytest.MonkeyPatch, output: bytes, returncode: int = 0, stderr: bytes = b"") -> None: + """Replace the tool run with one that writes `output` to the `-o` path.""" + + def run(cmd, **kwargs): + Path(cmd[cmd.index("-o") + 1]).write_bytes(output) + return subprocess.CompletedProcess(cmd, returncode, b"", stderr) + + monkeypatch.setattr(tool_runner.subprocess, "run", run) + + +def test_a_missing_tool_raises_and_keeps_the_file(make_drm, track: Path, tmp_path: Path) -> None: + with pytest.raises(RuntimeError, match="tool"): + make_drm(tmp_path / "absent").decrypt(track) + assert track.read_bytes() == b"cipher" + + +@pytest.mark.parametrize( + ("output", "returncode", "stderr"), + [(b"partial", 1, b"bad key"), (b"", 0, b"")], + ids=["non-zero exit", "empty output"], +) +def test_a_failed_run_keeps_the_ciphertext( + make_drm, track: Path, tool: Path, monkeypatch: pytest.MonkeyPatch, output: bytes, returncode: int, stderr: bytes +) -> None: + fake_tool(monkeypatch, output, returncode, stderr) + with pytest.raises(RuntimeError, match=f"failed: {stderr.decode()}"): + make_drm(tool).decrypt(track) + assert track.read_bytes() == b"cipher" + assert not track.with_name("track_decrypted.mp4").exists() + + +def test_a_timeout_keeps_the_ciphertext_and_hides_the_key( + make_drm, track: Path, tool: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + def hang(cmd, **kwargs): + raise subprocess.TimeoutExpired(cmd, kwargs.get("timeout")) + + monkeypatch.setattr(tool_runner.subprocess, "run", hang) + drm = make_drm(tool) + with pytest.raises(RuntimeError, match="timed out") as raised: + drm.decrypt(track) + assert drm.key_pair not in str(raised.value) + assert raised.value.__suppress_context__ + assert track.read_bytes() == b"cipher" + + +def test_a_good_run_replaces_the_file_and_drops_the_temp_output( + make_drm, track: Path, tool: Path, monkeypatch: pytest.MonkeyPatch +) -> None: + fake_tool(monkeypatch, b"plain") + make_drm(tool).decrypt(track) + assert track.read_bytes() == b"plain" + assert not track.with_name("track_decrypted.mp4").exists() diff --git a/unshackle/core/drm/monalisa.py b/unshackle/core/drm/monalisa.py index 32a96bd9..1d4b3f35 100644 --- a/unshackle/core/drm/monalisa.py +++ b/unshackle/core/drm/monalisa.py @@ -8,13 +8,12 @@ from __future__ import annotations import base64 import logging -import subprocess -import sys from pathlib import Path from typing import Any, Optional, Union from uuid import UUID from unshackle.core.cdm.monalisa import MonaLisaCDM +from unshackle.core.utils.subprocess import run_step log = logging.getLogger(__name__) @@ -220,57 +219,22 @@ class MonaLisa: self._execute_decrypt(path, output_path) def _execute_decrypt(self, path: Path, output_path: Optional[Path] = None) -> None: - """Execute external CLI tool via subprocess to decrypt the target file.""" + """Run the external tool into a temporary file, which replaces the target only on success.""" if not path or not path.exists(): raise ValueError(f"Target file does not exist: {path}") if not self._tool_path or not self._tool_path.exists(): - return + raise RuntimeError(f"MonaLisa decryption tool missing: {self._tool_path or 'no tool_path given'}") if not self.key or not self.kid: raise MonaLisa.Exceptions.KeyExtractionFailed("Missing content key for decryption.") target_output = output_path or path - temp_enc = path.with_name(f"{path.name}.temp_enc") - - try: - if path.exists(): - path.replace(temp_enc) - else: - raise FileNotFoundError(f"Target file disappeared: {path}") - - cmd = [ - str(self._tool_path), - "-i", - str(temp_enc), - "-o", - str(target_output), - "-k", - self.key_pair, - ] - - startupinfo = None - if sys.platform == "win32": - startupinfo = subprocess.STARTUPINFO() - startupinfo.dwFlags |= subprocess.STARTF_USESHOWWINDOW - - process = subprocess.run( - cmd, - stdout=subprocess.PIPE, - stderr=subprocess.PIPE, - text=True, - encoding="utf-8", - errors="replace", - startupinfo=startupinfo, - timeout=60, - ) - - if process.returncode != 0 or not target_output.exists(): - raise RuntimeError(f"MonaLisa decryption failed for {path.name}: {process.stderr or 'unknown error'}") - - finally: - if temp_enc.exists(): - try: - temp_enc.unlink() - except Exception: - pass + temp_out = target_output.with_stem(f"{target_output.stem}_decrypted") + run_step( + [self._tool_path, "-i", path, "-o", temp_out, "-k", self.key_pair], + output=temp_out, + label=f"MonaLisa decryption of {path.name}", + timeout=60, + ) + temp_out.replace(target_output) diff --git a/unshackle/core/utils/subprocess.py b/unshackle/core/utils/subprocess.py index 97598c68..f4dbeb6e 100644 --- a/unshackle/core/utils/subprocess.py +++ b/unshackle/core/utils/subprocess.py @@ -187,22 +187,30 @@ def run_step( status: Optional[str] = None, output: Optional[Path] = None, label: str = "subprocess step", + timeout: Optional[float] = None, ) -> bytes: """Operate a CLI step that writes to `output` (when provided). Returns stderr bytes. Raises RuntimeError with the stderr tail when the process exits non-zero, or when - you give `output` and it does not exist or is empty after the step. + you give `output` and it does not exist or is empty after the step. A timeout also + raises RuntimeError. """ if output is not None: output.unlink(missing_ok=True) str_args = [str(a) for a in args] start = time.monotonic() - if status: - with console.status(status, spinner="dots"): - p = subprocess.run(str_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) - else: - p = subprocess.run(str_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE) + try: + if status: + with console.status(status, spinner="dots"): + p = subprocess.run(str_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) + else: + p = subprocess.run(str_args, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=timeout) + except subprocess.TimeoutExpired: + if output is not None: + output.unlink(missing_ok=True) + # from None: the TimeoutExpired text holds the whole command line, which can carry a content key + raise RuntimeError(f"{label} timed out after {timeout}s") from None stderr = p.stderr or b"" bad_output = output is not None and (not output.exists() or output.stat().st_size == 0)