diff --git a/app/audio_recorder.py b/app/audio_recorder.py index cad826a..80b0a6a 100644 --- a/app/audio_recorder.py +++ b/app/audio_recorder.py @@ -5,8 +5,8 @@ Design: - Synchron (kein Thread): der aufrufende Thread blockiert nicht, parec laeuft als Hintergrundprozess. -- Stale-PID-Schutz: abgestuerzte parec-Prozesse werden erkannt - und der PID-File wird bereinigt, bevor der naechste Start versucht wird. +- Only the child started by this instance can be controlled, using a pidfd. + PID files are untrusted status markers, never signal targets. - XDG_RUNTIME_DIR ist Pflicht (kein /tmp-Fallback, sicherheitsrelevant). """ from __future__ import annotations @@ -16,6 +16,7 @@ import shutil import signal import subprocess +import sys import time from pathlib import Path from typing import Optional @@ -26,6 +27,14 @@ _RATE = 16000 _CHANNELS = 1 _LATENCY_MS = 100 +_STOP_TIMEOUT = 2.0 +# The child cannot capture audio until the parent has pinned its identity. EOF +# aborts startup without a signal; exec preserves the PID and its pidfd binding. +_START_GATE = ( + "import os, sys\n" + "if sys.stdin.buffer.read(1) == b'1':\n" + " os.execvp(sys.argv[1], sys.argv[1:])\n" +) class AudioRecorderError(Exception): @@ -63,13 +72,15 @@ def __init__(self, runtime_dir: Optional[Path] = None) -> None: self._pid_file = self._runtime_dir / f"blitztext-linux-{user}.pid" self._wav_file = self._runtime_dir / f"blitztext-linux-{user}.wav" self._parec_err = self._runtime_dir / f"blitztext-linux-{user}.parec.err" + self._process: Optional[subprocess.Popen] = None + self._pidfd: Optional[int] = None # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def is_recording(self) -> bool: - """True, wenn ein parec-Prozess aktiv laeuft (PID-File existiert und Prozess lebt).""" + """True only while this instance's recording child is alive.""" return self._live_pid() is not None def start_recording(self, device: str = "@DEFAULT_SOURCE@") -> Path: @@ -104,9 +115,9 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: # Stale PID guard for interrupted recording processes. self._cleanup_stale_pid() - if self._pid_file.is_file(): + if self._process is not None: raise AudioRecorderError( - "Aufnahme laeuft bereits (PID-File vorhanden). Zuerst stop() aufrufen." + "Aufnahme laeuft bereits. Zuerst stop() aufrufen." ) if not self._pipewire_available(): @@ -115,11 +126,21 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: if not self._device_exists(device): raise AudioRecorderError(f"Mikrofon nicht gefunden: {device!r}") + # Fail before starting capture on kernels without safe process handles. + if not hasattr(os, "pidfd_open") or not hasattr(signal, "pidfd_send_signal"): + raise AudioRecorderError("Sichere Aufnahme-Prozessverwaltung benoetigt Linux pidfd-Unterstuetzung.") + try: + probe_fd = os.pidfd_open(os.getpid()) + os.close(probe_fd) + except OSError as exc: + raise AudioRecorderError("Linux pidfd-Unterstuetzung ist nicht verfuegbar.") from exc + self._parec_err.unlink(missing_ok=True) err_fh = open(self._parec_err, "wb") try: proc = subprocess.Popen( [ + sys.executable, "-I", "-S", "-c", _START_GATE, "parec", f"--device={device}", "--file-format=wav", @@ -130,16 +151,49 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: ], stdout=subprocess.DEVNULL, stderr=err_fh, + stdin=subprocess.PIPE, ) + except OSError as exc: + raise AudioRecorderError(f"Aufnahme-Prozess konnte nicht gestartet werden: {exc}") from exc finally: err_fh.close() - self._pid_file.write_text(f"{proc.pid}\n") + # This local Popen child has not been polled/reaped or exposed to another + # caller yet. Its PID cannot be recycled before pidfd_open pins it. + try: + pidfd = os.pidfd_open(proc.pid) + except OSError as exc: + # EOF ends the gate without starting parec or trusting a numeric PID. + self._process = proc + proc.stdin.close() + try: + proc.wait(timeout=_STOP_TIMEOUT) + except subprocess.TimeoutExpired: + logger.warning("Aufnahme-Startbarriere ist noch nicht beendet (PID=%d).", proc.pid) + else: + self._release_process() + raise AudioRecorderError( + "Aufnahme-Prozess konnte nicht sicher vorbereitet werden; Aufnahme nicht gestartet." + ) from exc + self._process = proc + self._pidfd = pidfd + + try: + self._pid_file.write_text(f"{proc.pid}\n") + # No shell interpolation; all parec arguments cross exec as argv. + with proc.stdin: + proc.stdin.write(b"1") + except OSError as exc: + if not proc.stdin.closed: + proc.stdin.close() + if self._stop_owned_process(): + self._wav_file.unlink(missing_ok=True) + raise AudioRecorderError("Aufnahme-Prozess konnte nicht freigegeben werden.") from exc # Kurz warten und sicherstellen, dass parec nicht sofort abgestuerzt ist time.sleep(0.2) if proc.poll() is not None: - self._pid_file.unlink(missing_ok=True) + self._release_process() self._wav_file.unlink(missing_ok=True) detail = self._read_parec_err() msg = f"parec sofort abgestuerzt: {detail}" if detail else "parec sofort abgestuerzt." @@ -157,20 +211,11 @@ def stop(self) -> Optional[Path]: Raises: AudioRecorderError: Wenn keine Aufnahme lief. """ - self._cleanup_stale_pid() - - if not self._pid_file.is_file(): + if self._live_pid() is None: raise AudioRecorderError("stop() aufgerufen, obwohl keine Aufnahme laeuft.") - pid = self._read_pid() - self._pid_file.unlink(missing_ok=True) - - if pid is not None: - try: - os.kill(pid, signal.SIGTERM) - time.sleep(0.4) # parec Zeit geben, WAV-Header abzuschliessen - except (ProcessLookupError, PermissionError, OSError) as exc: - logger.debug("SIGTERM fehlgeschlagen (PID=%d): %s", pid, exc) + if not self._stop_owned_process(): + raise AudioRecorderError("Aufnahme-Prozess konnte nicht sicher gestoppt werden.") if not self._wav_file.is_file() or self._wav_file.stat().st_size == 0: detail = self._read_parec_err() @@ -188,13 +233,10 @@ def discard(self) -> None: """Bricht die laufende Aufnahme ab, ohne zu transkribieren. WAV-Datei wird geloescht. Kein Fehler wenn keine Aufnahme lief. """ - pid = self._live_pid() - self._pid_file.unlink(missing_ok=True) - if pid is not None: - try: - os.kill(pid, signal.SIGTERM) - except (ProcessLookupError, PermissionError, OSError): - pass + if self._live_pid() is not None and not self._stop_owned_process(): + # Preserve the handle and files rather than pretending capture ended. + logger.warning("Aufnahme konnte nicht sicher verworfen werden.") + return self._wav_file.unlink(missing_ok=True) logger.debug("Aufnahme verworfen.") @@ -203,39 +245,53 @@ def discard(self) -> None: # ------------------------------------------------------------------ def _live_pid(self) -> Optional[int]: - """Gibt PID zurueck wenn Prozess lebt, sonst None. Bereinigt Stale-PID.""" - if not self._pid_file.is_file(): - return None - pid = self._read_pid() - if pid is None: - self._pid_file.unlink(missing_ok=True) - return None - try: - os.kill(pid, 0) - return pid - except OSError: - self._pid_file.unlink(missing_ok=True) - return None + """Return only the owned child PID; never trust or probe a PID file.""" + if self._process is not None and self._process.poll() is None: + return self._process.pid + self._release_process() + return None def _cleanup_stale_pid(self) -> None: - """Entfernt PID-File wenn Prozess nicht mehr laeuft (Stale-PID-Schutz).""" - if not self._pid_file.is_file(): - return - pid = self._read_pid() - if pid is None: - self._pid_file.unlink(missing_ok=True) - return + """Clean up markers without adopting or signalling unowned processes.""" + self._live_pid() + + def _release_process(self) -> None: + """Release the identity handle only after the child has stopped.""" + if self._pidfd is not None: + os.close(self._pidfd) + self._pidfd = None + self._process = None try: - os.kill(pid, 0) # Signal 0 = nur Existenz-Check, kein Kill - except OSError: - logger.debug("Stale PID %d bereinigt.", pid) self._pid_file.unlink(missing_ok=True) + except OSError as exc: + logger.warning("Aufnahme-PID-Marker konnte nicht entfernt werden: %s", exc) - def _read_pid(self) -> Optional[int]: + def _stop_owned_process(self) -> bool: + """Signal the pinned child, even if its numeric PID is later recycled.""" + proc = self._process + if proc is None: + return True + if self._pidfd is None: + logger.warning("Kein verifizierter Prozess-Handle fuer Aufnahme-PID %d.", proc.pid) + return False try: - return int(self._pid_file.read_text().strip()) - except (OSError, ValueError): - return None + try: + signal.pidfd_send_signal(self._pidfd, signal.SIGTERM) + except ProcessLookupError: + pass # The pinned child exited before delivery; no PID fallback. + try: + proc.wait(timeout=_STOP_TIMEOUT) + except subprocess.TimeoutExpired: + try: + signal.pidfd_send_signal(self._pidfd, signal.SIGKILL) + except ProcessLookupError: + pass + proc.wait(timeout=_STOP_TIMEOUT) + except (OSError, subprocess.TimeoutExpired) as exc: + logger.warning("Aufnahme-Prozess konnte nicht gestoppt werden: %s", exc) + return False + self._release_process() + return True def _read_parec_err(self) -> str: try: diff --git a/app/blitztext_linux.py b/app/blitztext_linux.py index 7ecc0b8..86a8205 100644 --- a/app/blitztext_linux.py +++ b/app/blitztext_linux.py @@ -1177,6 +1177,11 @@ def gui_discard(self) -> None: """Laufende Aufnahme verwerfen, ohne zu transkribieren.""" if self.state == "RECORDING": self.audio_recorder.discard() + if self.audio_recorder.is_recording(): + self.show_tray_error(t("error.recording.title"), t("error.recording.stop_failed").format( + error="Aufnahme-Prozess konnte nicht sicher gestoppt werden." + )) + return self._recording_routes_to_compose = False self.current_workflow = None self._set_state("IDLE", "discarded via gui") @@ -1195,8 +1200,8 @@ def _stop_recording_and_process(self) -> None: route_to_compose = ( self._recording_routes_to_compose and self._compose_voice_routing_enabled() ) - self._recording_routes_to_compose = False wav_path = self.audio_recorder.stop() + self._recording_routes_to_compose = False if not wav_path: logger.warning("No audio was recorded") self.show_tray_warning("Blitztext", "Keine Audioaufnahme erfasst.") @@ -1240,6 +1245,8 @@ def _stop_recording_and_process(self) -> None: except AudioRecorderError as e: logger.error("Failed to stop recording: %s", e) self.show_tray_error(t("error.recording.title"), t("error.recording.stop_failed").format(error=e)) + if self.audio_recorder.is_recording(): + return self._recording_routes_to_compose = False self.current_workflow = None self._set_state("IDLE", "recording stop failed") @@ -1504,6 +1511,11 @@ def show_tray_warning(self, title: str, message: str) -> None: def quit_app(self) -> None: logger.info("Quitting application...") self.audio_recorder.discard() + if self.audio_recorder.is_recording(): + self.show_tray_error(t("error.recording.title"), t("error.recording.stop_failed").format( + error="Aufnahme-Prozess konnte nicht sicher gestoppt werden." + )) + return self.stop_hotkey_worker() if self._tts_window is not None: self._tts_window.close() diff --git a/tests/test_audio_recorder.py b/tests/test_audio_recorder.py new file mode 100644 index 0000000..1ab008f --- /dev/null +++ b/tests/test_audio_recorder.py @@ -0,0 +1,356 @@ +"""Recorder ownership regressions; no microphone or foreign process signals.""" +from __future__ import annotations + +import errno +import os +import signal +import subprocess +import sys +from contextlib import nullcontext +from pathlib import Path +from types import SimpleNamespace +from unittest.mock import patch + +import pytest + +from app.audio_recorder import AudioRecorder, AudioRecorderError + + +@pytest.mark.parametrize("pid_text", ["-1", "-123", "0", "1", str(os.getpid()), "999999999", "bad", "", "9" * 100]) +@pytest.mark.parametrize("operation", ["is_recording", "stop", "discard", "_cleanup_stale_pid"]) +def test_pid_file_never_authorizes_a_signal(tmp_path: Path, pid_text: str, operation: str): + recorder = AudioRecorder(tmp_path) + recorder._pid_file.write_text(pid_text) + + # Even signal 0 must not be sent to an unowned PID. Mock the unsafe old + # boundary so the regression is safe when run against the vulnerable code. + with patch("app.audio_recorder.os.kill") as kill, \ + patch("app.audio_recorder.signal.pidfd_send_signal") as send, \ + patch("app.audio_recorder.time.sleep"): + if operation == "stop": + with pytest.raises(AudioRecorderError): + recorder.stop() + else: + result = getattr(recorder, operation)() + if operation == "is_recording": + assert result is False + kill.assert_not_called() + send.assert_not_called() + assert not recorder._pid_file.exists() + + +@pytest.fixture +def recording(tmp_path: Path, monkeypatch): + """Exercise real Popen/pidfd with a disposable, microphone-free parec.""" + bin_dir = tmp_path / "bin" + bin_dir.mkdir() + executable = bin_dir / "parec" + executable.write_text( + f"#!{sys.executable}\n" + "import os, signal, sys, time\n" + "from pathlib import Path\n" + "wav = Path(sys.argv[-1])\n" + "mode = os.environ.get('BT_TEST_PAREC_MODE', 'normal')\n" + "if mode == 'crash':\n" + " sys.stderr.write('isolated recorder failure\\n')\n" + " sys.exit(7)\n" + "def finish(*_):\n" + " wav.write_bytes(b'finished' if mode != 'empty' else b'')\n" + " sys.exit(0)\n" + "signal.signal(signal.SIGTERM, signal.SIG_IGN if mode == 'ignore' else finish)\n" + "wav.write_bytes(b'recording' if mode != 'empty' else b'')\n" + "while True: time.sleep(0.01)\n" + ) + executable.chmod(0o755) + monkeypatch.setenv("PATH", f"{bin_dir}{os.pathsep}{os.environ['PATH']}") + monkeypatch.setattr(AudioRecorder, "_pipewire_available", staticmethod(lambda: True)) + monkeypatch.setattr(AudioRecorder, "_device_exists", staticmethod(lambda _device: True)) + processes = [] + popen = subprocess.Popen + pidfd_open = os.pidfd_open + send_signal = signal.pidfd_send_signal + + def spawn(*args, **kwargs): + proc = popen(*args, **kwargs) + processes.append(proc) + return proc + + def signal_test_child(fd, sig, *args): + # Refuse foreign targets even if a regression accidentally adopts a PID + # file. A dead pidfd reports -1 and can only raise ESRCH, never redirect. + info = Path(f"/proc/self/fdinfo/{fd}").read_text() + target = next(int(line.split()[1]) for line in info.splitlines() if line.startswith("Pid:")) + assert target == -1 or target in {proc.pid for proc in processes}, "refusing foreign signal target" + return send_signal(fd, sig, *args) + + monkeypatch.setattr("app.audio_recorder.subprocess.Popen", spawn) + monkeypatch.setattr("app.audio_recorder.signal.pidfd_send_signal", signal_test_child) + recorder = AudioRecorder(tmp_path) + yield recorder, processes + # Clean up only children created by this fixture, using an identity-bound + # descriptor even if a failing test left the recorder unable to stop them. + for proc in processes: + if proc.poll() is None: + fd = pidfd_open(proc.pid) + try: + send_signal(fd, signal.SIGKILL) + proc.wait(timeout=5) + finally: + os.close(fd) + fd = getattr(recorder, "_pidfd", None) + if fd is not None: + os.close(fd) + + +def test_normal_stop_waits_for_owned_recording_and_closes_pidfd(recording): + recorder, processes = recording + wav = recorder.start() + fd = recorder._pidfd + assert recorder.is_recording() + assert recorder.stop() == wav + assert wav.read_bytes() == b"finished" + assert processes[0].returncode == 0 + assert not recorder.is_recording() + assert not recorder._pid_file.exists() + with pytest.raises(OSError) as error: + os.fstat(fd) + assert error.value.errno == errno.EBADF + + +@pytest.mark.parametrize("pid_text", ["-1", "0", str(os.getpid()), "bad", None]) +@pytest.mark.parametrize("operation", ["stop", "discard"]) +def test_changed_or_missing_pid_file_cannot_redirect_owned_signal(recording, pid_text, operation): + recorder, processes = recording + wav = recorder.start() + if pid_text is None: + recorder._pid_file.unlink() + else: + recorder._pid_file.write_text(pid_text) + with patch("app.audio_recorder.os.kill") as kill: + getattr(recorder, operation)() + kill.assert_not_called() + assert processes[0].returncode == 0 + assert wav.exists() is (operation == "stop") + assert not recorder._pid_file.exists() + assert not recorder.is_recording() + + +def test_second_start_does_not_spawn_another_child(recording): + recorder, processes = recording + recorder.start() + with pytest.raises(AudioRecorderError): + recorder.start() + assert len(processes) == 1 + recorder.discard() + + +def test_foreign_marker_does_not_block_new_owned_recording(recording): + recorder, _ = recording + recorder._pid_file.write_text(str(os.getpid())) + with patch("app.audio_recorder.os.kill") as kill: + recorder.start() + recorder.discard() + kill.assert_not_called() + + +def test_exited_child_is_not_signalled_again(recording): + recorder, processes = recording + recorder.start() + signal.pidfd_send_signal(recorder._pidfd, signal.SIGTERM) + processes[0].wait(timeout=5) + with patch("app.audio_recorder.os.kill") as kill, patch("app.audio_recorder.signal.pidfd_send_signal") as send: + recorder.discard() + kill.assert_not_called() + send.assert_not_called() + assert not recorder.is_recording() + assert not recorder._pid_file.exists() + + +def test_crashed_recording_reports_error_and_releases_identity(recording, monkeypatch): + recorder, processes = recording + monkeypatch.setenv("BT_TEST_PAREC_MODE", "crash") + with pytest.raises(AudioRecorderError, match="isolated recorder failure"): + recorder.start() + assert processes[0].returncode == 7 + assert not recorder.is_recording() + assert recorder._pidfd is None + assert not recorder._pid_file.exists() + assert not recorder._wav_file.exists() + + +def test_empty_recording_returns_none_after_stopping_owned_child(recording, monkeypatch): + recorder, processes = recording + monkeypatch.setenv("BT_TEST_PAREC_MODE", "empty") + recorder.start() + assert recorder.stop() is None + assert processes[0].returncode == 0 + assert not recorder._wav_file.exists() + + +def test_unresponsive_child_is_killed_only_through_owned_pidfd(recording, monkeypatch): + recorder, processes = recording + monkeypatch.setenv("BT_TEST_PAREC_MODE", "ignore") + recorder.start() + with patch("app.audio_recorder.os.kill") as kill: + recorder.discard() + kill.assert_not_called() + assert processes[0].returncode == -signal.SIGKILL + assert not recorder.is_recording() + + +def test_unsupported_pidfd_prevents_spawning_recorder(recording): + recorder, processes = recording + with patch("app.audio_recorder.os.pidfd_open", side_effect=OSError(errno.ENOSYS, "unsupported")), patch("app.audio_recorder.os.kill") as kill: + with pytest.raises(AudioRecorderError): + recorder.start() + kill.assert_not_called() + assert processes == [] + + +def test_spawn_failure_reports_recorder_error_without_retaining_identity(recording): + recorder, processes = recording + with patch("app.audio_recorder.subprocess.Popen", side_effect=FileNotFoundError("isolated spawn failure")): + with pytest.raises(AudioRecorderError): + recorder.start() + assert processes == [] + assert not recorder.is_recording() + assert recorder._pidfd is None + assert not recorder._pid_file.exists() + + +def test_child_exit_between_check_and_delivery_cannot_redirect_signal(recording): + recorder, processes = recording + wav = recorder.start() + send_signal = signal.pidfd_send_signal + + def exit_before_delivery(fd, sig): + # Reap only this test's child after the recorder's liveness check. + # The descriptor stays bound to the dead child, never a reused PID. + send_signal(fd, signal.SIGTERM) + processes[0].wait(timeout=5) + return send_signal(fd, sig) + + with patch("app.audio_recorder.signal.pidfd_send_signal", side_effect=exit_before_delivery), patch("app.audio_recorder.os.kill") as kill: + assert recorder.stop() == wav + kill.assert_not_called() + assert not recorder.is_recording() + + +def test_pid_marker_write_failure_stops_only_owned_child(recording, monkeypatch): + recorder, processes = recording + write_text = Path.write_text + + def fail_marker_write(path, *args, **kwargs): + if path == recorder._pid_file: + raise PermissionError("isolated marker failure") + return write_text(path, *args, **kwargs) + + monkeypatch.setattr(Path, "write_text", fail_marker_write) + with patch("app.audio_recorder.os.kill") as kill: + with pytest.raises(AudioRecorderError): + recorder.start() + kill.assert_not_called() + assert processes[0].returncode is not None + assert not recorder.is_recording() + + +def test_failed_identity_acquisition_never_falls_back_to_numeric_pid(recording): + recorder, processes = recording + pidfd_open = os.pidfd_open + + def reject_child(pid, flags=0): + if pid == os.getpid(): + return pidfd_open(pid, flags) + raise PermissionError("isolated identity failure") + + with patch("app.audio_recorder.os.pidfd_open", side_effect=reject_child), patch("app.audio_recorder.os.kill") as kill: + with pytest.raises(AudioRecorderError): + recorder.start() + assert processes[0].wait(timeout=2) == 0 + with pytest.raises(AudioRecorderError): + recorder.stop() + recorder.discard() + kill.assert_not_called() + assert not recorder.is_recording() + assert not recorder._wav_file.exists() + + +def test_signal_failure_preserves_owned_process_for_retry(recording): + recorder, processes = recording + recorder.start() + with patch("app.audio_recorder.signal.pidfd_send_signal", side_effect=PermissionError("denied")), patch("app.audio_recorder.os.kill") as kill: + with pytest.raises(AudioRecorderError): + recorder.stop() + recorder.discard() + kill.assert_not_called() + assert processes[0].poll() is None + assert recorder.is_recording() + recorder.discard() + assert not recorder.is_recording() + + +@pytest.mark.parametrize("operation,denied", [ + ("gui_discard", False), ("gui_discard", True), + ("_stop_recording_and_process", True), + ("quit_app", False), ("quit_app", True), +]) +def test_application_preserves_state_until_recording_has_stopped(recording, operation, denied): + from app.blitztext_linux import BlitztextApp + + recorder, processes = recording + recorder.start() + errors = [] + shutdown = [] + controller = SimpleNamespace( + audio_recorder=recorder, + state="RECORDING", + current_workflow="isolated-workflow", + _recording_routes_to_compose=True, + _compose_voice_routing_enabled=lambda: True, + show_tray_error=lambda *args: errors.append(args), + stop_hotkey_worker=lambda: shutdown.append("hotkeys"), + _tts_window=None, _history_panel=None, _compose_window=None, _main_window=None, + tray_icon=SimpleNamespace(hide=lambda: shutdown.append("tray")), + app=SimpleNamespace(quit=lambda: shutdown.append("quit")), + ) + controller._set_state = lambda state, _reason: setattr(controller, "state", state) + failure = patch("app.audio_recorder.signal.pidfd_send_signal", side_effect=PermissionError("denied")) if denied else nullcontext() + with failure: + getattr(BlitztextApp, operation)(controller) + if denied: + assert processes[0].poll() is None + assert controller.state == "RECORDING" + assert controller.current_workflow == "isolated-workflow" + assert controller._recording_routes_to_compose is True + assert errors + assert shutdown == [] + else: + assert processes[0].returncode == 0 + assert not errors + if operation == "quit_app": + assert shutdown == ["hotkeys", "tray", "quit"] + else: + assert controller.state == "IDLE" + assert controller.current_workflow is None + assert controller._recording_routes_to_compose is False + + +def test_marker_cleanup_failure_does_not_hide_successful_process_stop(recording, monkeypatch): + recorder, processes = recording + wav = recorder.start() + fd = recorder._pidfd + unlink = Path.unlink + + def deny_marker_unlink(path, *args, **kwargs): + if path == recorder._pid_file: + raise PermissionError("isolated cleanup failure") + return unlink(path, *args, **kwargs) + + monkeypatch.setattr(Path, "unlink", deny_marker_unlink) + assert recorder.stop() == wav + assert processes[0].returncode == 0 + assert not recorder.is_recording() + assert recorder._pidfd is None + with pytest.raises(OSError) as error: + os.fstat(fd) + assert error.value.errno == errno.EBADF