diff --git a/install.py b/install.py index 9a8d3020..01a1b03c 100644 --- a/install.py +++ b/install.py @@ -696,9 +696,9 @@ def _declared_state(copy: dict, entry: dict, bios_path: Path) -> str: path = _entry_path(bios_path, entry) except ValueError: return "missing" - if not path.is_file(): - return "missing" if not path.exists() else "wrong_hash" try: + if not path.is_file(): + return "missing" if not path.exists() else "wrong_hash" verdict = _classify_local(copy, path) except OSError: return "wrong_hash" @@ -1443,7 +1443,7 @@ def _rule_matches(rule: _OsRule, release: dict[str, str]) -> bool: except OSError: return False return re.search(pattern, text, re.IGNORECASE | re.MULTILINE) is not None - return all(_on_device(value).exists() for value in rule.values) + return all(_exists(_on_device(value)) for value in rule.values) def _config_files(pattern: str) -> list[tuple[Path, Path]]: @@ -4090,7 +4090,13 @@ def _classify_local(entry: dict, dest: Path) -> str: for name in ("sha256", "sha1") if entry.get(name) } - if not dest.is_file(): + try: + is_file = dest.is_file() + except OSError: + # A folder above it this user cannot enter: what sits there is not + # known to be the right file. + return "mismatched" + if not is_file: # A folder at the file's place: --check says so, an install fails # on that file alone. return "mismatched" @@ -4107,6 +4113,39 @@ def _classify_local(entry: dict, dest: Path) -> str: return "mismatched" +def _held(path: Path) -> bool: + """Whether something may sit at path. A place that cannot be looked at + (a folder above it this user cannot enter, a card that answers a read + error) counts as held, so the file is reported unreadable and the run + goes on with the others: Path.exists() lets PermissionError through.""" + try: + return path.exists() + except OSError: + return True + + +def _exists(path: Path) -> bool: + """Path.exists(), False where the place cannot be looked at.""" + try: + return path.exists() + except OSError: + return False + + +def _is_file(path: Path) -> bool: + """Path.is_file(), False where the place cannot be looked at.""" + try: + return path.is_file() + except OSError: + return False + + +def unreachable_words(exc: OSError) -> str: + """Why the place of a file could not be looked at, in words.""" + return (f"could not be read ({exc.strerror or exc}): this user may not " + f"have access to its folder") + + def check_local( files: list[dict], bios_path: Path, jobs: int = 8, term: "Terminal | None" = None, @@ -4122,7 +4161,7 @@ def check_local( for f in files: dest = _entry_path(bios_path, f) - if dest.exists(): + if _held(dest): present.append((f, dest)) else: to_download.append(f) @@ -4362,7 +4401,7 @@ def _move_aside(dest: Path, kept: Path) -> None: os.rename(dest, kept) return except OSError: - if not dest.is_file(): + if not _is_file(dest): raise scratch = kept.parent / f".{kept.name}.part" try: @@ -4497,7 +4536,11 @@ def _download_one( if _write_refused(exc): return f["dest"], False, NOT_WRITABLE return f["dest"], False, exc.strerror or str(exc) - if dest.is_dir(): + try: + in_place_folder = dest.is_dir() + except OSError as exc: + return f["dest"], False, unreachable_words(exc) + if in_place_folder: # Fetched, it failed only at the rename, on an errno and two paths. return f["dest"], False, FOLDER_IN_PLACE @@ -4752,7 +4795,7 @@ def _replace_released(source: Path, dest: Path, expected: dict[str, str]) -> Non os.replace(source, dest) return except FileNotFoundError: - if expected and dest.is_file() and _digest_file(dest, tuple(expected)) == expected: + if expected and _is_file(dest) and _digest_file(dest, tuple(expected)) == expected: return raise except PermissionError as exc: @@ -4771,7 +4814,7 @@ def _drop_full_partial(dest: Path, expected_size: int) -> None: left to it. """ path = dest.parent / f".{dest.name}.part" - if not path.is_file(): + if not _is_file(path): return try: handle = _claim_partial(path) @@ -4913,7 +4956,7 @@ def _clear_read_only(dest: Path) -> None: def _read_only_file(dest: Path) -> bool: - return dest.is_file() and not os.access(dest, os.W_OK) and _folder_writable(dest.parent) + return _is_file(dest) and not os.access(dest, os.W_OK) and _folder_writable(dest.parent) def _folder_writable(path: Path) -> bool: @@ -5017,7 +5060,7 @@ def _on_read_only_mount(path: Path) -> bool: if not hasattr(os, "statvfs"): return False probe = path - while not probe.exists() and probe.parent != probe: + while not _held(probe) and probe.parent != probe: probe = probe.parent try: return bool(os.statvfs(probe).f_flag & os.ST_RDONLY) @@ -5247,7 +5290,7 @@ def _absent_required(omitted: list[dict], bios_path: Path) -> list[dict]: if not entry.get("required", True) or not dest: continue try: - placed = _entry_path(bios_path, {**entry, "dest": dest}).is_file() + placed = _is_file(_entry_path(bios_path, {**entry, "dest": dest})) except ValueError: # A path climbing out of the folder is never one we read, and one # below a save directory this run does not know is not looked at. @@ -5576,11 +5619,12 @@ def data_directory_state(entry: dict, bios_path: Path) -> tuple[Path, dict[str, present = 0 for rel, meta in entry["files"].items(): target = _destination_path(root, rel) - if ( - target.is_file() - and target.stat().st_size == meta["size"] - and _sha1_file(target) == meta["sha1"] - ): + try: + held = (target.is_file() and target.stat().st_size == meta["size"] + and _sha1_file(target) == meta["sha1"]) + except OSError: + held = False + if held: present += 1 else: needed[rel] = meta @@ -7701,9 +7745,9 @@ def _wrong_contents( owners: dict[str, list[dict]] | None = None, ) -> str: """Why the file at an entry's place is not the one the list installs.""" - if path.is_dir(): - return FOLDER_IN_PLACE try: + if path.is_dir(): + return FOLDER_IN_PLACE size = path.stat().st_size with open(path, "rb") as handle: head = handle.read(4) @@ -8721,7 +8765,7 @@ def _card_rule_matches(rule: _OsRule, root: Path) -> bool: except OSError: return False return re.search(pattern, text, re.MULTILINE) is not None - return all((root / value).exists() for value in rule.values) + return all(_exists(root / value) for value in rule.values) def card_system(root: Path) -> tuple[str, Path, str] | None: diff --git a/tests/test_install_experience.py b/tests/test_install_experience.py index 5f64b1e0..76dc68f4 100644 --- a/tests/test_install_experience.py +++ b/tests/test_install_experience.py @@ -507,6 +507,65 @@ class ARefusedFolderReportsEachFile(unittest.TestCase): list(range(1, len(downloaded) + 1))) +class AFolderThisUserCannotEnter(unittest.TestCase): + """A subfolder of the BIOS folder without search permission (made 0700 + by an emulator that ran as root, or denied by a Windows ACL) raised + PermissionError from Path.exists() before any file was handled, in + --check and in an install alike.""" + + def _locked(self, repo: FakeRepository) -> Path: + (repo.dest / "pcsx").mkdir(parents=True) + (repo.dest / "pcsx" / "extra.bin").write_bytes(b"old") + os.chmod(repo.dest / "pcsx", 0o000) + return repo.dest / "pcsx" + + @unittest.skipIf(os.name == "nt" or os.geteuid() == 0, + "a folder without search permission is POSIX") + def test_check_names_the_file_and_reads_the_others(self): + with FakeRepository() as repo: + locked = self._locked(repo) + try: + proc = repo.run("--check") + finally: + os.chmod(locked, 0o755) + self.assertNotIn("Traceback", proc.stderr) + self.assertEqual(proc.returncode, 1, proc.stdout + proc.stderr) + self.assertIn("wrong hash: pcsx/extra.bin", proc.stdout) + self.assertIn("could not be read (Permission denied)", proc.stdout) + self.assertIn("missing: scph5501.bin", proc.stdout) + + @unittest.skipIf(os.name == "nt" or os.geteuid() == 0, + "a folder without search permission is POSIX") + def test_install_fails_that_file_alone(self): + with FakeRepository() as repo: + locked = self._locked(repo) + try: + proc = repo.run("--events") + finally: + os.chmod(locked, 0o755) + installed = repo.installed() + self.assertNotIn("Traceback", proc.stderr) + events = [json.loads(line) for line in proc.stdout.splitlines()] + failed = {e["dest"]: e["reason"] for e in events + if e["event"] == "downloaded" and not e["ok"]} + self.assertEqual(list(failed), ["pcsx/extra.bin"], proc.stderr) + self.assertIn("this user may not have access", failed["pcsx/extra.bin"]) + self.assertIn("scph5501.bin", installed) + self.assertIn("mgba/extra.bin", installed) + + @unittest.skipIf(os.name == "nt" or os.geteuid() == 0, + "a folder without search permission is POSIX") + def test_declared_state_reads_it_as_wrong(self): + with FakeRepository() as repo: + locked = self._locked(repo) + entry = {"dest": "pcsx/extra.bin", "sha1": "0" * 40, "size": 3} + try: + state = install._declared_state(entry, entry, repo.dest) + finally: + os.chmod(locked, 0o755) + self.assertEqual(state, "wrong_hash") + + class _Tty(io.StringIO): def isatty(self) -> bool: return True