1 files changed, 207 insertions, 0 deletions
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()