From bf07b42f881b1967d3cd3aa9ff44fbd10fa7d13c Mon Sep 17 00:00:00 2001 From: Flowseal Date: Sat, 10 Oct 2026 22:08:57 +0300 Subject: [PATCH] #1223 fixes, timeout for downloads --- tests/test_linux_update.py | 57 ++++++++++++++++++++++++++++++++------ utils/linux_update.py | 2 ++ utils/update_check.py | 4 ++- windows.py | 22 ++------------- 4 files changed, 55 insertions(+), 30 deletions(-) diff --git a/tests/test_linux_update.py b/tests/test_linux_update.py index 7f5eeb2..cc9d4f2 100644 --- a/tests/test_linux_update.py +++ b/tests/test_linux_update.py @@ -1,4 +1,6 @@ import os +import stat +import sys import tempfile import unittest from pathlib import Path @@ -71,29 +73,31 @@ class DetectInstallKindTest(unittest.TestCase): class InstallCmdTest(unittest.TestCase): + package_path = Path(tempfile.gettempdir()) / 'pkg' + def _cmd(self, kind, tools): with mock.patch.object( linux_update.shutil, 'which', side_effect=lambda n: '/usr/bin/' + n if n in tools else None, ), mock.patch.object(linux_update, '_is_root', return_value=False): - return linux_update.install_cmd(kind, Path('/tmp/pkg')) + return linux_update.install_cmd(kind, self.package_path) def test_deb_prefers_apt_get_and_elevates(self): cmd = self._cmd('deb', ('apt-get',)) self.assertEqual(cmd[:3], ['pkexec', 'apt-get', 'install']) - self.assertEqual(cmd[-1], '/tmp/pkg') + self.assertEqual(cmd[-1], str(self.package_path)) def test_deb_falls_back_to_dpkg(self): - self.assertEqual(self._cmd('deb', ()), ['pkexec', 'dpkg', '-i', '/tmp/pkg']) + self.assertEqual(self._cmd('deb', ()), ['pkexec', 'dpkg', '-i', str(self.package_path)]) def test_rpm_prefers_dnf_without_removing_packages(self): cmd = self._cmd('rpm', ('dnf',)) - self.assertEqual(cmd, ['pkexec', 'dnf', 'install', '-y', '/tmp/pkg']) + self.assertEqual(cmd, ['pkexec', 'dnf', 'install', '-y', str(self.package_path)]) self.assertNotIn('--allowerasing', cmd) def test_rpm_falls_back_to_rpm_binary(self): self.assertEqual( - self._cmd('rpm', ()), ['pkexec', 'rpm', '-U', '--force', '/tmp/pkg'], + self._cmd('rpm', ()), ['pkexec', 'rpm', '-U', '--force', str(self.package_path)], ) def test_unknown_kind_is_rejected(self): @@ -104,8 +108,8 @@ class InstallCmdTest(unittest.TestCase): with mock.patch.object(linux_update.shutil, 'which', return_value=None), \ mock.patch.object(linux_update, '_is_root', return_value=True): self.assertEqual( - linux_update.install_cmd('deb', Path('/tmp/pkg')), - ['dpkg', '-i', '/tmp/pkg'], + linux_update.install_cmd('deb', self.package_path), + ['dpkg', '-i', str(self.package_path)], ) @@ -127,9 +131,18 @@ class ApplyUpdateTest(unittest.TestCase): self.assertIsNone(linux_update.apply_update('binary', 'https://x.invalid/a')) self.assertEqual(exe.read_bytes(), b'new') - self.assertTrue(os.access(str(exe), os.X_OK)) self.assertEqual(list(Path(tmp).iterdir()), [exe]) + @unittest.skipUnless(sys.platform == 'linux', 'Linux executable permissions') + def test_replaced_binary_has_executable_permissions(self): + with tempfile.TemporaryDirectory() as tmp: + exe = Path(tmp) / 'TgWsProxy' + exe.write_bytes(b'old') + with mock.patch.object(linux_update, 'exe_path', return_value=exe), \ + mock.patch('utils.update_check.download_asset', self._download()): + self.assertIsNone(linux_update.apply_update('binary', 'https://x.invalid/a')) + self.assertEqual(stat.S_IMODE(exe.stat().st_mode), 0o755) + def test_read_only_dir_binary_is_installed_with_elevation(self): exe = Path('/usr/local/bin/TgWsProxy') with tempfile.TemporaryDirectory() as tmp: @@ -205,6 +218,32 @@ class ApplyUpdateTest(unittest.TestCase): self.assertEqual(linux_update.apply_update('binary', 'https://x.invalid/a'), 'boom') +class ExternalCommandsEnvTest(unittest.TestCase): + def test_commands_use_system_libraries_without_changing_parent_environment(self): + bundle = str(Path(tempfile.gettempdir()) / '_MEItest') + probe = ( + 'import os, sys; ' + 'assert os.environ.get("LD_LIBRARY_PATH", "") == sys.argv[1]; ' + 'assert "_PYI_APPLICATION_HOME_DIR" not in os.environ' + ) + for original in ('/usr/lib', None): + for runner, expected in ((linux_update._run_quiet, True), + (linux_update._run_install, None)): + with self.subTest(original=original, runner=runner.__name__): + env = dict(os.environ) + env.update(LD_LIBRARY_PATH=bundle, _PYI_APPLICATION_HOME_DIR=bundle) + env.pop('LD_LIBRARY_PATH_ORIG', None) + if original is not None: + env['LD_LIBRARY_PATH_ORIG'] = original + with mock.patch.dict(os.environ, env, clear=True), \ + mock.patch.object(sys, '_MEIPASS', bundle, create=True): + self.assertEqual( + runner([sys.executable, '-c', probe, original or '']), expected, + ) + self.assertEqual(os.environ['LD_LIBRARY_PATH'], bundle) + self.assertEqual(os.environ['_PYI_APPLICATION_HOME_DIR'], bundle) + + class RelaunchEnvTest(unittest.TestCase): def _env(self, env, meipass='/tmp/_MEI1'): with mock.patch.dict(linux_update.os.environ, env, clear=True), \ @@ -218,7 +257,7 @@ class RelaunchEnvTest(unittest.TestCase): 'LD_LIBRARY_PATH': '/tmp/_MEI1', 'LD_LIBRARY_PATH_ORIG': '/usr/lib', 'GI_TYPELIB_PATH': '/tmp/_MEI1', - 'PATH': '/tmp/_MEI1:/usr/bin', + 'PATH': os.pathsep.join(('/tmp/_MEI1', '/usr/bin')), 'HOME': '/home/u', }) diff --git a/utils/linux_update.py b/utils/linux_update.py index f92bd43..e2ab163 100644 --- a/utils/linux_update.py +++ b/utils/linux_update.py @@ -34,6 +34,7 @@ def _run_quiet(cmd: List[str]) -> bool: stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, timeout=15, + env=relaunch_env(), ).returncode == 0 except (OSError, subprocess.SubprocessError): return False @@ -107,6 +108,7 @@ def _run_install(cmd: List[str], log: Optional[logging.Logger] = None) -> Option proc = subprocess.run( cmd, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=_INSTALL_TIMEOUT_SEC, + env=relaunch_env(), ) if proc.returncode == 0: return None diff --git a/utils/update_check.py b/utils/update_check.py index 08baa73..5dcd447 100644 --- a/utils/update_check.py +++ b/utils/update_check.py @@ -271,12 +271,14 @@ def _expected_length(resp: Any) -> Optional[int]: def download_asset( url: str, dest_dir: Path, suffix: str = ".tmp", digest: str = "", + timeout: float = 30.0, ) -> Path: """Download a release asset into a temporary file inside dest_dir. The size is checked against Content-Length and, when the GitHub API provides a "sha256:..." digest, the checksum too. A truncated or corrupted file is removed instead of being returned. + timeout limits blocking network operations, including response reads. Raises: OSError: network error, truncated download or checksum mismatch. @@ -287,7 +289,7 @@ def download_asset( try: sha = hashlib.sha256() written = 0 - with build_github_opener().open(url) as resp: + with build_github_opener().open(url, timeout=timeout) as resp: expected = _expected_length(resp) with open(str(path), "wb") as out: while True: diff --git a/windows.py b/windows.py index fcfdc0f..95323dc 100644 --- a/windows.py +++ b/windows.py @@ -8,11 +8,9 @@ import threading import time import webbrowser import winreg -import tempfile from pathlib import Path from typing import Optional -from proxy.utils import build_github_opener try: import pyperclip @@ -48,7 +46,7 @@ from utils.tray_common import ( save_config, start_proxy, stop_proxy, tg_proxy_url, ) from utils.update_check import ( - get_status, get_update_asset, run_check, RELEASES_PAGE_URL, + download_asset, get_status, get_update_asset, run_check, RELEASES_PAGE_URL, ) from ui.ctk_tray_ui import ( install_tray_config_buttons, install_tray_config_form, @@ -245,27 +243,11 @@ def _perform_update(download_url: str, set_status=None) -> None: _step(t("update.downloading")) cur_exe = Path(sys.executable) old_exe = cur_exe.with_name(cur_exe.stem + "_oldtgws.exe") - tmp_path = None try: - fd, tmp_name = tempfile.mkstemp(dir=cur_exe.parent, suffix=".tmp") - os.close(fd) - tmp_path = Path(tmp_name) log.info("Downloading update from %s", download_url) - opener = build_github_opener() - with opener.open(download_url) as _resp: - with open(str(tmp_path), "wb") as _fout: - while True: - _chunk = _resp.read(65536) - if not _chunk: - break - _fout.write(_chunk) + tmp_path = download_asset(download_url, cur_exe.parent) except Exception as exc: _err(t("update.download_fail", error=exc)) - if tmp_path: - try: - tmp_path.unlink(missing_ok=True) - except OSError: - pass return _step(t("update.replacing"))