diff options
Diffstat (limited to 'packages/meshbay-node/src/meshbay_node/transport')
5 files changed, 70 insertions, 26 deletions
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: """ |