From 07480eb3f8ad0bb4369ac8c41df7c4140b108d0e Mon Sep 17 00:00:00 2001 From: Christophe Besson Date: Mon, 28 Sep 2026 21:49:14 +0200 Subject: feat: nodes apply the content blocklist in their public groups A node hosting a public group syncs the hub's blocklist on every connection (paged, node token only) and applies pushed changes. A blocked file leaves the index and is refused (content_blocked); private groups are untouched. The unused per-hash check route is gone. Co-Authored-By: Claude Opus 5.5 --- docs/MESHBAY_DESIGN.md | 30 ++- docs/MESHBAY_NODE_PROTOCOL.md | 6 + packages/meshbay-hub/src/meshbay_hub/api/deps.py | 11 + .../meshbay-hub/src/meshbay_hub/api/moderation.py | 68 +++--- .../meshbay-hub/src/meshbay_hub/api/revocation.py | 29 +++ packages/meshbay-hub/tests/test_moderation.py | 103 ++++++++- .../tests/test_unauthenticated_surface.py | 2 - .../meshbay-node/src/meshbay_node/blocklist.py | 83 +++++++ packages/meshbay-node/src/meshbay_node/daemon.py | 64 +++++- .../meshbay-node/src/meshbay_node/hub_client.py | 31 +++ .../meshbay_node/transport/webrtc/apps/music.py | 2 + .../transport/webrtc/apps/streaming.py | 2 + .../transport/webrtc/apps/subtitles.py | 2 + .../src/meshbay_node/transport/webrtc/core.py | 19 ++ .../src/meshbay_node/transport/webrtc/files.py | 12 +- .../src/meshbay_node/transport/wire.py | 16 +- .../meshbay-node/tests/test_content_blocklist.py | 238 +++++++++++++++++++++ 17 files changed, 661 insertions(+), 57 deletions(-) create mode 100644 packages/meshbay-node/src/meshbay_node/blocklist.py create mode 100644 packages/meshbay-node/tests/test_content_blocklist.py diff --git a/docs/MESHBAY_DESIGN.md b/docs/MESHBAY_DESIGN.md index 85e436b..2a0e354 100644 --- a/docs/MESHBAY_DESIGN.md +++ b/docs/MESHBAY_DESIGN.md @@ -1987,6 +1987,32 @@ and is refused when public groups are off.** An unauthenticated endpoint that blocklists a content hash after two reports is a network-wide censorship and DoS primitive for anyone who learns a public file's id. +**The content blocklist is applied by the nodes that host a public group, in their +public groups only.** A node holds the list (`blocklist.py`, persisted beside the +denylist so a restart while the hub is unreachable does not serve again what had +stopped being served), fetches the whole of it on every connection to the hub — +`GET /v1/blocklist`, paged by hash, answered to a node's own token only — and +receives each addition and removal pushed on its hub socket (`blocklist_update`), +sent only to nodes registered for a public group. In a public group a blocked file +leaves the index members are sent, and a request for it, its thumbnail, a stream of +it, a subtitle track or an audio conversion of it is refused (`content_blocked`); +the members connected when the list changes are resent the index. Nothing is +deleted: the file is on the operator's disk, and what they keep is theirs. + +Stated per the convention at the top: + +- **Private groups are untouched**, by construction: no node sends the hub a + content hash (**H7**), so nothing in a private group can be on the list. +- **Which groups are public is the node's own configuration.** The hub and the + node are given the same value when a group is created and the hub never changes + it; an operator who edits `node.toml` to call a hub-listed group private takes + it out of the list's reach. +- **It is an exact match on the content id.** A file changed by one byte is + another id, and a file past the partial-hash threshold is identified by a sample + of its bytes (§6.3). It is a moderation tool, not a guarantee. +- **The QUIC transport does not apply it** — it is in development and serves no + client (§5.1, §15.3). + ### 7.6 Federation (MHP) > **Federation is closed in the code, and every MHP route refuses with a stated @@ -3267,7 +3293,7 @@ had already been asked. | **AV19** | **Nothing carries the path to the migrations.** `meshbay-hub migrate` derives it from the installed package, so the RPM, the DEB, a venv and a checkout all agree. A unit naming `alembic.ini` names a file whose `%(here)s` stops being true the moment packaging moves it | | **AV18** | **The hub runs on exactly one worker, and says so at startup.** `_connected_nodes`, `_node_groups`, `_webrtc_answers` and the relay registry are per-process: a second worker makes a node intermittently unreachable for half its members, which is a symptom that describes something else entirely | | **AV14** | **MHP binds its audience, and the hub reads its own identity at call time.** A token is minted for one peer and accepted by that peer only. `federation.py` bound `_hub_id` and `_hub_sk_pem` at import, which is before `load_hub_keypair` runs, so it signed with `None` and called itself `meshbay.org` whatever the instance was named — and the verifier named no audience for the `aud` the issuer sets, which PyJWT refuses outright. MHP could not complete one authenticated request between two hubs | -| **AV15** | **A hash is checked for shape before it is a key lookup**, on the unauthenticated blocklist endpoints a node consults | +| **AV15** | **A hash is checked for shape before it is a key lookup**, on every blocklist endpoint, the administrator's included | | **AV20** | **Chat is bounded in size and in rate, like every other member-supplied write** (§6.6). A message is a row on the operator's disk that nothing expires, a relayed copy for every connected member and a notification for every member of the group; the only ceiling was the frame size. Uploads had carried four protections and a cap since C5a because somebody asked what one member costs the others on that path, and nobody had asked it on this one | | **AV21** | **A lease is what the node granted, not what the client called it** (§5.5). `tr` was read as a boolean, so any non-empty string skipped the leaseless ceiling and every cap behind it, and a queued transfer was held back only by the honesty of the client waiting in the queue | | **AV22** | **The node's own controls take no authority from a hub token** (§6.7). `node_status`, `node_settings_set`, `roster_read`, `denylist_read`, `denylist_clear` and `node_reload` were gated on the account id in the JWT, which is the hub's to choose — NS4 and M3 with the check written the other way round. The gate is a proved operator device, which a hub holding no user keys cannot produce | @@ -3448,7 +3474,7 @@ process runs it — `systemctl --user` on Linux, Task Scheduler on Windows. | **A signed upload transcript** | Ownership is recorded by the node and verifiable by nobody else (§5.4). Making it provable is a transcript the uploader signs, stored with the entry — designed in outline, not built | | Forward secrecy in group chat | **Given up deliberately and on the record** (§4.5). If it becomes a requirement it belongs in 1:1 DM | | Metadata at the hub | Membership, and who posted in which group and when. A known leak, not a solved problem (§7.1) | -| **QUIC** | Off by default, and **not at parity**: it serves the index and file chunks with no transfer lease, no leaseless ceiling and no root-availability check, does its file I/O on the event loop, and returns exception text to the peer (**L3**). No client speaks it. Either it comes to parity or it goes; until then §5.1's "chat is the only gap" is the one sentence here that overstates the code | +| **QUIC** | Off by default, and **not at parity**: it serves the index and file chunks with no transfer lease, no leaseless ceiling, no root-availability check and no content blocklist, does its file I/O on the event loop, and returns exception text to the peer (**L3**). No client speaks it. Either it comes to parity or it goes; until then §5.1's "chat is the only gap" is the one sentence here that overstates the code | | **The relay registry** | **Closed in the code**: `relay.RELAYS_ENABLED` is False and every `/v1/relays` route answers 503, as federation does. Nothing in the tree calls them, node or client, and §11.1 measured two ISPs with no TURN relay needed. Kept code that nothing calls is what **L7** says not to keep; it stays only as the proof-of-possession design (**AV6**) until a node needs a relay or it is deleted | | **A very high bitrate wedges the player against a small buffer ceiling** | Where even the *floor* read-ahead does not fit — ninety seconds plus the minute kept behind, at the file's bitrate, above what the engine will hold — the film stalls: measured on the harness at 9.3 Mbit/s against a 100 MB ceiling, 100.8 s of film played in 900 s of wall clock. **Predates the byte budget and is unchanged by it**, to the tenth of a second; what the budget did change there is the refusal count, 1560 → 2. The fix is not a bound at all, it is a second stage of buffer outside the SourceBuffer, which means gating the append path — the riskiest change in this area and not one to make alongside another | | **The reconnect backoff only wakes on `visibilitychange`** | So a tab that stays visible through an outage — which is what a screen wake lock guarantees while a film is playing — waits out the full backoff, up to 30 s, after the network is already back. Nothing listens for `online` | diff --git a/docs/MESHBAY_NODE_PROTOCOL.md b/docs/MESHBAY_NODE_PROTOCOL.md index 5bf367b..0886a7e 100644 --- a/docs/MESHBAY_NODE_PROTOCOL.md +++ b/docs/MESHBAY_NODE_PROTOCOL.md @@ -159,6 +159,7 @@ Codes in use: | Chat | `chat_too_large`, `chat_rate_limited` (§11.7) | | Operator controls | `not_operator`, `too_many_pending`, `too_large` (§10.4) | | Metadata | `transcode_not_applicable`, `tmdb_search_rate_limited` (§11.9) | +| Moderation | `content_blocked` — a file the hub's content blocklist names, in a public group (§11.3) | Everything else refuses with `detail` alone. A code is added when a client has a different thing to *do* about the refusal — retry, re-authenticate, offer an update — @@ -1565,6 +1566,11 @@ ct = AES-256-GCM(chunk_key).encrypt(nonce, plaintext) no AAD * **A chunk that is not this shape aborts the download**, with an error. There is no fallback that decodes it some other way: a client that guesses at a chunk it does not recognise writes its guess into the file the person is saving. +* **In a public group, a file the hub's content blocklist names is not served.** It is + left out of `index_sync` and `index_delta`, and `file_req` for it or for its + thumbnail, `stream_req`, `subtitle_req` and `audio_transcode_req` are refused with + `content_blocked`. The node syncs the list from the hub and applies pushed changes + (`MESHBAY_DESIGN.md` §7.5); private groups are never affected. ### 11.4 Upload diff --git a/packages/meshbay-hub/src/meshbay_hub/api/deps.py b/packages/meshbay-hub/src/meshbay_hub/api/deps.py index 42f4101..2fd8b99 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/deps.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/deps.py @@ -90,6 +90,17 @@ async def require_user_scope( return current_user +async def require_node_scope( + payload: dict = Depends(_decode_token), + current_user: User = Depends(get_current_user), +) -> User: + """Only a node daemon's own token — for what a node fetches on its own behalf.""" + if payload.get("scope") != "node": + raise HTTPException(status_code=status.HTTP_403_FORBIDDEN, + detail="Node token required") + return current_user + + def user_is_admin(user: User) -> bool: """Admin by DB role or by the config allow-list. Use inside a handler that already depends on `require_moderator` but has to draw the admin line for diff --git a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py index 35c688c..c4e6af5 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/moderation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/moderation.py @@ -18,8 +18,9 @@ Admin endpoints: DELETE /v1/admin/blocklist/{hash} — remove a hash Node integration: - GET /v1/blocklist/check?hash= — check if a hash is blocked - GET /v1/blocklist — full blocklist (for node sync) + GET /v1/blocklist?after= — the list, paged, for a node's own token + WebSocket `blocklist_update` — additions and removals, pushed to nodes + hosting a public group """ import logging @@ -30,9 +31,10 @@ from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub import hub_settings -from meshbay_hub.api.deps import get_current_user, require_admin +from meshbay_hub.api.deps import get_current_user, require_admin, require_node_scope from meshbay_hub.api.middleware import limiter from meshbay_hub.api.netutil import client_ip +from meshbay_hub.api.revocation import broadcast_blocklist_update from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import ContentBlocklist, ContentReport, User @@ -47,6 +49,10 @@ router = APIRouter(tags=["moderation"]) AUTO_BLOCK_THRESHOLD = 3 +def _is_hash(value: str) -> bool: + return len(value) == 64 and all(c in "0123456789abcdef" for c in value) + + # ── Models ──────────────────────────────────────────────────────────────────── class ReportRequest(BaseModel): @@ -89,7 +95,7 @@ async def report_content( status_code=403, detail="This hub does not broker public content, so there is nothing to report here.") - if len(body.content_hash) != 64 or not all(c in "0123456789abcdef" for c in body.content_hash): + if not _is_hash(body.content_hash): raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)") # One vote per account per hash — a single reporter must not be able to walk @@ -115,6 +121,7 @@ async def report_content( .where(ContentReport.content_hash == body.content_hash)) or 0 action = "already_reported" if already else "logged" + auto_blocked = False if distinct_reporters >= AUTO_BLOCK_THRESHOLD: existing = await db.get(ContentBlocklist, body.content_hash) if not existing: @@ -124,10 +131,13 @@ async def report_content( added_by="auto", )) action = "auto_blocked" + auto_blocked = True log.warning("Content auto-blocked after %d distinct reporters: %s", distinct_reporters, body.content_hash[:16]) await db.commit() + if auto_blocked: + await broadcast_blocklist_update(db, add=[body.content_hash]) return { "status": action, "content_hash": body.content_hash, @@ -136,49 +146,31 @@ async def report_content( } -@router.get("/v1/blocklist/check") -@limiter.limit("120/minute") -async def check_blocklist( - hash: str, - request: Request, - db: AsyncSession = Depends(get_db), -): - """Check if a single hash is blocked. Used by nodes before serving public content. - - Unauthenticated, because a node consults it before serving public content - and does so on its own behalf. That makes the shape check worth having: - without it any string of any length became a primary-key lookup. - """ - if len(hash) != 64 or not all(c in "0123456789abcdef" for c in hash): - raise HTTPException(status_code=422, detail="hash must be 64 hex chars (blake3)") - blocked = await db.get(ContentBlocklist, hash) - return { - "blocked": blocked is not None, - "hash": hash, - "reason": blocked.reason if blocked else None, - } - - @router.get("/v1/blocklist") async def get_blocklist( + current_node: User = Depends(require_node_scope), db: AsyncSession = Depends(get_db), - # Bounded, like every other list. This one takes no authentication — a - # node syncs it at startup — and had no ceiling at all, so any stranger - # could ask for the table in one query, repeatedly. 10 000 is what a node - # asks for, so it is the default and also the most anyone may have. + after: str = Query(default="", max_length=64), limit: int = Query(default=10000, ge=1, le=10000), ): """ - Return the full blocklist. Nodes sync this on startup. - Returns hashes only (not reasons) to minimize data exposure. + The content blocklist, a page at a time, for a node hosting a public group. + + Hashes only, never the reasons. Ordered by hash so `after` (the last hash of + the previous page) is a stable cursor: a list longer than one page used to + be cut at 10 000 with no way to ask for the rest, and the node applying it + silently served everything past the cut. A node's own token, because this + is what a node fetches on its own behalf and nothing else asks for it. """ result = await db.execute( select(ContentBlocklist.content_hash) - .order_by(ContentBlocklist.added_at.desc()) + .where(ContentBlocklist.content_hash > after) + .order_by(ContentBlocklist.content_hash) .limit(limit) ) hashes = [row[0] for row in result.fetchall()] - return {"count": len(hashes), "hashes": hashes} + return {"hashes": hashes, + "next": hashes[-1] if len(hashes) == limit else None} # ── Admin endpoints ─────────────────────────────────────────────────────────── @@ -214,16 +206,19 @@ async def admin_add_blocklist( current_user: User = Depends(require_admin), db: AsyncSession = Depends(get_db), ): + if not _is_hash(body.content_hash): + raise HTTPException(status_code=422, detail="content_hash must be 64 hex chars (blake3)") existing = await db.get(ContentBlocklist, body.content_hash) if existing: raise HTTPException(status_code=409, detail="Hash already blocked") db.add(ContentBlocklist( content_hash=body.content_hash, - reason=body.reason, + reason=body.reason[:64], added_by=current_user.username, )) await db.commit() + await broadcast_blocklist_update(db, add=[body.content_hash]) return {"status": "blocked", "hash": body.content_hash} @@ -238,5 +233,6 @@ async def admin_remove_blocklist( raise HTTPException(status_code=404, detail="Hash not in blocklist") await db.delete(entry) await db.commit() + await broadcast_blocklist_update(db, remove=[content_hash]) return {"status": "unblocked", "hash": content_hash} diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index 2c0b8db..fe9edf3 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -165,6 +165,35 @@ async def broadcast_revocation(token: str) -> int: return sent +async def broadcast_blocklist_update(db: AsyncSession, *, add: list[str] = (), + remove: list[str] = ()) -> int: + """ + Tell every connected node that hosts a public group what changed on the + content blocklist. Returns how many were told. + + Only those nodes: the list names public content, and a node hosting only + private groups has nothing to apply it to. A node that is offline now syncs + the whole list when it next connects (`GET /v1/blocklist`), so a missed push + is only late, never lost. + """ + if not add and not remove: + return 0 + public = set((await db.execute( + select(Group.id).where(Group.visibility == "public"))).scalars().all()) + payload = json.dumps({"type": "blocklist_update", + "add": list(add), "remove": list(remove)}) + sent = 0 + for node_id, ws in list(_connected_nodes.items()): + if not public.intersection(_node_groups.get(node_id, [])): + continue + try: + await ws.send_text(payload) + sent += 1 + except Exception: + _connected_nodes.pop(node_id, None) + return sent + + def _sign_revocation(target: str, target_id: str, reason: str) -> str: """Issue a signed revocation token (JWT EdDSA).""" from meshbay_hub.auth import _hub_id, _hub_sk_pem diff --git a/packages/meshbay-hub/tests/test_moderation.py b/packages/meshbay-hub/tests/test_moderation.py index 3109e29..b328d67 100644 --- a/packages/meshbay-hub/tests/test_moderation.py +++ b/packages/meshbay-hub/tests/test_moderation.py @@ -22,6 +22,20 @@ async def _register_and_login(client, username: str) -> dict: return {"Authorization": f"Bearer {r.json()['access_token']}"} +async def _node_headers(client, username: str) -> dict: + """A node daemon's token for a fresh account — what a node syncs with.""" + from meshbay_hub.auth import issue_access_token + user = await _register_and_login(client, username) + me = (await client.get("/v1/users/me", headers=user)).json() + tok = issue_access_token(me["user_id"], ttl=3600, groups=[], scope="node") + return {"Authorization": f"Bearer {tok}"} + + +async def _blocked(client, h: str) -> bool: + node = await _node_headers(client, f"node_{h[:6]}_{len(h)}") + return h in (await client.get("/v1/blocklist", headers=node)).json()["hashes"] + + @pytest.fixture async def reporter(client): return await _register_and_login(client, "reporter_one") @@ -69,8 +83,7 @@ async def test_same_reporter_cannot_walk_the_threshold(client, reporter): assert r.json()["report_count"] == 1 assert r.json()["status"] == "already_reported" - check = await client.get(f"/v1/blocklist/check?hash={h}") - assert check.json()["blocked"] is False + assert not await _blocked(client, h) @pytest.mark.asyncio @@ -84,8 +97,7 @@ async def test_auto_block_on_distinct_reporters(client): assert r.json()["status"] == "auto_blocked" assert r.json()["report_count"] == 3 - check = await client.get(f"/v1/blocklist/check?hash={h}") - assert check.json()["blocked"] is True + assert await _blocked(client, h) @pytest.mark.asyncio @@ -118,14 +130,13 @@ async def test_admin_add_remove_blocklist(client, admin_headers): headers=admin_headers) assert r.status_code == 201 - r = await client.get(f"/v1/blocklist/check?hash={hash4}") - assert r.json()["blocked"] is True + assert await _blocked(client, hash4) r = await client.delete(f"/v1/admin/blocklist/{hash4}", headers=admin_headers) assert r.status_code == 200 - r = await client.get(f"/v1/blocklist/check?hash={hash4}") - assert r.json()["blocked"] is False + node = await _node_headers(client, "node_after_unblock") + assert hash4 not in (await client.get("/v1/blocklist", headers=node)).json()["hashes"] @pytest.mark.asyncio @@ -134,6 +145,80 @@ async def test_full_blocklist(client, admin_headers): await client.post("/v1/admin/blocklist", json={"content_hash": hash5, "reason": "test"}, headers=admin_headers) - r = await client.get("/v1/blocklist") + node = await _node_headers(client, "node_full") + r = await client.get("/v1/blocklist", headers=node) assert r.status_code == 200 assert hash5 in r.json()["hashes"] + + +@pytest.mark.asyncio +async def test_only_a_node_token_reads_the_blocklist(client, reporter): + assert (await client.get("/v1/blocklist")).status_code in (401, 422) + assert (await client.get("/v1/blocklist", headers=reporter)).status_code == 403 + + +@pytest.mark.asyncio +async def test_the_blocklist_pages_past_its_limit(client, admin_headers): + hashes = sorted(f"{i:064x}" for i in range(5)) + for h in hashes: + await client.post("/v1/admin/blocklist", + json={"content_hash": h, "reason": "test"}, + headers=admin_headers) + node = await _node_headers(client, "node_pager") + seen, after = [], "" + while True: + page = (await client.get(f"/v1/blocklist?limit=2&after={after}", + headers=node)).json() + seen += page["hashes"] + if not page["next"]: + break + after = page["next"] + assert [h for h in seen if h in hashes] == hashes + + +@pytest.mark.asyncio +async def test_an_admin_cannot_block_something_that_is_not_a_hash(client, admin_headers): + r = await client.post("/v1/admin/blocklist", + json={"content_hash": "x" * 5000, "reason": "test"}, + headers=admin_headers) + assert r.status_code == 422 + + +@pytest.mark.asyncio +async def test_a_change_is_pushed_to_nodes_hosting_a_public_group_only(db_session): + """A node hosting only private groups has nothing to apply the list to.""" + import json + + import meshbay_hub.api.revocation as rev + from meshbay_hub.db.models import Group, User + + db_session.add(User(id="owner-bl", username="owner_bl", email="x", hub_id="h", + pw_hash=b"x", pw_salt=b"x")) + db_session.add(Group(id="pub-bl", name="pub", admin_id="owner-bl", + visibility="public")) + db_session.add(Group(id="priv-bl", name="priv", admin_id="owner-bl", + visibility="private")) + await db_session.commit() + + class _WS: + def __init__(self): + self.sent = [] + + async def send_text(self, text): + self.sent.append(json.loads(text)) + + public_node, private_node = _WS(), _WS() + saved = dict(rev._connected_nodes), dict(rev._node_groups) + rev._connected_nodes.update({"n-pub": public_node, "n-priv": private_node}) + rev._node_groups.update({"n-pub": ["pub-bl"], "n-priv": ["priv-bl"]}) + try: + sent = await rev.broadcast_blocklist_update(db_session, add=["a" * 64]) + finally: + rev._connected_nodes.clear() + rev._connected_nodes.update(saved[0]) + rev._node_groups.clear() + rev._node_groups.update(saved[1]) + assert sent == 1 + assert public_node.sent == [{"type": "blocklist_update", "add": ["a" * 64], + "remove": []}] + assert private_node.sent == [] diff --git a/packages/meshbay-hub/tests/test_unauthenticated_surface.py b/packages/meshbay-hub/tests/test_unauthenticated_surface.py index f2da42b..78615ca 100644 --- a/packages/meshbay-hub/tests/test_unauthenticated_surface.py +++ b/packages/meshbay-hub/tests/test_unauthenticated_surface.py @@ -42,8 +42,6 @@ PUBLIC = { ("POST", "/v1/nodes/auth"): "node sign-in — Ed25519 signature over a fresh timestamp", ("WS", "/v1/nodes/ws"): "a node-scoped JWT in the first message, within a timeout", ("GET", "/v1/groups"): "the public directory — empty when public groups are off", - ("GET", "/v1/blocklist"): "nodes sync it on their own behalf — hashes only", - ("GET", "/v1/blocklist/check"): "nodes consult it on their own behalf", ("GET", "/v1/relays"): "closed: 503 while relay.RELAYS_ENABLED is False", ("POST", "/v1/relays/register"): "closed; when open, approved key + signature", ("GET", "/mhp/info"): "closed: 503 while federation.FEDERATION_ENABLED is False", diff --git a/packages/meshbay-node/src/meshbay_node/blocklist.py b/packages/meshbay-node/src/meshbay_node/blocklist.py new file mode 100644 index 0000000..29d8666 --- /dev/null +++ b/packages/meshbay-node/src/meshbay_node/blocklist.py @@ -0,0 +1,83 @@ +""" +The hub's content blocklist, as this node applies it (docs/MESHBAY_DESIGN.md §7.5). + +Hashes of public content that the hub's moderation has blocked. A node hosting a +public group stops serving those files there: they leave the index members are +sent, and a request for one is refused. Nothing is deleted from the operator's +disk — the node stops serving, and what the operator keeps is theirs to decide. + +Private groups are untouched. The hub never learns what a private group holds, +so there is nothing it could have blocked in one, and nothing here reaches them. + +Kept on disk as well as in memory, so a node that restarts while the hub is +unreachable does not serve again, meanwhile, what it had already stopped serving. +The hub's list is the authority: a full sync replaces this one. + +It is an exact match on the content id. A file changed by one byte is another +id, and files over the partial-hash threshold are identified by a sample of their +bytes (§6.3) — a moderation tool, not a guarantee. +""" + +import json +import logging +import os +import re +from pathlib import Path + +log = logging.getLogger(__name__) + +_HASH = re.compile(r"^[0-9a-f]{64}$") + + +class ContentBlocklist: + def __init__(self, path: Path | None = None): + self._path = path + self._hashes: set[str] = set() + if path is not None: + try: + data = json.loads(path.read_text(encoding="utf-8")) + self._hashes = {h for h in data.get("hashes", []) if _HASH.match(h)} + except FileNotFoundError: + pass + except (OSError, ValueError) as e: + # A damaged file is not a reason to refuse to start; the next + # sync with the hub rewrites it. + log.warning("Content blocklist %s unreadable: %s", path, e) + + def __contains__(self, content_hash: object) -> bool: + return content_hash in self._hashes + + def __len__(self) -> int: + return len(self._hashes) + + def __iter__(self): + return iter(self._hashes) + + def replace(self, hashes) -> bool: + """The hub's full list. True when it differs from what was applied.""" + new = {h for h in hashes if isinstance(h, str) and _HASH.match(h)} + if new == self._hashes: + return False + self._hashes = new + self._save() + return True + + def apply(self, add=(), remove=()) -> bool: + """One pushed change. True when it changed anything.""" + before = set(self._hashes) + self._hashes |= {h for h in add if isinstance(h, str) and _HASH.match(h)} + self._hashes -= {h for h in remove if isinstance(h, str)} + if self._hashes == before: + return False + self._save() + return True + + def _save(self) -> None: + if self._path is None: + return + tmp = self._path.with_suffix(".tmp") + try: + tmp.write_text(json.dumps({"hashes": sorted(self._hashes)}), encoding="utf-8") + os.replace(tmp, self._path) + except OSError as e: + log.warning("Content blocklist %s not saved: %s", self._path, e) diff --git a/packages/meshbay-node/src/meshbay_node/daemon.py b/packages/meshbay-node/src/meshbay_node/daemon.py index a29aa9f..45a6b9a 100644 --- a/packages/meshbay-node/src/meshbay_node/daemon.py +++ b/packages/meshbay-node/src/meshbay_node/daemon.py @@ -44,6 +44,7 @@ from meshbay_common.protocol import MNP from meshbay_node import uploads as uploads_mod from meshbay_node.audit import RETENTION_DAYS as AUDIT_RETENTION_DAYS from meshbay_node.audit import AuditStore +from meshbay_node.blocklist import ContentBlocklist from meshbay_node.bundle_store import BundleStore from meshbay_node.chat.store import ChatStore from meshbay_node.cli.dispatch import run, start @@ -114,6 +115,10 @@ class NodeDaemon(EnrichmentMixin): # Persisted so a restart does not silently un-revoke everyone (H4) self._denylist = ( Denylist(path=config.data_dir / "denylist.json") if Denylist else None) + # The hub's content blocklist, applied in public groups only (§7.5). + # Persisted for the same reason: a restart while the hub is unreachable + # must not serve again what had stopped being served. + self._blocklist = ContentBlocklist(config.data_dir / "blocklist.json") self._chat_stores: dict[str, ChatStore] = {} # One instance, shared by every group's DirectoryIndexer — see # indexer/cache.py's docstring for why this stopped being per-group. @@ -525,6 +530,7 @@ class NodeDaemon(EnrichmentMixin): self._webrtc._ctx["pk_x25519_b64"] = keys.pk_x25519_b64 self._webrtc._ctx["roster"] = self._roster + self._webrtc._ctx["blocklist"] = self._blocklist # The MNP adapter calls the same operations as the loopback API # (meshbay_node.ops), and those take the daemon's state. Handing # the transport a second set of lookups is how two paths to one @@ -611,11 +617,16 @@ class NodeDaemon(EnrichmentMixin): except Exception as e: log.warning("Invalid revocation token: %s", e) + async def on_connected(): + await self._sync_blocklist(hub) + ws_task = asyncio.create_task(hub.maintain_ws( on_incoming=on_incoming, on_revocation=on_revocation, on_webrtc_offer=on_webrtc_offer, group_ids=lambda: list((self._state.get("groups_ctx") or {}).keys()), + on_connected=on_connected, + on_blocklist=self._on_blocklist_update, )) self._tasks.append(ws_task) log.info("Hub WS task started") @@ -1448,9 +1459,10 @@ class NodeDaemon(EnrichmentMixin): # node refuses every handshake while the GEK is None (NS8) — so this is # "nobody is listening", not a case to send in clear for. if peers and idx.gek: - msg = (index_delta_message(idx, delta, indexer.roots) + hidden = self._blocklist if self._is_public(group_id) else () + msg = (index_delta_message(idx, delta, indexer.roots, hidden) if delta is not None - else index_sync_message(idx, indexer.roots)) + else index_sync_message(idx, indexer.roots, hidden)) pushed = 0 for session in peers: try: @@ -1462,6 +1474,54 @@ class NodeDaemon(EnrichmentMixin): log.info("Index %s pushed to %d WebRTC peers", "delta" if delta is not None else "sync", pushed) + # ── Content blocklist (public groups, §7.5) ────────────────────────────── + + def _is_public(self, group_id: str) -> bool: + return any(g.id == group_id and g.visibility == "public" + for g in self._config.groups) + + def _hosts_public_group(self) -> bool: + return any(g.visibility == "public" for g in self._config.groups) + + async def _sync_blocklist(self, hub) -> None: + """The hub's whole list, on every connection. Only a node hosting a + public group asks: nothing else here could have been blocked.""" + if not self._hosts_public_group(): + return + try: + hashes = await hub.fetch_blocklist() + except Exception as e: + # Keep applying the list on disk; the next connection tries again. + log.warning("Content blocklist sync failed: %s", e) + return + if self._blocklist.replace(hashes): + log.info("Content blocklist: %d hashes", len(self._blocklist)) + self._push_public_indexes() + + def _on_blocklist_update(self, add: list, remove: list) -> None: + if not self._hosts_public_group(): + return + if self._blocklist.apply(add, remove): + log.info("Content blocklist updated: +%d -%d", len(add), len(remove)) + self._push_public_indexes() + + def _push_public_indexes(self) -> None: + """Resend each public group's whole index, so what the list now hides + leaves every connected member's view, and what it released comes back.""" + if not self._webrtc: + return + for indexer in self._indexers: + idx = indexer.index + if not self._is_public(idx.group_id) or not idx.gek: + continue + msg = index_sync_message(idx, indexer.roots, self._blocklist) + for session in list(self._webrtc._sessions.values()): + if session._group_id == idx.group_id: + try: + session._send(msg) + except Exception: + pass + def _drop_group_sessions(self, group_id: str) -> None: """Close live sessions for a revoked group (H4).""" if not self._webrtc or not group_id: diff --git a/packages/meshbay-node/src/meshbay_node/hub_client.py b/packages/meshbay-node/src/meshbay_node/hub_client.py index 5f47090..2ce8993 100644 --- a/packages/meshbay-node/src/meshbay_node/hub_client.py +++ b/packages/meshbay-node/src/meshbay_node/hub_client.py @@ -334,6 +334,8 @@ class HubClient: on_revocation: Any = None, on_webrtc_offer: Any = None, group_ids: list[str] | None = None, # static list or callable returning one + on_connected: Any = None, + on_blocklist: Any = None, ) -> None: """ Maintain a persistent WebSocket connection to the hub. @@ -393,6 +395,12 @@ class HubClient: self._ws = ws log.info("Hub WS connected") + if on_connected: + # Off the read loop, for the reason offers are: a slow + # fetch must not stop this socket being read. + task = asyncio.create_task(on_connected()) + pending.add(task) + task.add_done_callback(pending.discard) async for raw in ws: msg = json.loads(raw) @@ -405,6 +413,9 @@ class HubClient: elif mtype == "revocation" and on_revocation: on_revocation(msg.get("token", "")) + elif mtype == "blocklist_update" and on_blocklist: + on_blocklist(msg.get("add") or [], msg.get("remove") or []) + elif mtype == "webrtc_offer" and on_webrtc_offer: # Answered off the read loop on purpose. Awaiting the # handler here meant one slow negotiation stopped the @@ -460,6 +471,26 @@ class HubClient: log.warning("Could not deliver WebRTC answer to %s: %s", str(msg.get("peer_id"))[:8], e) + # ── Content blocklist (public groups) ───────────────────────────────── + + async def fetch_blocklist(self, max_pages: int = 100) -> set[str]: + """The hub's content blocklist, every page of it (docs/MESHBAY_DESIGN.md §7.5).""" + if self._session is None: + raise RuntimeError("Not logged in") + await self.ensure_fresh_token() + hashes: set[str] = set() + after = "" + for _ in range(max_pages): + r = await self._http.get("/v1/blocklist", params={"after": after}, + headers=self._session.auth_headers) + r.raise_for_status() + page = r.json() + hashes.update(page.get("hashes") or []) + after = page.get("next") or "" + if not after: + return hashes + raise RuntimeError("Content blocklist longer than expected; not applied") + # ── Convenience: full startup sequence ─────────────────────────────────── async def startup(self, endpoint_hint: str | None = None) -> HubSession: diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py index f2b6fa5..857db33 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/music.py @@ -64,6 +64,8 @@ class MusicMixin: """ ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py index a425c65..4337e24 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/streaming.py @@ -228,6 +228,8 @@ class StreamingMixin: async def _stream_video_inner(self, msg: dict) -> None: ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py index bd2ab52..70781eb 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/apps/subtitles.py @@ -39,6 +39,8 @@ class SubtitlesMixin: """ ctx = self._group_ctx() file_id = msg.get("file_id", "") + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: self._send({"type": "error", "detail": "File not found"}) diff --git a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py index c3fb623..882ddbc 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/core.py @@ -246,6 +246,25 @@ class SessionCore: return self._ctx["groups"].get(self._group_id) or {} return self._ctx + def _hidden_ids(self): + """The content blocklist, in a public group; nothing anywhere else. + + A private group's content never reaches the hub, so nothing in it can have + been blocked there (docs/MESHBAY_DESIGN.md §7.5, `meshbay_node.blocklist`). + """ + if self._group_ctx().get("visibility") != "public": + return () + return self._ctx.get("blocklist") or () + + def _refuse_blocked(self, file_id) -> bool: + """Refuse a file the blocklist names, in a public group. True if refused.""" + if not isinstance(file_id, str) or file_id not in self._hidden_ids(): + return False + self._send({"type": "error", "detail": "This file is not available here.", + "code": "content_blocked", "file_id": file_id}) + self._audit("content_blocked", file_id[:16]) + return True + def _register_peer(self) -> None: """Add this connection to its group's peer set. 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 e76e23e..f629563 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc/files.py @@ -179,7 +179,7 @@ class FilesMixin: def _do_index_sync(self) -> None: ctx = self._group_ctx() - self._send(index_sync_message(ctx["index"], ctx.get("roots"))) + self._send(index_sync_message(ctx["index"], ctx.get("roots"), self._hidden_ids())) async def _try_serve_thumbnail( self, thumb_hash: str, chunk_index: int, gek: bytes | None, @@ -236,8 +236,18 @@ class FilesMixin: self._note_unleased(tr) file_id = msg["file_id"] chunk_index = msg["chunk_index"] + if self._refuse_blocked(file_id): + return entry = ctx["index"].get_entry(file_id) if not entry: + # A blocked file's own thumbnail is a preview of it. + hidden = self._hidden_ids() + if hidden and file_id in { + e.thumb_hash for e in map(ctx["index"].get_entry, hidden) + if e is not None and e.thumb_hash}: + self._send({"type": "error", "detail": "This file is not available here.", + "code": "content_blocked", "file_id": file_id}) + return thumb = await self._try_serve_thumbnail(file_id, chunk_index, ctx.get("gek")) if thumb is not None: log.debug("file_req file_id=%s chunk=%s: served as thumbnail", diff --git a/packages/meshbay-node/src/meshbay_node/transport/wire.py b/packages/meshbay-node/src/meshbay_node/transport/wire.py index 1f9c925..258edb5 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/wire.py +++ b/packages/meshbay-node/src/meshbay_node/transport/wire.py @@ -65,7 +65,7 @@ def list_dirs(roots: RootSet | None) -> list[str]: return sorted(out)[:MAX_DIRS] -def index_sync_message(index, roots: RootSet | None) -> dict: +def index_sync_message(index, roots: RootSet | None, hidden=()) -> dict: """ The full `index_sync` message for one group. @@ -73,10 +73,13 @@ def index_sync_message(index, roots: RootSet | None) -> dict: without them a folder someone just created, or one they emptied, does not exist as far as a client is concerned, and a member cannot tell "the drive is unplugged" from "it is all still there". + + `hidden` is the content blocklist in a public group (`meshbay_node.blocklist`): + those entries are not sent at all. """ payload = { "version": index.version, - "entries": [index_entry_wire(e) for e in index.entries], + "entries": [index_entry_wire(e) for e in index.entries if e.id not in hidden], "dirs": list_dirs(roots), "roots": roots.describe() if roots else [], } @@ -88,7 +91,7 @@ def index_sync_message(index, roots: RootSet | None) -> dict: } -def index_delta_message(index, delta, roots=None) -> dict: +def index_delta_message(index, delta, roots=None, hidden=()) -> dict: """ One `index_delta` — what changed since the last thing this node broadcast. @@ -103,13 +106,16 @@ def index_delta_message(index, delta, roots=None) -> dict: page: the delta that told them something had changed was the one message that could not say what. It is a handful of dicts, bounded by the number of directories a group has, and it is sealed with the rest. + + `hidden`, as for `index_sync_message`: a blocked entry is never added or + updated; its deletion still goes out. """ payload = { "base_version": delta.base_version, "version": delta.version, - "additions": [index_entry_wire(e) for e in delta.additions], + "additions": [index_entry_wire(e) for e in delta.additions if e.id not in hidden], "deletions": list(delta.deletions), - "updates": [index_entry_wire(e) for e in delta.updates], + "updates": [index_entry_wire(e) for e in delta.updates if e.id not in hidden], } if roots is not None: payload["roots"] = roots.describe() diff --git a/packages/meshbay-node/tests/test_content_blocklist.py b/packages/meshbay-node/tests/test_content_blocklist.py new file mode 100644 index 0000000..0e07787 --- /dev/null +++ b/packages/meshbay-node/tests/test_content_blocklist.py @@ -0,0 +1,238 @@ +""" +The hub's content blocklist, applied by a node in its public groups +(docs/MESHBAY_DESIGN.md §7.5, `meshbay_node.blocklist`). + +A blocked file leaves the index members are sent and is refused if asked for — +in a public group, and nowhere else: a private group's content never reaches the +hub, so nothing there can have been blocked. +""" + +import struct + +import msgpack +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from meshbay_common.crypto import generate_gek +from meshbay_common.groupbox import PURPOSE_INDEX, unseal +from meshbay_common.protocol import IndexEntry +from meshbay_node.blocklist import ContentBlocklist +from meshbay_node.hub_client import HubClient +from meshbay_node.indexer import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from meshbay_node.transport.wire import index_delta_message, index_sync_message + +GROUP = "g" * 32 +BLOCKED = "b" * 64 +KEPT = "c" * 64 +THUMB = "d" * 64 + + +# ── The list ────────────────────────────────────────────────────────────────── + +def test_the_list_survives_a_restart(tmp_path): + path = tmp_path / "blocklist.json" + assert ContentBlocklist(path).replace([BLOCKED]) + assert BLOCKED in ContentBlocklist(path) + + +def test_a_full_sync_replaces_and_says_whether_anything_changed(tmp_path): + bl = ContentBlocklist(tmp_path / "blocklist.json") + assert bl.replace([BLOCKED, KEPT]) + assert not bl.replace([KEPT, BLOCKED]) + assert bl.replace([KEPT]) + assert BLOCKED not in bl and KEPT in bl + + +def test_a_pushed_change_applies_and_ignores_what_is_not_a_hash(tmp_path): + bl = ContentBlocklist(tmp_path / "blocklist.json") + assert bl.apply(add=[BLOCKED, "../etc", 7, "B" * 64]) + assert list(bl) == [BLOCKED] + assert not bl.apply(add=[BLOCKED]) + assert bl.apply(remove=[BLOCKED]) + assert len(bl) == 0 + + +def test_a_damaged_file_does_not_stop_the_node(tmp_path): + path = tmp_path / "blocklist.json" + path.write_text("{not json", encoding="utf-8") + assert len(ContentBlocklist(path)) == 0 + + +# ── What members are sent ───────────────────────────────────────────────────── + +def _index(gek) -> GroupIndex: + idx = GroupIndex(group_id=GROUP, sk_node=Ed25519PrivateKey.generate(), gek=gek) + idx.add_entry(IndexEntry(id=BLOCKED, name="a.jpg", path="r", size=1, type="image", + added_at=0, thumb_hash=THUMB)) + idx.add_entry(IndexEntry(id=KEPT, name="b.jpg", path="r", size=1, type="image", + added_at=0)) + return idx + + +def test_a_hidden_entry_is_not_in_the_index_or_its_deltas(): + gek = generate_gek() + idx = _index(gek) + msg = index_sync_message(idx, None, {BLOCKED}) + ids = [e["id"] for e in unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, msg)["entries"]] + assert ids == [KEPT] + + delta = idx.diff(GroupIndex(group_id=GROUP, sk_node=idx.sk_node, gek=gek, version=0)) + payload = unseal(gek, PURPOSE_INDEX, "index_delta", GROUP, + index_delta_message(idx, delta, None, {BLOCKED})) + assert [e["id"] for e in payload["additions"]] == [KEPT] + + +# ── What a session serves ──────────────────────────────────────────────────── + +class _Channel: + readyState = "open" + + def __init__(self): + self.sent = [] + + def send(self, data: bytes) -> None: + (n,) = struct.unpack(">I", data[:4]) + self.sent.append(msgpack.unpackb(data[4:4 + n], raw=False)) + + +class _PC: + connectionState = "connected" + iceConnectionState = "connected" + remoteDescription = None + localDescription = None + sctp = None + + +def _session(visibility: str, tmp_path): + gek = generate_gek() + bl = ContentBlocklist(tmp_path / "blocklist.json") + bl.replace([BLOCKED]) + ctx = {"sk_node": Ed25519PrivateKey.from_private_bytes(b"\x01" * 32), + "groups": {GROUP: {"visibility": visibility, "index": _index(gek), + "gek": gek, "roots": None}}, + "blocklist": bl} + s = WebRTCPeerSession(_PC(), ctx, peer_id="peer") + s._channel = _Channel() + s._audit = lambda *a, **k: None + s._user_id, s._group_id = "member-1", GROUP + return s, gek + + +@pytest.mark.asyncio +async def test_a_public_group_refuses_a_blocked_file_and_its_thumbnail(tmp_path): + s, _ = _session("public", tmp_path) + for file_id in (BLOCKED, THUMB): + await s._do_file_request({"file_id": file_id, "chunk_index": 0}) + assert s._channel.sent[-1]["code"] == "content_blocked" + + +@pytest.mark.asyncio +async def test_a_public_group_does_not_list_a_blocked_file(tmp_path): + s, gek = _session("public", tmp_path) + s._do_index_sync() + payload = unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, s._channel.sent[-1]) + assert [e["id"] for e in payload["entries"]] == [KEPT] + + +@pytest.mark.asyncio +async def test_a_private_group_is_untouched(tmp_path): + s, gek = _session("private", tmp_path) + s._do_index_sync() + payload = unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, s._channel.sent[-1]) + assert sorted(e["id"] for e in payload["entries"]) == [BLOCKED, KEPT] + assert not s._refuse_blocked(BLOCKED) + + +@pytest.mark.asyncio +async def test_streaming_subtitles_and_transcoding_are_refused_too(tmp_path): + s, _ = _session("public", tmp_path) + await s._stream_video_inner({"file_id": BLOCKED}) + assert s._channel.sent[-1]["code"] == "content_blocked" + await s._do_subtitle_request({"file_id": BLOCKED, "track": 0}) + assert s._channel.sent[-1]["code"] == "content_blocked" + await s._do_audio_transcode_request({"file_id": BLOCKED}) + assert s._channel.sent[-1]["code"] == "content_blocked" + + +# ── How the node gets the list ─────────────────────────────────────────────── + +@pytest.mark.asyncio +async def test_the_whole_list_is_fetched_page_by_page(): + pages = {"": {"hashes": [BLOCKED], "next": BLOCKED}, + BLOCKED: {"hashes": [KEPT], "next": None}} + + class _Resp: + def __init__(self, body): + self._body = body + + def raise_for_status(self): + pass + + def json(self): + return self._body + + class _Http: + async def get(self, path, params, headers): + assert path == "/v1/blocklist" + return _Resp(pages[params["after"]]) + + class _Session: + auth_headers = {} + + hub = HubClient.__new__(HubClient) + hub._session, hub._http = _Session(), _Http() + + async def fresh(): + return None + hub.ensure_fresh_token = fresh + assert await hub.fetch_blocklist() == {BLOCKED, KEPT} + + +# ── A change reaches members already connected ─────────────────────────────── + +def _daemon(tmp_path, visibility: str): + from unittest.mock import MagicMock + + from meshbay_node.config import Config, GroupConfig, HubConfig, KeystoreConfig, NodeConfig + from meshbay_node.daemon import NodeDaemon + + shared = tmp_path / "shared" + shared.mkdir() + config = Config( + hub=HubConfig(url="http://localhost:9999", username="t"), + node=NodeConfig(), + groups=[GroupConfig(id=GROUP, name="g", shared_dir=str(shared), + visibility=visibility)], + keystore=KeystoreConfig(path=tmp_path / "keystore.enc"), + data_dir=tmp_path / "data", + ) + daemon = NodeDaemon(config) + gek = generate_gek() + indexer = MagicMock() + indexer.index, indexer.roots = _index(gek), None + daemon._indexers = [indexer] + session = MagicMock() + session._group_id = GROUP + daemon._webrtc = MagicMock() + daemon._webrtc._sessions = {"p": session} + return daemon, session, gek + + +def test_a_pushed_block_resends_the_index_without_the_file(tmp_path): + daemon, session, gek = _daemon(tmp_path, "public") + daemon._on_blocklist_update([BLOCKED], []) + msg = session._send.call_args[0][0] + ids = [e["id"] for e in unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, msg)["entries"]] + assert ids == [KEPT] + + daemon._on_blocklist_update([], [BLOCKED]) + msg = session._send.call_args[0][0] + ids = sorted(e["id"] for e in unseal(gek, PURPOSE_INDEX, "index_sync", GROUP, msg)["entries"]) + assert ids == [BLOCKED, KEPT] + + +def test_a_node_with_only_private_groups_ignores_the_list(tmp_path): + daemon, session, _ = _daemon(tmp_path, "private") + daemon._on_blocklist_update([BLOCKED], []) + session._send.assert_not_called() + assert BLOCKED not in daemon._blocklist -- cgit v1.2.3