diff options
Diffstat (limited to 'packages')
11 files changed, 809 insertions, 28 deletions
diff --git a/packages/meshbay-hub/src/meshbay_hub/api/groups.py b/packages/meshbay-hub/src/meshbay_hub/api/groups.py index 88125c0..b1a22f7 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/groups.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/groups.py @@ -1,5 +1,7 @@ """Group endpoints — /v1/groups/*""" +import re + from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel from datetime import datetime, timezone @@ -10,6 +12,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from meshbay_hub import hub_settings, mail from meshbay_hub.auth import decrypt_email from meshbay_hub.api.deps import get_current_user, require_user_scope +from meshbay_hub.api.middleware import limiter from meshbay_hub.api.netutil import client_ip from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import ( @@ -192,12 +195,31 @@ async def list_public_groups( class SwarmRegisterRequest(BaseModel): content_hash: str # blake3 hex - endpoint: str # "ip:port" + endpoint: str # "<scheme>:<port>" — a port on the caller, never a host + + +# A transport and a port, and deliberately no host. The field used to be free +# text documented as "ip:port", so a caller could name *someone else's* +# address as a source; nothing dials a swarm source today, which is the only +# reason that was not already a reflection primitive. A reader learns where a +# node is from the node record, which is stamped with the address the announce +# actually came from — so a host here would be a second, weaker, answer to a +# question already settled elsewhere. +_SWARM_ENDPOINT = re.compile(r"^(webrtc|quic):([0-9]{1,5})$") + +# One account, this many public hashes. Rows are keyed (hash, account) with no +# cap, so a loop of invented hashes was unbounded storage growth on a hub +# shared with everyone else. A public library far larger than this is a real +# thing — but it is one a hub operator should be asked about, not something a +# client establishes by writing rows. +MAX_SWARM_HASHES_PER_ACCOUNT = 10_000 @swarm_router.post("/register", status_code=201) +@limiter.limit("120/minute") async def swarm_register( body: SwarmRegisterRequest, + request: Request, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): @@ -209,11 +231,21 @@ async def swarm_register( node's calls 404'd and the leak was masked by a routing bug rather than prevented. Nodes now filter by group visibility before calling, and the path is correct, so the filter has to be right. + + Availability: the endpoint is a port, not an address, and the number of + hashes one account may claim is bounded. See the two constants above. """ from meshbay_hub.csam import check_content_hash if check_content_hash(body.content_hash): raise HTTPException(status_code=451, detail="Content blocked") + m = _SWARM_ENDPOINT.match(body.endpoint or "") + if not m or not (0 < int(m.group(2)) < 65536): + raise HTTPException( + status_code=422, + detail="endpoint must be '<webrtc|quic>:<port>' — a port on the " + "registering node, not an address") + from datetime import datetime, timezone existing = await db.get(SwarmSource, (body.content_hash, current_user.id)) now = datetime.now(timezone.utc) @@ -221,6 +253,14 @@ async def swarm_register( existing.endpoint = body.endpoint existing.last_seen = now else: + held = (await db.execute( + select(func.count()).select_from(SwarmSource) + .where(SwarmSource.node_id == current_user.id))).scalar() or 0 + if held >= MAX_SWARM_HASHES_PER_ACCOUNT: + raise HTTPException( + status_code=429, + detail="This account already claims the maximum number of " + "public content hashes") db.add(SwarmSource( content_hash=body.content_hash, node_id=current_user.id, @@ -672,16 +712,27 @@ async def delete_group( return {"status": "deleted", "group_id": group_id} +# `XXXX-XXXX`, as `roster.generate_code` produces. Checked because this string +# is placed in an email the hub sends under its own domain, and the endpoint +# used to accept any text at all. +_INVITE_CODE = re.compile(r"^[0-9A-Za-z]{4}-[0-9A-Za-z]{4}$") + + class InviteNotifyRequest(BaseModel): username: str code: str - group_name: str + # `group_name` used to be here and went straight into the email's subject + # line. The hub knows the group's name — it is reading the row two lines + # into the handler — so accepting a second answer only let the sender + # choose the subject of a message the hub signs with its own domain. @router.post("/{group_id}/invite-notify") +@limiter.limit("20/hour") async def invite_notify( group_id: str, body: InviteNotifyRequest, + request: Request, current_user: User = Depends(require_user_scope), db: AsyncSession = Depends(get_db), ): @@ -691,6 +742,15 @@ async def invite_notify( because the inviter's browser sends it here. The hub looks up the invitee's encrypted email, decrypts it, and sends the notification. The inviter never sees the email address. + + Availability: the target is any account on the hub — it has to be, since + an invitee is by definition not yet a member — so this is the one endpoint + where one user causes mail to be sent to another. It was unmetered and the + subject line came from the request. Anyone who created a group, which is to + say anyone, could send any registered account arbitrary text from the hub's + own domain, as fast as they liked. The rate limit and the two checks below + are what keep that from being a phishing kit with the hub's reputation + attached. """ group = await db.get(Group, group_id) if not group: @@ -699,6 +759,9 @@ async def invite_notify( raise HTTPException(status_code=403, detail="Only the group owner can send invitations") + if not _INVITE_CODE.match(body.code or ""): + raise HTTPException(status_code=422, detail="Not an invite code") + target = (await db.execute( select(User).where(User.username == body.username))).scalar_one_or_none() if not target: @@ -715,7 +778,7 @@ async def invite_notify( try: mail.send_invite_notification( - email, body.code, current_user.username, body.group_name) + email, body.code, current_user.username, group.name) except Exception: return {"status": "send_failed"} diff --git a/packages/meshbay-hub/src/meshbay_hub/api/relay.py b/packages/meshbay-hub/src/meshbay_hub/api/relay.py index f82495b..c6ef26e 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/relay.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/relay.py @@ -8,24 +8,25 @@ Relay registration: POST /v1/relays/register — relay announces itself (signed JWT) GET /v1/relays — list active relays (for nodes) -Relay authentication: relay generates an Ed25519 keypair at install time. -It registers its public key with the hub admin, then signs keepalive JWTs. +Relay authentication: relay generates an Ed25519 keypair at install time. An +admin approves the public key, and every register call carries an Ed25519 +signature over "meshbay:relay_register:<relay_id>:<endpoint>:<timestamp>" — +the same proof-of-possession shape as /v1/nodes/announce. Relay is responsible for E2E encrypted QUIC traffic only (it cannot read the application-layer content, only forward UDP packets). """ +import base64 import logging import time -import uuid -import jwt -from fastapi import APIRouter, Depends, Header, HTTPException +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PublicKey +from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel from sqlalchemy.ext.asyncio import AsyncSession -from meshbay_hub.api.deps import get_current_user, require_admin -from meshbay_hub.auth import _hub_id, hub_public_key_pem +from meshbay_hub.api.deps import require_admin from meshbay_hub.db.engine import get_db from meshbay_hub.db.models import User @@ -40,11 +41,13 @@ _relays: dict[str, dict] = {} # relay_id → {endpoint, pk, last_seen, capacit # ── Models ──────────────────────────────────────────────────────────────────── class RelayRegisterRequest(BaseModel): - """Relay self-registers with a signed JWT.""" + """Relay self-registers, proving possession of its approved key.""" relay_id: str endpoint: str # "ip:port" (UDP) pk_relay: str # base64 Ed25519 public key capacity: int = 100 # max concurrent connections + timestamp: int | None = None # unix seconds + signature: str | None = None # base64 Ed25519 over the register message class RelayAdminApproveRequest(BaseModel): @@ -54,20 +57,48 @@ class RelayAdminApproveRequest(BaseModel): # ── Relay endpoints ─────────────────────────────────────────────────────────── +REGISTER_TIMESTAMP_WINDOW = 300 # seconds either side, as /v1/nodes/announce + + @router.post("/register", status_code=201) async def relay_register( body: RelayRegisterRequest, db: AsyncSession = Depends(get_db), ): """ - Relay announces itself. Must be pre-approved by a hub admin. - The relay's public key must already be in the approved list. + Relay announces itself. Must be pre-approved by a hub admin, and must prove + it holds the private key that approval registered. + + This endpoint has no `Depends` on an account on purpose — a relay is not a + user — but it had no proof of anything either: it compared `pk_relay` + against the approved value, which is a **public** key, so anyone who could + read it could rewrite where the hub tells nodes to send relayed traffic. + The module docstring said "signs keepalive JWTs" and nothing verified a + signature; `jwt` was imported and never used. A key is not a password, and + the fix is the proof-of-possession pattern already used by + /v1/nodes/announce and /v1/nodes/auth. """ approved = _relays.get(body.relay_id) if not approved or approved.get("pk") != body.pk_relay: raise HTTPException(status_code=403, detail="Relay not approved — ask hub admin to run POST /v1/relays/approve") + if body.timestamp is None or not body.signature: + raise HTTPException( + status_code=400, + detail="register requires timestamp and signature (proof of possession)") + if abs(int(time.time()) - body.timestamp) > REGISTER_TIMESTAMP_WINDOW: + raise HTTPException(status_code=401, detail="Timestamp too old or too far ahead") + + message = (f"meshbay:relay_register:{body.relay_id}:" + f"{body.endpoint}:{body.timestamp}").encode() + try: + pk = Ed25519PublicKey.from_public_bytes(base64.b64decode(body.pk_relay)) + pk.verify(base64.b64decode(body.signature), message) + except Exception: + log.warning("Relay %s failed proof of possession", body.relay_id[:8]) + raise HTTPException(status_code=401, detail="Invalid relay key proof of possession") + _relays[body.relay_id].update({ "endpoint": body.endpoint, "capacity": body.capacity, diff --git a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py index 7c37a1e..0e274f9 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/revocation.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/revocation.py @@ -68,6 +68,53 @@ def get_online_nodes_for_group(group_id: str) -> list[str]: return [nid for nid, gids in _node_groups.items() if group_id in gids] +# A chat_notify costs one query per member of the group plus a write for each, +# and nothing on the node's side paces it. Without a budget, one node can keep +# the hub's database busy on behalf of a group it belongs to — a cost borne by +# every other group on the instance. Generous enough that a lively conversation +# never meets it: notifications aggregate to one row per person per group, so +# the useful rate is far below this. +NOTIFY_BURST = 30 # messages +NOTIFY_WINDOW_SECONDS = 60 + +_notify_window: dict[str, tuple[float, int]] = {} # node_id → (window start, count) + + +def forget_node(node_id: str) -> None: + """Drop everything a disconnected node's socket owned. + + One function rather than three lines in a `finally`, so that what a + disconnect does can be asserted by running it instead of by re-enacting it + — a test that re-enacts cleanup tests its own re-enactment, and would not + have noticed `_notify_window` being added here. + + Which is the point: `_notify_window` is **not** cleared. Dropping it would + make reconnecting the way to refill the budget, and a node's token stays + valid for an hour. It expires by time, in `_notify_budget`. + """ + _connected_nodes.pop(node_id, None) + _node_groups.pop(node_id, None) + + +def _notify_budget(node_id: str) -> bool: + """True if this node may send one more chat_notify now.""" + now = time.monotonic() + if len(_notify_window) > 1000: + # Swept here rather than on disconnect, which would let a node refill + # its budget by reconnecting — the same token stays valid for an hour. + for nid, (started, _) in list(_notify_window.items()): + if now - started >= NOTIFY_WINDOW_SECONDS: + _notify_window.pop(nid, None) + start, count = _notify_window.get(node_id, (now, 0)) + if now - start >= NOTIFY_WINDOW_SECONDS: + start, count = now, 0 + if count >= NOTIFY_BURST: + _notify_window[node_id] = (start, count) + return False + _notify_window[node_id] = (start, count + 1) + return True + + async def _mark_hosted(group_ids: list[str]) -> None: """Stamp the first time a node announced it hosts each of these groups. @@ -133,10 +180,27 @@ def _sign_revocation(target: str, target_id: str, reason: str) -> str: # ── WebSocket endpoint ──────────────────────────────────────────────────────── -async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str) -> None: - """Node informs hub that a chat message was posted — create notifications for offline members.""" +async def _handle_chat_notify(group_id: str, sender_name: str, sender_user_id: str, + *, node_id: str) -> None: + """Node informs hub that a chat message was posted — create notifications for offline members. + + `group_id` arrives in the node's own message and is checked against what + that node is registered for. Without the check, any connected node could + write a notification to every member of **any** group on the hub, carrying + a display string of its choosing, with its account having no relation to + that group at all. Same shape as the empty group claim: something believed + about a group the sender has nothing to do with. + + `node_id` is keyword-**required** rather than defaulted. A default here + would mean "unchecked when the caller forgets", which is the failure this + whole review is about. + """ if not group_id: return + if group_id not in _node_groups.get(node_id, ()): + log.warning("Node %s sent chat_notify for a group it does not host", + (node_id or "?")[:8]) + return try: from meshbay_hub.db.engine import get_session_factory from meshbay_hub.db.models import GroupMember, Group @@ -318,7 +382,7 @@ async def node_websocket(ws: WebSocket): event.set() elif msg.get("type") == "webrtc_answer": from meshbay_hub.api.signaling import handle_webrtc_answer - handle_webrtc_answer(msg) + handle_webrtc_answer(msg, node_id) elif msg.get("type") == "update_groups": # Through the same gate as the registration above. This used to # assign the message's list verbatim, so the ceiling that makes @@ -330,6 +394,9 @@ async def node_websocket(ws: WebSocket): await _mark_hosted(new_gids) log.info("Node %s updated groups: %d", node_id[:8], len(new_gids)) elif msg.get("type") == "chat_notify": + if not _notify_budget(node_id): + log.warning("Node %s exceeded its chat_notify rate", node_id[:8]) + continue asyncio.ensure_future(_handle_chat_notify( msg.get("group_id", ""), msg.get("sender_name", ""), @@ -339,6 +406,7 @@ async def node_websocket(ws: WebSocket): # that lied here could only suppress one notification, which # is the same power it has by not sending the message at all. msg.get("sender_user_id", ""), + node_id=node_id, )) except WebSocketDisconnect: @@ -347,8 +415,7 @@ async def node_websocket(ws: WebSocket): log.error("Node WS error: %s", e) finally: if node_id: - _connected_nodes.pop(node_id, None) - _node_groups.pop(node_id, None) + forget_node(node_id) # ── Admin revocation endpoint ───────────────────────────────────────────────── diff --git a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py index a8feae8..b4be3f2 100644 --- a/packages/meshbay-hub/src/meshbay_hub/api/signaling.py +++ b/packages/meshbay-hub/src/meshbay_hub/api/signaling.py @@ -34,6 +34,9 @@ log = logging.getLogger(__name__) router = APIRouter(prefix="/v1/nodes", tags=["signaling"]) _webrtc_answers: dict[str, asyncio.Future] = {} +# peer_id → the node the offer was relayed to. An answer is only accepted +# from that node (see handle_webrtc_answer). +_answer_owner: dict[str, str] = {} class WebRTCOfferRequest(BaseModel): @@ -133,6 +136,7 @@ async def webrtc_offer( peer_id = str(uuid.uuid4()) answer_future: asyncio.Future = asyncio.get_event_loop().create_future() _webrtc_answers[peer_id] = answer_future + _answer_owner[peer_id] = node_id _pending_per_user[current_user.id] = _pending_per_user.get(current_user.id, 0) + 1 try: @@ -157,6 +161,7 @@ async def webrtc_offer( ) finally: _webrtc_answers.pop(peer_id, None) + _answer_owner.pop(peer_id, None) remaining = _pending_per_user.get(current_user.id, 1) - 1 if remaining > 0: _pending_per_user[current_user.id] = remaining @@ -164,13 +169,26 @@ async def webrtc_offer( _pending_per_user.pop(current_user.id, None) -def handle_webrtc_answer(msg: dict) -> None: - """Called from the node WebSocket message loop when a webrtc_answer arrives.""" +def handle_webrtc_answer(msg: dict, node_id: str) -> None: + """Called from the node WebSocket message loop when a webrtc_answer arrives. + + `node_id` is the socket this arrived on, and the answer is accepted only for + a `peer_id` the hub issued to **that** node. The answer carries the SDP the + browser then connects to, so without the check any connected node could + resolve any pending offer and stand in for the node the client asked for. + That it had not happened rested on a uuid4 being unguessable, which is a + reason it was hard, not a reason it was refused. + """ peer_id = msg.get("peer_id") if not peer_id: log.warning("webrtc_answer without peer_id") return + if _answer_owner.get(peer_id) != node_id: + log.warning("Node %s answered an offer it was never sent (peer=%s)", + (node_id or "?")[:8], str(peer_id)[:8]) + return + future = _webrtc_answers.get(peer_id) if future and not future.done(): future.set_result({ diff --git a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js index 8e11088..0f8ccdd 100644 --- a/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js +++ b/packages/meshbay-hub/src/meshbay_hub/static/group-settings.js @@ -794,7 +794,10 @@ function GroupSettingsPanel({ groupId, group, token, transportRef, gekRef, try { const notif = await hubFetch(`/v1/groups/${groupId}/invite-notify`, { method: 'POST', token, - body: { username, code: result.code, group_name: group?.name || '' }, + // No group_name: the hub reads it from the group row it has already + // loaded. Sending one offered a second answer to a settled question, + // and that answer was the subject line of an email the hub signs. + body: { username, code: result.code }, }); emailStatus = notif.status; } catch { /* best effort */ } diff --git a/packages/meshbay-hub/tests/test_account_deletion.py b/packages/meshbay-hub/tests/test_account_deletion.py index b97417d..4cbca8b 100644 --- a/packages/meshbay-hub/tests/test_account_deletion.py +++ b/packages/meshbay-hub/tests/test_account_deletion.py @@ -135,7 +135,10 @@ async def test_deletion_clears_device_keys_and_swarm_sources(client, db_session) """ The privacy statement says every account row goes but the IP log. Swarm sources are keyed by the *user* id despite the column's name, and carry the - node's ip:port. + node's transport and port — `webrtc:<port>`, which is what `daemon.py` + actually sends. This asked with `192.0.2.7:4433`, from the days when the + field was free text documented as "ip:port": a shape no node has ever + produced, and one that let a caller name a third party's address. """ from meshbay_hub.db.models import SwarmSource, UserDevice @@ -148,7 +151,7 @@ async def test_deletion_clears_device_keys_and_swarm_sources(client, db_session) json={"pk_auth_ed25519": _device_pk(), "label": "desktop"}) assert r.status_code == 201, r.text r = await client.post("/v1/swarm/register", headers=headers, - json={"content_hash": "ab" * 32, "endpoint": "192.0.2.7:4433"}) + json={"content_hash": "ab" * 32, "endpoint": "webrtc:4433"}) assert r.status_code == 201, r.text # Present before, or the emptiness asserted below proves nothing. diff --git a/packages/meshbay-hub/tests/test_availability_between_members.py b/packages/meshbay-hub/tests/test_availability_between_members.py new file mode 100644 index 0000000..f2c9a4f --- /dev/null +++ b/packages/meshbay-hub/tests/test_availability_between_members.py @@ -0,0 +1,388 @@ +""" +Availability: what one participant can do to the others. + +The three security reviews asked who can *read* what, who can impersonate whom, +and what a hostile node can forge. None of them asked what a legitimate but +misconfigured or careless member costs everyone else — and that is the question +a group platform lives or dies on, because every member is invited by someone +who trusted them and none of them is an attacker. + +Written after 2026-09-11, where a member's unconfigured node was registered by +the hub as a host for a group it could not serve, and — being answered first — +made that group unopenable for everyone in it. Nothing was compromised and +nothing was forged. The group was simply gone. + +Each test below is two accounts, because that is the shape the single-node +suite could not express: `_make_user` twice, a group owned by one, the other +holding whatever the defect needs. A one-member test proves a one-member +property, and every finding here needed a second person to exist at all. +""" + +import base64 +import time + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey +from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey + +from meshbay_common.crypto import pk_to_b64 + + +async def _make_user(client, username: str) -> dict: + sk_ed, sk_x = Ed25519PrivateKey.generate(), X25519PrivateKey.generate() + pk_ed, pk_x = pk_to_b64(sk_ed.public_key()), pk_to_b64(sk_x.public_key()) + r = await client.post("/v1/users/register", json={ + "username": username, + "email": f"{username}@example.test", + "auth_key": base64.b64encode(b"k" * 32).decode(), + "pk_user_ed25519": pk_ed, + "pk_user_x25519": pk_x, + }) + assert r.status_code == 201, r.text + user_id = r.json()["user_id"] + r = await client.post("/v1/users/login", json={ + "username": username, + "auth_key": base64.b64encode(b"k" * 32).decode(), + }) + assert r.status_code == 200, r.text + return {"user_id": user_id, "username": username, + "token": r.json()["access_token"], "pk_ed": pk_ed, "sk_ed": sk_ed} + + +async def _announce_node(client, user: dict) -> str: + ts = int(time.time()) + msg = f"meshbay:node_announce:{user['user_id']}:{user['pk_ed']}:{ts}".encode() + r = await client.post("/v1/nodes/announce", json={ + "pk_node": user["pk_ed"], "endpoint_hint": "test", "timestamp": ts, + "signature": base64.b64encode(user["sk_ed"].sign(msg)).decode(), + }, headers={"Authorization": f"Bearer {user['token']}"}) + assert r.status_code == 201, r.text + return r.json()["node_id"] + + +async def _make_group(client, owner: dict, name: str) -> str: + r = await client.post( + "/v1/groups", + json={"name": name, "visibility": "private", "join_policy": "invite"}, + headers={"Authorization": f"Bearer {owner['token']}"}) + assert r.status_code == 201, r.text + return r.json()["group_id"] + + +async def _add_member(client, owner: dict, group_id: str, member: dict) -> None: + r = await client.post( + f"/v1/groups/{group_id}/members/{member['username']}", + headers={"Authorization": f"Bearer {owner['token']}"}) + assert r.status_code == 201, r.text + + +# ── A member's node must not speak for a group it does not host ────────────── + +@pytest.mark.asyncio +async def test_a_members_node_cannot_notify_a_group_it_does_not_host(client): + """ + `chat_notify` carried a `group_id` the hub believed, so any connected node + could write a notification to every member of any group on the hub, with a + display string of its own choosing. The node's account needed no relation + to the group whatsoever — this is the same defect as the group claim, one + message further along the same socket. + """ + from meshbay_hub.api import revocation as rev + + owner = await _make_user(client, "av_owner1") + outsider = await _make_user(client, "av_outsider1") + owner_node = await _announce_node(client, owner) + outsider_node = await _announce_node(client, outsider) + group_id = await _make_group(client, owner, "not-yours") + + rev._node_groups[owner_node] = [group_id] + rev._node_groups[outsider_node] = [] # hosts nothing, as in the incident + try: + await rev._handle_chat_notify( + group_id, "Someone you do not know", outsider["user_id"], + node_id=outsider_node) + + r = await client.get("/v1/notifications", + headers={"Authorization": f"Bearer {owner['token']}"}) + assert r.status_code == 200, r.text + assert r.json()["notifications"] == [], ( + "an unrelated node wrote into this account's notifications") + + # And the node that does host it is still able to. + await rev._handle_chat_notify( + group_id, "A real member", outsider["user_id"], node_id=owner_node) + r = await client.get("/v1/notifications", + headers={"Authorization": f"Bearer {owner['token']}"}) + assert len(r.json()["notifications"]) == 1 + finally: + rev._node_groups.pop(owner_node, None) + rev._node_groups.pop(outsider_node, None) + + +@pytest.mark.asyncio +async def test_a_member_whose_node_hosts_nothing_still_cannot_notify(client): + """ + The incident's own shape, and the one a check on *membership* would have + missed: the second account is a real member of the group, invited by its + owner. Their node is simply not the one holding the files. Being entitled + to be in a group is not being entitled to speak for it. + """ + from meshbay_hub.api import revocation as rev + + owner = await _make_user(client, "av_owner2") + member = await _make_user(client, "av_member2") + owner_node = await _announce_node(client, owner) + member_node = await _announce_node(client, member) + group_id = await _make_group(client, owner, "shared-room") + await _add_member(client, owner, group_id, member) + + rev._node_groups[owner_node] = [group_id] + rev._node_groups[member_node] = [] + try: + await rev._handle_chat_notify( + group_id, "Sounds like a member", member["user_id"], + node_id=member_node) + r = await client.get("/v1/notifications", + headers={"Authorization": f"Bearer {owner['token']}"}) + assert r.json()["notifications"] == [], ( + "a member's empty node wrote into the group owner's notifications") + finally: + rev._node_groups.pop(owner_node, None) + rev._node_groups.pop(member_node, None) + + +@pytest.mark.asyncio +async def test_one_node_cannot_spend_the_hubs_database_on_notifications(): + """ + Each chat_notify is a query and a write per member of the group, spawned + without backpressure. The budget is what stops one group's node from + costing every other group on the instance. + """ + from meshbay_hub.api.revocation import ( + NOTIFY_BURST, _notify_budget, _notify_window) + + node = "budget-node" + _notify_window.pop(node, None) + try: + assert all(_notify_budget(node) for _ in range(NOTIFY_BURST)) + assert not _notify_budget(node), "the burst was not bounded" + + # A second node is unaffected — the budget is per node, not global. + assert _notify_budget("another-node") + finally: + _notify_window.pop(node, None) + _notify_window.pop("another-node", None) + + +@pytest.mark.asyncio +async def test_the_notify_budget_is_not_refilled_by_reconnecting(client): + """ + The obvious place to clear this is the socket's `finally`, beside the other + two registries — and that would make reconnecting the way around it. The + same node token is valid for an hour. + """ + from meshbay_hub.api import revocation as rev + + user = await _make_user(client, "av_reconnect") + node_id = await _announce_node(client, user) + rev._notify_window.pop(node_id, None) + try: + for _ in range(rev.NOTIFY_BURST): + rev._notify_budget(node_id) + assert not rev._notify_budget(node_id) + + # The disconnect path itself, not a re-enactment of it: `forget_node` + # is what the socket's `finally` calls, so adding a line there is + # caught here. + rev.forget_node(node_id) + assert node_id not in rev._connected_nodes + assert node_id not in rev._node_groups + + assert not rev._notify_budget(node_id), ( + "disconnecting refilled the budget, so reconnecting defeats it") + finally: + rev._notify_window.pop(node_id, None) + + +# ── A member must not aim other people's traffic ───────────────────────────── + +@pytest.mark.asyncio +async def test_a_swarm_source_cannot_name_someone_elses_address(client): + """ + `endpoint` was free text documented as "ip:port", so an account could + publish a third party's address as a source for any content. Nothing dials + a swarm source today, which is the only reason this was not already the + reflection primitive that `notify_incoming` was fixed for (H6). A port is + all a reader needs: where the node is comes from the node record, which is + stamped with the address its announce arrived from. + """ + user = await _make_user(client, "av_swarm1") + headers = {"Authorization": f"Bearer {user['token']}"} + + for bad in ("192.0.2.7:4433", "evil.example:53", "webrtc:0", "webrtc:70000", + "webrtc:4433 ", "http://example.test"): + r = await client.post("/v1/swarm/register", headers=headers, + json={"content_hash": "ab" * 32, "endpoint": bad}) + assert r.status_code == 422, f"{bad!r} was accepted: {r.text}" + + r = await client.post("/v1/swarm/register", headers=headers, + json={"content_hash": "ab" * 32, "endpoint": "webrtc:19010"}) + assert r.status_code == 201, r.text + + +@pytest.mark.asyncio +async def test_one_account_cannot_fill_the_swarm_table(client, monkeypatch): + """Rows are keyed (hash, account) with no cap — an invented hash each time.""" + import meshbay_hub.api.groups as groups_api + monkeypatch.setattr(groups_api, "MAX_SWARM_HASHES_PER_ACCOUNT", 3) + + user = await _make_user(client, "av_swarm2") + headers = {"Authorization": f"Bearer {user['token']}"} + for i in range(3): + r = await client.post("/v1/swarm/register", headers=headers, + json={"content_hash": f"{i:064x}", + "endpoint": "webrtc:19010"}) + assert r.status_code == 201, r.text + + r = await client.post("/v1/swarm/register", headers=headers, + json={"content_hash": f"{99:064x}", "endpoint": "webrtc:19010"}) + assert r.status_code == 429, r.text + + # Refreshing one already held is not a new claim and must still work. + r = await client.post("/v1/swarm/register", headers=headers, + json={"content_hash": f"{0:064x}", "endpoint": "webrtc:19011"}) + assert r.status_code == 201, r.text + + +# ── A member's node must not answer for another's ──────────────────────────── + +def test_a_node_cannot_answer_an_offer_it_was_never_sent(): + """ + `handle_webrtc_answer` resolved any pending `peer_id` from any node's + socket. The answer is the SDP the browser then connects to, so the check is + what keeps one node from standing in for the node a client asked for. That + it had not happened rested on a uuid4 being unguessable. + """ + import asyncio + + from meshbay_hub.api.signaling import ( + _answer_owner, _webrtc_answers, handle_webrtc_answer) + + loop = asyncio.new_event_loop() + try: + future = loop.create_future() + _webrtc_answers["peer-1"] = future + _answer_owner["peer-1"] = "the-node-asked-for" + + handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 impostor"}, + "a-different-node") + assert not future.done(), "another node resolved this offer" + + handle_webrtc_answer({"peer_id": "peer-1", "sdp": "v=0 genuine"}, + "the-node-asked-for") + assert future.done() and future.result()["sdp"] == "v=0 genuine" + finally: + _webrtc_answers.pop("peer-1", None) + _answer_owner.pop("peer-1", None) + loop.close() + + +# ── One account must not be able to mail another at will ───────────────────── + +@pytest.mark.asyncio +async def test_an_invite_email_says_what_the_hub_knows_not_what_it_is_told( + client, monkeypatch): + """ + `group_name` went from the request body into the subject line of an email + the hub sends under its own domain, and the invitee need not be a member of + anything — as they cannot be, being invited. So any account, having made a + group, could mail any other account arbitrary text. The name now comes from + the group row, and the code has to look like one. + """ + sent: list = [] + import meshbay_hub.mail as mail_mod + monkeypatch.setattr(mail_mod, "send_invite_notification", + lambda *a, **kw: sent.append(a)) + + owner = await _make_user(client, "av_inviter") + await _make_user(client, "av_invitee") + group_id = await _make_group(client, owner, "real-group-name") + headers = {"Authorization": f"Bearer {owner['token']}"} + + r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers, + json={"username": "av_invitee", "code": "not a code", + "group_name": "Your account is suspended"}) + assert r.status_code == 422, r.text + assert not sent + + r = await client.post(f"/v1/groups/{group_id}/invite-notify", headers=headers, + json={"username": "av_invitee", "code": "AB12-CD34", + "group_name": "Your account is suspended"}) + assert r.status_code == 200, r.text + assert sent, "the legitimate path stopped working" + assert sent[0][3] == "real-group-name", ( + "the sender chose the subject line of a message the hub signs") + + +# ── A relay is not authenticated by the key it publishes ───────────────────── + +@pytest.mark.asyncio +async def test_a_relay_must_prove_it_holds_the_approved_key(client, monkeypatch): + """ + `relay_register` had no `Depends` and verified nothing: it compared + `pk_relay` against the approved value, which is a **public** key. Anyone + who could read it could rewrite where the hub tells nodes to send relayed + traffic — an unauthenticated write to state other people's machines act + on. The module docstring said the relay "signs keepalive JWTs"; `jwt` was + imported and never used. + """ + from meshbay_hub.api import relay as relay_mod + + sk = Ed25519PrivateKey.generate() + pk = pk_to_b64(sk.public_key()) + relay_mod._relays["r1"] = {"pk": pk, "active": False} + try: + # The public key alone, which used to be enough. + r = await client.post("/v1/relays/register", json={ + "relay_id": "r1", "endpoint": "198.51.100.9:9999", + "pk_relay": pk, "capacity": 100}) + assert r.status_code == 400, r.text + assert relay_mod._relays["r1"].get("endpoint") is None + + # A signature over someone else's endpoint does not carry either: the + # endpoint is inside the signed message. + ts = int(time.time()) + sig = sk.sign(f"meshbay:relay_register:r1:10.0.0.1:4433:{ts}".encode()) + r = await client.post("/v1/relays/register", json={ + "relay_id": "r1", "endpoint": "198.51.100.9:9999", "pk_relay": pk, + "timestamp": ts, "signature": base64.b64encode(sig).decode()}) + assert r.status_code == 401, r.text + + endpoint = "203.0.113.4:4433" + sig = sk.sign(f"meshbay:relay_register:r1:{endpoint}:{ts}".encode()) + r = await client.post("/v1/relays/register", json={ + "relay_id": "r1", "endpoint": endpoint, "pk_relay": pk, + "timestamp": ts, "signature": base64.b64encode(sig).decode()}) + assert r.status_code == 201, r.text + assert relay_mod._relays["r1"]["endpoint"] == endpoint + finally: + relay_mod._relays.pop("r1", None) + + +@pytest.mark.asyncio +async def test_a_captured_relay_registration_is_not_replayable(client): + """Same reason /v1/nodes/announce bounds its timestamp.""" + from meshbay_hub.api import relay as relay_mod + + sk = Ed25519PrivateKey.generate() + pk = pk_to_b64(sk.public_key()) + relay_mod._relays["r2"] = {"pk": pk, "active": False} + try: + ts = int(time.time()) - 3600 + sig = sk.sign(f"meshbay:relay_register:r2:203.0.113.5:4433:{ts}".encode()) + r = await client.post("/v1/relays/register", json={ + "relay_id": "r2", "endpoint": "203.0.113.5:4433", "pk_relay": pk, + "timestamp": ts, "signature": base64.b64encode(sig).decode()}) + assert r.status_code == 401, r.text + finally: + relay_mod._relays.pop("r2", None) diff --git a/packages/meshbay-hub/tests/test_hub_api.py b/packages/meshbay-hub/tests/test_hub_api.py index 724c8ec..5bb6af9 100644 --- a/packages/meshbay-hub/tests/test_hub_api.py +++ b/packages/meshbay-hub/tests/test_hub_api.py @@ -658,9 +658,14 @@ async def test_webrtc_signaling_roundtrip(client, app): from meshbay_hub.api.revocation import _connected_nodes from meshbay_hub.api.signaling import handle_webrtc_answer + node_id = "test-node-sig" + class FakeWS: - def __init__(self): + def __init__(self, answering_as): self.sent = [] + # An answer is accepted only from the node the offer was relayed + # to, so the stand-in has to say which node it is. + self._node_id = answering_as async def send_text(self, text): self.sent.append(json.loads(text)) @@ -672,10 +677,9 @@ async def test_webrtc_signaling_roundtrip(client, app): "peer_id": msg["peer_id"], "sdp": "v=0\r\nanswer-sdp", "ice_candidates": [{"candidate": "test"}], - }) + }, self._node_id) - fake_ws = FakeWS() - node_id = "test-node-sig" + fake_ws = FakeWS(node_id) _connected_nodes[node_id] = fake_ws try: diff --git a/packages/meshbay-hub/tests/test_notifications_behaviour.py b/packages/meshbay-hub/tests/test_notifications_behaviour.py index 376b9b0..5fe9170 100644 --- a/packages/meshbay-hub/tests/test_notifications_behaviour.py +++ b/packages/meshbay-hub/tests/test_notifications_behaviour.py @@ -161,6 +161,7 @@ async def test_you_are_not_notified_of_your_own_message(client, db_session): the operator and nobody else: everyone was told about their own messages, and the operator was told about no one's. """ + from meshbay_hub.api import revocation as rev from meshbay_hub.api.revocation import _handle_chat_notify owner = await _user(client, "operator") @@ -177,7 +178,14 @@ async def test_you_are_not_notified_of_your_own_message(client, db_session): select(User).where(User.username.in_(["operator", "chatty", "quiet"])) )).scalars().all()} - await _handle_chat_notify(gid, "chatty", ids["chatty"]) + # A node registered for this group, because a notification is now written + # only for a group the sending node actually hosts (AV3). + node_id = "notify-test-node" + rev._node_groups[node_id] = [gid] + try: + await _handle_chat_notify(gid, "chatty", ids["chatty"], node_id=node_id) + finally: + rev._node_groups.pop(node_id, None) async def chat_rows(uid): return (await db_session.execute( 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 a0776d2..29d6e7f 100644 --- a/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py +++ b/packages/meshbay-node/src/meshbay_node/transport/webrtc_server.py @@ -192,6 +192,18 @@ MAX_UPLOAD_BYTES = 4 * 1024 * 1024 * 1024 # 4 GB per file # Budget for an unauthenticated peer: enough for a handshake and a bundle fetch, # nowhere near enough to be a memory-exhaustion primitive (H6). PRE_HANDSHAKE_MAX_MSG = 64 * 1024 + +# How many peer connections this node holds at once, and how long one may stay +# without completing the MNP handshake. The budget above bounds what *one* +# unauthenticated peer costs; these bound how many there may be and how long +# each lasts, which is the other half and was missing. The hub caps three +# offers in flight per account — a limit on each caller, not on this machine — +# so the cost to an operator grew with the number of people in their groups. +# Sized to be unreachable in ordinary use: a browser holds one connection per +# open group, and a handshake unfinished after a minute is not going to finish. +MAX_PEER_SESSIONS = 64 +UNAUTHENTICATED_SESSION_TIMEOUT = 60 # seconds + # ffmpeg is spawned per stream request; without a cap any member can fork-bomb # the node by requesting many streams at once (H6). # @@ -5882,6 +5894,7 @@ class WebRTCTransport: from meshbay_node.config import DEFAULT_STUN_SERVERS self._stun = stun_servers or list(DEFAULT_STUN_SERVERS) self._sessions: dict[str, WebRTCPeerSession] = {} + self._reapers: set[asyncio.Task] = set() def set_capacity(self, *, max_concurrent_streams: int | None = None, max_concurrent_downloads: int | None = None, @@ -5991,9 +6004,21 @@ class WebRTCTransport: config = RTCConfiguration( iceServers=[RTCIceServer(urls=s) for s in self._stun] if self._stun else [] ) + # Before anything is allocated. Every offer costs an RTCPeerConnection + # with its own DTLS and SCTP stacks, and nothing here used to bound how + # many a node would hold: the hub caps three in flight *per account*, + # which is a limit on each caller and not on this machine, so the cost + # grew with the number of members in the group. An operator's node must + # not be exhaustible by the people they invited. + if len(self._sessions) >= MAX_PEER_SESSIONS: + log.warning("Refusing WebRTC offer: %d peer sessions already open", + len(self._sessions)) + raise RuntimeError("Node is at its peer-connection limit") + pc = RTCPeerConnection(configuration=config) session = WebRTCPeerSession(pc, self._ctx, peer_id=peer_id) self._sessions[peer_id] = session + self._reap_if_unauthenticated(peer_id) @pc.on("datachannel") def on_datachannel(channel: RTCDataChannel): @@ -6048,12 +6073,47 @@ class WebRTCTransport: ", ".join(sorted(host_addrs)) or "none", srflx) return answer_sdp, [] + def _reap_if_unauthenticated(self, peer_id: str) -> None: + """Close a session that never completes the handshake. + + A peer that connects and then says nothing is indistinguishable from a + working one until it is asked to prove something, and it was never + asked: `connectionstatechange` reaps a connection that *fails*, and one + that succeeds and stays silent was held for the node's lifetime. That + is the cheapest way to spend someone else's memory — no GEK, no token, + no group, just an open connection. `_user_id` is set by the GEK proof + (`_do_handshake_response`), so it is the one honest test of whether + this peer ever became anybody. + """ + async def reap() -> None: + try: + await asyncio.sleep(UNAUTHENTICATED_SESSION_TIMEOUT) + session = self._sessions.get(peer_id) + if session is not None and not session._user_id: + log.warning("Closing peer %s: no handshake within %ds", + peer_id[:8], UNAUTHENTICATED_SESSION_TIMEOUT) + await self.close_peer(peer_id) + except asyncio.CancelledError: + raise + except Exception as e: + log.warning("Reaping peer %s failed: %s", peer_id[:8], e) + + # Held in a set for the same reason every other task here is: asyncio + # keeps only a weak reference, and a reaper collected mid-sleep reaps + # nothing (see WebRTCPeerSession.__init__). + task = asyncio.ensure_future(reap()) + self._reapers.add(task) + task.add_done_callback(self._reapers.discard) + async def close_peer(self, peer_id: str) -> None: session = self._sessions.pop(peer_id, None) if session: await session.close() async def close_all(self) -> None: + for task in list(self._reapers): + task.cancel() + self._reapers.clear() for session in list(self._sessions.values()): await session.close() self._sessions.clear() diff --git a/packages/meshbay-node/tests/test_peer_session_limits.py b/packages/meshbay-node/tests/test_peer_session_limits.py new file mode 100644 index 0000000..64961ad --- /dev/null +++ b/packages/meshbay-node/tests/test_peer_session_limits.py @@ -0,0 +1,136 @@ +""" +Availability, on the node: what the people you invited can cost you. + +The message budget for an unauthenticated peer (`PRE_HANDSHAKE_MAX_MSG`, H6) +bounds what *one* of them spends. Nothing bounded how many there could be, or +how long one could stay without ever proving anything — and the hub's cap is +three offers in flight **per account**, which is a limit on each caller rather +than on this machine, so an operator's exposure grew with the number of people +in their groups. A member who never meant any harm — a tab left open through a +sleep, a client retrying a connection it cannot complete — arrives here the +same way as one who does. +""" + +import asyncio +from unittest.mock import MagicMock + +import pytest + +from meshbay_node.transport import webrtc_server as ws_mod +from meshbay_node.transport.webrtc_server import ( + MAX_PEER_SESSIONS, UNAUTHENTICATED_SESSION_TIMEOUT, WebRTCTransport) + + +def _transport() -> WebRTCTransport: + return WebRTCTransport(sk_node=MagicMock(), hub_pk_pem=b"", gek=None, + roots=None, index=None) + + +class _FakeSession: + """Stands in for a peer that connected and then said nothing.""" + + def __init__(self, user_id=None): + self._user_id = user_id + self.closed = False + + async def close(self): + self.closed = True + + +@pytest.mark.asyncio +async def test_the_node_refuses_more_peers_than_it_will_hold(): + """The refusal comes before an RTCPeerConnection is allocated, or the cap + would be counting the thing it is meant to prevent.""" + tp = _transport() + for i in range(MAX_PEER_SESSIONS): + tp._sessions[f"peer-{i}"] = _FakeSession(user_id="someone") + + with pytest.raises(RuntimeError, match="peer-connection limit"): + await tp.handle_offer("v=0", "one-too-many") + + assert "one-too-many" not in tp._sessions + + +@pytest.mark.asyncio +async def test_a_peer_that_never_handshakes_is_closed(monkeypatch): + """`connectionstatechange` reaps a connection that *fails*. One that + succeeds and stays silent was held for the life of the daemon.""" + monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05) + tp = _transport() + session = _FakeSession(user_id=None) + tp._sessions["quiet"] = session + + tp._reap_if_unauthenticated("quiet") + await asyncio.sleep(0.2) + + assert session.closed, "a peer that never proved anything was kept" + assert "quiet" not in tp._sessions + + +@pytest.mark.asyncio +async def test_a_peer_that_handshaked_is_left_alone(monkeypatch): + """`_user_id` is set by the GEK proof, and is the one honest test of + whether this peer ever became anybody.""" + monkeypatch.setattr(ws_mod, "UNAUTHENTICATED_SESSION_TIMEOUT", 0.05) + tp = _transport() + session = _FakeSession(user_id=None) + tp._sessions["real"] = session + + tp._reap_if_unauthenticated("real") + session._user_id = "a-real-member" # the handshake completes + await asyncio.sleep(0.2) + + assert not session.closed, "a member who completed the handshake was cut off" + assert tp._sessions["real"] is session + + +@pytest.mark.asyncio +async def test_handle_offer_arms_the_reaper(monkeypatch): + """ + The seam, driven rather than described. The three tests above call + `_reap_if_unauthenticated` themselves, so every one of them would still + pass with the call removed from `handle_offer` and no peer reaped at all — + which is the defect, not the helper. aiortc is stubbed because a real + RTCPeerConnection wants a real SDP; everything else here is the shipped + code path. + """ + class _FakePC: + def __init__(self, *a, **kw): + self.localDescription = MagicMock(sdp="v=0\r\n") + self.remoteDescription = None + + def on(self, _event): + return lambda fn: fn + + async def setRemoteDescription(self, _d): pass + async def createAnswer(self): return MagicMock() + async def setLocalDescription(self, _d): pass + async def close(self): pass + + monkeypatch.setattr(ws_mod, "RTCPeerConnection", _FakePC) + monkeypatch.setattr(ws_mod, "RTCSessionDescription", + lambda **kw: MagicMock(**kw)) + + tp = _transport() + await tp.handle_offer("v=0\r\n", "fresh-peer") + try: + assert tp._reapers, ( + "handle_offer allocated a peer session and armed nothing to " + "close it if the handshake never comes") + finally: + await tp.close_all() + + +@pytest.mark.asyncio +async def test_the_reaper_is_held_and_cancelled_with_the_transport(): + """asyncio keeps only a weak reference to a task, and a reaper collected + mid-sleep reaps nothing — the same trap as every other task in this file.""" + tp = _transport() + tp._sessions["held"] = _FakeSession() + tp._reap_if_unauthenticated("held") + assert tp._reapers, "the reaper was fired and forgotten" + + await tp.close_all() + assert not tp._reapers + # The timeout is a real duration, not something a test has to wait out. + assert UNAUTHENTICATED_SESSION_TIMEOUT >= 30 |