diff --git a/app/audio_recorder.py b/app/audio_recorder.py index 80b0a6a..f007b24 100644 --- a/app/audio_recorder.py +++ b/app/audio_recorder.py @@ -15,8 +15,10 @@ import os import shutil import signal +import stat import subprocess import sys +import tempfile import time from pathlib import Path from typing import Optional @@ -57,23 +59,174 @@ def __init__(self, runtime_dir: Optional[Path] = None) -> None: runtime_dir: Pfad fuer PID- und WAV-Dateien. Standard: $XDG_RUNTIME_DIR (Pflicht). """ - if runtime_dir is not None: - self._runtime_dir = runtime_dir - else: - rt = os.environ.get("XDG_RUNTIME_DIR", "") - if not rt or not Path(rt).is_dir(): - raise AudioRecorderError( - "XDG_RUNTIME_DIR ist nicht gesetzt oder existiert nicht. " - "AudioRecorder kann nicht initialisiert werden." - ) - self._runtime_dir = Path(rt) - - user = os.environ.get("USER", "user") - 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" + rt = runtime_dir if runtime_dir is not None else os.environ.get("XDG_RUNTIME_DIR", "") + if not rt: + raise AudioRecorderError("XDG_RUNTIME_DIR ist nicht gesetzt.") + self._runtime_dir = Path(rt) self._process: Optional[subprocess.Popen] = None self._pidfd: Optional[int] = None + self._runtime_fd: int | None = None + self._job_fd: int | None = None + self._new_job() + + @staticmethod + def _open_runtime(path: Path) -> int: + """Walk an absolute path without following links, then check privacy. + + Parents must be owned by root or this effective UID and not writable by + others. Root-owned sticky directories (e.g. /tmp in isolated tests) are + safe parents: other users cannot rename our private child directory. + """ + if not path.is_absolute() or ".." in path.parts: + raise AudioRecorderError("Recorder-Runtime muss ein absoluter Pfad ohne '..' sein.") + flags = os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC + fd = os.open("/", flags) + try: + for component in path.parts[1:]: + parent = os.fstat(fd) + sticky_root = parent.st_uid == 0 and bool(parent.st_mode & stat.S_ISVTX) + if parent.st_uid not in (0, os.geteuid()) or (parent.st_mode & 0o022 and not sticky_root): + raise AudioRecorderError("Unsicheres Elternverzeichnis der Recorder-Runtime.") + child = os.open(component, flags, dir_fd=fd) + os.close(fd) + fd = child + info = os.fstat(fd) + if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: + raise AudioRecorderError("Recorder-Runtime muss dem Benutzer gehoeren und Rechte 0700 haben.") + return fd + except BaseException: + os.close(fd) + raise + + def _new_job(self) -> None: + """Create a unique private job through a verified directory handle.""" + runtime_fd = None + job_fd = None + created = None + try: + runtime_fd = self._open_runtime(self._runtime_dir) + # The procfs path pins creation to the checked inode, including when + # a pathname is renamed during startup. mkdtemp uses atomic mkdir. + created = Path(tempfile.mkdtemp(prefix="blitztext-recording-", dir=f"/proc/self/fd/{runtime_fd}")) + job_fd = os.open(created.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=runtime_fd) + os.fchmod(job_fd, 0o700) + except OSError as exc: + if job_fd is not None: + os.close(job_fd) + if runtime_fd is not None: + try: + if created is not None: + os.rmdir(created.name, dir_fd=runtime_fd) + finally: + os.close(runtime_fd) + raise AudioRecorderError("Sichere Recorder-Runtime konnte nicht erstellt werden.") from exc + except BaseException: + if runtime_fd is not None: + os.close(runtime_fd) + raise + self._runtime_fd = runtime_fd + self._job_fd = job_fd + self._job_name = created.name + self._job_dir = self._runtime_dir / self._job_name + self._pid_file = self._job_dir / "recorder.pid" + self._wav_file = self._job_dir / "recording.wav" + self._parec_err = self._job_dir / "parec.err" + + def _check_job(self) -> None: + """Recheck permissions and pathname identity before handing out paths.""" + try: + fd = self._open_runtime(self._runtime_dir) + try: + current = os.fstat(fd) + pinned = os.fstat(self._runtime_fd) + if (current.st_dev, current.st_ino) != (pinned.st_dev, pinned.st_ino): + raise AudioRecorderError("Recorder-Runtime wurde ersetzt.") + job = os.stat(self._job_name, dir_fd=fd, follow_symlinks=False) + pinned_job = os.fstat(self._job_fd) + if ( + (job.st_dev, job.st_ino) != (pinned_job.st_dev, pinned_job.st_ino) + or not stat.S_ISDIR(job.st_mode) + or job.st_uid != os.geteuid() + or stat.S_IMODE(job.st_mode) != 0o700 + ): + raise AudioRecorderError("Recorder-Jobverzeichnis ist unsicher oder wurde ersetzt.") + finally: + os.close(fd) + except OSError as exc: + raise AudioRecorderError("Recorder-Runtime konnte nicht sicher geprueft werden.") from exc + + def _create_file(self, name: str): + fd = os.open(name, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW | os.O_CLOEXEC, 0o600, dir_fd=self._job_fd) + try: + # Also retain 0600 under a restrictive caller umask. + os.fchmod(fd, 0o600) + return os.fdopen(fd, "wb") + except BaseException: + os.close(fd) + raise + + def _unlink_file(self, name: str) -> None: + if self._job_fd is None: + return + try: + os.unlink(name, dir_fd=self._job_fd) + except FileNotFoundError: + pass + + def _finish_job(self, *, keep_wav: bool = False) -> None: + """Never remove a successful WAV: its consumer owns that lifecycle.""" + if self._job_fd is None: + return + try: + for name in ("recorder.pid", "parec.err") + (() if keep_wav else ("recording.wav",)): + self._unlink_file(name) + if not keep_wav: + entry = os.stat(self._job_name, dir_fd=self._runtime_fd, follow_symlinks=False) + pinned = os.fstat(self._job_fd) + if (entry.st_dev, entry.st_ino) == (pinned.st_dev, pinned.st_ino): + os.rmdir(self._job_name, dir_fd=self._runtime_fd) + except OSError as exc: + logger.warning("Recorder-Job konnte nicht vollstaendig entfernt werden: %s", exc) + finally: + self._close_job_handles() + + def _close_job_handles(self) -> None: + for attr in ("_job_fd", "_runtime_fd"): + fd = getattr(self, attr, None) + if fd is not None: + os.close(fd) + setattr(self, attr, None) + + def __del__(self) -> None: + # Successful stop has already detached the job, preserving its WAV. + # Remove unused jobs, but never unlink files of a retained child. + if getattr(self, "_job_fd", None) is not None and getattr(self, "_process", None) is None: + self._finish_job() + else: + self._close_job_handles() + + @classmethod + def cleanup_recording(cls, wav: Path) -> None: + """Worker-only cleanup after transcription, independent of later jobs.""" + runtime_fd = cls._open_runtime(wav.parent.parent) + job_fd = None + try: + if wav.name != "recording.wav" or not wav.parent.name.startswith("blitztext-recording-"): + raise AudioRecorderError("Ungueltiger Recorder-Jobpfad.") + job_fd = os.open(wav.parent.name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | os.O_CLOEXEC, dir_fd=runtime_fd) + info = os.fstat(job_fd) + if info.st_uid != os.geteuid() or stat.S_IMODE(info.st_mode) != 0o700: + raise AudioRecorderError("Unsicheres Recorder-Jobverzeichnis.") + for name in ("recording.wav", "recorder.pid", "parec.err"): + try: + os.unlink(name, dir_fd=job_fd) + except FileNotFoundError: + pass + os.rmdir(wav.parent.name, dir_fd=runtime_fd) + finally: + if job_fd is not None: + os.close(job_fd) + os.close(runtime_fd) # ------------------------------------------------------------------ # Public API @@ -135,8 +288,19 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: 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") + if self._job_fd is None: + self._new_job() + self._check_job() + try: + wav_fh = self._create_file("recording.wav") + try: + err_fh = self._create_file("parec.err") + except BaseException: + wav_fh.close() + raise + except OSError as exc: + self._finish_job() + raise AudioRecorderError("Recorder-Dateien konnten nicht sicher erstellt werden.") from exc try: proc = subprocess.Popen( [ @@ -147,16 +311,19 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: f"--rate={_RATE}", f"--channels={_CHANNELS}", f"--latency-msec={_LATENCY_MS}", - str(self._wav_file), + f"/proc/self/fd/{wav_fh.fileno()}", ], stdout=subprocess.DEVNULL, stderr=err_fh, stdin=subprocess.PIPE, + pass_fds=(wav_fh.fileno(),), ) except OSError as exc: + self._finish_job() raise AudioRecorderError(f"Aufnahme-Prozess konnte nicht gestartet werden: {exc}") from exc finally: err_fh.close() + wav_fh.close() # 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. @@ -172,6 +339,7 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: logger.warning("Aufnahme-Startbarriere ist noch nicht beendet (PID=%d).", proc.pid) else: self._release_process() + self._finish_job() raise AudioRecorderError( "Aufnahme-Prozess konnte nicht sicher vorbereitet werden; Aufnahme nicht gestartet." ) from exc @@ -179,7 +347,8 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: self._pidfd = pidfd try: - self._pid_file.write_text(f"{proc.pid}\n") + with self._create_file("recorder.pid") as marker: + marker.write(f"{proc.pid}\n".encode("ascii")) # No shell interpolation; all parec arguments cross exec as argv. with proc.stdin: proc.stdin.write(b"1") @@ -187,18 +356,25 @@ def start(self, device: str = "@DEFAULT_SOURCE@") -> Path: if not proc.stdin.closed: proc.stdin.close() if self._stop_owned_process(): - self._wav_file.unlink(missing_ok=True) + self._finish_job() 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._release_process() - self._wav_file.unlink(missing_ok=True) detail = self._read_parec_err() + self._finish_job() msg = f"parec sofort abgestuerzt: {detail}" if detail else "parec sofort abgestuerzt." raise AudioRecorderError(msg) + try: + self._check_job() + except AudioRecorderError: + if self._stop_owned_process(): + self._finish_job() + raise + logger.debug("Aufnahme gestartet: PID=%d device=%s", proc.pid, device) return self._wav_file @@ -217,16 +393,27 @@ def stop(self) -> Optional[Path]: 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: + try: + self._check_job() + info = os.stat("recording.wav", dir_fd=self._job_fd, follow_symlinks=False) + if ( + not stat.S_ISREG(info.st_mode) + or info.st_uid != os.geteuid() + or info.st_nlink != 1 + or stat.S_IMODE(info.st_mode) != 0o600 + ): + raise AudioRecorderError("Unsichere WAV-Datei.") + except (OSError, AudioRecorderError) as exc: + self._finish_job() + raise AudioRecorderError("WAV-Datei konnte nicht sicher geprueft werden.") from exc + if info.st_size == 0: detail = self._read_parec_err() - self._wav_file.unlink(missing_ok=True) - if detail: - logger.warning("Kein Audio aufgezeichnet: %s", detail) - else: - logger.warning("Kein Audio aufgezeichnet (leere WAV-Datei).") + self._finish_job() + logger.warning("Kein Audio aufgezeichnet: %s", detail or "leere WAV-Datei") return None - logger.debug("Aufnahme gestoppt: %s (%d Bytes)", self._wav_file, self._wav_file.stat().st_size) + logger.debug("Aufnahme gestoppt: %s (%d Bytes)", self._job_name, info.st_size) + self._finish_job(keep_wav=True) return self._wav_file def discard(self) -> None: @@ -237,7 +424,7 @@ def discard(self) -> None: # 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) + self._finish_job() logger.debug("Aufnahme verworfen.") # ------------------------------------------------------------------ @@ -248,7 +435,10 @@ def _live_pid(self) -> Optional[int]: """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 + had_process = self._process is not None self._release_process() + if had_process: + self._finish_job() return None def _cleanup_stale_pid(self) -> None: @@ -262,7 +452,7 @@ def _release_process(self) -> None: self._pidfd = None self._process = None try: - self._pid_file.unlink(missing_ok=True) + self._unlink_file("recorder.pid") except OSError as exc: logger.warning("Aufnahme-PID-Marker konnte nicht entfernt werden: %s", exc) @@ -294,14 +484,19 @@ def _stop_owned_process(self) -> bool: return True def _read_parec_err(self) -> str: + if self._job_fd is None: + return "" try: - if self._parec_err.is_file() and self._parec_err.stat().st_size > 0: - text = self._parec_err.read_text(errors="replace").strip() + fd = os.open("parec.err", os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK | os.O_CLOEXEC, dir_fd=self._job_fd) + with os.fdopen(fd, "rb") as stream: + info = os.fstat(stream.fileno()) + if not stat.S_ISREG(info.st_mode) or info.st_uid != os.geteuid() or info.st_nlink != 1: + return "" + text = stream.read(4096).decode(errors="replace").strip() first = text.splitlines()[0] if text else "" return (first[:157] + "...") if len(first) > 160 else first except OSError: - pass - return "" + return "" @staticmethod def _pipewire_available() -> bool: diff --git a/app/blitztext_linux.py b/app/blitztext_linux.py index 86a8205..8aba4ad 100644 --- a/app/blitztext_linux.py +++ b/app/blitztext_linux.py @@ -11,6 +11,7 @@ import os import sys import time +from collections.abc import Callable from pathlib import Path from typing import Optional @@ -674,10 +675,12 @@ def __init__( paste_service: PasteService, custom_terms: Optional[list[str]] = None, route_to_compose: bool = False, + cleanup_wav: Callable[[], None] | None = None, ) -> None: super().__init__() self.signals = _WorkerSignals() self.wav_file = wav_file + self._cleanup_wav = cleanup_wav self.model = model self.language = language self.backend = backend @@ -732,9 +735,11 @@ def run(self) -> None: finally: # Clean up WAV file try: - if self.wav_file.is_file(): - self.wav_file.unlink() - except OSError as exc: + if self._cleanup_wav is not None: + self._cleanup_wav() + else: + self.wav_file.unlink(missing_ok=True) + except (OSError, AudioRecorderError) as exc: logger.warning("Failed to delete temp WAV file %s: %s", self.wav_file, exc) self._emit("finished", self) @@ -1227,6 +1232,7 @@ def _stop_recording_and_process(self) -> None: paste_service=self.paste_service, custom_terms=self.config.custom_terms, route_to_compose=route_to_compose, + cleanup_wav=lambda: AudioRecorder.cleanup_recording(wav_path), ) worker.signals.status_changed.connect(self._on_worker_status_changed) diff --git a/tests/test_audio_recorder.py b/tests/test_audio_recorder.py index 1ab008f..702bb67 100644 --- a/tests/test_audio_recorder.py +++ b/tests/test_audio_recorder.py @@ -238,14 +238,14 @@ def exit_before_delivery(fd, sig): def test_pid_marker_write_failure_stops_only_owned_child(recording, monkeypatch): recorder, processes = recording - write_text = Path.write_text + create_file = recorder._create_file - def fail_marker_write(path, *args, **kwargs): - if path == recorder._pid_file: + def fail_marker_write(name, *args, **kwargs): + if name == "recorder.pid": raise PermissionError("isolated marker failure") - return write_text(path, *args, **kwargs) + return create_file(name, *args, **kwargs) - monkeypatch.setattr(Path, "write_text", fail_marker_write) + monkeypatch.setattr(recorder, "_create_file", fail_marker_write) with patch("app.audio_recorder.os.kill") as kill: with pytest.raises(AudioRecorderError): recorder.start() @@ -339,14 +339,14 @@ def test_marker_cleanup_failure_does_not_hide_successful_process_stop(recording, recorder, processes = recording wav = recorder.start() fd = recorder._pidfd - unlink = Path.unlink + unlink = os.unlink def deny_marker_unlink(path, *args, **kwargs): - if path == recorder._pid_file: + if path == "recorder.pid" and kwargs.get("dir_fd") == recorder._job_fd: raise PermissionError("isolated cleanup failure") return unlink(path, *args, **kwargs) - monkeypatch.setattr(Path, "unlink", deny_marker_unlink) + monkeypatch.setattr("app.audio_recorder.os.unlink", deny_marker_unlink) assert recorder.stop() == wav assert processes[0].returncode == 0 assert not recorder.is_recording() @@ -354,3 +354,289 @@ def deny_marker_unlink(path, *args, **kwargs): with pytest.raises(OSError) as error: os.fstat(fd) assert error.value.errno == errno.EBADF + + +@pytest.mark.parametrize("mode", [0o777, 0o770, 0o755, 0o700 | 0o2000]) +def test_runtime_requires_private_permissions(tmp_path, mode): + runtime = tmp_path / "runtime" + runtime.mkdir(mode=mode) + runtime.chmod(mode) + with pytest.raises(AudioRecorderError): + AudioRecorder(runtime) + assert list(runtime.iterdir()) == [] + + +def test_relative_runtime_is_rejected(tmp_path, monkeypatch): + monkeypatch.chdir(tmp_path) + Path("runtime").mkdir(mode=0o700) + with pytest.raises(AudioRecorderError): + AudioRecorder(Path("runtime")) + + +@pytest.mark.parametrize("ancestor", [False, True]) +def test_symlink_in_runtime_path_is_rejected(tmp_path, ancestor): + real = tmp_path / "real" + real.mkdir(mode=0o700) + (real / "runtime").mkdir(mode=0o700) + link = tmp_path / "link" + link.symlink_to(real if ancestor else real / "runtime", target_is_directory=True) + with pytest.raises(AudioRecorderError): + AudioRecorder(link / "runtime" if ancestor else link) + assert list((real / "runtime").iterdir()) == [] + + +def test_runtime_with_foreign_owner_is_rejected(tmp_path, monkeypatch): + # No chown or access to another user's directory is needed. + monkeypatch.setattr("app.audio_recorder.os.geteuid", lambda: os.getuid() + 1) + with pytest.raises(AudioRecorderError): + AudioRecorder(tmp_path) + + +def test_runtime_below_nonsticky_writable_parent_is_rejected(tmp_path): + parent = tmp_path / "shared" + parent.mkdir(mode=0o777) + parent.chmod(0o777) + runtime = parent / "runtime" + runtime.mkdir(mode=0o700) + with pytest.raises(AudioRecorderError): + AudioRecorder(runtime) + + +def test_runtime_permissions_are_rechecked_before_start(recording): + recorder, processes = recording + recorder._runtime_dir.chmod(0o777) + try: + with pytest.raises(AudioRecorderError): + recorder.start() + assert not processes + finally: + recorder._runtime_dir.chmod(0o700) + + +def test_user_cannot_redirect_recording_and_files_are_private(recording, monkeypatch): + recorder, _ = recording + monkeypatch.setenv("USER", "../../isolated-target") + previous_umask = os.umask(0) + try: + wav = recorder.start() + finally: + os.umask(previous_umask) + assert wav.parent != recorder._runtime_dir + assert wav.parent.parent == recorder._runtime_dir + assert wav.parent.stat().st_mode & 0o7777 == 0o700 + for path in (wav, recorder._pid_file, recorder._parec_err): + assert path.stat().st_mode & 0o7777 == 0o600 + recorder.discard() + assert not wav.parent.exists() + + +@pytest.mark.parametrize("kind", ["symlink", "hardlink", "regular"]) +def test_preplanted_error_file_never_overwrites_target(recording, tmp_path, kind): + recorder, processes = recording + target = tmp_path / "isolated-target" + target.write_bytes(b"keep me") + if kind == "symlink": + recorder._parec_err.symlink_to(target) + elif kind == "hardlink": + os.link(target, recorder._parec_err) + else: + recorder._parec_err.write_bytes(b"keep me") + with pytest.raises(AudioRecorderError): + recorder.start() + assert not processes + assert target.read_bytes() == b"keep me" + + +def test_successful_wav_survives_discard_and_next_recording(recording): + recorder, _ = recording + first = recorder.start() + assert recorder.stop() == first + recorder.discard() + assert first.read_bytes() == b"finished" + second = recorder.start() + assert second != first + recorder.discard() + assert first.read_bytes() == b"finished" + assert not second.parent.exists() + + +@pytest.mark.parametrize("mode", ["normal", "empty", "crash"]) +def test_failed_and_discarded_jobs_do_not_leave_files(recording, monkeypatch, mode): + recorder, _ = recording + monkeypatch.setenv("BT_TEST_PAREC_MODE", mode) + if mode == "crash": + with pytest.raises(AudioRecorderError): + recorder.start() + job = recorder._wav_file.parent + else: + job = recorder.start().parent + if mode == "empty": + assert recorder.stop() is None + else: + recorder.discard() + assert not job.exists() + + +@pytest.mark.parametrize("outcome", ["success", "silence", "error"]) +def test_worker_keeps_wav_until_transcription_finishes_then_removes_job(recording, outcome): + from app.blitztext_linux import _TranscribeWorker + from app.llm_service import WorkflowType + + recorder, _ = recording + wav = recorder.start() + recorder.stop() + + observed = [] + + def transcribe(**kwargs): + observed.append(kwargs["wav_file"].read_bytes()) + recorder.discard() + observed.append(wav.read_bytes()) + if outcome == "error": + raise RuntimeError("isolated transcription failure") + return "" if outcome == "silence" else "isolated transcript" + + worker = _TranscribeWorker( + wav_file=wav, model="base", language="de", backend="openai-whisper", + workflow=WorkflowType.TRANSCRIPTION, llm_service=None, autopaste=False, + paste_service=SimpleNamespace(clipboard_only=lambda _: None), + cleanup_wav=lambda: AudioRecorder.cleanup_recording(wav), + ) + with patch("app.blitztext_linux.transcribe", side_effect=transcribe): + worker.run() + assert observed == [b"finished", b"finished"] + assert not wav.parent.exists() + + +@pytest.mark.parametrize("filename", ["recording.wav", "recorder.pid", "parec.err"]) +def test_symlinks_in_job_cannot_overwrite_isolated_target(recording, tmp_path, filename): + recorder, processes = recording + target = tmp_path / "isolated-target" + target.write_bytes(b"untouched") + path = recorder._wav_file.parent / filename + path.symlink_to(target) + if filename == "recorder.pid": + # Simulate a link planted after stale-marker cleanup and before create. + cleanup = recorder._cleanup_stale_pid + def plant(): + cleanup() + path.symlink_to(target) + with patch.object(recorder, "_cleanup_stale_pid", side_effect=plant), pytest.raises(AudioRecorderError): + recorder.start() + else: + with pytest.raises(AudioRecorderError): + recorder.start() + assert target.read_bytes() == b"untouched" + assert all(proc.poll() is not None for proc in processes) + + +def test_successful_job_cleanup_does_not_affect_active_recording(recording): + recorder, _ = recording + first = recorder.start() + recorder.stop() + second = recorder.start() + AudioRecorder.cleanup_recording(first) + assert not first.parent.exists() + assert recorder.is_recording() + assert second.exists() + recorder.discard() + + +@pytest.mark.parametrize("operation", ["chmod", "fdopen"]) +def test_file_preparation_failure_does_not_leak_descriptors(recording, operation): + recorder, _ = recording + before = len(list(Path("/proc/self/fd").iterdir())) + boundary = "os.fchmod" if operation == "chmod" else "os.fdopen" + with patch(f"app.audio_recorder.{boundary}", side_effect=PermissionError("isolated file setup failure")), pytest.raises(AudioRecorderError): + recorder.start() + # The two directory descriptors are released when startup fails. + assert len(list(Path("/proc/self/fd").iterdir())) == before - 2 + assert not recorder._wav_file.parent.exists() + + +def test_job_preparation_failure_does_not_leave_directory(tmp_path): + with patch("app.audio_recorder.os.fchmod", side_effect=PermissionError("isolated job setup failure")), pytest.raises(AudioRecorderError): + AudioRecorder(tmp_path) + assert list(tmp_path.iterdir()) == [] + + +@pytest.mark.parametrize("check_status", [False, True]) +def test_new_start_cleans_job_of_unexpectedly_exited_child(recording, check_status): + recorder, processes = recording + first = recorder.start() + signal.pidfd_send_signal(recorder._pidfd, signal.SIGTERM) + processes[0].wait(timeout=5) + if check_status: + assert not recorder.is_recording() + second = recorder.start() + assert first != second + assert not first.parent.exists() + assert recorder.is_recording() + recorder.discard() + + +def test_job_swap_during_start_cannot_redirect_capture(recording, tmp_path): + recorder, processes = recording + job = recorder._wav_file.parent + moved = tmp_path / "moved-private-job" + victim = tmp_path / "isolated-victim" + victim.mkdir(mode=0o700) + target = victim / "recording.wav" + target.write_bytes(b"untouched") + spawn = subprocess.Popen + + def swap_then_spawn(*args, **kwargs): + job.rename(moved) + job.symlink_to(victim, target_is_directory=True) + return spawn(*args, **kwargs) + + with patch("app.audio_recorder.subprocess.Popen", side_effect=swap_then_spawn), pytest.raises(AudioRecorderError): + recorder.start() + assert target.read_bytes() == b"untouched" + assert all(proc.poll() is not None for proc in processes) + assert not recorder.is_recording() + + +def test_worker_cleanup_refuses_symlink_job_directory(recording, tmp_path): + recorder, _ = recording + wav = recorder.start() + recorder.stop() + victim = tmp_path / "isolated-victim" + victim.mkdir(mode=0o700) + target = victim / "recording.wav" + target.write_bytes(b"untouched") + wav.parent.rename(tmp_path / "moved-private-job") + wav.parent.symlink_to(victim, target_is_directory=True) + with pytest.raises((OSError, AudioRecorderError)): + AudioRecorder.cleanup_recording(wav) + assert target.read_bytes() == b"untouched" + + +def test_stop_rejects_wav_with_nonprivate_permissions(recording): + recorder, processes = recording + wav = recorder.start() + wav.chmod(0o644) + with pytest.raises(AudioRecorderError, match="WAV-Datei"): + recorder.stop() + assert processes[0].poll() is not None + assert recorder._pidfd is None + assert not wav.parent.exists() + + +def test_unused_recorder_releases_empty_job(tmp_path): + import gc + + recorder = AudioRecorder(tmp_path) + job = recorder._wav_file.parent + del recorder + gc.collect() + assert not job.exists() + + +def test_destroying_stopped_recorder_preserves_successful_wav(recording): + recorder, _ = recording + wav = recorder.start() + recorder.stop() + recorder.__del__() + assert wav.read_bytes() == b"finished" + AudioRecorder.cleanup_recording(wav)