diff options
Diffstat (limited to 'packages/meshbay-hub/src')
5 files changed, 203 insertions, 21 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 */ } |