diff options
Diffstat (limited to 'packages')
19 files changed, 616 insertions, 85 deletions
diff --git a/packages/meshbay-common/src/meshbay_common/paths.py b/packages/meshbay-common/src/meshbay_common/paths.py index 8be4680..059bf09 100644 --- a/packages/meshbay-common/src/meshbay_common/paths.py +++ b/packages/meshbay-common/src/meshbay_common/paths.py @@ -19,7 +19,7 @@ from __future__ import annotations import os import re import unicodedata -from pathlib import Path, PurePath +from pathlib import Path # Windows refuses these as a basename, with or without an extension: `AUX.txt` # is as impossible as `AUX`. A group indexed on Linux can hold them, and a @@ -39,6 +39,9 @@ _RESERVED_CHARS = set('<>:"/\\|?*') | {chr(c) for c in range(32)} | BIDI_CONTROL # Windows without long-path support. A deep media library reaches this. MAX_PATH_WINDOWS = 260 +# Creating a directory is refused twelve characters earlier (room for an 8.3 +# name), and listing one adds `\*`: the extended form starts there. +_EXTENDED_FROM = MAX_PATH_WINDOWS - 12 def nfc(text: str) -> str: @@ -102,19 +105,22 @@ def long_path(path: Path | str) -> str: """ A path string safe to hand to the OS. - On Windows, prefix with `\\\\?\\` so `MAX_PATH` does not truncate a deep - library. The prefix requires an absolute, already-normalized path, and it is - a no-op everywhere else. + On Windows, a path that reaches `MAX_PATH` is refused by every call unless + the system's `LongPathsEnabled` is set, which it is not by default and which + a program cannot set for itself. The extended form (`\\\\?\\`) is accepted + whatever the setting, so a long path gets it. The form skips the system's + normalisation, so the path is normalised here first. A no-op everywhere + else, and for a path short enough to need nothing. """ text = str(path) if os.name != "nt" or text.startswith("\\\\?\\"): return text - resolved = str(PurePath(text)) - if len(resolved) < MAX_PATH_WINDOWS and not text.startswith("\\\\"): + full = os.path.normpath(text) if os.path.isabs(text) else os.path.abspath(text) + if len(full) < _EXTENDED_FROM: return text - if text.startswith("\\\\"): # UNC: \\server\share → \\?\UNC\server\share - return "\\\\?\\UNC" + text[1:] - return "\\\\?\\" + text + if full.startswith("\\\\"): # UNC: \\server\share → \\?\UNC\server\share + return "\\\\?\\UNC" + full[1:] + return "\\\\?\\" + full def find_fold_collisions(names: list[str]) -> dict[str, list[str]]: diff --git a/packages/meshbay-common/tests/test_paths.py b/packages/meshbay-common/tests/test_paths.py index 012a32e..cbaf1f5 100644 --- a/packages/meshbay-common/tests/test_paths.py +++ b/packages/meshbay-common/tests/test_paths.py @@ -6,12 +6,15 @@ these are not compatibility niceties — they decide whether two index entries a one file, and whether a member can save what they downloaded. """ +import os + import pytest from meshbay_common.paths import ( find_fold_collisions, fold, fold_path, is_portable_name, + long_path, nfc, portable_name_problem, sanitize_for_download, @@ -126,3 +129,49 @@ def test_a_bidi_override_cannot_hide_an_extension(): disguised = "invoicefdp.exe" # displays as "invoiceexe.pdf" assert sanitize_for_download(disguised) == "invoice_fdp.exe" assert portable_name_problem(disguised) + + +# ── Long paths on Windows ──────────────────────────────────────────────────── + +on_windows = pytest.mark.skipif(os.name != "nt", reason="the extended form is Windows'") +BS = chr(92) +EXT = BS * 2 + "?" + BS +DEEP = BS.join(["C:", *["a folder name of some length"] * 9, "track.flac"]) + + +@on_windows +def test_a_short_path_is_left_as_it_is(): + assert long_path(f"C:{BS}Music{BS}track.flac") == f"C:{BS}Music{BS}track.flac" + + +@on_windows +def test_a_long_path_gets_the_extended_form(): + assert long_path(DEEP) == EXT + DEEP + + +@on_windows +def test_the_extended_form_is_normalised_first(): + """The system normalises nothing behind the prefix: a forward slash or a + `..` left in would name a file that does not exist.""" + messy = DEEP.replace(BS, "/").replace("/track.flac", "/x/../track.flac") + assert long_path(messy) == EXT + DEEP + + +@on_windows +def test_a_long_share_path_gets_the_unc_form(): + share = BS * 2 + "server" + BS + "share" + DEEP[2:] + assert long_path(share) == EXT + "UNC" + BS + share[2:] + + +@on_windows +def test_a_path_in_the_extended_form_is_left_alone(): + assert long_path(EXT + DEEP) == EXT + DEEP + + +@on_windows +def test_a_directory_gets_it_before_max_path(): + """Creating a directory is refused twelve characters before MAX_PATH.""" + at_248 = "C:" + BS + "d" * 245 + assert len(at_248) == 248 + assert long_path(at_248) == EXT + at_248 + assert long_path(at_248[:-1]) == at_248[:-1] diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index 022553e..557b523 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -39,7 +39,7 @@ from pathlib import Path import uvicorn from meshbay_common import MNP_VERSION from meshbay_common.background import spawn -from meshbay_common.paths import fold +from meshbay_common.paths import fold, long_path from meshbay_common.protocol import MNP from meshbay_node import uploads as uploads_mod @@ -1209,8 +1209,8 @@ class NodeDaemon(EnrichmentMixin): for path in uploads_mod.orphaned_parts( uploads_mod.find_parts(roots.roots), live, when): try: - size = path.stat().st_size - path.unlink() + size = os.stat(long_path(path)).st_size + os.unlink(long_path(path)) except OSError as exc: log.warning("Could not remove abandoned upload %s: %s", path.name, exc) diff --git a/packages/meshbay-node/src/meshbay_node/enrichment.py b/packages/meshbay-node/src/meshbay_node/enrichment.py index 19aafba..c09c5e5 100644 --- a/packages/meshbay-node/src/meshbay_node/enrichment.py +++ b/packages/meshbay-node/src/meshbay_node/enrichment.py @@ -2,8 +2,11 @@ thumbnail for Videos, tags and cover for Music, thumbnail and EXIF for Photos. NodeDaemon runs it; the enrichers themselves are in indexer/.""" +import os from dataclasses import replace +from meshbay_common.paths import long_path + from meshbay_node.indexer import DirectoryIndexer, GroupIndex from meshbay_node.roots import entry_abs_path @@ -67,7 +70,7 @@ class EnrichmentMixin: if not _under_any_directory(entry.path, video_dirs): continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) @@ -153,7 +156,7 @@ class EnrichmentMixin: if owner is None: continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) @@ -236,7 +239,7 @@ class EnrichmentMixin: if not _under_any_directory(entry.path, photo_dirs): continue file_path = entry_abs_path(indexer.roots, entry) - if not file_path or not file_path.exists(): + if not file_path or not os.path.exists(long_path(file_path)): continue self._enriched_attempted.add((indexer.group_id, entry.id)) diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py index 9881298..53f1b27 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich.py @@ -13,14 +13,16 @@ the rest asynchronously and hands the result back via a callback. import asyncio import logging +import os from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from meshbay_node.indexer import title_parse -from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS +from meshbay_node.indexer.indexer import MEDIA_EXTENSIONS, walk_files from meshbay_node.media_cache import MediaCache from meshbay_node.media_probe import probe_video @@ -102,7 +104,8 @@ def _title_from_siblings(file_path: Path) -> str | None: from another Special just like this one. """ try: - names = sorted(p.name for p in file_path.parent.iterdir() if p.is_file()) + with os.scandir(long_path(file_path.parent)) as listing: + names = sorted(e.name for e in listing if e.is_file()) except OSError: return None checked = 0 @@ -142,8 +145,8 @@ def _synthetic_episode_number(file_path: Path, show_root: Path, season: int) -> """ try: candidates = sorted( - (p for p in show_root.rglob("*") - if p.is_file() and p.suffix.lower() in MEDIA_EXTENSIONS["video"]), + (p for p in walk_files(show_root) + if p.suffix.lower() in MEDIA_EXTENSIONS["video"]), key=lambda p: (str(p.parent), p.name), ) except OSError: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py index 397f760..ecc3b5e 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich_audio.py @@ -60,11 +60,13 @@ image-file fallback for cover art, same as everything else. import asyncio import logging +import os import re from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from mutagen import File as MutagenFile from mutagen.asf import ASF @@ -155,8 +157,9 @@ _WMP_ALBUMART_GUID_RE = re.compile(r"^albumart_\{([0-9a-f-]{36})\}_(large|small) def _find_sibling_cover(folder: Path) -> Path | None: try: - candidates = [p for p in folder.iterdir() - if p.is_file() and p.suffix.lower() in _COVER_EXTS] + with os.scandir(long_path(folder)) as listing: + candidates = [folder / e.name for e in listing + if e.is_file() and Path(e.name).suffix.lower() in _COVER_EXTS] except OSError: return None if not candidates: @@ -240,12 +243,12 @@ def _read_format_specific_tags(path: Path) -> tuple[dict, float | None] | None: suffix = path.suffix.lower() if suffix == ".wma": try: - return _asf_tags_from_object(ASF(str(path))) + return _asf_tags_from_object(ASF(long_path(path))) except Exception: return {}, None if suffix == ".mpc": try: - return _musepack_tags_from_object(Musepack(str(path))) + return _musepack_tags_from_object(Musepack(long_path(path))) except Exception: return {}, None return None @@ -257,7 +260,7 @@ def _read_generic_tags_and_cover( tags: dict = {} duration: float | None = None try: - easy = MutagenFile(str(path), easy=True) + easy = MutagenFile(long_path(path), easy=True) except Exception: easy = None if easy is not None: @@ -277,7 +280,7 @@ def _read_generic_tags_and_cover( cover: bytes | None = None if not skip_cover: try: - raw = MutagenFile(str(path)) + raw = MutagenFile(long_path(path)) except Exception: raw = None if raw is not None: @@ -339,7 +342,8 @@ def _read_sibling_cover(path: Path) -> bytes | None: if sibling is None: return None try: - return sibling.read_bytes() + with open(long_path(sibling), "rb") as f: + return f.read() except OSError: return None diff --git a/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py b/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py index 5cacef9..a9f85d0 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/enrich_photo.py @@ -26,6 +26,7 @@ from collections.abc import Awaitable, Callable from pathlib import Path import blake3 +from meshbay_common.paths import long_path from meshbay_common.protocol import IndexEntry from PIL import ExifTags, Image, ImageOps @@ -74,7 +75,7 @@ def _read_image(file_path: Path) -> tuple[bytes, int, int, int | None, str | Non *original* image's own dimensions — the thumbnail is a separate, resized copy, never what width/height describe. """ - with Image.open(file_path) as img: + with Image.open(long_path(file_path)) as img: taken_at = None camera = None try: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index e4679d4..55ec56a 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -82,7 +82,7 @@ def _is_indexable_name(name: str) -> bool: def _is_indexable(path: Path) -> bool: - return path.is_file() and _is_indexable_name(path.name) + return os.path.isfile(long_path(path)) and _is_indexable_name(path.name) # ── A disk that stops answering ─────────────────────────────────────────────── @@ -348,13 +348,25 @@ def _walk_root(root: Root) -> list[Path]: every other thing the daemon is doing — WebRTC sessions, chat, the admin UI — for as long as it takes. """ + return walk_files(root.path) + + +def walk_files(top: Path) -> list[Path]: + """ + Every file below `top`, links to files included, links to directories not + entered. Raises if `top` itself cannot be listed. Blocking. + + Each directory is listed under its long form and the paths returned are + built from the plain one, so a directory past Windows' MAX_PATH is listed + and everything downstream still sees, stores and compares plain paths. + """ files: list[Path] = [] - pending = [root.path] + pending = [top] while pending: directory = pending.pop() _beat(directory) try: - with os.scandir(directory) as listing: + with os.scandir(long_path(directory)) as listing: below: list[Path] = [] for entry in listing: _beat() @@ -363,13 +375,13 @@ def _walk_root(root: Root) -> list[Path]: # a file counts whether or not it is one. try: if entry.is_dir(follow_symlinks=False): - below.append(Path(entry.path)) + below.append(directory / entry.name) elif entry.is_file(): - files.append(Path(entry.path)) + files.append(directory / entry.name) except OSError: continue except OSError as e: - if directory == root.path: + if directory == top: raise # One directory that cannot be listed costs that directory. It # used to be silent for a refusal and to end the whole walk for @@ -406,7 +418,7 @@ def _size_files(files: list[Path]) -> list[tuple[Path, os.stat_result]]: for p in files: _beat(p.parent) try: - sized.append((p, p.stat())) + sized.append((p, os.stat(long_path(p)))) except OSError: continue return sized @@ -420,7 +432,7 @@ def _scan_file(root: Root, file_path: Path) -> IndexEntry | None: if not _is_indexable(file_path): return None try: - stat = file_path.stat() + stat = os.stat(long_path(file_path)) if not _is_indexable_size(file_path, stat.st_size): return None if stat.st_size > _PARTIAL_THRESHOLD: @@ -903,7 +915,7 @@ class DirectoryIndexer: if not _is_indexable(file_path): return None try: - st = file_path.stat() + st = os.stat(long_path(file_path)) except OSError: return None elif not _is_indexable_name(file_path.name): @@ -988,7 +1000,7 @@ class DirectoryIndexer: if self._cache is None or not user_id: return try: - st = file_path.stat() + st = os.stat(long_path(file_path)) except OSError: # Gone between the rename and here. Nothing to attribute, and # nothing for anyone to delete either. @@ -1565,7 +1577,7 @@ class DirectoryIndexer: return True for path, size, mtime in known: try: - st = Path(path).stat() + st = os.stat(long_path(path)) except OSError: continue if st.st_size == size and (mtime is None or st.st_mtime == mtime): @@ -1735,7 +1747,7 @@ class DirectoryIndexer: # bookkeeping still has to reach the `fire()` that actually runs. if not deleted and key not in self._burst_sizes: try: - size = file_path.stat().st_size + size = os.stat(long_path(file_path)).st_size except OSError: size = 0 if self._burst_inflight <= 0: @@ -1790,7 +1802,8 @@ class DirectoryIndexer: covers this directory too. """ try: - return not dir_path.exists() and dir_path.parent.is_dir() + return (not os.path.exists(long_path(dir_path)) + and os.path.isdir(long_path(dir_path.parent))) except OSError: return False diff --git a/packages/meshbay-node/src/meshbay_node/ops/files.py b/packages/meshbay-node/src/meshbay_node/ops/files.py index 4609627..a448103 100644 --- a/packages/meshbay-node/src/meshbay_node/ops/files.py +++ b/packages/meshbay-node/src/meshbay_node/ops/files.py @@ -4,8 +4,11 @@ from __future__ import annotations import asyncio import logging +import os from pathlib import Path +from meshbay_common.paths import long_path + from meshbay_node.ops.core import OpError, _group_ctx log = logging.getLogger("meshbay_node.ops") @@ -41,7 +44,7 @@ async def delete_file(state: dict, group_id: str, file_id: str) -> dict: f"not readable right now — the file is frozen, not gone", status=409) try: - path.unlink() + os.unlink(long_path(path)) except FileNotFoundError: # Already gone from disk; drop the stale entry rather than refusing. log.warning("Index named a file that is not on disk: %s", path) @@ -104,7 +107,7 @@ async def prune_index_cache(state: dict) -> dict: return True if any(not r.available for r in owning): return False - return not path.exists() + return not os.path.exists(long_path(path)) paths = await cache.all_paths() stale = await asyncio.to_thread(lambda: [p for p in paths if _is_stale(p)]) diff --git a/packages/meshbay-node/src/meshbay_node/roots.py b/packages/meshbay-node/src/meshbay_node/roots.py index 61fdc9b..94560f5 100644 --- a/packages/meshbay-node/src/meshbay_node/roots.py +++ b/packages/meshbay-node/src/meshbay_node/roots.py @@ -36,7 +36,7 @@ from concurrent.futures import ThreadPoolExecutor from dataclasses import dataclass, field from pathlib import Path -from meshbay_common.paths import fold, portable_name_problem +from meshbay_common.paths import fold, long_path, portable_name_problem log = logging.getLogger(__name__) @@ -78,7 +78,7 @@ def _free_name(directory: Path, filename: str, or another upload (C5a). """ def free(name: str) -> bool: - return name not in taken and not (directory / name).exists() + return name not in taken and not os.path.exists(long_path(directory / name)) if free(filename): return filename @@ -109,17 +109,17 @@ def publish_upload(part: Path, directory: Path, stored_name: str, filename: str, for _ in range(8): target = directory / name try: - os.link(part, target) + os.link(long_path(part), long_path(target)) except FileExistsError: name = _free_name(directory, filename, taken) continue except OSError: - if target.exists(): + if os.path.exists(long_path(target)): name = _free_name(directory, filename, taken) continue - part.rename(target) + os.rename(long_path(part), long_path(target)) return name - part.unlink() + os.unlink(long_path(part)) return name raise FileExistsError(stored_name) @@ -195,7 +195,7 @@ class Root: def is_live(self) -> bool: """Readable right now. The question `available` caches.""" try: - return self.path.is_dir() + return os.path.isdir(long_path(self.path)) except OSError: return False diff --git a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py index 715448f..3974a2f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/quic_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/quic_server.py @@ -48,6 +48,7 @@ from meshbay_common.handshake import ( quic_binding, verify_proof, ) +from meshbay_common.paths import long_path from meshbay_common.protocol import MNP, file_chunk_wire from meshbay_node.indexer import GroupIndex @@ -434,7 +435,7 @@ class _MNPServerProtocol(QuicConnectionProtocol): if file_path is None: self._send(stream_id, {"type": "error", "detail": ROOT_NOT_SERVED}) return - if not file_path.exists(): + if not os.path.exists(long_path(file_path)): self._send(stream_id, {"type": "error", "detail": "File not on disk"}) return @@ -463,7 +464,7 @@ def _read_and_encrypt( file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it, in the one shape every transport uses.""" - with open(file_path, "rb") as f: + with open(long_path(file_path), "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py index e298942..386b81f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/disk.py @@ -1,8 +1,15 @@ -"""Blocking disk work the session hands to `off_disk`.""" +""" +Blocking disk work the session hands to `off_disk`. +Every call takes the path's long form (`long_path`): a file past Windows' +MAX_PATH is indexed, so it is served, deleted and written beside like any other. +""" + +import os import shutil from pathlib import Path +from meshbay_common.paths import long_path from meshbay_common.protocol import file_chunk_wire from meshbay_node.roots import ROOT_NOT_SERVED, RootSet, entry_abs_path @@ -22,7 +29,7 @@ def _locate(roots: RootSet, entry) -> tuple[Path | None, str | None]: path = entry_abs_path(roots, entry) if path is None: return None, ROOT_NOT_SERVED - if not path.exists(): + if not os.path.exists(long_path(path)): return None, "File not on disk" return path, None @@ -37,15 +44,16 @@ def _mkdir_if_absent(target: Path) -> str | None: `mkdir` would raise where a refusal was meant. Blocking; called through `off_disk`. """ - if target.exists(): + if os.path.exists(long_path(target)): return "Already exists" - target.mkdir(parents=False) + os.mkdir(long_path(target)) return None def _is_empty_dir(target: Path) -> bool: """Blocking; called through `off_disk`.""" - return not any(target.iterdir()) + with os.scandir(long_path(target)) as listing: + return not any(listing) def _rmdir_if_empty(target: Path) -> bool: @@ -57,9 +65,10 @@ def _rmdir_if_empty(target: Path) -> bool: operator's browser, and a file can land in between. Two awaits here would reopen the same window one size smaller. Blocking; called through `off_disk`. """ - if any(target.iterdir()): - return False - target.rmdir() + with os.scandir(long_path(target)) as listing: + if any(listing): + return False + os.rmdir(long_path(target)) return True @@ -71,7 +80,7 @@ def _read_and_encrypt( file_id: str = "", ) -> dict: """Read one chunk off disk and encrypt it. Blocking; called through `off_disk`.""" - with open(file_path, "rb") as f: + with open(long_path(file_path), "rb") as f: f.seek(chunk_index * CHUNK_SIZE) plaintext = f.read(CHUNK_SIZE) return file_chunk_wire(gek, plaintext, chunk_index, file_hash, file_id) @@ -80,10 +89,24 @@ def _read_and_encrypt( def _free_bytes(directory: Path) -> int: """Space left on the filesystem holding `directory`. Blocking; called through `off_disk`, since a `statvfs` wakes a sleeping disk like any stat.""" - return shutil.disk_usage(directory).free + return shutil.disk_usage(long_path(directory)).free def _append_chunk(tmp_path: Path, chunk_bytes: bytes, first: bool) -> None: """Add one chunk to a partial upload. Blocking; called through `off_disk`.""" - with open(tmp_path, "wb" if first else "ab") as f: + with open(long_path(tmp_path), "wb" if first else "ab") as f: f.write(chunk_bytes) + + +def _unlink(path: Path, missing_ok: bool = False) -> None: + """Remove one file. Blocking; called through `off_disk`.""" + try: + os.unlink(long_path(path)) + except FileNotFoundError: + if not missing_ok: + raise + + +def _exists(path: Path) -> bool: + """Blocking; called through `off_disk`.""" + return os.path.exists(long_path(path)) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py index eafdaed..9c5fab5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -17,6 +17,7 @@ from meshbay_node.transport.webrtc.disk import ( _mkdir_if_absent, _read_and_encrypt, _rmdir_if_empty, + _unlink, ) from meshbay_node.transport.webrtc.limits import CHUNK_SIZE, LEASE_GRANTED, LEASE_QUEUED from meshbay_node.transport.wire import index_sync_message, list_dirs_off_loop @@ -439,7 +440,7 @@ class FilesMixin: self._send({"type": "error", "detail": ROOT_NOT_SERVED}) return if file_path is not None: - await off_disk(ctx["roots"], file_path.unlink) + await off_disk(ctx["roots"], _unlink, file_path) log.info("File deleted: %s", entry.name) self._audit("file_delete", entry.name) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py index c052703..0d5bb0e 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/upload_handlers.py @@ -17,7 +17,7 @@ from meshbay_node.roots import ( publish_upload, shell_active, ) -from meshbay_node.transport.webrtc.disk import _append_chunk, _free_bytes +from meshbay_node.transport.webrtc.disk import _append_chunk, _exists, _free_bytes, _unlink from meshbay_node.transport.webrtc.limits import LEASE_GRANTED, LEASE_NONE, LEASE_QUEUED log = logging.getLogger("meshbay_node.transport.webrtc_server") @@ -365,7 +365,7 @@ class UploadMixin: if chunk_index == 0: # Backstop: _free_name already guarantees this, and it stays because # it asserts the invariant where the write happens. - if await off_disk(roots, final_path.exists): + if await off_disk(roots, _exists, final_path): _refuse("File already exists", "already_exists") return # Asked once, before the first byte, from the size the client @@ -396,7 +396,7 @@ class UploadMixin: if state.bytes + len(chunk_bytes) > self._max_upload_bytes(): uploads.drop(user_id, rel_dir, filename) - await off_disk(roots, tmp_path.unlink, True) + await off_disk(roots, _unlink, tmp_path, True) _refuse("Upload exceeds size limit", "too_large") return @@ -410,7 +410,7 @@ class UploadMixin: # space that ran out. The client starts over from zero, which is # what it would have to do for a resume against a full disk anyway. uploads.drop(user_id, rel_dir, filename) - await off_disk(roots, tmp_path.unlink, True) + await off_disk(roots, _unlink, tmp_path, True) log.warning("Upload refused: the disk holding %s is full", rel_dir) self._audit("upload_refused", "disk_full") _refuse("The node's disk is full", "disk_full") diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py index 77c5d62..75d27c1 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/wire.py +++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py @@ -32,6 +32,7 @@ from pathlib import Path from meshbay_common import MNP_VERSION from meshbay_common.groupbox import PURPOSE_INDEX, seal +from meshbay_common.paths import long_path from meshbay_common.protocol import MNP, index_entry_wire from meshbay_node.roots import RootSet, off_disk @@ -75,22 +76,40 @@ def _dir_roots(roots: RootSet) -> list[tuple[str, Path, bool]]: def _walk_dirs(dir_roots: list[tuple[str, Path, bool]]) -> list[str]: - """Blocking. Directories only: `os.walk` reads each one once and never - stats a file, which `rglob("*")` and an `is_dir()` per entry did.""" + """Blocking. Directories only: each is read once and no file is stat'ed, + which `rglob("*")` and an `is_dir()` per entry did. What `os.walk` did, + except that each directory is listed under its long form: `os.walk` skipped + whatever lay past Windows' MAX_PATH, without a word.""" out: list[str] = [] for name, path, available in dir_roots: out.append(name) if not available: continue - for top, subdirs, _files in os.walk(path): - # Pruned in place: nothing under a hidden directory is listed either. - subdirs[:] = [d for d in subdirs if not d.startswith(".")] - rel = Path(top).relative_to(path) - for d in subdirs: - out.append(f"{name}/{(rel / d).as_posix()}") + pending = [path] + while pending: + top = pending.pop() + try: + with os.scandir(long_path(top)) as listing: + # Nothing under a hidden directory is listed either. + subdirs = [e for e in listing + if not e.name.startswith(".") and _is_dir(e)] + except OSError: + continue + rel = top.relative_to(path) + for entry in subdirs: + out.append(f"{name}/{(rel / entry.name).as_posix()}") + if not entry.is_symlink(): # listed, not entered, as os.walk did + pending.append(top / entry.name) return sorted(out)[:MAX_DIRS] +def _is_dir(entry: os.DirEntry) -> bool: + try: + return entry.is_dir() + except OSError: + return False + + def index_sync_message(index, roots: RootSet | None, hidden=(), dirs: list[str] | None = None) -> dict: """ diff --git a/packages/meshbay-node/src/meshbay_node/uploads.py b/packages/meshbay-node/src/meshbay_node/uploads.py index c485e39..a0f65dc 100644 --- a/packages/meshbay-node/src/meshbay_node/uploads.py +++ b/packages/meshbay-node/src/meshbay_node/uploads.py @@ -24,6 +24,7 @@ build first. from __future__ import annotations +import os import re import secrets import time @@ -31,6 +32,10 @@ from collections.abc import Iterable from dataclasses import dataclass, field from pathlib import Path +from meshbay_common.paths import long_path + +from meshbay_node.indexer.indexer import walk_files + # What an unfinished upload is called on disk while it is being written. The # node has always used this; it is named here because the reaper below has to # recognise one, and a second spelling of it would be a bug nobody could see. @@ -208,14 +213,14 @@ def find_parts(roots: Iterable) -> list[tuple[Path, float]]: if not getattr(root, "available", False): continue try: - candidates = root.path.rglob(f"*{PART_SUFFIX}") + candidates = walk_files(root.path) except OSError: continue for path in candidates: + if not path.name.endswith(PART_SUFFIX): + continue try: - if not path.is_file(): - continue - found.append((path, path.stat().st_mtime)) + found.append((path, os.stat(long_path(path)).st_mtime)) except OSError: continue return found diff --git a/packages/meshbay-node/tests/test_disk_io_off_loop.py b/packages/meshbay-node/tests/test_disk_io_off_loop.py index c0ab052..cacf936 100644 --- a/packages/meshbay-node/tests/test_disk_io_off_loop.py +++ b/packages/meshbay-node/tests/test_disk_io_off_loop.py @@ -260,7 +260,7 @@ def test_no_handler_touches_the_disk_on_the_loop(): # Written to block, and reached only through `off_disk`. on_the_disk_thread = {"_locate", "_append_chunk", "_read_and_encrypt", "_mkdir_if_absent", "_is_empty_dir", "_rmdir_if_empty", - "safe_subdir"} + "_unlink", "_exists", "safe_subdir"} # ffmpeg's own output goes through `_read_scratch_capped` and # `_discard_scratch` on a worker thread — `asyncio.to_thread` and not # `off_disk`, because a temp file is not a group root and has no platter to diff --git a/packages/meshbay-node/tests/test_indexer_silent_disk.py b/packages/meshbay-node/tests/test_indexer_silent_disk.py index bb618fc..4073ac5 100644 --- a/packages/meshbay-node/tests/test_indexer_silent_disk.py +++ b/packages/meshbay-node/tests/test_indexer_silent_disk.py @@ -133,14 +133,14 @@ async def test_a_file_the_system_refuses_is_listed_with_what_it_said( tmp_path, monkeypatch): films = _folder(tmp_path, "Films", ["a.mkv", "locked.mkv"]) idx = _indexer(RootSet.build([{"path": str(films)}])) - real = indexer_mod.long_path - def refuse(path): + def refuse(path, *args, **kw): + # What a locked file does: it lists and stats, and refuses to open. if Path(path).name == "locked.mkv": raise PermissionError(13, "Permission denied") - return real(path) + return open(path, *args, **kw) - monkeypatch.setattr(indexer_mod, "long_path", refuse) + monkeypatch.setattr(indexer_mod, "open", refuse, raising=False) await idx.initial_scan() assert _names(idx) == {"a.mkv"} @@ -149,7 +149,7 @@ async def test_a_file_the_system_refuses_is_listed_with_what_it_said( # A refusal costs nothing to ask again, and permissions change without the # file changing: it is not remembered the way a silence is. - monkeypatch.setattr(indexer_mod, "long_path", real) + monkeypatch.delattr(indexer_mod, "open") await idx.reconcile() assert "locked.mkv" in _names(idx) assert idx.unread == {} diff --git a/packages/meshbay-node/tests/test_long_paths.py b/packages/meshbay-node/tests/test_long_paths.py new file mode 100644 index 0000000..9f7876a --- /dev/null +++ b/packages/meshbay-node/tests/test_long_paths.py @@ -0,0 +1,400 @@ +""" +Files whose full path is longer than Windows' MAX_PATH (260 characters). + +A deep library reaches it without trying: "Artist / Album (year, edition) / +01 - a long track title.flac" below a folder that is already deep. With the +system's default setting (`LongPathsEnabled` off) every call given a plain path +that long refuses it, whatever the program declares, and such a file was +invisible to the members: the walk could not list its directory, and said so +only in the log. The node now hands the system the extended form (`\\\\?\\`) at +each call that touches a group's disk, which works whatever the setting, and +keeps the plain form everywhere else: in the index, the hash cache, the log. + +ffmpeg and ffprobe are given the plain form on purpose: they extend a long path +themselves, which the probe and thumbnail tests below hold. + +These run only where the limit exists: with the setting on, a plain path works +too and they would prove nothing. +""" + +import asyncio +import os +import shutil +import sys +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.protocol import file_chunk_plaintext +from meshbay_node import ops +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.roots import RootSet, publish_upload, safe_subdir +from meshbay_node.transport.webrtc.disk import ( + _append_chunk, + _free_bytes, + _is_empty_dir, + _locate, + _mkdir_if_absent, + _read_and_encrypt, + _rmdir_if_empty, +) + + +def _limit_in_force() -> bool: + if sys.platform != "win32": + return False + import winreg + try: + with winreg.OpenKey(winreg.HKEY_LOCAL_MACHINE, + r"SYSTEM\CurrentControlSet\Control\FileSystem") as key: + return winreg.QueryValueEx(key, "LongPathsEnabled")[0] == 0 + except OSError: + return True + + +pytestmark = pytest.mark.skipif( + not _limit_in_force(), reason="MAX_PATH binds only on Windows with LongPathsEnabled off") + +GID = "g" * 32 +SEGMENT = "A folder name as long as the ones a real library has" + + +def _x(path: Path) -> str: + """The extended form, for the test's own setup and checks.""" + return "\\\\?\\" + str(path) + + +def _write(path: Path, data: bytes) -> Path: + with open(_x(path), "wb") as f: + f.write(data) + return path + + +def _read(path: Path) -> bytes: + with open(_x(path), "rb") as f: + return f.read() + + +@pytest.fixture +def library(tmp_path): + """A root with one short file at the top and a directory past the limit. + Removed here: the plain-path cleanup pytest does would stop at it.""" + root = tmp_path / "Library" + deep = root + while len(str(deep)) < 280: + deep = deep / SEGMENT + os.makedirs(_x(deep)) + _write(root / "short.txt", b"near the top") + yield root, deep + shutil.rmtree(_x(root), ignore_errors=True) + + +def _indexer(roots: RootSet, **kw) -> DirectoryIndexer: + return DirectoryIndexer(roots=roots, group_id=GID, + sk_node=Ed25519PrivateKey.generate(), gek=None, **kw) + + +def _names(idx: DirectoryIndexer) -> set[str]: + return {e.name for e in idx.index.entries} + + +def _entry(idx: DirectoryIndexer, name: str): + (entry,) = [e for e in idx.index.entries if e.name == name] + return entry + + +def _ffmpeg() -> str: + exe = shutil.which("ffmpeg") + if exe is None: + pytest.skip("ffmpeg is not on PATH") + return exe + + +async def _make_media(args: list[str], out: Path) -> None: + proc = await asyncio.create_subprocess_exec( + _ffmpeg(), "-v", "error", "-y", *args, str(out), + stdout=asyncio.subprocess.DEVNULL, stderr=asyncio.subprocess.PIPE) + _, err = await proc.communicate() + assert proc.returncode == 0, err.decode(errors="replace") + + +def test_the_fixture_is_past_the_limit(library): + """The premise: a plain path this long is refused on this machine.""" + _, deep = library + track = _write(deep / "track.mkv", b"deep down") + assert len(str(track)) > 260 + assert not os.path.exists(str(track)) + with pytest.raises(OSError): + os.scandir(str(deep)) + + +# ── Indexing ──────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_a_scan_indexes_a_file_past_the_limit(library): + root, deep = library + _write(deep / "track.mkv", b"deep down" * 100) + idx = _indexer(RootSet.build([{"path": str(root)}])) + + await idx.initial_scan() + + assert _names(idx) == {"short.txt", "track.mkv"} + assert _entry(idx, "track.mkv").size == 900 + assert idx.unread == {} + + +@pytest.mark.asyncio +async def test_reconcile_finds_a_new_one_and_leaves_a_known_one_alone(library): + """Reconcile compares the disk with the index by path. Were the two sides + to spell a long path differently, it would rewrite the entry at every + pass and push the change to every member.""" + root, deep = library + _write(deep / "track.mkv", b"deep down") + idx = _indexer(RootSet.build([{"path": str(root)}])) + await idx.initial_scan() + + _write(deep / "later.mkv", b"added while the node ran") + await idx.reconcile() + assert "later.mkv" in _names(idx) + + version = idx.index.version + await idx.reconcile() + assert idx.index.version == version, "a pass with nothing new changed the index" + + +@pytest.mark.asyncio +async def test_a_watcher_event_indexes_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.01) + await idx.initial_scan() + idx._loop = asyncio.get_running_loop() + + _write(deep / "arrived.mkv", b"one more") + idx._schedule_update(deep / "arrived.mkv") + for _ in range(200): + if "arrived.mkv" in _names(idx): + break + await asyncio.sleep(0.01) + + assert "arrived.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_the_real_watcher_sees_it(library): + root, deep = library + idx = _indexer(RootSet.build([{"path": str(root)}]), debounce_secs=0.05) + await idx.start() + try: + _write(deep / "watched.mkv", b"seen by the system's own notification") + for _ in range(100): + if "watched.mkv" in _names(idx): + break + await asyncio.sleep(0.05) + finally: + await idx.stop() + + assert "watched.mkv" in _names(idx) + + +@pytest.mark.asyncio +async def test_a_root_that_is_itself_past_the_limit(library): + _, deep = library + _write(deep / "track.mkv", b"deep down") + roots = RootSet.build([{"path": str(deep)}]) + assert roots.roots[0].is_live() + idx = _indexer(roots) + + await idx.initial_scan() + + assert _names(idx) == {"track.mkv"} + + +# ── Serving and deleting ──────────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_it_is_served_to_a_member(library): + root, deep = library + data = os.urandom(5000) + _write(deep / "track.mkv", data) + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + assert refusal is None + msg = _read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == data + + +@pytest.mark.asyncio +async def test_it_is_served_over_quic_too(library): + from meshbay_node.transport import quic_server + + root, deep = library + _write(deep / "track.mkv", b"over the other transport") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + entry = _entry(idx, "track.mkv") + gek = os.urandom(32) + + path, refusal = _locate(roots, entry) + msg = quic_server._read_and_encrypt(gek, path, 0, bytes.fromhex(entry.id), entry.id) + + assert file_chunk_plaintext(gek, msg) == b"over the other transport" + + +@pytest.mark.asyncio +async def test_the_operator_can_delete_it(library): + root, deep = library + track = _write(deep / "track.mkv", b"to be deleted") + roots = RootSet.build([{"path": str(root)}]) + idx = _indexer(roots) + await idx.initial_scan() + state = {"groups_ctx": {GID: {"index": idx.index, "roots": roots}}} + + await ops.delete_file(state, GID, _entry(idx, "track.mkv").id) + + assert not os.path.exists(_x(track)) + assert "track.mkv" not in _names(idx) + + +# ── Uploads and folders ───────────────────────────────────────────────────── + +def test_an_upload_lands_in_a_directory_past_the_limit(library): + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + rel = "/".join(["Library", *deep.relative_to(root).parts]) + target = safe_subdir(roots, rel) + assert target == deep + + assert _free_bytes(target) > 0 + part = target / "up.bin.0123abcd.part" + _append_chunk(part, b"first ", True) + _append_chunk(part, b"second", False) + assert publish_upload(part, target, "up.bin", "up.bin") == "up.bin" + assert _read(target / "up.bin") == b"first second" + + again = target / "up.bin.4567cdef.part" + _append_chunk(again, b"same name", True) + assert publish_upload(again, target, "up.bin", "up.bin") == "up (2).bin", \ + "a long path made the existing file look absent, and it was replaced" + assert _read(target / "up.bin") == b"first second" + + +def test_an_abandoned_upload_is_found(library): + from meshbay_node import uploads + + root, deep = library + roots = RootSet.build([{"path": str(root), "writable": True}]) + part = _write(deep / "film.mkv.89abcdef.part", b"half a film") + + found = [p for p, _ in uploads.find_parts(roots.roots)] + + assert found == [part] + + +def test_members_see_the_folders_past_the_limit(library): + """The folder tree a member browses and uploads into.""" + from meshbay_node.transport.wire import list_dirs + + root, deep = library + os.makedirs(_x(deep / "Deeper still")) + roots = RootSet.build([{"path": str(root)}]) + + dirs = list_dirs(roots) + + rel = "/".join(["Library", *deep.relative_to(root).parts]) + assert rel in dirs + assert f"{rel}/Deeper still" in dirs + + +def test_a_folder_is_created_and_removed_past_the_limit(library): + _, deep = library + folder = deep / "New folder" + assert _mkdir_if_absent(folder) is None + assert _mkdir_if_absent(folder) == "Already exists" + assert _is_empty_dir(folder) + assert _rmdir_if_empty(folder) + assert not os.path.exists(_x(folder)) + + +# ── Enrichment ────────────────────────────────────────────────────────────── + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_audio_tags_and_the_cover_beside_it_are_read(library): + from meshbay_node.indexer.enrich_audio import _read_tags_and_cover + from PIL import Image + + _, deep = library + track = deep / "01 - A Long Track Title.flac" + await _make_media(["-f", "lavfi", "-i", "sine=duration=1", + "-metadata", "title=Some Song", "-metadata", "artist=Some Artist", + "-metadata", "album=Some Album"], track) + with open(_x(deep / "cover.jpg"), "wb") as f: + Image.new("RGB", (8, 8), "red").save(f, "JPEG") + + tags, duration, cover = await asyncio.to_thread(_read_tags_and_cover, track) + + assert tags["title"] == "Some Song" and tags["album"] == "Some Album" + assert duration == pytest.approx(1.0, abs=0.1) + assert cover == _read(deep / "cover.jpg") + + +@pytest.mark.asyncio +async def test_a_photo_is_read(library): + from meshbay_node.indexer.enrich_photo import _read_image + from PIL import Image + + _, deep = library + with open(_x(deep / "photo.jpg"), "wb") as f: + Image.new("RGB", (64, 48), "blue").save(f, "JPEG") + + thumb, width, height, _, _ = await asyncio.to_thread(_read_image, deep / "photo.jpg") + + assert (width, height) == (64, 48) and thumb + + +@pytest.mark.asyncio +@pytest.mark.daemon_loop # a subprocess needs the proactor loop +async def test_ffmpeg_probes_and_thumbnails_it_from_the_plain_path(library): + from meshbay_node.indexer.enrich import _make_thumbnail + from meshbay_node.media_probe import probe_video + + _, deep = library + clip = deep / "clip.mp4" + await _make_media(["-f", "lavfi", "-i", "testsrc=size=64x48:rate=5:duration=2", + "-pix_fmt", "yuv420p"], clip) + + probe = await probe_video(str(clip)) + thumb = await _make_thumbnail(clip, probe.duration) + + assert probe.duration == pytest.approx(2.0, abs=0.3) + assert thumb and thumb[:2] == b"\xff\xd8" + + +@pytest.mark.asyncio +async def test_a_show_folder_past_the_limit_numbers_its_episodes(library): + from meshbay_node.indexer.enrich import ( + _season_and_show_from_ancestors, + _synthetic_episode_number, + _title_from_siblings, + ) + + _, deep = library + season = deep / "Some Show" / "Season 1" + os.makedirs(_x(season)) + _write(season / "Some Show S01E01.mkv", b"") + _write(season / "Pilot without a number.mkv", b"") + _write(season / "Second without a number.mkv", b"") + later = season / "Second without a number.mkv" + + found = _season_and_show_from_ancestors(later) + assert found is not None + number, show = found + assert await asyncio.to_thread(_synthetic_episode_number, later, show, number) == 2 + assert await asyncio.to_thread(_title_from_siblings, later) == "Some Show" |