Copilot commented on code in PR #65513: URL: https://github.com/apache/airflow/pull/65513#discussion_r3107576735
########## airflow-core/tests/unit/utils/test_pidfile.py: ########## @@ -0,0 +1,193 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os + +import pytest + +from airflow.utils.pidfile import ( + PIDLockFile, + TimeoutPIDLockFile, + read_pid_from_pidfile, + remove_existing_pidfile, + write_pid_to_pidfile, +) + + +class TestReadPidFromPidfile: + def test_reads_valid_pid(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("12345\n") + assert read_pid_from_pidfile(str(pidfile)) == 12345 + + def test_reads_pid_without_trailing_newline(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("99") + assert read_pid_from_pidfile(str(pidfile)) == 99 + + def test_ignores_leading_whitespace(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text(" 42 \n") + assert read_pid_from_pidfile(str(pidfile)) == 42 + + def test_returns_none_for_missing_file(self, tmp_path): + assert read_pid_from_pidfile(str(tmp_path / "nonexistent.pid")) is None + + def test_returns_none_for_non_numeric_content(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("not-a-number\n") + assert read_pid_from_pidfile(str(pidfile)) is None + + def test_returns_none_for_empty_file(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("") + assert read_pid_from_pidfile(str(pidfile)) is None + + +class TestWritePidToPidfile: + def test_writes_current_pid(self, tmp_path): + pidfile = str(tmp_path / "test.pid") + write_pid_to_pidfile(pidfile) + content = open(pidfile).read() + assert content.strip() == str(os.getpid()) + + def test_raises_on_existing_file(self, tmp_path): + pidfile = str(tmp_path / "test.pid") + write_pid_to_pidfile(pidfile) + with pytest.raises(OSError, match="File exists"): + write_pid_to_pidfile(pidfile) + + def test_creates_file_with_correct_permissions(self, tmp_path): + pidfile = str(tmp_path / "test.pid") + write_pid_to_pidfile(pidfile) Review Comment: `test_creates_file_with_correct_permissions` asserts an exact mode of `0o644`, but file creation mode is masked by the process umask. This makes the test environment-dependent/flaky. Set a known umask in the test (and restore it) or assert based on the current umask. ```suggestion previous_umask = os.umask(0o022) try: write_pid_to_pidfile(pidfile) finally: os.umask(previous_umask) ``` ########## airflow-core/src/airflow/utils/daemon.py: ########## @@ -0,0 +1,315 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Minimal daemon context replacing the ``python-daemon`` distribution. + +Implements the subset of :pep:`3143` that Airflow actually uses: +double-fork detach, file-descriptor cleanup, stream redirection, +signal-handler installation, umask/chroot/cwd changes, UID/GID +switching, PID-file management and core-dump prevention. +""" + +from __future__ import annotations + +import atexit +import os +import pwd +import resource +import signal +import sys + + +class DaemonContext: + """Context manager that turns the current process into a Unix daemon.""" + + def __init__( + self, + *, + chroot_directory: str | None = None, + working_directory: str = "/", + umask: int = 0, + uid: int | None = None, + gid: int | None = None, + initgroups: bool = False, + prevent_core: bool = True, + detach_process: bool | None = None, + files_preserve: list | None = None, + pidfile=None, + stdin=None, + stdout=None, + stderr=None, + signal_map: dict | None = None, + ): + self.chroot_directory = chroot_directory + self.working_directory = working_directory + self.umask = umask + self.prevent_core = prevent_core + self.files_preserve = files_preserve + self.pidfile = pidfile + self.stdin = stdin + self.stdout = stdout + self.stderr = stderr + + self.uid = uid if uid is not None else os.getuid() + self.gid = gid if gid is not None else os.getgid() + self.initgroups = initgroups + + if detach_process is None: + detach_process = _is_detach_process_context_required() + self.detach_process = detach_process + + if signal_map is None: + signal_map = _make_default_signal_map() + self.signal_map = signal_map + + self._is_open = False + + @property + def is_open(self) -> bool: + return self._is_open + + # -- context-manager protocol ------------------------------------------ + + def open(self) -> None: + if self.is_open: + return + + if self.chroot_directory is not None: + _change_root_directory(self.chroot_directory) + + if self.prevent_core: + _prevent_core_dump() + + _change_file_creation_mask(self.umask) + _change_working_directory(self.working_directory) + _change_process_owner(self.uid, self.gid, self.initgroups) + + if self.detach_process: + _detach_process_context() + + signal_handler_map = self._make_signal_handler_map() + _set_signal_handlers(signal_handler_map) + + exclude_fds = self._get_exclude_file_descriptors() + _close_all_open_files(exclude=exclude_fds) + + _redirect_stream(sys.stdin, self.stdin) + _redirect_stream(sys.stdout, self.stdout) + _redirect_stream(sys.stderr, self.stderr) + + if self.pidfile is not None: + self.pidfile.__enter__() + + self._is_open = True + atexit.register(self.close) + + def close(self) -> None: + if not self.is_open: + return + if self.pidfile is not None: + self.pidfile.__exit__(None, None, None) + self._is_open = False + + def __enter__(self): + self.open() + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + # -- signal handling --------------------------------------------------- + + def terminate(self, signal_number, stack_frame): + raise SystemExit(f"Terminating on signal {signal_number!r}") + + def _make_signal_handler(self, target): + if target is None: + return signal.SIG_IGN + if isinstance(target, str): + return getattr(self, target) + return target + + def _make_signal_handler_map(self) -> dict: + return { + signal_number: self._make_signal_handler(target) + for signal_number, target in self.signal_map.items() + } + + # -- file-descriptor helpers ------------------------------------------- + + def _get_exclude_file_descriptors(self) -> set[int]: + files_preserve = list(self.files_preserve or []) + files_preserve.extend( + item for item in (self.stdin, self.stdout, self.stderr) if hasattr(item, "fileno") + ) + + exclude: set[int] = set() + for item in files_preserve: + if item is None: + continue + fd = _get_file_descriptor(item) + if fd is not None: + exclude.add(fd) + else: + exclude.add(item) + return exclude + + +# --------------------------------------------------------------------------- +# Helper functions +# --------------------------------------------------------------------------- + + +def _get_file_descriptor(obj) -> int | None: + if hasattr(obj, "fileno"): + try: + return obj.fileno() + except ValueError: + pass + return None + + +def _change_working_directory(directory: str) -> None: + try: + os.chdir(directory) + except Exception as exc: + raise OSError(f"Unable to change working directory ({exc})") from exc + + +def _change_root_directory(directory: str) -> None: + try: + os.chdir(directory) + os.chroot(directory) + except Exception as exc: + raise OSError(f"Unable to change root directory ({exc})") from exc + + +def _change_file_creation_mask(mask: int) -> None: + try: + os.umask(mask) + except Exception as exc: + raise OSError(f"Unable to change file creation mask ({exc})") from exc + + +def _change_process_owner(uid: int, gid: int, initgroups: bool = False) -> None: + try: + username = pwd.getpwuid(uid).pw_name + except KeyError: + initgroups = False + username = None + + try: + if initgroups and username is not None: + os.initgroups(username, gid) + else: + os.setgid(gid) + os.setuid(uid) + except Exception as exc: + raise OSError(f"Unable to change process owner ({exc})") from exc + + +def _prevent_core_dump() -> None: + core_resource = resource.RLIMIT_CORE + try: + resource.getrlimit(core_resource) + except ValueError as exc: + raise OSError(f"System does not support RLIMIT_CORE resource limit ({exc})") from exc + resource.setrlimit(core_resource, (0, 0)) + + +def _detach_process_context() -> None: + def _fork_then_exit_parent(error_message: str) -> None: + try: + pid = os.fork() + if pid > 0: + os._exit(0) + except OSError as exc: + raise OSError(f"{error_message}: [{exc.errno}] {exc.strerror}") from exc + + _fork_then_exit_parent("Failed first fork") + os.setsid() + _fork_then_exit_parent("Failed second fork") + + +def _is_detach_process_context_required() -> bool: + if os.getppid() == 1: + return False + try: + if sys.__stdin__ is not None and hasattr(sys.__stdin__, "fileno"): + fd = sys.__stdin__.fileno() + import socket + + sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW) + sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) + return False + except (OSError, ValueError): Review Comment: `_is_detach_process_context_required` does an inline `import socket` (against the repo guidance to keep imports at module level) and creates a socket via `socket.fromfd(...)` without closing it. Please move the import to module scope and ensure any temporary socket objects are closed to avoid leaking file descriptors. ########## airflow-core/tests/unit/utils/test_pidfile.py: ########## @@ -0,0 +1,193 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os + +import pytest + +from airflow.utils.pidfile import ( + PIDLockFile, + TimeoutPIDLockFile, + read_pid_from_pidfile, + remove_existing_pidfile, + write_pid_to_pidfile, +) + + +class TestReadPidFromPidfile: + def test_reads_valid_pid(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("12345\n") + assert read_pid_from_pidfile(str(pidfile)) == 12345 + + def test_reads_pid_without_trailing_newline(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("99") + assert read_pid_from_pidfile(str(pidfile)) == 99 + + def test_ignores_leading_whitespace(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text(" 42 \n") + assert read_pid_from_pidfile(str(pidfile)) == 42 + + def test_returns_none_for_missing_file(self, tmp_path): + assert read_pid_from_pidfile(str(tmp_path / "nonexistent.pid")) is None + + def test_returns_none_for_non_numeric_content(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("not-a-number\n") + assert read_pid_from_pidfile(str(pidfile)) is None + + def test_returns_none_for_empty_file(self, tmp_path): + pidfile = tmp_path / "test.pid" + pidfile.write_text("") + assert read_pid_from_pidfile(str(pidfile)) is None + + +class TestWritePidToPidfile: + def test_writes_current_pid(self, tmp_path): + pidfile = str(tmp_path / "test.pid") + write_pid_to_pidfile(pidfile) + content = open(pidfile).read() Review Comment: This test reads the pidfile using `open(...).read()` without a context manager, leaving the file handle to be closed by GC. Use a `with open(...)` block to avoid leaking file descriptors during the test run. ```suggestion with open(pidfile) as file: content = file.read() ``` ########## airflow-core/src/airflow/utils/daemon.py: ########## @@ -0,0 +1,315 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Minimal daemon context replacing the ``python-daemon`` distribution. + +Implements the subset of :pep:`3143` that Airflow actually uses: +double-fork detach, file-descriptor cleanup, stream redirection, +signal-handler installation, umask/chroot/cwd changes, UID/GID +switching, PID-file management and core-dump prevention. +""" + +from __future__ import annotations + +import atexit +import os +import pwd +import resource +import signal +import sys Review Comment: `airflow.utils.daemon` unconditionally imports Unix-only modules (`pwd`, `resource`) and is imported at module import time by `airflow.cli.commands.daemon_utils`. This will raise `ModuleNotFoundError` on non-POSIX platforms even when `--daemon` isn’t used. Consider guarding these imports / limiting this module to POSIX (e.g., import-time fallback that raises a clear error only when daemonization is attempted). ########## airflow-core/tests/unit/utils/test_daemon.py: ########## @@ -0,0 +1,373 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +from __future__ import annotations + +import os +import resource +import signal +import sys +from unittest import mock + +import pytest + +from airflow.utils.daemon import ( + DaemonContext, + _change_file_creation_mask, + _change_working_directory, + _close_all_open_files, + _get_file_descriptor, + _get_maximum_file_descriptors, + _is_detach_process_context_required, + _make_default_signal_map, + _prevent_core_dump, + _redirect_stream, + _set_signal_handlers, +) + + +class TestGetFileDescriptor: + def test_returns_fd_for_file_like_object(self, tmp_path): + path = tmp_path / "test.txt" + path.write_text("hello") + with open(str(path)) as f: + assert _get_file_descriptor(f) == f.fileno() + + def test_returns_none_for_non_file_object(self): + assert _get_file_descriptor("not-a-file") is None + + def test_returns_none_for_closed_file(self, tmp_path): + path = tmp_path / "test.txt" + path.write_text("hello") + f = open(str(path)) + f.close() + assert _get_file_descriptor(f) is None + + +class TestChangeWorkingDirectory: + def test_changes_cwd(self, tmp_path): + original = os.getcwd() + try: + _change_working_directory(str(tmp_path)) + assert os.getcwd() == str(tmp_path) + finally: + os.chdir(original) + + def test_raises_on_invalid_directory(self): + with pytest.raises(OSError, match="Unable to change working directory"): + _change_working_directory("/nonexistent_dir_abc123") + + +class TestChangeFileCreationMask: + def test_sets_umask(self): + old = os.umask(0o022) + os.umask(old) + try: + _change_file_creation_mask(0o077) + assert os.umask(0o077) == 0o077 + finally: + os.umask(old) + + +class TestPreventCoreDump: + def test_sets_core_limit_to_zero(self): + original = resource.getrlimit(resource.RLIMIT_CORE) + try: + _prevent_core_dump() + soft, hard = resource.getrlimit(resource.RLIMIT_CORE) + assert soft == 0 + assert hard == 0 + finally: + try: + resource.setrlimit(resource.RLIMIT_CORE, original) + except ValueError: + pass + + +class TestGetMaximumFileDescriptors: + def test_returns_hard_limit(self): + __, hard = resource.getrlimit(resource.RLIMIT_NOFILE) + result = _get_maximum_file_descriptors() + if hard == resource.RLIM_INFINITY: + assert result == 2048 + else: + assert result == hard + + +class TestCloseAllOpenFiles: + def test_closes_file_descriptors_except_excluded(self): + r_fd, w_fd = os.pipe() + try: + _close_all_open_files(exclude={r_fd, w_fd, 0, 1, 2}) + os.write(w_fd, b"alive") + assert os.read(r_fd, 5) == b"alive" + finally: + os.close(r_fd) + os.close(w_fd) + + def test_empty_exclude_does_not_raise(self): + _close_all_open_files(exclude={0, 1, 2}) + + +class TestRedirectStream: + def test_redirects_to_target(self, tmp_path): + target_path = tmp_path / "output.txt" + with open(str(target_path), "w") as target: + original_fd = os.dup(1) + try: + _redirect_stream(sys.stdout, target) + os.write(1, b"redirected\n") + finally: + os.dup2(original_fd, 1) + os.close(original_fd) + assert "redirected" in target_path.read_text() + + def test_redirects_to_devnull_when_none(self): + import sys + Review Comment: Redundant inner `import sys` inside `test_redirects_to_devnull_when_none` (the module already imports `sys` at top). Please remove the inner import to keep imports consistent and avoid function-local imports. ```suggestion ``` ########## airflow-core/src/airflow/utils/daemon.py: ########## @@ -0,0 +1,315 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +""" +Minimal daemon context replacing the ``python-daemon`` distribution. + +Implements the subset of :pep:`3143` that Airflow actually uses: +double-fork detach, file-descriptor cleanup, stream redirection, +signal-handler installation, umask/chroot/cwd changes, UID/GID +switching, PID-file management and core-dump prevention. +""" + +from __future__ import annotations + +import atexit +import os +import pwd +import resource +import signal +import sys + + +class DaemonContext: + """Context manager that turns the current process into a Unix daemon.""" + + def __init__( + self, + *, + chroot_directory: str | None = None, + working_directory: str = "/", + umask: int = 0, + uid: int | None = None, + gid: int | None = None, + initgroups: bool = False, + prevent_core: bool = True, + detach_process: bool | None = None, + files_preserve: list | None = None, + pidfile=None, + stdin=None, + stdout=None, + stderr=None, + signal_map: dict | None = None, + ): + self.chroot_directory = chroot_directory + self.working_directory = working_directory + self.umask = umask + self.prevent_core = prevent_core + self.files_preserve = files_preserve + self.pidfile = pidfile + self.stdin = stdin + self.stdout = stdout + self.stderr = stderr + + self.uid = uid if uid is not None else os.getuid() + self.gid = gid if gid is not None else os.getgid() + self.initgroups = initgroups + + if detach_process is None: + detach_process = _is_detach_process_context_required() + self.detach_process = detach_process + + if signal_map is None: + signal_map = _make_default_signal_map() + self.signal_map = signal_map + + self._is_open = False + + @property + def is_open(self) -> bool: + return self._is_open + + # -- context-manager protocol ------------------------------------------ + + def open(self) -> None: + if self.is_open: + return + + if self.chroot_directory is not None: + _change_root_directory(self.chroot_directory) + + if self.prevent_core: + _prevent_core_dump() + + _change_file_creation_mask(self.umask) + _change_working_directory(self.working_directory) + _change_process_owner(self.uid, self.gid, self.initgroups) + + if self.detach_process: + _detach_process_context() + + signal_handler_map = self._make_signal_handler_map() + _set_signal_handlers(signal_handler_map) + + exclude_fds = self._get_exclude_file_descriptors() + _close_all_open_files(exclude=exclude_fds) + + _redirect_stream(sys.stdin, self.stdin) + _redirect_stream(sys.stdout, self.stdout) + _redirect_stream(sys.stderr, self.stderr) + + if self.pidfile is not None: + self.pidfile.__enter__() + + self._is_open = True + atexit.register(self.close) + + def close(self) -> None: + if not self.is_open: + return + if self.pidfile is not None: + self.pidfile.__exit__(None, None, None) + self._is_open = False + + def __enter__(self): + self.open() + return self + + def __exit__(self, exc_type, exc_value, traceback): + self.close() + + # -- signal handling --------------------------------------------------- + + def terminate(self, signal_number, stack_frame): + raise SystemExit(f"Terminating on signal {signal_number!r}") + + def _make_signal_handler(self, target): + if target is None: + return signal.SIG_IGN + if isinstance(target, str): + return getattr(self, target) + return target + + def _make_signal_handler_map(self) -> dict: + return { + signal_number: self._make_signal_handler(target) + for signal_number, target in self.signal_map.items() + } + + # -- file-descriptor helpers ------------------------------------------- + + def _get_exclude_file_descriptors(self) -> set[int]: + files_preserve = list(self.files_preserve or []) + files_preserve.extend( + item for item in (self.stdin, self.stdout, self.stderr) if hasattr(item, "fileno") + ) + + exclude: set[int] = set() + for item in files_preserve: + if item is None: + continue + fd = _get_file_descriptor(item) + if fd is not None: + exclude.add(fd) + else: + exclude.add(item) + return exclude + + +# --------------------------------------------------------------------------- +# Helper functions +# --------------------------------------------------------------------------- + + +def _get_file_descriptor(obj) -> int | None: + if hasattr(obj, "fileno"): + try: + return obj.fileno() + except ValueError: + pass + return None + + +def _change_working_directory(directory: str) -> None: + try: + os.chdir(directory) + except Exception as exc: + raise OSError(f"Unable to change working directory ({exc})") from exc + + +def _change_root_directory(directory: str) -> None: + try: + os.chdir(directory) + os.chroot(directory) + except Exception as exc: + raise OSError(f"Unable to change root directory ({exc})") from exc + + +def _change_file_creation_mask(mask: int) -> None: + try: + os.umask(mask) + except Exception as exc: + raise OSError(f"Unable to change file creation mask ({exc})") from exc + + +def _change_process_owner(uid: int, gid: int, initgroups: bool = False) -> None: + try: + username = pwd.getpwuid(uid).pw_name + except KeyError: + initgroups = False + username = None + + try: + if initgroups and username is not None: + os.initgroups(username, gid) + else: + os.setgid(gid) + os.setuid(uid) + except Exception as exc: + raise OSError(f"Unable to change process owner ({exc})") from exc + + +def _prevent_core_dump() -> None: + core_resource = resource.RLIMIT_CORE + try: + resource.getrlimit(core_resource) + except ValueError as exc: + raise OSError(f"System does not support RLIMIT_CORE resource limit ({exc})") from exc + resource.setrlimit(core_resource, (0, 0)) + + +def _detach_process_context() -> None: + def _fork_then_exit_parent(error_message: str) -> None: + try: + pid = os.fork() + if pid > 0: + os._exit(0) + except OSError as exc: + raise OSError(f"{error_message}: [{exc.errno}] {exc.strerror}") from exc + + _fork_then_exit_parent("Failed first fork") + os.setsid() + _fork_then_exit_parent("Failed second fork") + + +def _is_detach_process_context_required() -> bool: + if os.getppid() == 1: + return False + try: + if sys.__stdin__ is not None and hasattr(sys.__stdin__, "fileno"): + fd = sys.__stdin__.fileno() + import socket + + sock = socket.fromfd(fd, socket.AF_INET, socket.SOCK_RAW) + sock.getsockopt(socket.SOL_SOCKET, socket.SO_TYPE) + return False + except (OSError, ValueError): + pass + return True + + +_MAXFD = 2048 + + +def _get_maximum_file_descriptors() -> int: + __, hard_limit = resource.getrlimit(resource.RLIMIT_NOFILE) + if hard_limit == resource.RLIM_INFINITY: + return _MAXFD + return hard_limit + + +def _close_all_open_files(exclude: set[int] | None = None) -> None: + if exclude is None: + exclude = set() + maxfd = _get_maximum_file_descriptors() + ranges: list[tuple[int, int]] = [] + remaining = range(0, maxfd) + for fd in sorted(exclude): + if fd > remaining.stop: + break + if fd < remaining.start: + continue + if fd != remaining.start: + ranges.append((remaining.start, fd)) + remaining = range(fd + 1, remaining.stop) + if remaining.start < remaining.stop: + ranges.append((remaining.start, remaining.stop)) + for low, high in ranges: + os.closerange(low, high) + + +def _redirect_stream(system_stream, target_stream) -> None: + if target_stream is None: + target_fd = os.open(os.devnull, os.O_RDWR) + else: + target_fd = target_stream.fileno() + os.dup2(target_fd, system_stream.fileno()) + + Review Comment: `_redirect_stream` opens `os.devnull` when `target_stream` is `None` but never closes the returned fd after `dup2`. This leaks a file descriptor for each redirected stream. Close the temporary fd after duplication (taking care not to close stdio fds if they’re reused). ```suggestion system_fd = system_stream.fileno() if target_stream is None: target_fd = os.open(os.devnull, os.O_RDWR) try: if target_fd != system_fd: os.dup2(target_fd, system_fd) finally: if target_fd != system_fd: os.close(target_fd) return os.dup2(target_stream.fileno(), system_fd) ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
