diff options
Diffstat (limited to 'packages')
5 files changed, 349 insertions, 14 deletions
diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index dcfa8f8..9428ddd 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -393,6 +393,11 @@ class NodeDaemon: # _reconcile_loop) so the backstop is prompt again now # that someone is actually looking. "note_activity": indexer.note_activity, + # Bound method, called when an upload finishes. The entry + # it belongs to does not exist yet (see + # webrtc_server._register_uploader), so the indexer keeps + # the record and stamps the entry when it creates it. + "record_upload": indexer.record_upload, # Shown to the operator in Settings, and kept current in # place by set_scan_settings (ops.py) — same reasoning as # enabled_apps below. @@ -879,6 +884,7 @@ class NodeDaemon: "index": indexer.index, "progress": indexer.progress, "note_activity": indexer.note_activity, + "record_upload": indexer.record_upload, "reconcile_interval_secs": scan_settings["reconcile_interval_secs"], "debounce_secs": scan_settings["debounce_secs"], "visibility": group_cfg.visibility, diff --git a/packages/meshbay-node/src/meshbay_node/indexer/cache.py b/packages/meshbay-node/src/meshbay_node/indexer/cache.py index c167db9..6c87f40 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/cache.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/cache.py @@ -23,6 +23,7 @@ only daemon.py's wiring changed. """ import logging +import time from dataclasses import dataclass from pathlib import Path @@ -40,6 +41,25 @@ CREATE TABLE IF NOT EXISTS files ( added_at INTEGER NOT NULL, hash_version INTEGER NOT NULL DEFAULT 1 ); + +-- Who sent a file. Written when an upload finishes, read when the file is +-- indexed — two different moments, and the second is much later: the watchdog +-- debounces for two seconds and then hashes, so the entry does not exist yet +-- when the last chunk lands. Recording it in memory would also lose it at every +-- restart, where the index is rebuilt from disk, and an owner the node forgets +-- is an owner who cannot delete their own file tomorrow. +-- +-- Validated against a live stat() exactly as `files` is: a row whose size or +-- mtime no longer match is a different file at that path, and attributes +-- nothing. That is what makes a row left behind by a deleted file harmless. +CREATE TABLE IF NOT EXISTS uploads ( + path TEXT PRIMARY KEY, + size INTEGER NOT NULL, + mtime REAL NOT NULL, + user_id TEXT NOT NULL, + pk_ed25519 TEXT NOT NULL DEFAULT '', + at INTEGER NOT NULL +); """ _MIGRATE_V2 = "ALTER TABLE files ADD COLUMN hash_version INTEGER NOT NULL DEFAULT 1" @@ -59,6 +79,11 @@ class IndexCache: def __init__(self, db_path: Path): self._db_path = db_path self._db: aiosqlite.Connection | None = None + # Whether `uploads` holds anything at all. Every entry the indexer + # builds asks this cache who sent the file, and on a node that has + # never received an upload — most of them, most of the time — that is + # one query per file per scan for an answer that is always None. + self._has_uploads = False async def open(self) -> None: self._db_path.parent.mkdir(parents=True, exist_ok=True) @@ -69,6 +94,10 @@ class IndexCache: except Exception: pass # column already exists await self._db.commit() + async with self._db.execute( + "SELECT EXISTS(SELECT 1 FROM uploads)") as cur: + row = await cur.fetchone() + self._has_uploads = bool(row and row[0]) async def close(self) -> None: if self._db: @@ -114,6 +143,45 @@ class IndexCache: (path, mtime, size, hash, type, added_at, hash_version)) await self._db.commit() + # ── Who sent a file ────────────────────────────────────────────────────── + + async def record_upload(self, path: str, size: int, mtime: float, + user_id: str, pk_ed25519: str) -> None: + """Remember that this account put this file here. + + Written once the upload is complete and the file is at its final name, + never partway through — a `.part` is not indexable and would key a row + to a path that is about to change. + """ + await self._db.execute( + "INSERT INTO uploads (path, size, mtime, user_id, pk_ed25519, at) " + "VALUES (?, ?, ?, ?, ?, ?) " + "ON CONFLICT(path) DO UPDATE SET " + "size = excluded.size, mtime = excluded.mtime, " + "user_id = excluded.user_id, pk_ed25519 = excluded.pk_ed25519, " + "at = excluded.at", + (path, size, mtime, user_id, pk_ed25519, int(time.time()))) + await self._db.commit() + self._has_uploads = True + + async def uploader(self, path: str, size: int, + mtime: float) -> tuple[str, str] | None: + """Who sent the file currently at this path, or None. + + The size and mtime are matched exactly, like `lookup`: the path alone + would credit whoever last uploaded *a* file of that name for whatever + occupies the name now — including something the operator put there + themselves afterwards, which would hand a member the right to delete it. + """ + if not self._has_uploads: + return None + async with self._db.execute( + "SELECT user_id, pk_ed25519 FROM uploads " + "WHERE path = ? AND size = ? AND mtime = ?", + (path, size, mtime)) as cur: + row = await cur.fetchone() + return (row[0], row[1]) if row else None + # ── Maintenance (node admin UI "prune index cache") ────────────────────── async def count(self) -> int: diff --git a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py index 33e7210..852c061 100644 --- a/packages/meshbay-node/src/meshbay_node/indexer/indexer.py +++ b/packages/meshbay-node/src/meshbay_node/indexer/indexer.py @@ -436,7 +436,7 @@ class DirectoryIndexer: cached = await self._cache.lookup( str(file_path), st.st_size, st.st_mtime, expected_hv) if cached is not None: - return IndexEntry( + return await self._attribute(IndexEntry( id=cached.hash, name=file_path.name, path=_virtual_dir(root, file_path), @@ -444,7 +444,7 @@ class DirectoryIndexer: type=cached.type, added_at=cached.added_at, hash_version=cached.hash_version, - ) + ), file_path, st) loop = asyncio.get_event_loop() entry = await loop.run_in_executor(self._executor, _scan_file, root, file_path) @@ -452,8 +452,51 @@ class DirectoryIndexer: await self._cache.put(str(file_path), st.st_size, st.st_mtime, entry.id, entry.type, entry.added_at, entry.hash_version) + return await self._attribute(entry, file_path, st) + + async def _attribute(self, entry: IndexEntry | None, file_path: Path, + st) -> IndexEntry | None: + """Stamp an entry with whoever sent the file, if a member did. + + Here, rather than beside each `add_entry`, because this is the one + funnel every entry passes through: the initial scan, the watchdog, + reconcile and a replug all build theirs from `_hash_or_cached`. + + The attribution used to be written at the end of the *upload* instead, + by walking the index for an entry that by construction did not exist + yet — the watchdog has not fired, and the `.part` the file was until the + rename is not indexable. It matched nothing, silently, so every uploaded + file was owned by nobody and `file_delete` refused everyone but the + operator, where MESHBAY_DESIGN.md §5.4 grants it to any non-revoked + device of the uploading account. + """ + if entry is None or self._cache is None: + return entry + who = await self._cache.uploader(str(file_path), st.st_size, st.st_mtime) + if who is not None: + entry.uploader_id, entry.uploader_pk = who return entry + async def record_upload(self, file_path: Path, user_id: str, + pk_ed25519: str) -> None: + """Remember who sent this file, for the entry that does not exist yet. + + Called by the transport once the last chunk has landed and the file is + at its final name. Durable rather than in-memory: the index is rebuilt + from disk at every start, and an owner the node forgets on restart is an + owner who cannot delete their own file tomorrow. + """ + if self._cache is None or not user_id: + return + try: + st = file_path.stat() + except OSError: + # Gone between the rename and here. Nothing to attribute, and + # nothing for anyone to delete either. + return + await self._cache.record_upload( + str(file_path), st.st_size, st.st_mtime, user_id, pk_ed25519 or "") + def _report_collisions(self) -> None: """ Names that are the same file on a case-insensitive filesystem. @@ -749,7 +792,15 @@ class DirectoryIndexer: if old is None: continue for field in self._ENRICHED_FIELDS: - setattr(entry, field, getattr(old, field)) + # What the rescan itself established wins. `uploader_id` and + # `uploader_pk` now come off the durable record (`_attribute`), + # and the entry being replaced is memory this process happens + # to still hold — so copying over them would let a stale blank + # beat the thing that survives a restart. Every other field is + # None on a freshly scanned entry, so for those this is exactly + # the carry-over it has always been. + if getattr(entry, field) is None: + setattr(entry, field, getattr(old, field)) # It came back intact, so it is not one of the entries the daemon # needs to enrich again. self.rescanned_ids.discard(entry.id) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py index 29d6e7f..8df88d8 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -4914,26 +4914,29 @@ class WebRTCPeerSession: log.info("Upload complete: %s (%d chunks, %d bytes)", stored_name, total_chunks, state.bytes) self._audit("file_upload", f"{rel_dir}/{stored_name}") - self._register_uploader(ctx, rel_dir, stored_name) + self._register_uploader(ctx, final_path) - def _register_uploader(self, ctx: dict, rel_dir: str, filename: str) -> None: + def _register_uploader(self, ctx: dict, file_path: Path) -> None: """ - Tag the index entry with the uploader's identity after upload completes. + Record who sent this file, for the index entry that does not exist yet. - The key recorded here is the one this node pinned, not the one the token + The key recorded is the one this node pinned, not the one the token carried. `pk_user` was a hub-chosen claim, and it decided who could later delete the file: a hub issuing a token naming its own key could delete anyone's uploads on any node. Deletion is supposed to be authorized by the node, and this closes the last place where it was not. + + **The entry is not here to be tagged.** This used to walk `ctx["index"]` + for the name just written and set the fields on it; at this point the + watchdog has not fired (it debounces for two seconds and then hashes) + and the file was a `.part` until the line above, which is not indexable + — so the walk matched nothing, every time, and said nothing about it. + The indexer stamps the entry from this record when it creates it. """ - idx = ctx.get("index") - if not idx: + record = ctx.get("record_upload") + if record is None: return - for entry in idx.entries: - if entry.name == filename and entry.path == rel_dir: - entry.uploader_id = self._user_id - entry.uploader_pk = self._pinned_pk - return + self._spawn(record(file_path, self._user_id or "", self._pinned_pk or "")) def _do_file_delete(self, msg: dict) -> None: ctx = self._group_ctx() diff --git a/packages/meshbay-node/tests/test_upload_attribution.py b/packages/meshbay-node/tests/test_upload_attribution.py new file mode 100644 index 0000000..dc8d989 --- /dev/null +++ b/packages/meshbay-node/tests/test_upload_attribution.py @@ -0,0 +1,207 @@ +""" +Who owns an uploaded file, and who may therefore delete it. + +MESHBAY_DESIGN.md §5.4 grants `file_delete` to "the operator, **or any +non-revoked device of the uploading account**". That second half needs the +index entry to record an uploader, and nothing recorded one: the transport +tagged the entry at the end of the upload, walking `ctx["index"]` for the name +it had just written — at a moment when, by construction, no such entry exists. +The file was a `.part` until the rename on the line above (excluded from the +index), and the watchdog that will index it debounces for two seconds and then +hashes. The walk matched nothing, returned silently, and every uploaded file in +every group was owned by nobody: `_do_file_delete` refuses a caller with no +admin authority when the entry records no uploader, so an ordinary member could +not delete what they had just sent. + +The suite did not see it because every test of ownership sets `uploader_id` on +an entry by hand — which tests `_verify_uploader_sig`, and nothing about how a +real upload ever comes to have an uploader. These tests cross that seam: a real +`file_upload` through the real handler, a real `DirectoryIndexer` over the same +directory, and the entry it produces. + +The second property is the one a restart decides. The index is rebuilt from +disk at every start, so an attribution held in memory is an owner the node +forgets overnight — the file would be deletable by its uploader today and not +tomorrow, which is worse than never having offered it. +""" + +import asyncio +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_node.indexer.cache import IndexCache +from meshbay_node.indexer.indexer import DirectoryIndexer +from meshbay_node.transport.webrtc_server import WebRTCPeerSession + +from conftest import one_root, sealed_upload + +GROUP = "g" * 32 +UPLOADER = "user-1" +UPLOADER_PK = "cGluc2V0LWtleQ==" # the key this node pinned, base64 + + +async def _indexer(shared: Path, cache: IndexCache) -> DirectoryIndexer: + """A real indexer over `shared`, watching, with a short debounce. + + Short but not zero: the debounce is the thing that puts the entry's creation + after the upload's last chunk, which is the whole subject here. + """ + indexer = DirectoryIndexer( + roots=one_root(shared), group_id=GROUP, + sk_node=Ed25519PrivateKey.generate(), gek=generate_gek(), + cache=cache, debounce_secs=0.2, + ) + await indexer.initial_scan() + await indexer.start() + return indexer + + +def _session(shared: Path, indexer: DirectoryIndexer, gek: bytes, + *, user_id: str = UPLOADER) -> WebRTCPeerSession: + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": one_root(shared), + "index": indexer.index, + "sk_node": indexer.sk_node, + "gek": gek, + # The seam under test: the transport hands the record to the indexer, + # which stamps the entry when it finally creates it. + # + # `getattr` rather than the attribute, deliberately: against the source + # this test was written for there is no such seam at all, and a test + # that dies of AttributeError there proves only that a method is + # missing. Tolerating its absence makes the pre-fix run reach the + # assertions and fail on the property — no uploader on the entry — + # which is the thing being guarded. + "record_upload": getattr(indexer, "record_upload", None), + } + session._group_id = GROUP + session._user_id = user_id + session._pinned_pk = UPLOADER_PK + session._pk_user = "" + session._uploads = {} + session._tasks = set() + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +async def _upload(session, shared: Path, name: str, data: bytes) -> None: + root_name = shared.name + session._do_file_upload( + sealed_upload(session, filename=name, data=data, dir=root_name)) + errors = [m for m in session.sent if m.get("type") == "error"] + assert not errors, errors + # The record is written by a task the handler spawned. + await asyncio.gather(*list(session._tasks)) + + +async def _entry_for(indexer: DirectoryIndexer, name: str, timeout: float = 5.0): + """The index entry for a file, once the indexer has got to it.""" + deadline = asyncio.get_event_loop().time() + timeout + while asyncio.get_event_loop().time() < deadline: + for entry in indexer.index.entries: + if entry.name == name: + return entry + await asyncio.sleep(0.05) + return None + + +@pytest.mark.asyncio +async def test_an_uploaded_file_records_who_sent_it(tmp_path): + shared = tmp_path / "shared" + shared.mkdir() + gek = generate_gek() + async with IndexCache(tmp_path / "cache.db") as cache: + indexer = await _indexer(shared, cache) + try: + session = _session(shared, indexer, gek) + await _upload(session, shared, "holiday.jpg", b"JPEGDATA" * 64) + + entry = await _entry_for(indexer, "holiday.jpg") + assert entry is not None, "the file was never indexed at all" + assert entry.uploader_id == UPLOADER, ( + "an uploaded file with no uploader is a file its own sender " + "cannot delete — §5.4 grants that to the uploading account") + assert entry.uploader_pk == UPLOADER_PK + finally: + await indexer.stop() + + +@pytest.mark.asyncio +async def test_the_uploader_survives_a_restart(tmp_path): + """A second indexer over the same directory and the same cache. + + This is what a node restart is: the index is rebuilt from disk, and every + field not on the disk has to come from somewhere durable. + """ + shared = tmp_path / "shared" + shared.mkdir() + gek = generate_gek() + async with IndexCache(tmp_path / "cache.db") as cache: + first = await _indexer(shared, cache) + try: + session = _session(shared, first, gek) + await _upload(session, shared, "report.txt", b"TEXT" * 64) + assert await _entry_for(first, "report.txt") is not None + finally: + await first.stop() + + second = await _indexer(shared, cache) + try: + entry = await _entry_for(second, "report.txt") + assert entry is not None + assert entry.uploader_id == UPLOADER, ( + "the attribution did not survive the rebuild, so the uploader " + "could delete their file today and not tomorrow") + finally: + await second.stop() + + +@pytest.mark.asyncio +async def test_a_different_file_at_the_same_path_inherits_nothing(tmp_path): + """The record is keyed by path, and a path is not an identity. + + A member uploads, the operator deletes it and puts a file of their own + there under the same name. Nothing about that second file was sent by the + member, and crediting them would hand them the right to delete it. + """ + shared = tmp_path / "shared" + shared.mkdir() + gek = generate_gek() + async with IndexCache(tmp_path / "cache.db") as cache: + indexer = await _indexer(shared, cache) + try: + session = _session(shared, indexer, gek) + await _upload(session, shared, "notes.txt", b"SENT" * 64) + assert await _entry_for(indexer, "notes.txt") is not None + + (shared / "notes.txt").unlink() + (shared / "notes.txt").write_bytes(b"THE OPERATOR'S OWN FILE") + + await asyncio.sleep(0.6) + entry = await _entry_for(indexer, "notes.txt") + assert entry is not None + assert not entry.uploader_id, ( + "a file the operator put there is not the member's to delete") + finally: + await indexer.stop() + + +@pytest.mark.asyncio +async def test_a_file_nobody_uploaded_has_no_uploader(tmp_path): + """The operator's own library is not attributed to anyone.""" + shared = tmp_path / "shared" + shared.mkdir() + (shared / "already-here.txt").write_bytes(b"ON DISK BEFORE ANY MEMBER") + async with IndexCache(tmp_path / "cache.db") as cache: + indexer = await _indexer(shared, cache) + try: + entry = await _entry_for(indexer, "already-here.txt") + assert entry is not None + assert not entry.uploader_id and not entry.uploader_pk + finally: + await indexer.stop() |